| Taskflow: A General-purpose Task-parallel Programming System

Loading...
Searching...
No Matches
flow_builder.hpp
1#pragma once
2
3#include "task.hpp"
4#include "../algorithm/partitioner.hpp"
5
10
11 namespace tf {
12
22 class FlowBuilder {
23
24 friend class Executor;
25
26 public:
27
31 FlowBuilder(Graph& graph);
32
51 template <StaticTask C>
52 Task emplace(C&& callable);
53
72 template <RuntimeTask C>
73 Task emplace(C&& callable);
74
97 template <SubflowTask C>
98 Task emplace(C&& callable);
99
130 template <ConditionTask C>
131 Task emplace(C&& callable);
132
164 template <MultiConditionTask C>
165 Task emplace(C&& callable);
166
191 template <typename... C> requires (sizeof...(C) > 1)
192 auto emplace(C&&... callables);
193
214 void erase(Task task);
215
268 template <HasGraph T>
269 Task composed_of(T& object);
270
288 Task adopt(Graph&& graph);
289
314 Task placeholder();
315
333 void linearize(std::vector<Task>& tasks);
334
350 void linearize(std::initializer_list<Task> tasks);
351
352
353 // ------------------------------------------------------------------------
354 // parallel iterations
355 // ------------------------------------------------------------------------
356
389 template <typename B, typename E, typename C, typename P = DefaultPartitioner>
390 Task for_each(B first, E last, C callable, P part = P());
391
431 template <typename B, typename E, typename S, typename C, typename P = DefaultPartitioner>
432 Task for_each_index(B first, E last, S step, C callable, P part = P());
433
470 template <typename R, typename C, typename P = DefaultPartitioner>
471 Task for_each_by_index(R range, C callable, P part = P());
472
473 // ------------------------------------------------------------------------
474 // transform
475 // ------------------------------------------------------------------------
476
511 template <typename B, typename E, typename O, typename C,
512 typename P = DefaultPartitioner>
513 requires Partitioner<std::decay_t<P>>
514 Task transform(B first1, E last1, O d_first, C c, P part = P());
515
552 template <typename B1, typename E1, typename B2, typename O, typename C,
553 typename P = DefaultPartitioner>
554 requires (!Partitioner<std::decay_t<C>>)
555 Task transform(B1 first1, E1 last1, B2 first2, O d_first, C c, P part = P());
556
557 // ------------------------------------------------------------------------
558 // reduction
559 // ------------------------------------------------------------------------
560
594 template <typename B, typename E, typename T, typename O, typename P = DefaultPartitioner>
595 Task reduce(B first, E last, T& init, O bop, P part = P());
596
651 template <typename R, typename T, typename L, typename G, typename P = DefaultPartitioner>
652 Task reduce_by_index(R range, T& init, L lop, G gop, P part = P());
653
654 // ------------------------------------------------------------------------
655 // transform and reduction
656 // ------------------------------------------------------------------------
657
693 template <typename B, typename E, typename T, typename BOP, typename UOP,
694 typename P = DefaultPartitioner>
695 requires Partitioner<std::decay_t<P>>
696 Task transform_reduce(B first, E last, T& init, BOP bop, UOP uop, P part = P());
697
734
735 template <typename B1, typename E1, typename B2, typename T,
736 typename BOP_R, typename BOP_T, typename P = DefaultPartitioner>
737 requires (!Partitioner<std::decay_t<BOP_T>>)
738 Task transform_reduce(
739 B1 first1, E1 last1, B2 first2, T& init, BOP_R bop_r, BOP_T bop_t, P part = P()
740 );
741
742 // ------------------------------------------------------------------------
743 // scan
744 // ------------------------------------------------------------------------
745
784 template <typename B, typename E, typename D, typename BOP>
785 Task inclusive_scan(B first, E last, D d_first, BOP bop);
786
828 template <typename B, typename E, typename D, typename BOP, typename T>
829 Task inclusive_scan(B first, E last, D d_first, BOP bop, T init);
830
871 template <typename B, typename E, typename D, typename T, typename BOP>
872 Task exclusive_scan(B first, E last, D d_first, T init, BOP bop);
873
874 // ------------------------------------------------------------------------
875 // transform scan
876 // ------------------------------------------------------------------------
877
919 template <typename B, typename E, typename D, typename BOP, typename UOP>
920 Task transform_inclusive_scan(B first, E last, D d_first, BOP bop, UOP uop);
921
966 template <typename B, typename E, typename D, typename BOP, typename UOP, typename T>
967 Task transform_inclusive_scan(B first, E last, D d_first, BOP bop, UOP uop, T init);
968
1012 template <typename B, typename E, typename D, typename T, typename BOP, typename UOP>
1013 Task transform_exclusive_scan(B first, E last, D d_first, T init, BOP bop, UOP uop);
1014
1015 // ------------------------------------------------------------------------
1016 // find
1017 // ------------------------------------------------------------------------
1018
1064 template <typename B, typename E, typename T, typename UOP, typename P = DefaultPartitioner>
1065 Task find_if(B first, E last, T &result, UOP predicate, P part = P());
1066
1112 template <typename B, typename E, typename T, typename UOP, typename P = DefaultPartitioner>
1113 Task find_if_not(B first, E last, T &result, UOP predicate, P part = P());
1114
1164 template <typename B, typename E, typename T, typename C, typename P>
1165 Task min_element(B first, E last, T& result, C comp, P part);
1166
1216 template <typename B, typename E, typename T, typename C, typename P>
1217 Task max_element(B first, E last, T& result, C comp, P part);
1218
1219 // ------------------------------------------------------------------------
1220 // sort
1221 // ------------------------------------------------------------------------
1222
1242 template <typename B, typename E, typename C>
1243 Task sort(B first, E last, C cmp);
1244
1264 template <typename B, typename E>
1265 Task sort(B first, E last);
1266
1292
1293 template <typename B1, typename E1, typename B2, typename E2,
1294 typename O, typename P = DefaultPartitioner>
1295 requires Partitioner<std::decay_t<P>>
1296 Task merge(B1 first1, E1 last1, B2 first2, E2 last2, O d_first, P part = P());
1297
1326
1327 template <typename B1, typename E1, typename B2, typename E2,
1328 typename O, typename C, typename P = DefaultPartitioner>
1329 requires (!Partitioner<std::decay_t<C>>)
1330 Task merge(B1 first1, E1 last1, B2 first2, E2 last2, O d_first, C cmp, P part = P());
1331
1332 protected:
1333
1337 Graph& _graph;
1338
1339 private:
1340
1341 template <typename L>
1342 void _linearize(L&);
1343};
1344
1345// Constructor
1346 inline FlowBuilder::FlowBuilder(Graph& graph) :
1347 _graph {graph} {
1348}
1349
1350// Function: emplace
1351 template <StaticTask C>
1352 Task FlowBuilder::emplace(C&& c) {
1353 return Task(_graph._emplace_back(NSTATE::NONE, ESTATE::NONE, DefaultTaskParams{}, nullptr, nullptr, 0,
1354 std::in_place_type_t<Node::Static>{}, std::forward<C>(c)
1355 ));
1356}
1357
1358// Function: emplace
1359 template <RuntimeTask C>
1360 Task FlowBuilder::emplace(C&& c) {
1361 if constexpr (std::is_invocable_v<C, tf::Runtime&>) {
1362 return Task(_graph._emplace_back(NSTATE::NONE, ESTATE::NONE, DefaultTaskParams{}, nullptr, nullptr, 0,
1363 std::in_place_type_t<Node::Runtime>{}, std::forward<C>(c)
1364 ));
1365 }
1366 else if constexpr (std::is_invocable_v<C, tf::NonpreemptiveRuntime&>) {
1367 return Task(_graph._emplace_back(NSTATE::NONE, ESTATE::NONE, DefaultTaskParams{}, nullptr, nullptr, 0,
1368 std::in_place_type_t<Node::NonpreemptiveRuntime>{}, std::forward<C>(c)
1369 ));
1370 }
1371 else {
1372 static_assert(dependent_false_v<C>, "invalid runtime task callable");
1373 }
1374}
1375
1376// Function: emplace
1377 template <SubflowTask C>
1378 Task FlowBuilder::emplace(C&& c) {
1379 return Task(_graph._emplace_back(NSTATE::NONE, ESTATE::NONE, DefaultTaskParams{}, nullptr, nullptr, 0,
1380 std::in_place_type_t<Node::Subflow>{}, std::forward<C>(c)
1381 ));
1382}
1383
1384// Function: emplace
1385 template <ConditionTask C>
1386 Task FlowBuilder::emplace(C&& c) {
1387 return Task(_graph._emplace_back(NSTATE::NONE, ESTATE::NONE, DefaultTaskParams{}, nullptr, nullptr, 0,
1388 std::in_place_type_t<Node::Condition>{}, std::forward<C>(c)
1389 ));
1390}
1391
1392// Function: emplace
1393 template <MultiConditionTask C>
1394 Task FlowBuilder::emplace(C&& c) {
1395 return Task(_graph._emplace_back(NSTATE::NONE, ESTATE::NONE, DefaultTaskParams{}, nullptr, nullptr, 0,
1396 std::in_place_type_t<Node::MultiCondition>{}, std::forward<C>(c)
1397 ));
1398}
1399
1400// Function: composed_of
1401 template <HasGraph T>
1402 Task FlowBuilder::composed_of(T& target) {
1403 return Task(_graph._emplace_back(NSTATE::NONE, ESTATE::NONE, DefaultTaskParams{}, nullptr, nullptr, 0,
1404 std::in_place_type_t<Node::Module>{}, retrieve_graph(target)
1405 ));
1406}
1407
1408// Function: adopt
1409 inline Task FlowBuilder::adopt(Graph&& graph) {
1410 return Task(_graph._emplace_back(NSTATE::NONE, ESTATE::NONE, DefaultTaskParams{}, nullptr, nullptr, 0,
1411 std::in_place_type_t<Node::AdoptedModule>{}, std::move(graph)
1412 ));
1413}
1414
1415// Function: placeholder
1416 inline Task FlowBuilder::placeholder() {
1417 auto node = _graph._emplace_back(NSTATE::NONE, ESTATE::NONE, DefaultTaskParams{}, nullptr, nullptr, 0,
1418 std::in_place_type_t<Node::Placeholder>{}
1419 );
1420 return Task(node);
1421}
1422
1423// Function: emplace
1424 template <typename... C> requires (sizeof...(C) > 1)
1425 auto FlowBuilder::emplace(C&&... cs) {
1426 return std::make_tuple(emplace(std::forward<C>(cs))...);
1427}
1428
1429// Function: erase
1430 inline void FlowBuilder::erase(Task task) {
1431
1432 if (!task._node) {
1433 return;
1434 }
1435
1436 // remove task from its successors' predecessor list
1437 for(size_t i=0; i<task._node->_num_successors; ++i) {
1438 task._node->_edges[i]->_remove_predecessors(task._node);
1439 }
1440
1441 // remove task from its precedessors' successor list
1442 for(size_t i=task._node->_num_successors; i<task._node->_edges.size(); ++i) {
1443 task._node->_edges[i]->_remove_successors(task._node);
1444 }
1445
1446 _graph._erase(task._node);
1447}
1448
1449
1450// Procedure: _linearize
1451 template <typename L>
1452 void FlowBuilder::_linearize(L& keys) {
1453
1454 auto itr = keys.begin();
1455 auto end = keys.end();
1456
1457 if(itr == end) {
1458 return;
1459 }
1460
1461 auto nxt = itr;
1462
1463 for(++nxt; nxt != end; ++nxt, ++itr) {
1464 itr->_node->_precede(nxt->_node);
1465 }
1466}
1467
1468// Procedure: linearize
1469 inline void FlowBuilder::linearize(std::vector<Task>& keys) {
1470 _linearize(keys);
1471}
1472
1473// Procedure: linearize
1474 inline void FlowBuilder::linearize(std::initializer_list<Task> keys) {
1475 _linearize(keys);
1476}
1477
1478// ----------------------------------------------------------------------------
1479
1516 class Subflow : public FlowBuilder {
1517
1518 friend class Executor;
1519 friend class FlowBuilder;
1520
1521 public:
1522
1538 void join();
1539
1555 bool joinable() const noexcept;
1556
1560 Executor& executor() noexcept;
1561
1565 Graph& graph() { return _graph; }
1566
1576 void retain(bool flag) noexcept;
1577
1585 bool retain() const;
1586
1587 private:
1588
1589 Subflow(Executor&, Worker&, Node*, Graph&);
1590
1591 Subflow() = delete;
1592 Subflow(const Subflow&) = delete;
1593 Subflow(Subflow&&) = delete;
1594
1595 Executor& _executor;
1596 Worker& _worker;
1597 Node* _node;
1598};
1599
1600// Constructor
1601 inline Subflow::Subflow(Executor& executor, Worker& worker, Node* node, Graph& graph) :
1602 FlowBuilder {graph},
1603 _executor {executor},
1604 _worker {worker},
1605 _node {node} {
1606
1607 // need to reset since there could have iterative control flow
1608 _node->_nstate &= ~(NSTATE::JOINED_SUBFLOW | NSTATE::RETAIN_SUBFLOW);
1609
1610 // clear the graph
1611 graph.clear();
1612}
1613
1614// Function: joinable
1615 inline bool Subflow::joinable() const noexcept {
1616 return !(_node->_nstate & NSTATE::JOINED_SUBFLOW);
1617}
1618
1619// Function: executor
1620 inline Executor& Subflow::executor() noexcept {
1621 return _executor;
1622}
1623
1624// Function: retain
1625 inline void Subflow::retain(bool flag) noexcept {
1626 // default value is not to retain
1627 if(flag == true) {
1628 _node->_nstate |= NSTATE::RETAIN_SUBFLOW;
1629 }
1630 else {
1631 _node->_nstate &= ~NSTATE::RETAIN_SUBFLOW;
1632 }
1633
1634 //_node->_nstate = (_node->_nstate & ~NSTATE::RETAIN_SUBFLOW) |
1635 // (-static_cast<int>(flag) & NSTATE::RETAIN_SUBFLOW);
1636}
1637
1638// Function: retain
1639 inline bool Subflow::retain() const {
1640 return _node->_nstate & NSTATE::RETAIN_SUBFLOW;
1641}
1642
1643} // end of namespace tf. ---------------------------------------------------
tf::DefaultTaskParams
class to create an empty task parameter for compile-time optimization
Definition graph.hpp:166
tf::Executor
class to create an executor
Definition executor.hpp:62
tf::FlowBuilder::inclusive_scan
Task inclusive_scan(B first, E last, D d_first, BOP bop, T init)
creates an STL-styled parallel inclusive-scan task with an initial value
tf::FlowBuilder::inclusive_scan
Task inclusive_scan(B first, E last, D d_first, BOP bop)
creates an STL-styled parallel inclusive-scan task
tf::FlowBuilder::for_each_by_index
Task for_each_by_index(R range, C callable, P part=P())
constructs an index range-based parallel-for task
tf::FlowBuilder::sort
Task sort(B first, E last, C cmp)
constructs a dynamic task to perform STL-styled parallel sort
tf::FlowBuilder::adopt
Task adopt(Graph &&graph)
creates an adopted module task from the given graph with move semantics
Definition flow_builder.hpp:1409
tf::FlowBuilder::for_each_index
Task for_each_index(B first, E last, S step, C callable, P part=P())
constructs an index-based parallel-for task
tf::FlowBuilder::reduce_by_index
Task reduce_by_index(R range, T &init, L lop, G gop, P part=P())
constructs an index range-based parallel-reduction task
tf::FlowBuilder::find_if
Task find_if(B first, E last, T &result, UOP predicate, P part=P())
constructs a task to perform STL-styled find-if algorithm
tf::FlowBuilder::transform_inclusive_scan
Task transform_inclusive_scan(B first, E last, D d_first, BOP bop, UOP uop, T init)
creates an STL-styled parallel transform-inclusive scan task
tf::FlowBuilder::emplace
Task emplace(C &&callable)
creates a static task
Definition flow_builder.hpp:1352
tf::FlowBuilder::exclusive_scan
Task exclusive_scan(B first, E last, D d_first, T init, BOP bop)
creates an STL-styled parallel exclusive-scan task
tf::FlowBuilder::erase
void erase(Task task)
removes a task from a taskflow
Definition flow_builder.hpp:1430
tf::FlowBuilder::transform_reduce
Task transform_reduce(B first, E last, T &init, BOP bop, UOP uop, P part=P())
constructs an STL-styled parallel transform-reduce task
tf::FlowBuilder::merge
Task merge(B1 first1, E1 last1, B2 first2, E2 last2, O d_first, P part=P())
merges two sorted ranges into a single sorted output using the std::less comparator
tf::FlowBuilder::FlowBuilder
FlowBuilder(Graph &graph)
constructs a flow builder with a graph
Definition flow_builder.hpp:1346
tf::FlowBuilder::max_element
Task max_element(B first, E last, T &result, C comp, P part)
constructs a task to perform STL-styled max-element algorithm
tf::FlowBuilder::min_element
Task min_element(B first, E last, T &result, C comp, P part)
constructs a task to perform STL-styled min-element algorithm
tf::FlowBuilder::sort
Task sort(B first, E last)
constructs a dynamic task to perform STL-styled parallel sort using the std::less<T> comparator,...
tf::FlowBuilder::transform_inclusive_scan
Task transform_inclusive_scan(B first, E last, D d_first, BOP bop, UOP uop)
creates an STL-styled parallel transform-inclusive scan task
tf::FlowBuilder::transform_exclusive_scan
Task transform_exclusive_scan(B first, E last, D d_first, T init, BOP bop, UOP uop)
creates an STL-styled parallel transform-exclusive scan task
tf::FlowBuilder::linearize
void linearize(std::vector< Task > &tasks)
adds adjacent dependency links to a linear list of tasks
Definition flow_builder.hpp:1469
tf::FlowBuilder::_graph
Graph & _graph
associated graph object
Definition flow_builder.hpp:1337
tf::FlowBuilder::transform
Task transform(B first1, E last1, O d_first, C c, P part=P())
constructs a parallel-transform task
tf::FlowBuilder::find_if_not
Task find_if_not(B first, E last, T &result, UOP predicate, P part=P())
constructs a task to perform STL-styled find-if-not algorithm
tf::FlowBuilder::merge
Task merge(B1 first1, E1 last1, B2 first2, E2 last2, O d_first, C cmp, P part=P())
merges two sorted ranges into a single sorted output using the supplied comparator
tf::FlowBuilder::for_each
Task for_each(B first, E last, C callable, P part=P())
constructs an STL-styled parallel-for task
tf::FlowBuilder::transform_reduce
Task transform_reduce(B1 first1, E1 last1, B2 first2, T &init, BOP_R bop_r, BOP_T bop_t, P part=P())
constructs an STL-styled parallel transform-reduce task
tf::FlowBuilder::composed_of
Task composed_of(T &object)
creates a module task for the target object
Definition flow_builder.hpp:1402
tf::FlowBuilder::placeholder
Task placeholder()
creates a placeholder task
Definition flow_builder.hpp:1416
tf::FlowBuilder::reduce
Task reduce(B first, E last, T &init, O bop, P part=P())
constructs an STL-styled parallel-reduction task
tf::FlowBuilder::transform
Task transform(B1 first1, E1 last1, B2 first2, O d_first, C c, P part=P())
constructs a parallel-transform task
tf::Graph
class to create a graph object
Definition graph.hpp:47
tf::Graph::clear
void clear()
clears the graph
Definition graph.hpp:881
tf::Subflow
class to construct a subflow graph from the execution of a dynamic task
Definition flow_builder.hpp:1516
tf::Subflow::executor
Executor & executor() noexcept
acquires the associated executor
Definition flow_builder.hpp:1620
tf::Subflow::graph
Graph & graph()
acquires the associated graph
Definition flow_builder.hpp:1565
tf::Subflow::join
void join()
enables the subflow to join its parent task
tf::Subflow::joinable
bool joinable() const noexcept
queries if the subflow is joinable
Definition flow_builder.hpp:1615
tf::Subflow::retain
bool retain() const
queries if the subflow will be retained after it is joined
Definition flow_builder.hpp:1639
tf::Task
class to create a task handle over a taskflow node
Definition task.hpp:263
tf::Worker
class to create a worker in an executor
Definition worker.hpp:55
tf::Partitioner
determines if a type is a partitioner
Definition partitioner.hpp:815
tf
taskflow namespace
Definition small_vector.hpp:20
tf::retrieve_graph
auto & retrieve_graph(T &t)
retrieves a reference to the underlying tf::Graph from an object
Definition graph.hpp:987
tf::DefaultPartitioner
GuidedPartitioner<> DefaultPartitioner
default partitioner set to tf::GuidedPartitioner
Definition partitioner.hpp:807