FazBrowse GitHub Viewer | Trending |
URL:
| Home
Tools: [Download Repo ZIP]   [Original HTTPS Page]

made subflow preemptable · HibsStatement/taskflow@06d13fd · GitHub

  1. HibsStatement
  2. taskflow

Repository navigation

Commit 06d13fd

Browse files
TSUNG-WEI HUANG
committed
made subflow preemptable
1 parent 1f3084f commit 06d13fd

11 files changed

Lines changed: 140 additions & 267 deletions

File tree

‎examples/CMakeLists.txt‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,7 @@ list(APPEND TF_EXAMPLES
22
simple
33
attach_data
44
async
5-
subflow_async
5+
runtime_async
66
dependent_async
77
dependent_async_algorithm
88
observer

‎examples/async.cpp‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -35,11 +35,11 @@ int main() {
3535

3636
std::atomic<int> counter {0};
3737

38-
taskflow.emplace([&](tf::Subflow& sf){
38+
taskflow.emplace([&](tf::Runtime& rt){
3939
for(int i=0; i<100; i++) {
40-
sf.silent_async([&](){ counter.fetch_add(1, std::memory_order_relaxed); });
40+
rt.silent_async([&](){ counter.fetch_add(1, std::memory_order_relaxed); });
4141
}
42-
sf.join();
42+
rt.corun_all();
4343

4444
// when subflow joins, all spawned tasks from the subflow will finish
4545
if(counter == 100) {

examples/subflow_async.cpp renamed to examples/runtime_async.cpp

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -9,18 +9,18 @@ int main() {
99

1010
std::atomic<int> counter{0};
1111

12-
taskflow.emplace([&](tf::Subflow& sf){
12+
taskflow.emplace([&](tf::Runtime& rt){
1313
for(int i=0; i<10; i++) {
1414
// Here, we use "silent_async" instead of "async" because we do
1515
// not care the return value. The method "silent_async" gives us
1616
// less overhead compared to "async".
1717
// The 10 asynchronous tasks run concurrently.
18-
sf.silent_async([&](){
18+
rt.silent_async([&](){
1919
std::cout << "async task from the subflow\n";
2020
counter.fetch_add(1, std::memory_order_relaxed);
2121
});
2222
}
23-
sf.join();
23+
rt.corun_all();
2424
std::cout << counter << " = 10\n";
2525
});
2626

‎examples/simple.cpp‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -13,7 +13,7 @@
1313
#include <taskflow/taskflow.hpp> // the only include you need
1414

1515
int main(){
16-
16+
1717
tf::Executor executor;
1818
tf::Taskflow taskflow("simple");
1919

‎taskflow/algorithm/launch.hpp‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -48,7 +48,7 @@ TF_FORCE_INLINE void launch_loop(
4848
break;
4949
}
5050
else {
51-
rt.silent_async_unchecked([=](){ launch_loop(part, loop); });
51+
rt.silent_async([=](){ launch_loop(part, loop); });
5252
}
5353
}
5454

@@ -69,7 +69,7 @@ TF_FORCE_INLINE void launch_loop(
6969
launch_loop(part, loop);
7070
}
7171
else {
72-
rt.silent_async_unchecked([=](){ launch_loop(part, loop); });
72+
rt.silent_async([=](){ launch_loop(part, loop); });
7373
}
7474
}
7575

‎taskflow/core/executor.hpp‎

Lines changed: 88 additions & 64 deletions
Original file line numberDiff line numberDiff line change
@@ -1075,10 +1075,9 @@ class Executor {
10751075
void _explore_task(Worker&, Node*&);
10761076
void _schedule(Worker&, Node*);
10771077
void _schedule(Node*);
1078-
void _schedule(Worker&, const SmallVector<Node*>&);
1079-
void _schedule(const SmallVector<Node*>&);
1078+
void _schedule(Worker&, const std::vector<Node*>&, size_t);
1079+
void _schedule(const std::vector<Node*>&, size_t);
10801080
void _set_up_topology(Worker*, Topology*);
1081-
void _set_up_graph(Graph&, Node*, Topology*, int, SmallVector<Node*>&);
10821081
void _tear_down_topology(Worker&, Topology*);
10831082
void _tear_down_async(Node*);
10841083
void _tear_down_dependent_async(Worker&, Node*);
@@ -1087,7 +1086,6 @@ class Executor {
10871086
void _decrement_topology();
10881087
void _invoke(Worker&, Node*);
10891088
void _invoke_static_task(Worker&, Node*);
1090-
void _invoke_subflow_task(Worker&, Node*);
10911089
void _detach_subflow_task(Worker&, Node*, Graph&);
10921090
void _invoke_condition_task(Worker&, Node*, SmallVector<int>&);
10931091
void _invoke_multi_condition_task(Worker&, Node*, SmallVector<int>&);
@@ -1098,7 +1096,10 @@ class Executor {
10981096
void _schedule_async_task(Node*);
10991097
void _corun_graph(Worker&, Node*, Graph&);
11001098

1099+
bool _invoke_subflow_task(Worker&, Node*);
11011100
bool _invoke_module_task(Worker&, Node*);
1101+
1102+
size_t _set_up_graph(Graph&, Node*, Topology*, int);
11021103

11031104
template <typename P>
11041105
void _corun_until(Worker&, P&&);
@@ -1434,11 +1435,9 @@ inline void Executor::_schedule(Node* node) {
14341435
}
14351436

14361437
// Procedure: _schedule
1437-
inline void Executor::_schedule(Worker& worker, const SmallVector<Node*>& nodes) {
1438+
inline void Executor::_schedule(Worker& worker, const std::vector<Node*>& nodes, size_t num_nodes) {
14381439

1439-
// We need to cacth the node count to avoid accessing the nodes
1440-
// vector while the parent topology is removed!
1441-
const auto num_nodes = nodes.size();
1440+
//assert(nodes.size() >= num_nodes);
14421441

14431442
if(num_nodes == 0) {
14441443
return;
@@ -1462,10 +1461,9 @@ inline void Executor::_schedule(Worker& worker, const SmallVector<Node*>& nodes)
14621461
}
14631462

14641463
// Procedure: _schedule
1465-
inline void Executor::_schedule(const SmallVector<Node*>& nodes) {
1466-
1467-
// parent topology may be removed!
1468-
const auto num_nodes = nodes.size();
1464+
inline void Executor::_schedule(const std::vector<Node*>& nodes, size_t num_nodes) {
1465+
1466+
//assert(nodes.size() >= num_nodes);
14691467

14701468
if(num_nodes == 0) {
14711469
return;
@@ -1511,7 +1509,9 @@ inline void Executor::_invoke(Worker& worker, Node* node) {
15111509

15121510
// subflow task
15131511
case Node::SUBFLOW: {
1514-
_invoke_subflow_task(worker, node);
1512+
if(_invoke_subflow_task(worker, node)) {
1513+
return;
1514+
}
15151515
}
15161516
break;
15171517

@@ -1630,6 +1630,17 @@ inline void Executor::_tear_down_invoke(Worker& worker, Node* node) {
16301630
}
16311631
else {
16321632
switch(parent->_handle.index()) {
1633+
1634+
case Node::SUBFLOW: {
1635+
if(parent->_join_counter.fetch_sub(1, std::memory_order_acq_rel) == 1) {
1636+
// only preempted parent needs to be re-invoked
1637+
if(parent->_state.load(std::memory_order_relaxed) & Node::PREEMPTED) {
1638+
//assert(worker._cache == nullptr);
1639+
worker._cache = parent;
1640+
}
1641+
}
1642+
} break;
1643+
16331644
case Node::MODULE: {
16341645
if(parent->_join_counter.fetch_sub(1, std::memory_order_acq_rel) == 1) {
16351646
//assert(worker._cache == nullptr);
@@ -1703,19 +1714,37 @@ inline void Executor::_invoke_static_task(Worker& worker, Node* node) {
17031714
}
17041715

17051716
// Procedure: _invoke_subflow_task
1706-
inline void Executor::_invoke_subflow_task(Worker& w, Node* node) {
1707-
_observer_prologue(w, node);
1708-
TF_EXECUTOR_EXCEPTION_HANDLER(w, node, {
1717+
inline bool Executor::_invoke_subflow_task(Worker& w, Node* node) {
1718+
//TF_EXECUTOR_EXCEPTION_HANDLER(w, node, {
1719+
1720+
if((node->_state.load(std::memory_order_relaxed) & Node::PREEMPTED) == 0) {
17091721
auto handle = std::get_if<Node::Subflow>(&node->_handle);
17101722
handle->subgraph._clear();
17111723
Subflow sf(*this, w, node, handle->subgraph);
1724+
1725+
_observer_prologue(w, node);
17121726
handle->work(sf);
1713-
if(sf._joinable) {
1714-
_corun_graph(w, node, handle->subgraph);
1727+
_observer_epilogue(w, node);
1728+
1729+
// spawn the subflow if it is joinable and its graph is non-empty
1730+
if(sf._joinable && handle->subgraph.size()) {
1731+
1732+
// signal the executor to preempt this node
1733+
node->_state.fetch_or(Node::PREEMPTED, std::memory_order_relaxed);
1734+
1735+
// set up and schedule the graph
1736+
auto N = _set_up_graph(handle->subgraph, node, node->_topology, 0);
1737+
node->_join_counter.fetch_add(N, std::memory_order_relaxed);
1738+
_schedule(w, handle->subgraph._nodes, N);
1739+
return true;
17151740
}
1716-
node->_process_exception();
1717-
});
1718-
_observer_epilogue(w, node);
1741+
//node->_process_exception();
1742+
}
1743+
else {
1744+
node->_state.fetch_and(~Node::PREEMPTED, std::memory_order_relaxed);
1745+
}
1746+
1747+
return false;
17191748
}
17201749

17211750
// Procedure: _detach_subflow_task
@@ -1726,16 +1755,20 @@ inline void Executor::_detach_subflow_task(Worker& w, Node* p, Graph& g) {
17261755
return;
17271756
}
17281757

1729-
SmallVector<Node*> src;
1730-
_set_up_graph(g, nullptr, p->_topology, Node::DETACHED, src);
1758+
auto N = _set_up_graph(g, nullptr, p->_topology, Node::DETACHED);
1759+
1760+
// need to fetch the node to a local vector since the graph will be
1761+
// merged to the parent taskflow with move semantics
1762+
std::vector<Node*> local(N);
1763+
std::copy(g._nodes.begin(), g._nodes.begin() + N, local.begin());
17311764

17321765
{
17331766
std::lock_guard<std::mutex> lock(p->_topology->_taskflow._mutex);
17341767
p->_topology->_taskflow._graph._merge(std::move(g));
17351768
}
1736-
1737-
p->_topology->_join_counter.fetch_add(src.size(), std::memory_order_relaxed);
1738-
_schedule(w, src);
1769+
1770+
p->_topology->_join_counter.fetch_add(N, std::memory_order_relaxed);
1771+
_schedule(w, local, N);
17391772
}
17401773

17411774
// Procedure: _corun_graph
@@ -1748,12 +1781,9 @@ inline void Executor::_corun_graph(Worker& w, Node* p, Graph& g) {
17481781
return;
17491782
}
17501783

1751-
SmallVector<Node*> src;
1752-
1753-
_set_up_graph(g, p, p->_topology, 0, src);
1754-
p->_join_counter.fetch_add(src.size(), std::memory_order_relaxed);
1755-
1756-
_schedule(w, src);
1784+
auto N = _set_up_graph(g, p, p->_topology, 0);
1785+
p->_join_counter.fetch_add(N, std::memory_order_relaxed);
1786+
_schedule(w, g._nodes, N);
17571787

17581788
_corun_until(w, [p] () -> bool {
17591789
return p->_join_counter.load(std::memory_order_acquire) == 0; }
@@ -1806,15 +1836,19 @@ inline void Executor::_invoke_multi_condition_task(
18061836

18071837
// Procedure: _invoke_module_task
18081838
inline bool Executor::_invoke_module_task(Worker& w, Node* node) {
1839+
18091840
// spawns only non-empty module
18101841
if(auto m = std::get_if<Node::Module>(&node->_handle); m->graph.size()) {
1842+
18111843
// first entry - not spawned yet
18121844
if((node->_state.load(std::memory_order_relaxed) & Node::PREEMPTED) == 0) {
1845+
1846+
// signal the executor to preempt this node
18131847
node->_state.fetch_or(Node::PREEMPTED, std::memory_order_relaxed);
1814-
SmallVector<Node*> src;
1815-
_set_up_graph(m->graph, node, node->_topology, 0, src);
1816-
node->_join_counter.fetch_add(src.size(), std::memory_order_relaxed);
1817-
_schedule(w, src);
1848+
1849+
auto N = _set_up_graph(m->graph, node, node->_topology, 0);
1850+
node->_join_counter.fetch_add(N, std::memory_order_relaxed);
1851+
_schedule(w, m->graph._nodes, N);
18181852
return true;
18191853
}
18201854
// second entry - already spawned
@@ -2047,29 +2081,31 @@ inline void Executor::wait_for_all() {
20472081
inline void Executor::_set_up_topology(Worker* worker, Topology* tpg) {
20482082

20492083
// ---- under taskflow lock ----
2050-
2051-
tpg->_sources.clear();
20522084
tpg->_taskflow._graph._clear_detached();
2053-
_set_up_graph(tpg->_taskflow._graph, nullptr, tpg, 0, tpg->_sources);
2054-
tpg->_join_counter.store(tpg->_sources.size(), std::memory_order_relaxed);
2085+
tpg->_num_sources = _set_up_graph(tpg->_taskflow._graph, nullptr, tpg, 0);
2086+
tpg->_join_counter.store(tpg->_num_sources, std::memory_order_relaxed);
20552087

2056-
worker ? _schedule(*worker, tpg->_sources) : _schedule(tpg->_sources);
2088+
worker ? _schedule(*worker, tpg->_taskflow._graph._nodes, tpg->_num_sources) :
2089+
_schedule( tpg->_taskflow._graph._nodes, tpg->_num_sources);
20572090
}
20582091

20592092
// Function: _set_up_graph
2060-
inline void Executor::_set_up_graph(
2061-
Graph& g, Node* parent, Topology* tpg, int state, SmallVector<Node*>& src
2093+
inline size_t Executor::_set_up_graph(
2094+
Graph& g, Node* parent, Topology* tpg, int state
20622095
) {
2063-
for(auto node : g._nodes) {
2096+
size_t num_sources = 0;
2097+
for(size_t i=0; i<g._nodes.size(); i++) {
2098+
auto node = g._nodes[i];
20642099
node->_topology = tpg;
20652100
node->_parent = parent;
20662101
node->_state.store(state, std::memory_order_relaxed);
2067-
if(node->num_dependents() == 0) {
2068-
src.push_back(node);
2069-
}
20702102
node->_set_up_join_counter();
20712103
node->_exception_ptr = nullptr;
2104+
if(node->num_dependents() == 0) {
2105+
std::swap(g._nodes[num_sources++], g._nodes[i]);
2106+
}
20722107
}
2108+
return num_sources;
20732109
}
20742110

20752111
// Function: _tear_down_topology
@@ -2083,8 +2119,8 @@ inline void Executor::_tear_down_topology(Worker& worker, Topology* tpg) {
20832119
if(!tpg->_exception_ptr && !tpg->cancelled() && !tpg->_pred()) {
20842120
//assert(tpg->_join_counter == 0);
20852121
std::lock_guard<std::mutex> lock(f._mutex);
2086-
tpg->_join_counter.store(tpg->_sources.size(), std::memory_order_relaxed);
2087-
_schedule(worker, tpg->_sources);
2122+
tpg->_join_counter.store(tpg->_num_sources, std::memory_order_relaxed);
2123+
_schedule(worker, tpg->_taskflow._graph._nodes, tpg->_num_sources);
20882124
}
20892125
// case 2: the final run of this topology
20902126
else {
@@ -2255,9 +2291,9 @@ void Runtime::release(I begin, I end) {
22552291

22562292
// Destructor
22572293
inline Runtime::~Runtime() {
2258-
_executor._corun_until(_worker, [this] () -> bool {
2259-
return _parent->_join_counter.load(std::memory_order_acquire) == 0;
2260-
});
2294+
//_executor._corun_until(_worker, [this] () -> bool {
2295+
// return _parent->_join_counter.load(std::memory_order_acquire) == 0;
2296+
//});
22612297
}
22622298

22632299
// ------------------------------------
@@ -2290,18 +2326,6 @@ void Runtime::silent_async(P&& params, F&& f) {
22902326
_silent_async(*pt::worker, std::forward<P>(params), std::forward<F>(f));
22912327
}
22922328

2293-
// Function: silent_async_unchecked
2294-
template <typename F>
2295-
void Runtime::silent_async_unchecked(F&& f) {
2296-
_silent_async(_worker, DefaultTaskParams{}, std::forward<F>(f));
2297-
}
2298-
2299-
// Function: silent_async_unchecked
2300-
template <typename P, typename F>
2301-
void Runtime::silent_async_unchecked(P&& params, F&& f) {
2302-
_silent_async(_worker, std::forward<P>(params), std::forward<F>(f));
2303-
}
2304-
23052329
// ------------------------------------
23062330
// Runtime::async series
23072331
// ------------------------------------

0 commit comments

Comments
 (0)
Back | FazBrowse Home | New Git URL