| 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 <StaticTaskLike C>
52 Task emplace(C&& callable);
53
72 template <RuntimeTaskLike C>
73 Task emplace(C&& callable);
74
97 template <SubflowTaskLike C>
98 Task emplace(C&& callable);
99
130 template <ConditionTaskLike C>
131 Task emplace(C&& callable);
132
164 template <MultiConditionTaskLike 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
279 template <GraphLike T>
280 Task composed_of(T& object);
281
303 Task adopt(Graph&& graph);
304
329 Task placeholder();
330
348 void linearize(std::vector<Task>& tasks);
349
365 void linearize(std::initializer_list<Task> tasks);
366
367
368 // ------------------------------------------------------------------------
369 // parallel iterations
370 // ------------------------------------------------------------------------
371
404 template <typename B, typename E, typename C, PartitionerLike P = DefaultPartitioner>
405 Task for_each(B first, E last, C callable, P part = P());
406
446 template <typename B, typename E, typename S, typename C, PartitionerLike P = DefaultPartitioner>
447 Task for_each_index(B first, E last, S step, C callable, P part = P());
448
565 template <IndexRangesLike R, typename C, PartitionerLike P = DefaultPartitioner>
566 Task for_each_by_index(R range, C callable, P part = P());
567
568 // ------------------------------------------------------------------------
569 // transform
570 // ------------------------------------------------------------------------
571
606 template <typename B, typename E, typename O, typename C,
607 PartitionerLike P = DefaultPartitioner>
608 Task transform(B first1, E last1, O d_first, C c, P part = P());
609
646 template <typename B1, typename E1, typename B2, typename O, typename C,
647 PartitionerLike P = DefaultPartitioner>
648 requires (!PartitionerLike<std::decay_t<C>>)
649 Task transform(B1 first1, E1 last1, B2 first2, O d_first, C c, P part = P());
650
651 // ------------------------------------------------------------------------
652 // reduction
653 // ------------------------------------------------------------------------
654
688 template <typename B, typename E, typename T, typename O, PartitionerLike P = DefaultPartitioner>
689 Task reduce(B first, E last, T& init, O bop, P part = P());
690
745 template <IndexRanges1DLike R, typename T, typename L, typename G, PartitionerLike P = DefaultPartitioner>
746 Task reduce_by_index(R range, T& init, L lop, G gop, P part = P());
747
748 // ------------------------------------------------------------------------
749 // transform and reduction
750 // ------------------------------------------------------------------------
751
787 template <typename B, typename E, typename T, typename BOP, typename UOP,
788 PartitionerLike P = DefaultPartitioner>
789 Task transform_reduce(B first, E last, T& init, BOP bop, UOP uop, P part = P());
790
827
828 template <typename B1, typename E1, typename B2, typename T,
829 typename BOP_R, typename BOP_T, PartitionerLike P = DefaultPartitioner>
830 requires (!PartitionerLike<std::decay_t<BOP_T>>)
831 Task transform_reduce(
832 B1 first1, E1 last1, B2 first2, T& init, BOP_R bop_r, BOP_T bop_t, P part = P()
833 );
834
835 // ------------------------------------------------------------------------
836 // scan
837 // ------------------------------------------------------------------------
838
877 template <typename B, typename E, typename D, typename BOP>
878 Task inclusive_scan(B first, E last, D d_first, BOP bop);
879
921 template <typename B, typename E, typename D, typename BOP, typename T>
922 Task inclusive_scan(B first, E last, D d_first, BOP bop, T init);
923
964 template <typename B, typename E, typename D, typename T, typename BOP>
965 Task exclusive_scan(B first, E last, D d_first, T init, BOP bop);
966
967 // ------------------------------------------------------------------------
968 // transform scan
969 // ------------------------------------------------------------------------
970
1012 template <typename B, typename E, typename D, typename BOP, typename UOP>
1013 Task transform_inclusive_scan(B first, E last, D d_first, BOP bop, UOP uop);
1014
1059 template <typename B, typename E, typename D, typename BOP, typename UOP, typename T>
1060 Task transform_inclusive_scan(B first, E last, D d_first, BOP bop, UOP uop, T init);
1061
1105 template <typename B, typename E, typename D, typename T, typename BOP, typename UOP>
1106 Task transform_exclusive_scan(B first, E last, D d_first, T init, BOP bop, UOP uop);
1107
1108 // ------------------------------------------------------------------------
1109 // find
1110 // ------------------------------------------------------------------------
1111
1157 template <typename B, typename E, typename T, typename UOP, PartitionerLike P = DefaultPartitioner>
1158 Task find_if(B first, E last, T &result, UOP predicate, P part = P());
1159
1205 template <typename B, typename E, typename T, typename UOP, PartitionerLike P = DefaultPartitioner>
1206 Task find_if_not(B first, E last, T &result, UOP predicate, P part = P());
1207
1257 template <typename B, typename E, typename T, typename C, PartitionerLike P>
1258 Task min_element(B first, E last, T& result, C comp, P part);
1259
1309 template <typename B, typename E, typename T, typename C, PartitionerLike P>
1310 Task max_element(B first, E last, T& result, C comp, P part);
1311
1312 // ------------------------------------------------------------------------
1313 // sort
1314 // ------------------------------------------------------------------------
1315
1335 template <typename B, typename E, typename C>
1336 Task sort(B first, E last, C cmp);
1337
1357 template <typename B, typename E>
1358 Task sort(B first, E last);
1359
1393 template <typename B1, typename E1, typename B2, typename E2, typename O>
1394 Task merge(B1 first1, E1 last1, B2 first2, E2 last2, O d_first);
1395
1431 template <typename B1, typename E1, typename B2, typename E2,
1432 typename O, typename C>
1433 Task merge(B1 first1, E1 last1, B2 first2, E2 last2, O d_first, C cmp);
1434
1460 template<typename B, typename E, typename V, PartitionerLike P = DefaultPartitioner>
1461 Task fill(B first, E last, V value, P part = P());
1462
1488 template<typename B, std::integral C, typename V, PartitionerLike P = DefaultPartitioner>
1489 Task fill_n(B first, C count, V value, P part = P());
1490
1518 template <typename B, typename E, typename G, PartitionerLike P= DefaultPartitioner>
1519 Task generate(B first, E last, G gen, P part = P());
1520
1548 template <typename B, std::integral C, typename G, PartitionerLike P = DefaultPartitioner>
1549 Task generate_n(B first, C count, G gen, P part = P());
1550
1551 protected:
1552
1556 Graph& _graph;
1557
1558 private:
1559
1560 template <typename L>
1561 void _linearize(L&);
1562};
1563
1564// Constructor
1565 inline FlowBuilder::FlowBuilder(Graph& graph) :
1566 _graph {graph} {
1567}
1568
1569// Function: emplace
1570 template <StaticTaskLike C>
1571 Task FlowBuilder::emplace(C&& c) {
1572 return Task(_graph._emplace_back(NSTATE::NONE, ESTATE::NONE, DefaultTaskParams{}, nullptr, nullptr, 0,
1573 std::in_place_type_t<Node::Static>{}, std::forward<C>(c)
1574 ));
1575}
1576
1577// Function: emplace
1578 template <RuntimeTaskLike C>
1579 Task FlowBuilder::emplace(C&& c) {
1580 if constexpr (std::is_invocable_v<C, tf::Runtime&>) {
1581 return Task(_graph._emplace_back(NSTATE::NONE, ESTATE::NONE, DefaultTaskParams{}, nullptr, nullptr, 0,
1582 std::in_place_type_t<Node::Runtime>{}, std::forward<C>(c)
1583 ));
1584 }
1585 else if constexpr (std::is_invocable_v<C, tf::NonpreemptiveRuntime&>) {
1586 return Task(_graph._emplace_back(NSTATE::NONE, ESTATE::NONE, DefaultTaskParams{}, nullptr, nullptr, 0,
1587 std::in_place_type_t<Node::NonpreemptiveRuntime>{}, std::forward<C>(c)
1588 ));
1589 }
1590 else {
1591 static_assert(dependent_false_v<C>, "invalid runtime task callable");
1592 }
1593}
1594
1595// Function: emplace
1596 template <SubflowTaskLike C>
1597 Task FlowBuilder::emplace(C&& c) {
1598 return Task(_graph._emplace_back(NSTATE::NONE, ESTATE::NONE, DefaultTaskParams{}, nullptr, nullptr, 0,
1599 std::in_place_type_t<Node::Subflow>{}, std::forward<C>(c)
1600 ));
1601}
1602
1603// Function: emplace
1604 template <ConditionTaskLike C>
1605 Task FlowBuilder::emplace(C&& c) {
1606 return Task(_graph._emplace_back(NSTATE::NONE, ESTATE::NONE, DefaultTaskParams{}, nullptr, nullptr, 0,
1607 std::in_place_type_t<Node::Condition>{}, std::forward<C>(c)
1608 ));
1609}
1610
1611// Function: emplace
1612 template <MultiConditionTaskLike C>
1613 Task FlowBuilder::emplace(C&& c) {
1614 return Task(_graph._emplace_back(NSTATE::NONE, ESTATE::NONE, DefaultTaskParams{}, nullptr, nullptr, 0,
1615 std::in_place_type_t<Node::MultiCondition>{}, std::forward<C>(c)
1616 ));
1617}
1618
1619// Function: composed_of
1620 template <GraphLike T>
1621 Task FlowBuilder::composed_of(T& target) {
1622 return Task(_graph._emplace_back(NSTATE::NONE, ESTATE::NONE, DefaultTaskParams{}, nullptr, nullptr, 0,
1623 std::in_place_type_t<Node::Module>{}, retrieve_graph(target)
1624 ));
1625}
1626
1627// Function: adopt
1628 inline Task FlowBuilder::adopt(Graph&& graph) {
1629 return Task(_graph._emplace_back(NSTATE::NONE, ESTATE::NONE, DefaultTaskParams{}, nullptr, nullptr, 0,
1630 std::in_place_type_t<Node::AdoptedModule>{}, std::move(graph)
1631 ));
1632}
1633
1634// Function: placeholder
1635 inline Task FlowBuilder::placeholder() {
1636 auto node = _graph._emplace_back(NSTATE::NONE, ESTATE::NONE, DefaultTaskParams{}, nullptr, nullptr, 0,
1637 std::in_place_type_t<Node::Placeholder>{}
1638 );
1639 return Task(node);
1640}
1641
1642// Function: emplace
1643 template <typename... C> requires (sizeof...(C) > 1)
1644 auto FlowBuilder::emplace(C&&... cs) {
1645 return std::make_tuple(emplace(std::forward<C>(cs))...);
1646}
1647
1648// Function: erase
1649 inline void FlowBuilder::erase(Task task) {
1650
1651 if (!task._node) {
1652 return;
1653 }
1654
1655 // remove task from its successors' predecessor list
1656 for(size_t i=0; i<task._node->_num_successors; ++i) {
1657 task._node->_edges[i]->_remove_predecessors(task._node);
1658 }
1659
1660 // remove task from its precedessors' successor list
1661 for(size_t i=task._node->_num_successors; i<task._node->_edges.size(); ++i) {
1662 task._node->_edges[i]->_remove_successors(task._node);
1663 }
1664
1665 _graph._erase(task._node);
1666}
1667
1668
1669// Procedure: _linearize
1670 template <typename L>
1671 void FlowBuilder::_linearize(L& keys) {
1672
1673 auto itr = keys.begin();
1674 auto end = keys.end();
1675
1676 if(itr == end) {
1677 return;
1678 }
1679
1680 auto nxt = itr;
1681
1682 for(++nxt; nxt != end; ++nxt, ++itr) {
1683 itr->_node->_precede(nxt->_node);
1684 }
1685}
1686
1687// Procedure: linearize
1688 inline void FlowBuilder::linearize(std::vector<Task>& keys) {
1689 _linearize(keys);
1690}
1691
1692// Procedure: linearize
1693 inline void FlowBuilder::linearize(std::initializer_list<Task> keys) {
1694 _linearize(keys);
1695}
1696
1697// ----------------------------------------------------------------------------
1698
1735 class Subflow : public FlowBuilder {
1736
1737 friend class Executor;
1738 friend class FlowBuilder;
1739
1740 public:
1741
1757 void join();
1758
1774 bool joinable() const noexcept;
1775
1779 Executor& executor() noexcept;
1780
1784 Graph& graph() { return _graph; }
1785
1795 void retain(bool flag) noexcept;
1796
1804 bool retain() const;
1805
1806 private:
1807
1808 Subflow(Executor&, Worker&, Node*, Graph&);
1809
1810 Subflow() = delete;
1811 Subflow(const Subflow&) = delete;
1812 Subflow(Subflow&&) = delete;
1813
1814 Executor& _executor;
1815 Worker& _worker;
1816 Node* _node;
1817};
1818
1819// Constructor
1820 inline Subflow::Subflow(Executor& executor, Worker& worker, Node* node, Graph& graph) :
1821 FlowBuilder {graph},
1822 _executor {executor},
1823 _worker {worker},
1824 _node {node} {
1825
1826 // need to reset since there could have iterative control flow
1827 _node->_nstate &= ~(NSTATE::JOINED_SUBFLOW | NSTATE::RETAIN_SUBFLOW);
1828
1829 // clear the graph
1830 graph.clear();
1831}
1832
1833// Function: joinable
1834 inline bool Subflow::joinable() const noexcept {
1835 return !(_node->_nstate & NSTATE::JOINED_SUBFLOW);
1836}
1837
1838// Function: executor
1839 inline Executor& Subflow::executor() noexcept {
1840 return _executor;
1841}
1842
1843// Function: retain
1844 inline void Subflow::retain(bool flag) noexcept {
1845 // default value is not to retain
1846 if(flag == true) {
1847 _node->_nstate |= NSTATE::RETAIN_SUBFLOW;
1848 }
1849 else {
1850 _node->_nstate &= ~NSTATE::RETAIN_SUBFLOW;
1851 }
1852
1853 //_node->_nstate = (_node->_nstate & ~NSTATE::RETAIN_SUBFLOW) |
1854 // (-static_cast<int>(flag) & NSTATE::RETAIN_SUBFLOW);
1855}
1856
1857// Function: retain
1858 inline bool Subflow::retain() const {
1859 return _node->_nstate & NSTATE::RETAIN_SUBFLOW;
1860}
1861
1862} // end of namespace tf. ---------------------------------------------------
tf::DefaultTaskParams
class to create an empty task parameter for compile-time optimization
Definition graph.hpp:191
tf::Executor
class to create an executor
Definition executor.hpp:62
tf::FlowBuilder::transform
Task transform(B first1, E last1, O d_first, C c, P part=P())
constructs a parallel-transform task
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::transform
Task transform(B1 first1, E1 last1, B2 first2, O d_first, C c, P part=P())
constructs a parallel-transform task
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::merge
Task merge(B1 first1, E1 last1, B2 first2, E2 last2, O d_first, C cmp)
merges two sorted ranges into a single sorted output using a custom comparator
tf::FlowBuilder::for_each_by_index
Task for_each_by_index(R range, C callable, P part=P())
constructs a parallel-for task over a one- or multi-dimensional index range
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 a module task from a graph by taking over its ownership
Definition flow_builder.hpp:1628
tf::FlowBuilder::generate_n
Task generate_n(B first, C count, G gen, P part=P())
generates N values into a range in parallel using a callable
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:1571
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::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::erase
void erase(Task task)
removes a task from a taskflow
Definition flow_builder.hpp:1649
tf::FlowBuilder::fill
Task fill(B first, E last, V value, P part=P())
fills a range with a given value in parallel
tf::FlowBuilder::FlowBuilder
FlowBuilder(Graph &graph)
constructs a flow builder with a graph
Definition flow_builder.hpp:1565
tf::FlowBuilder::fill_n
Task fill_n(B first, C count, V value, P part=P())
fills N elements with a given value in parallel
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::generate
Task generate(B first, E last, G gen, P part=P())
generates values into a range in parallel using a callable
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:1688
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::for_each
Task for_each(B first, E last, C callable, P part=P())
constructs an STL-styled parallel-for task
tf::FlowBuilder::composed_of
Task composed_of(T &object)
creates a module task for the target object
Definition flow_builder.hpp:1621
tf::FlowBuilder::placeholder
Task placeholder()
creates a placeholder task
Definition flow_builder.hpp:1635
tf::FlowBuilder::merge
Task merge(B1 first1, E1 last1, B2 first2, E2 last2, O d_first)
merges two sorted ranges into a single sorted output using the std::less comparator
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::reduce
Task reduce(B first, E last, T &init, O bop, P part=P())
constructs an STL-styled parallel-reduction task
tf::Graph
class to create a graph object
Definition graph.hpp:47
tf::Graph::clear
void clear()
clears the graph
Definition graph.hpp:929
tf::Subflow
class to construct a subflow graph from the execution of a dynamic task
Definition flow_builder.hpp:1735
tf::Subflow::executor
Executor & executor() noexcept
acquires the associated executor
Definition flow_builder.hpp:1839
tf::Subflow::graph
Graph & graph()
acquires the associated graph
Definition flow_builder.hpp:1784
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:1834
tf::Subflow::retain
bool retain() const
queries if the subflow will be retained after it is joined
Definition flow_builder.hpp:1858
tf::Task
class to create a task handle over a taskflow node
Definition task.hpp:569
tf::Worker
class to create a worker in an executor
Definition worker.hpp:55
tf::PartitionerLike
determines if a type is a partitioner
Definition partitioner.hpp:906
tf
taskflow namespace
Definition small_vector.hpp:20
tf::retrieve_graph
Graph & retrieve_graph(T &target)
retrieves a reference to the underlying tf::Graph from an object
Definition graph.hpp:1067
tf::DefaultPartitioner
GuidedPartitioner<> DefaultPartitioner
default partitioner set to tf::GuidedPartitioner
Definition partitioner.hpp:898