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

Loading...
Searching...
No Matches
executor.hpp
1#pragma once
2
3#include "../observer/tfprof.hpp"
4#include "taskflow.hpp"
5#include "async_task.hpp"
6
12 namespace tf {
13
14// ----------------------------------------------------------------------------
15// Executor Definition
16// ----------------------------------------------------------------------------
17
62 class Executor {
63
64 friend class FlowBuilder;
65 friend class Subflow;
66 friend class Runtime;
67 friend class NonpreemptiveRuntime;
68 friend class Algorithm;
69 friend class TaskGroup;
70
71 public:
72
91 explicit Executor(
92 size_t N = std::thread::hardware_concurrency(),
93 std::shared_ptr<WorkerInterface> wif = nullptr
94 );
95
103 ~Executor();
104
123 tf::Future<void> run(Taskflow& taskflow);
124
144 tf::Future<void> run(Taskflow&& taskflow);
145
168 template<typename C>
169 tf::Future<void> run(Taskflow& taskflow, C&& callable);
170
195 template<typename C>
196 tf::Future<void> run(Taskflow&& taskflow, C&& callable);
197
217 tf::Future<void> run_n(Taskflow& taskflow, size_t N);
218
241 tf::Future<void> run_n(Taskflow&& taskflow, size_t N);
242
268 template<typename C>
269 tf::Future<void> run_n(Taskflow& taskflow, size_t N, C&& callable);
270
296 template<typename C>
297 tf::Future<void> run_n(Taskflow&& taskflow, size_t N, C&& callable);
298
322 template<typename P>
323 tf::Future<void> run_until(Taskflow& taskflow, P&& pred);
324
350 template<typename P>
351 tf::Future<void> run_until(Taskflow&& taskflow, P&& pred);
352
379 template<typename P, typename C>
380 tf::Future<void> run_until(Taskflow& taskflow, P&& pred, C&& callable);
381
410 template<typename P, typename C>
411 tf::Future<void> run_until(Taskflow&& taskflow, P&& pred, C&& callable);
412
453 template <typename T>
454 void corun(T& target);
455
484 template <typename P>
485 void corun_until(P&& predicate);
486
500 void wait_for_all();
501
512 size_t num_workers() const noexcept;
513
520 size_t num_waiters() const noexcept;
521
525 size_t num_queues() const noexcept;
526
540 size_t num_topologies() const;
541
558 Worker* this_worker();
559
577 int this_worker_id() const;
578
579 // --------------------------------------------------------------------------
580 // Observer methods
581 // --------------------------------------------------------------------------
582
600 template <typename Observer, typename... ArgsT>
601 std::shared_ptr<Observer> make_observer(ArgsT&&... args);
602
608 template <typename Observer>
609 void remove_observer(std::shared_ptr<Observer> observer);
610
614 size_t num_observers() const noexcept;
615
616 // --------------------------------------------------------------------------
617 // Async Task Methods
618 // --------------------------------------------------------------------------
619
645 template <TaskParamsLike P, typename F>
646 auto async(P&& params, F&& func);
647
671 template <typename F>
672 auto async(F&& func);
673
698 template <TaskParamsLike P, typename F>
699 void silent_async(P&& params, F&& func);
700
723 template <typename F>
724 void silent_async(F&& func);
725
726 // --------------------------------------------------------------------------
727 // Silent Dependent Async Methods
728 // --------------------------------------------------------------------------
729
757 template <typename F, AsyncTaskHandleLike... Tasks>
758 tf::AsyncTask silent_dependent_async(F&& func, Tasks&&... tasks);
759
792 template <TaskParamsLike P, typename F, AsyncTaskHandleLike... Tasks>
793 tf::AsyncTask silent_dependent_async(P&& params, F&& func, Tasks&&... tasks);
794
827 template <typename F, std::input_iterator I>
828 tf::AsyncTask silent_dependent_async(F&& func, I first, I last);
829
865 template <TaskParamsLike P, typename F, std::input_iterator I>
866 tf::AsyncTask silent_dependent_async(P&& params, F&& func, I first, I last);
867
868 // --------------------------------------------------------------------------
869 // Dependent Async Methods
870 // --------------------------------------------------------------------------
871
909 template <typename F, AsyncTaskHandleLike... Tasks>
910 auto dependent_async(F&& func, Tasks&&... tasks);
911
953 template <TaskParamsLike P, typename F, AsyncTaskHandleLike... Tasks>
954 auto dependent_async(P&& params, F&& func, Tasks&&... tasks);
955
996 template <typename F, std::input_iterator I>
997 auto dependent_async(F&& func, I first, I last);
998
1043 template <TaskParamsLike P, typename F, std::input_iterator I>
1044 auto dependent_async(P&& params, F&& func, I first, I last);
1045
1046 // ----------------------------------------------------------------------------------------------
1047 // Task Group
1048 // ----------------------------------------------------------------------------------------------
1049
1095 TaskGroup task_group();
1096
1097 private:
1098
1099 struct Buffer {
1100 std::mutex mutex;
1101 UnboundedWSQ<Node*> queue;
1102 };
1103
1104 std::vector<Worker> _workers;
1105 std::vector<Buffer> _buffers;
1106
1107 // notifier's state variable and num_topologies should sit on different cachelines
1108 // or the false sharing can cause serious performance drop
1109 alignas(TF_CACHELINE_SIZE) DefaultNotifier _notifier;
1110 alignas(TF_CACHELINE_SIZE) std::atomic<size_t> _num_topologies {0};
1111
1112 std::unordered_map<std::thread::id, Worker*> _t2w;
1113 std::unordered_set<std::shared_ptr<ObserverInterface>> _observers;
1114
1115 void _shutdown();
1116 void _observer_prologue(Worker&, Node*);
1117 void _observer_epilogue(Worker&, Node*);
1118 void _spawn(size_t, std::shared_ptr<WorkerInterface>);
1119 void _exploit_task(Worker&, Node*&);
1120 bool _explore_task(Worker&, Node*&);
1121 void _schedule(Worker&, Node*);
1122 void _schedule(Node*);
1123 void _schedule_graph(Worker&, Graph&, Topology*, NodeBase*);
1124 void _spill(Node*);
1125 void _set_up_topology(Worker*, Topology*);
1126 void _tear_down_topology(Worker&, Topology*, Node*&);
1127 void _tear_down_async(Worker&, Node*, Node*&);
1128 void _tear_down_dependent_async(Worker&, Node*, Node*&);
1129 void _tear_down_nonasync(Worker&, Node*, Node*&);
1130 void _tear_down_invoke(Worker&, Node*, Node*&);
1131 void _increment_topology();
1132 void _decrement_topology();
1133 void _invoke(Worker&, Node*);
1134 void _invoke_static_task(Worker&, Node*);
1135 void _invoke_nonpreemptive_runtime_task(Worker&, Node*);
1136 void _invoke_condition_task(Worker&, Node*, SmallVector<int>&);
1137 void _invoke_multi_condition_task(Worker&, Node*, SmallVector<int>&);
1138 void _process_dependent_async(Node*, tf::AsyncTask&, size_t&);
1139 void _process_exception(Worker&, Node*);
1140 void _update_cache(Worker&, Node*&, Node*);
1141 void _corun_graph(Worker&, Graph&, Topology*, NodeBase*);
1142
1143 bool _wait_for_task(Worker&, Node*&);
1144 bool _invoke_subflow_task(Worker&, Node*);
1145 bool _invoke_module_task(Worker&, Node*);
1146 bool _invoke_adopted_module_task(Worker&, Node*);
1147 bool _invoke_module_task_impl(Worker&, Node*, Graph&);
1148 bool _invoke_async_task(Worker&, Node*);
1149 bool _invoke_dependent_async_task(Worker&, Node*);
1150 bool _invoke_runtime_task(Worker&, Node*);
1151 bool _invoke_runtime_task_impl(Worker&, Node*, std::function<void(Runtime&)>&);
1152 bool _invoke_runtime_task_impl(Worker&, Node*, std::function<void(Runtime&, bool)>&);
1153
1154 size_t _set_up_graph(Graph&, Topology*, NodeBase*);
1155
1156 template <typename P>
1157 void _corun_until(Worker&, P&&);
1158
1159 template <std::input_iterator I>
1160 void _bulk_schedule(Worker&, I, size_t);
1161
1162 template <std::input_iterator I>
1163 void _bulk_schedule(I, size_t);
1164
1165 template <std::input_iterator I>
1166 void _bulk_spill(I, size_t);
1167
1168 template <std::input_iterator I>
1169 void _bulk_spill_round_robin(I, size_t);
1170
1171 template <size_t N>
1172 void _bulk_update_cache(Worker&, Node*&, Node*, std::array<Node*, N>&, size_t&);
1173
1174 template <TaskParamsLike P, typename F>
1175 auto _async(P&&, F&&, Topology*, NodeBase*);
1176
1177 template <TaskParamsLike P, typename F>
1178 void _silent_async(P&&, F&&, Topology*, NodeBase*);
1179
1180 template <TaskParamsLike P, typename F, std::input_iterator I>
1181 auto _dependent_async(P&&, F&&, I, I, Topology*, NodeBase*);
1182
1183 template <TaskParamsLike P, typename F, std::input_iterator I>
1184 auto _silent_dependent_async(P&&, F&&, I, I, Topology*, NodeBase*);
1185
1186 template <typename... ArgsT>
1187 void _schedule_async_task(ArgsT&&...);
1188
1189 template <std::input_iterator I, typename... ArgsT>
1190 AsyncTask _schedule_dependent_async_task(I, I, size_t, ArgsT&&...);
1191};
1192
1193#ifndef DOXYGEN_GENERATING_OUTPUT
1194
1195// Constructor
1196 inline Executor::Executor(size_t N, std::shared_ptr<WorkerInterface> wif) :
1197 _workers (N),
1198 _buffers (std::bit_width(N)), // Empirically, we find that log2(N) performs best.
1199 _notifier (N) {
1200
1201 if(N == 0) {
1202 TF_THROW("executor must define at least one worker");
1203 }
1204
1205 // If spawning N threads fails, shut down any created threads before
1206 // rethrowing the exception.
1207#ifndef TF_DISABLE_EXCEPTION_HANDLING
1208 try {
1209#endif
1210 _spawn(N, std::move(wif));
1211#ifndef TF_DISABLE_EXCEPTION_HANDLING
1212 }
1213 catch(...) {
1214 _shutdown();
1215 std::rethrow_exception(std::current_exception());
1216 }
1217#endif
1218
1219 // initialize the default observer if requested
1220 if(has_env(TF_ENABLE_PROFILER)) {
1221 TFProfManager::get()._manage(make_observer<TFProfObserver>());
1222 }
1223}
1224
1225// Destructor
1226 inline Executor::~Executor() {
1227 _shutdown();
1228}
1229
1230// Function: _shutdown
1231 inline void Executor::_shutdown() {
1232
1233 // wait for all topologies to complete
1234 wait_for_all();
1235
1236 // shut down the scheduler
1237 for(size_t i=0; i<_workers.size(); ++i) {
1238 _workers[i]._done.test_and_set(std::memory_order_relaxed);
1239 }
1240
1241 _notifier.notify_all();
1242
1243 // Only join the thread if it is joinable, as std::thread construction
1244 // may fail and throw an exception.
1245 for(auto& w : _workers) {
1246 if(w._thread.joinable()) {
1247 w._thread.join();
1248 }
1249 }
1250}
1251
1252// Function: num_workers
1253 inline size_t Executor::num_workers() const noexcept {
1254 return _workers.size();
1255}
1256
1257// Function: num_waiters
1258 inline size_t Executor::num_waiters() const noexcept {
1259 return _notifier.num_waiters();
1260}
1261
1262// Function: num_queues
1263 inline size_t Executor::num_queues() const noexcept {
1264 return _workers.size() + _buffers.size();
1265}
1266
1267// Function: num_topologies
1268 inline size_t Executor::num_topologies() const {
1269 return _num_topologies.load(std::memory_order_relaxed);
1270}
1271
1272// Function: this_worker
1273 inline Worker* Executor::this_worker() {
1274 auto itr = _t2w.find(std::this_thread::get_id());
1275 return itr == _t2w.end() ? nullptr : itr->second;
1276}
1277
1278// Function: this_worker_id
1279 inline int Executor::this_worker_id() const {
1280 auto i = _t2w.find(std::this_thread::get_id());
1281 return i == _t2w.end() ? -1 : static_cast<int>(i->second->_id);
1282}
1283
1284// Procedure: _spawn
1285 inline void Executor::_spawn(size_t N, std::shared_ptr<WorkerInterface> wif) {
1286
1287 for(size_t id=0; id<N; ++id) {
1288 _workers[id]._thread = std::thread([&, id, wif] () {
1289
1290 auto& worker = _workers[id];
1291
1292 worker._id = id;
1293 worker._sticky_victim = id;
1294 worker._rdgen.seed(static_cast<uint32_t>(std::hash<std::thread::id>()(std::this_thread::get_id())));
1295
1296 // before entering the work-stealing loop, call the scheduler prologue
1297 if(wif) {
1298 wif->scheduler_prologue(worker);
1299 }
1300
1301 Node* t = nullptr;
1302 std::exception_ptr ptr = nullptr;
1303
1304 // must use 1 as condition instead of !done because
1305 // the previous worker may stop while the following workers
1306 // are still preparing for entering the scheduling loop
1307#ifndef TF_DISABLE_EXCEPTION_HANDLING
1308 try {
1309#endif
1310 // work-stealing loop
1311 while(1) {
1312
1313 // drains out the local queue first
1314 _exploit_task(worker, t);
1315
1316 // steals and waits for tasks
1317 if(_wait_for_task(worker, t) == false) {
1318 break;
1319 }
1320 }
1321
1322#ifndef TF_DISABLE_EXCEPTION_HANDLING
1323 }
1324 catch(...) {
1325 ptr = std::current_exception();
1326 }
1327#endif
1328
1329 // call the user-specified epilogue function
1330 if(wif) {
1331 wif->scheduler_epilogue(worker, ptr);
1332 }
1333
1334 });
1335
1336 // We avoid using thread-local storage to track the mapping between a thread
1337 // and its corresponding worker in an executor. On Windows, thread-local
1338 // storage can be unreliable in certain situations (see issue #727).
1339 //
1340 // Instead, we maintain a per-executor mapping from threads to workers.
1341 // This approach has an additional advantage: according to the C++ Standard,
1342 // std::thread::id uniquely identifies a thread object. Therefore, once the map
1343 // returns a valid worker, we can be certain that the worker belongs to this
1344 // executor. This eliminates the need for additional executor validation
1345 // required by using thread-local storage.
1346 //
1347 // Example:
1348 //
1349 // Worker* w = this_worker();
1350 // // Using thread-local storage, we would need additional executor validation:
1351 // if (w == nullptr || w->_executor != this) { /* caller is not a worker of this executor */ }
1352 //
1353 // // Using per-executor mapping, it suffices to check:
1354 // if (w == nullptr) { /* caller is not a worker of this executor */ }
1355 //
1356 _t2w.emplace(_workers[id]._thread.get_id(), &_workers[id]);
1357 }
1358}
1359
1360// Function: _explore_task
1361 inline bool Executor::_explore_task(Worker& w, Node*& t) {
1362
1363 // Fast path: if no topologies are live, all queues are guaranteed empty
1364 // by the executor's invariant (num_topologies reaches zero only after all
1365 // nodes have been scheduled and their queues flushed). Skip the entire
1366 // steal loop and return immediately so the caller enters _wait_for_task
1367 // to sleep. relaxed ordering is sufficient — this is a hint, and any
1368 // missed update is caught safely by the 2PC guard in _wait_for_task.
1369 if(_num_topologies.load(std::memory_order_relaxed) == 0) {
1370 return true;
1371 }
1372
1373 const size_t MAX_VICTIM = num_queues(); // guaranteed >= 2 by constructor
1374 const size_t MAX_STEALS = ((MAX_VICTIM + 1) << 1);
1375
1376 // local aliases for steal protocol sentinels — these are properties of the
1377 // steal protocol, not of any specific queue type
1378 size_t num_steals = 0;
1379 size_t vtm = w._sticky_victim;
1380
1381 while(true) {
1382
1383 t = (vtm < _workers.size())
1384 ? _workers[vtm]._wsq.steal()
1385 : _buffers[vtm - _workers.size()].queue.steal();
1386
1387 if(t) {
1388 w._sticky_victim = vtm;
1389 break;
1390 }
1391
1392 // EMPTY: pick a new victim excluding self since our own queue is likely empty.
1393 // map [0, MAX_VICTIM-1) over [0, MAX_VICTIM) \ {w._id} — always safe since MAX_VICTIM >= 2.
1394 vtm = w._rdgen() % (MAX_VICTIM - 1);
1395 if(vtm >= w._id) vtm++;
1396
1397 if(++num_steals > MAX_STEALS) {
1398 std::this_thread::yield();
1399 if(num_steals > 150 + MAX_STEALS) {
1400 break;
1401 }
1402 }
1403
1404 if(w._done.test(std::memory_order_relaxed)) {
1405 return false;
1406 }
1407 }
1408
1409 return true;
1410}
1411
1412/*
1413// Function: _explore_task
1414 inline bool Executor::_explore_task(Worker& w, Node*& t) {
1415
1416 // Fast path: if no topologies are live, all queues are guaranteed empty
1417 // by the executor's invariant (num_topologies reaches zero only after all
1418 // nodes have been scheduled and their queues flushed). Skip the entire
1419 // steal loop and return immediately so the caller enters _wait_for_task
1420 // to sleep. relaxed ordering is sufficient — this is a hint, and any
1421 // missed update is caught safely by the 2PC guard in _wait_for_task.
1422 if(_num_topologies.load(std::memory_order_relaxed) == 0) {
1423 return true;
1424 }
1425
1426 //assert(!t);
1427 const size_t MAX_VICTIM = num_queues();
1428 const size_t MAX_STEALS = ((MAX_VICTIM + 1) << 1);
1429
1430 size_t num_steals = 0;
1431 size_t vtm = w._sticky_victim;
1432
1433 // Make the worker steal immediately from the assigned victim.
1434 while(true) {
1435
1436 // If the worker's victim thread is within the worker pool, steal from the worker's queue.
1437 // Otherwise, steal from the buffer, adjusting the victim index based on the worker pool size.
1438 t = (vtm < _workers.size())
1439 ? _workers[vtm]._wsq.steal()
1440 : _buffers[vtm - _workers.size()].queue.steal();
1441
1442 if(t) {
1443 w._sticky_victim = vtm;
1444 break;
1445 }
1446
1447 // Increment the steal count, and if it exceeds MAX_STEALS, yield the thread.
1448 // If the number of empty steals reaches MAX_STEALS, exit the loop.
1449 if (++num_steals > MAX_STEALS) {
1450 std::this_thread::yield();
1451 if(num_steals > 150 + MAX_STEALS) {
1452 break;
1453 }
1454 }
1455
1456 if(w._done.test(std::memory_order_relaxed)) {
1457 return false;
1458 }
1459
1460 // Randomely generate a next victim.
1461 vtm = w._rdgen() % MAX_VICTIM;
1462 }
1463 return true;
1464}
1465*/
1466
1467// Procedure: _exploit_task
1468 inline void Executor::_exploit_task(Worker& w, Node*& t) {
1469 while(t) {
1470 _invoke(w, t);
1471 t = w._wsq.pop();
1472 }
1473}
1474
1475// Function: _wait_for_task
1476 inline bool Executor::_wait_for_task(Worker& w, Node*& t) {
1477
1478 explore_task:
1479
1480 if(_explore_task(w, t) == false) {
1481 return false;
1482 }
1483
1484 // Go exploit the task if we successfully steal one.
1485 if(t) {
1486 return true;
1487 }
1488
1489 // Entering the 2PC guard as all queues are likely empty after many stealing attempts.
1490 _notifier.prepare_wait(w._id);
1491
1492 // Fast path: if no topologies are live, all queues are guaranteed empty.
1493 // Skip the O(N) buffer and worker queue scans and go directly to sleep.
1494 // This is safe because prepare_wait has already been called — any notify
1495 // that arrives after this check but before commit_wait will be caught by
1496 // the 2PC guarantee of the notifier.
1497 if(_num_topologies.load(std::memory_order_relaxed) == 0) {
1498 // still check done flag before committing to sleep
1499 if(w._done.test(std::memory_order_relaxed)) {
1500 _notifier.cancel_wait(w._id);
1501 return false;
1502 }
1503 _notifier.commit_wait(w._id);
1504 goto explore_task;
1505 }
1506
1507 // Condition #1: buffers should be empty
1508 for(size_t b=0; b<_buffers.size(); ++b) {
1509 if(!_buffers[b].queue.empty()) {
1510 _notifier.cancel_wait(w._id);
1511 w._sticky_victim = b + _workers.size();
1512 goto explore_task;
1513 }
1514 }
1515
1516 // Condition #2: worker queues should be empty
1517 // Note: We need to use index-based looping to avoid data race with _spawn
1518 // which initializes other worker data structure at the same time.
1519 // Also, due to the property of a work-stealing queue, we don't need to check
1520 // this worker's work-stealing queue.
1521 for(size_t k=0; k<_workers.size()-1; ++k) {
1522 if(size_t vtm = k + (k >= w._id); !_workers[vtm]._wsq.empty()) {
1523 _notifier.cancel_wait(w._id);
1524 w._sticky_victim = vtm;
1525 goto explore_task;
1526 }
1527 }
1528
1529 // Condition #3: worker should be alive
1530 if(w._done.test(std::memory_order_relaxed)) {
1531 _notifier.cancel_wait(w._id);
1532 return false;
1533 }
1534
1535 // Now I really need to relinquish myself to others.
1536 _notifier.commit_wait(w._id);
1537 goto explore_task;
1538}
1539
1540// Function: make_observer
1541 template<typename Observer, typename... ArgsT>
1542std::shared_ptr<Observer> Executor::make_observer(ArgsT&&... args) {
1543
1544 static_assert(
1545 std::is_base_of_v<ObserverInterface, Observer>,
1546 "Observer must be derived from ObserverInterface"
1547 );
1548
1549 // use a local variable to mimic the constructor
1550 auto ptr = std::make_shared<Observer>(std::forward<ArgsT>(args)...);
1551
1552 ptr->set_up(_workers.size());
1553
1554 _observers.emplace(std::static_pointer_cast<ObserverInterface>(ptr));
1555
1556 return ptr;
1557}
1558
1559// Procedure: remove_observer
1560 template <typename Observer>
1561 void Executor::remove_observer(std::shared_ptr<Observer> ptr) {
1562
1563 static_assert(
1564 std::is_base_of_v<ObserverInterface, Observer>,
1565 "Observer must be derived from ObserverInterface"
1566 );
1567
1568 _observers.erase(std::static_pointer_cast<ObserverInterface>(ptr));
1569}
1570
1571// Function: num_observers
1572 inline size_t Executor::num_observers() const noexcept {
1573 return _observers.size();
1574}
1575
1576// Procedure: _spill
1577 inline void Executor::_spill(Node* item) {
1578 // Since pointers are aligned to 8 bytes, we perform a simple hash to avoid
1579 // contention caused by hashing to the same slot.
1580 auto b = (reinterpret_cast<uintptr_t>(item) >> 16) % _buffers.size();
1581 std::scoped_lock lock(_buffers[b].mutex);
1582 _buffers[b].queue.push(item);
1583}
1584
1585// Procedure: _bulk_spill (single batch to one buffer)
1586// Uses Knuth multiplicative hash on the first pointer to select a buffer,
1587// providing better bit diffusion than the shift-based approach, especially
1588// when the allocator returns pointers with regular low-bit patterns.
1589 template <std::input_iterator I>
1590 void Executor::_bulk_spill(I first, size_t N) {
1591 //assert(N != 0);
1592 auto b = ((reinterpret_cast<uintptr_t>(*first) * 2654435761ULL) >> 32) % _buffers.size();
1593 std::scoped_lock lock(_buffers[b].mutex);
1594 _buffers[b].queue.bulk_push(first, N);
1595}
1596
1597// Procedure: _bulk_spill
1598// Distributes a batch of N spilled nodes across all buffers in round-robin
1599// order starting from a hash of the first node's pointer. Each buffer's lock
1600// is held only for its chunk, reducing contention compared to sending the
1601// entire batch to a single buffer.
1602 template <std::input_iterator I>
1603 void Executor::_bulk_spill_round_robin(I first, size_t N) {
1604
1605 // assert(N != 0);
1606 const size_t B = _buffers.size();
1607 const size_t start = ((reinterpret_cast<uintptr_t>(*first) * 2654435761ULL) >> 32) % B;
1608 const size_t per_buf = (N + B - 1) / B;
1609 size_t remaining = N;
1610 for(size_t i = 0; i < B && remaining > 0; ++i) {
1611 size_t b = (start + i) % B;
1612 size_t chunk = std::min(per_buf, remaining);
1613 {
1614 std::scoped_lock lock(_buffers[b].mutex);
1615 _buffers[b].queue.bulk_push(first, chunk);
1616 }
1617 // terminates early via remaining > 0, so we don't acquire unnecessary locks on empty chunks.
1618 remaining -= chunk;
1619 }
1620}
1621
1622// Procedure: _schedule
1623 inline void Executor::_schedule(Worker& worker, Node* node) {
1624 // starting at v3.5 we do not use any complicated notification mechanism
1625 // as the experimental result has shown no significant advantage.
1626 if(worker._wsq.try_push(node) == false) {
1627 _spill(node);
1628 }
1629 _notifier.notify_one();
1630}
1631
1632// Procedure: _schedule
1633 inline void Executor::_schedule(Node* node) {
1634 _spill(node);
1635 _notifier.notify_one();
1636}
1637
1638// Procedure: _schedule
1639 template <std::input_iterator I>
1640 void Executor::_bulk_schedule(Worker& worker, I first, size_t num_nodes) {
1641
1642 if(num_nodes == 0) {
1643 return;
1644 }
1645
1646 // NOTE: We cannot use first/last in the for-loop (e.g., for(; first != last; ++first)).
1647 // This is because when a node v is inserted into the queue, v can run and finish
1648 // immediately. If v is the last node in the graph, it will tear down the parent task vector
1649 // which cause the last ++first to fail. This problem is specific to MSVC which has a stricter
1650 // iterator implementation in std::vector than GCC/Clang.
1651 if(auto n = worker._wsq.try_bulk_push(first, num_nodes); n != num_nodes) {
1652 _bulk_spill(first, num_nodes - n);
1653 }
1654 _notifier.notify_n(num_nodes);
1655
1656 // notify first before spilling to hopefully wake up workers earlier
1657 // however, the experiment does not show any benefit for doing this.
1658 //auto n = worker._wsq.try_bulk_push(first, num_nodes);
1659 //_notifier.notify_n(n);
1660 //_bulk_schedule(first + n, num_nodes - n);
1661}
1662
1663// Procedure: _schedule
1664 template <std::input_iterator I>
1665 inline void Executor::_bulk_schedule(I first, size_t num_nodes) {
1666
1667 if(num_nodes == 0) {
1668 return;
1669 }
1670
1671 // NOTE: We cannot use first/last in the for-loop (e.g., for(; first != last; ++first)).
1672 // This is because when a node v is inserted into the queue, v can run and finish
1673 // immediately. If v is the last node in the graph, it will tear down the parent task vector
1674 // which cause the last ++first to fail. This problem is specific to MSVC which has a stricter
1675 // iterator implementation in std::vector than GCC/Clang.
1676 _bulk_spill(first, num_nodes);
1677 _notifier.notify_n(num_nodes);
1678}
1679
1680// Function: _update_cache
1681TF_FORCE_INLINE void Executor::_update_cache(Worker& worker, Node*& cache, Node* node) {
1682 if(cache) {
1683 _schedule(worker, cache);
1684 }
1685 cache = node;
1686}
1687
1688// Function: _bulk_update_cache
1689 template <size_t N>
1690TF_FORCE_INLINE void Executor::_bulk_update_cache(
1691 Worker& worker, Node*& cache, Node* node, std::array<Node*, N>& array, size_t& n
1692) {
1693 // experimental results show no benefit of using bulk_update_cache
1694 if(cache) {
1695 array[n++] = cache;
1696 if(n == N) {
1697 _bulk_schedule(worker, array, n);
1698 n = 0;
1699 }
1700 }
1701 cache = node;
1702}
1703
1704// Procedure: _invoke
1705 inline void Executor::_invoke(Worker& worker, Node* node) {
1706
1707 #define TF_INVOKE_CONTINUATION() \
1708 if (cache) { \
1709 node = cache; \
1710 goto begin_invoke; \
1711 }
1712
1713 begin_invoke:
1714
1715 Node* cache {nullptr};
1716
1717 // if this is the second invoke due to preemption, directly jump to invoke task
1718 if(node->_nstate & NSTATE::PREEMPTED) {
1719 goto invoke_task;
1720 }
1721
1722 // If the work has been cancelled, there is no need to continue.
1723 // Here, we do tear_down_invoke since async tasks may also get cancelled where
1724 // we need to recycle the node.
1725 if(node->_is_parent_cancelled()) {
1726 _tear_down_invoke(worker, node, cache);
1727 TF_INVOKE_CONTINUATION();
1728 return;
1729 }
1730
1731 // if acquiring semaphore(s) exists, acquire them first
1732 if(node->_semaphores && !node->_semaphores->to_acquire.empty()) {
1733 SmallVector<Node*> waiters;
1734 if(!node->_acquire_all(waiters)) {
1735 _bulk_schedule(worker, waiters.begin(), waiters.size());
1736 return;
1737 }
1738 }
1739
1740 invoke_task:
1741
1742 SmallVector<int> conds;
1743
1744 // switch is faster than nested if-else due to jump table
1745 switch(node->_handle.index()) {
1746 // static task
1747 case Node::STATIC:{
1748 _invoke_static_task(worker, node);
1749 }
1750 break;
1751
1752 // runtime task
1753 case Node::RUNTIME:{
1754 if(_invoke_runtime_task(worker, node)) {
1755 return;
1756 }
1757 }
1758 break;
1759
1760 // non-preemptive runtime task
1761 case Node::NONPREEMPTIVE_RUNTIME:{
1762 _invoke_nonpreemptive_runtime_task(worker, node);
1763 }
1764 break;
1765
1766 // subflow task
1767 case Node::SUBFLOW: {
1768 if(_invoke_subflow_task(worker, node)) {
1769 return;
1770 }
1771 }
1772 break;
1773
1774 // condition task
1775 case Node::CONDITION: {
1776 _invoke_condition_task(worker, node, conds);
1777 }
1778 break;
1779
1780 // multi-condition task
1781 case Node::MULTI_CONDITION: {
1782 _invoke_multi_condition_task(worker, node, conds);
1783 }
1784 break;
1785
1786 // module task
1787 case Node::MODULE: {
1788 if(_invoke_module_task(worker, node)) {
1789 return;
1790 }
1791 }
1792 break;
1793
1794 // adopted module task
1795 case Node::ADOPTED_MODULE: {
1796 if(_invoke_adopted_module_task(worker, node)) {
1797 return;
1798 }
1799 }
1800 break;
1801
1802 // async task
1803 case Node::ASYNC: {
1804 if(_invoke_async_task(worker, node)) {
1805 return;
1806 }
1807 _tear_down_async(worker, node, cache);
1808 TF_INVOKE_CONTINUATION();
1809 return;
1810 }
1811 break;
1812
1813 // dependent async task
1814 case Node::DEPENDENT_ASYNC: {
1815 if(_invoke_dependent_async_task(worker, node)) {
1816 return;
1817 }
1818 _tear_down_dependent_async(worker, node, cache);
1819 TF_INVOKE_CONTINUATION();
1820 return;
1821 }
1822 break;
1823
1824 // monostate (placeholder)
1825 default:
1826 break;
1827 }
1828
1829 // if releasing semaphores exist, release them
1830 if(node->_semaphores && !node->_semaphores->to_release.empty()) {
1831 SmallVector<Node*> waiters;
1832 node->_release_all(waiters);
1833 _bulk_schedule(worker, waiters.begin(), waiters.size());
1834 }
1835
1836 // Reset the join counter with strong dependencies to support cycles.
1837 // + We must do this before scheduling the successors to avoid race
1838 // condition on _predecessors.
1839 // + We must use fetch_add instead of direct assigning
1840 // because the user-level call on "invoke" may explicitly schedule
1841 // this task again (e.g., pipeline) which can access the join_counter.
1842 node->_join_counter.fetch_add(
1843 node->_nstate & NSTATE::STRONG_DEPENDENCIES_MASK, std::memory_order_relaxed
1844 );
1845
1846 // Invoke the task based on the corresponding type
1847 switch(node->_handle.index()) {
1848
1849 // condition and multi-condition tasks
1850 case Node::CONDITION:
1851 case Node::MULTI_CONDITION: {
1852 for(auto cond : conds) {
1853 if(cond >= 0 && static_cast<size_t>(cond) < node->_num_successors) {
1854 auto s = node->_edges[cond];
1855 // zeroing the join counter for invariant
1856 s->_join_counter.store(0, std::memory_order_relaxed);
1857 node->_parent->_join_counter.fetch_add(1, std::memory_order_relaxed);
1858 _update_cache(worker, cache, s);
1859 }
1860 }
1861 }
1862 break;
1863
1864 // non-condition task
1865 default: {
1866 for(size_t i=0; i<node->_num_successors; ++i) {
1867 if(auto s = node->_edges[i]; s->_join_counter.fetch_sub(1, std::memory_order_acq_rel) == 1) {
1868 node->_parent->_join_counter.fetch_add(1, std::memory_order_relaxed);
1869 _update_cache(worker, cache, s);
1870 }
1871 }
1872 }
1873 break;
1874 }
1875
1876 // clean up the node after execution
1877 _tear_down_nonasync(worker, node, cache);
1878 TF_INVOKE_CONTINUATION();
1879}
1880
1881// Procedure: _tear_down_nonasync
1882 inline void Executor::_tear_down_nonasync(Worker& worker, Node* node, Node*& cache) {
1883
1884 // we must check parent first before subtracting the join counter,
1885 // or it can introduce data race
1886 if(auto parent = node->_parent; parent == node->_topology) {
1887 if(parent->_join_counter.fetch_sub(1, std::memory_order_acq_rel) == 1) {
1888 _tear_down_topology(worker, node->_topology, cache);
1889 }
1890 }
1891 else {
1892 // needs to fetch every data before join counter becomes zero at which
1893 // the node may be deleted
1894 auto state = parent->_nstate;
1895 if(parent->_join_counter.fetch_sub(1, std::memory_order_acq_rel) == 1) {
1896 // this task is spawned from a preempted parent, so we need to resume it
1897 if(state & NSTATE::PREEMPTED) {
1898 _update_cache(worker, cache, static_cast<Node*>(parent));
1899 }
1900 }
1901 }
1902}
1903
1904// Procedure: _tear_down_invoke
1905 inline void Executor::_tear_down_invoke(Worker& worker, Node* node, Node*& cache) {
1906 switch(node->_handle.index()) {
1907 case Node::ASYNC:
1908 _tear_down_async(worker, node, cache);
1909 break;
1910
1911 case Node::DEPENDENT_ASYNC:
1912 _tear_down_dependent_async(worker, node, cache);
1913 break;
1914
1915 default:
1916 _tear_down_nonasync(worker, node, cache);
1917 break;
1918 }
1919}
1920
1921// Procedure: _observer_prologue
1922 inline void Executor::_observer_prologue(Worker& worker, Node* node) {
1923 for(auto& observer : _observers) {
1924 observer->on_entry(WorkerView(worker), TaskView(*node));
1925 }
1926}
1927
1928// Procedure: _observer_epilogue
1929 inline void Executor::_observer_epilogue(Worker& worker, Node* node) {
1930 for(auto& observer : _observers) {
1931 observer->on_exit(WorkerView(worker), TaskView(*node));
1932 }
1933}
1934
1935// Procedure: _process_exception
1936 inline void Executor::_process_exception(Worker&, Node* node) {
1937
1938 // Finds the anchor and mark the entire path with exception,
1939 // so recursive tasks can be cancelled properly.
1940 // Since exception can come from asynchronous task (with runtime), the node itself can be anchored.
1941 NodeBase* ea = node; // explicit anchor
1942 NodeBase* ia = nullptr; // implicit anchor
1943
1944 while(ea && (ea->_estate.load(std::memory_order_relaxed) & ESTATE::EXPLICITLY_ANCHORED) == 0) {
1945 ea->_estate.fetch_or(ESTATE::EXCEPTION, std::memory_order_relaxed);
1946 // we only want the inner-most implicit anchor
1947 if(ia == nullptr && (ea->_nstate & NSTATE::IMPLICITLY_ANCHORED)) {
1948 ia = ea;
1949 }
1950 ea = ea->_parent;
1951 }
1952
1953 // flag used to ensure execution is caught in a thread-safe manner
1954 constexpr static auto flag = ESTATE::EXCEPTION | ESTATE::CAUGHT;
1955
1956 // The exception occurs under a blocking call (e.g., corun, join).
1957 if(ea) {
1958 // multiple tasks may throw, and we only take the first thrown exception
1959 if((ea->_estate.fetch_or(flag, std::memory_order_relaxed) & ESTATE::CAUGHT) == 0) {
1960 ea->_exception_ptr = std::current_exception();
1961 return;
1962 }
1963 }
1964 // Implicit anchor has the lowest priority
1965 else if(ia){
1966 if((ia->_estate.fetch_or(flag, std::memory_order_relaxed) & ESTATE::CAUGHT) == 0) {
1967 ia->_exception_ptr = std::current_exception();
1968 return;
1969 }
1970 }
1971
1972 // For now, we simply store the exception in this node; this can happen in an
1973 // execution that does not have any external control to capture the exception,
1974 // such as silent async task without any parent.
1975 node->_exception_ptr = std::current_exception();
1976}
1977
1978// Procedure: _invoke_static_task
1979 inline void Executor::_invoke_static_task(Worker& worker, Node* node) {
1980 _observer_prologue(worker, node);
1981 TF_EXECUTOR_EXCEPTION_HANDLER(worker, node, {
1982 std::get_if<Node::Static>(&node->_handle)->work();
1983 });
1984 _observer_epilogue(worker, node);
1985}
1986
1987// Procedure: _invoke_subflow_task
1988 inline bool Executor::_invoke_subflow_task(Worker& worker, Node* node) {
1989
1990 auto& h = *std::get_if<Node::Subflow>(&node->_handle);
1991 auto& g = h.subgraph;
1992
1993 if((node->_nstate & NSTATE::PREEMPTED) == 0) {
1994
1995 // set up the subflow
1996 Subflow sf(*this, worker, node, g);
1997
1998 // invoke the subflow callable
1999 _observer_prologue(worker, node);
2000 TF_EXECUTOR_EXCEPTION_HANDLER(worker, node, {
2001 h.work(sf);
2002 });
2003 _observer_epilogue(worker, node);
2004
2005 // spawn the subflow if it is joinable and its graph is non-empty
2006 // implicit join is faster than Subflow::join as it does not involve corun
2007 if(sf.joinable() && !g.empty()) {
2008
2009 // signal the executor to preempt this node
2010 node->_nstate |= NSTATE::PREEMPTED;
2011
2012 // set up and schedule the graph
2013 _schedule_graph(worker, g, node->_topology, node);
2014 return true;
2015 }
2016 }
2017 else {
2018 node->_nstate &= ~NSTATE::PREEMPTED;
2019 }
2020
2021 // The subflow has finished or joined.
2022 // By default, we clear the subflow storage as applications can perform recursive
2023 // subflow tasking which accumulates a huge amount of memory overhead, hampering
2024 // the performance.
2025 if((node->_nstate & NSTATE::RETAIN_SUBFLOW) == 0) {
2026 g.clear();
2027 }
2028
2029 return false;
2030}
2031
2032// Procedure: _invoke_condition_task
2033 inline void Executor::_invoke_condition_task(
2034 Worker& worker, Node* node, SmallVector<int>& conds
2035) {
2036 _observer_prologue(worker, node);
2037 TF_EXECUTOR_EXCEPTION_HANDLER(worker, node, {
2038 auto& work = std::get_if<Node::Condition>(&node->_handle)->work;
2039 conds = { work() };
2040 });
2041 _observer_epilogue(worker, node);
2042}
2043
2044// Procedure: _invoke_multi_condition_task
2045 inline void Executor::_invoke_multi_condition_task(
2046 Worker& worker, Node* node, SmallVector<int>& conds
2047) {
2048 _observer_prologue(worker, node);
2049 TF_EXECUTOR_EXCEPTION_HANDLER(worker, node, {
2050 conds = std::get_if<Node::MultiCondition>(&node->_handle)->work();
2051 });
2052 _observer_epilogue(worker, node);
2053}
2054
2055// Procedure: _invoke_module_task
2056 inline bool Executor::_invoke_module_task(Worker& w, Node* node) {
2057 return _invoke_module_task_impl(w, node, std::get_if<Node::Module>(&node->_handle)->graph);
2058}
2059
2060// Procedure: _invoke_adopted_module_task
2061 inline bool Executor::_invoke_adopted_module_task(Worker& w, Node* node) {
2062 return _invoke_module_task_impl(w, node, std::get_if<Node::AdoptedModule>(&node->_handle)->graph);
2063}
2064
2065// Procedure: _invoke_module_task_impl
2066 inline bool Executor::_invoke_module_task_impl(Worker& w, Node* node, Graph& graph) {
2067
2068 // No need to do anything for empty graph
2069 if(graph.empty()) {
2070 return false;
2071 }
2072
2073 // first entry - not spawned yet
2074 if((node->_nstate & NSTATE::PREEMPTED) == 0) {
2075 // signal the executor to preempt this node
2076 node->_nstate |= NSTATE::PREEMPTED;
2077 _schedule_graph(w, graph, node->_topology, node);
2078 return true;
2079 }
2080
2081 // second entry - already spawned
2082 node->_nstate &= ~NSTATE::PREEMPTED;
2083
2084 return false;
2085}
2086
2087
2088// Procedure: _invoke_async_task
2089 inline bool Executor::_invoke_async_task(Worker& worker, Node* node) {
2090 auto& work = std::get_if<Node::Async>(&node->_handle)->work;
2091 switch(work.index()) {
2092 // void()
2093 case 0:
2094 _observer_prologue(worker, node);
2095 TF_EXECUTOR_EXCEPTION_HANDLER(worker, node, {
2096 std::get_if<0>(&work)->operator()();
2097 });
2098 _observer_epilogue(worker, node);
2099 break;
2100
2101 // void(Runtime&)
2102 case 1:
2103 if(_invoke_runtime_task_impl(worker, node, *std::get_if<1>(&work))) {
2104 return true;
2105 }
2106 break;
2107
2108 // void(Runtime&, bool)
2109 case 2:
2110 if(_invoke_runtime_task_impl(worker, node, *std::get_if<2>(&work))) {
2111 return true;
2112 }
2113 break;
2114 }
2115
2116 return false;
2117}
2118
2119// Procedure: _invoke_dependent_async_task
2120 inline bool Executor::_invoke_dependent_async_task(Worker& worker, Node* node) {
2121 auto& work = std::get_if<Node::DependentAsync>(&node->_handle)->work;
2122 switch(work.index()) {
2123 // void()
2124 case 0:
2125 _observer_prologue(worker, node);
2126 TF_EXECUTOR_EXCEPTION_HANDLER(worker, node, {
2127 std::get_if<0>(&work)->operator()();
2128 });
2129 _observer_epilogue(worker, node);
2130 break;
2131
2132 // void(Runtime&) - silent async
2133 case 1:
2134 if(_invoke_runtime_task_impl(worker, node, *std::get_if<1>(&work))) {
2135 return true;
2136 }
2137 break;
2138
2139 // void(Runtime&, bool) - async
2140 case 2:
2141 if(_invoke_runtime_task_impl(worker, node, *std::get_if<2>(&work))) {
2142 return true;
2143 }
2144 break;
2145 }
2146 return false;
2147}
2148
2149// Function: run
2150 inline tf::Future<void> Executor::run(Taskflow& f) {
2151 return run_n(f, 1, [](){});
2152}
2153
2154// Function: run
2155 inline tf::Future<void> Executor::run(Taskflow&& f) {
2156 return run_n(std::move(f), 1, [](){});
2157}
2158
2159// Function: run
2160 template <typename C>
2161 tf::Future<void> Executor::run(Taskflow& f, C&& c) {
2162 return run_n(f, 1, std::forward<C>(c));
2163}
2164
2165// Function: run
2166 template <typename C>
2167 tf::Future<void> Executor::run(Taskflow&& f, C&& c) {
2168 return run_n(std::move(f), 1, std::forward<C>(c));
2169}
2170
2171// Function: run_n
2172 inline tf::Future<void> Executor::run_n(Taskflow& f, size_t repeat) {
2173 return run_n(f, repeat, [](){});
2174}
2175
2176// Function: run_n
2177 inline tf::Future<void> Executor::run_n(Taskflow&& f, size_t repeat) {
2178 return run_n(std::move(f), repeat, [](){});
2179}
2180
2181// Function: run_n
2182 template <typename C>
2183 tf::Future<void> Executor::run_n(Taskflow& f, size_t repeat, C&& c) {
2184 return run_until(
2185 f, [repeat]() mutable { return repeat-- == 0; }, std::forward<C>(c)
2186 );
2187}
2188
2189// Function: run_n
2190 template <typename C>
2191 tf::Future<void> Executor::run_n(Taskflow&& f, size_t repeat, C&& c) {
2192 return run_until(
2193 std::move(f), [repeat]() mutable { return repeat-- == 0; }, std::forward<C>(c)
2194 );
2195}
2196
2197// Function: run_until
2198 template<typename P>
2199 tf::Future<void> Executor::run_until(Taskflow& f, P&& pred) {
2200 return run_until(f, std::forward<P>(pred), [](){});
2201}
2202
2203// Function: run_until
2204 template<typename P>
2205 tf::Future<void> Executor::run_until(Taskflow&& f, P&& pred) {
2206 return run_until(std::move(f), std::forward<P>(pred), [](){});
2207}
2208
2209// Function: run_until
2210 template <typename P, typename C>
2211 tf::Future<void> Executor::run_until(Taskflow& f, P&& p, C&& c) {
2212
2213 // No need to create a real topology but returns an dummy future for invariant.
2214 if(f.empty() || p()) {
2215 c();
2216 std::promise<void> promise;
2217 promise.set_value();
2218 return tf::Future<void>(promise.get_future());
2219 }
2220
2221 _increment_topology();
2222
2223 // create a topology for this run
2224 auto t = std::make_shared<Topology>(f, std::forward<P>(p), std::forward<C>(c));
2225 //auto t = std::make_shared<DerivedTopology<P, C>>(f, std::forward<P>(p), std::forward<C>(c));
2226
2227 // need to create future before the topology got torn down quickly
2228 tf::Future<void> future(t->_promise.get_future(), t);
2229
2230 // modifying topology needs to be protected under the lock
2231 if(f._fetch_enqueue(t) == 0) {
2232 _set_up_topology(this_worker(), t.get());
2233 }
2234
2235 return future;
2236}
2237
2238
2239
2240// Function: corun_until
2241 template <typename P>
2242 void Executor::corun_until(P&& predicate) {
2243
2244 Worker* w = this_worker();
2245 if(w == nullptr) {
2246 TF_THROW("corun_until must be called by a worker of the executor");
2247 }
2248
2249 _corun_until(*w, std::forward<P>(predicate));
2250}
2251
2252/*
2253// Function: _corun_until
2254 template <typename P>
2255 void Executor::_corun_until(Worker& w, P&& stop_predicate) {
2256
2257 const size_t MAX_VICTIM = num_queues();
2258 const size_t MAX_STEALS = ((MAX_VICTIM + 1) << 1);
2259
2260 bool stop = false;
2261
2262 while(!stop && !(stop = stop_predicate())) {
2263
2264 // try local queue first — only one task at a time to avoid deep
2265 // recursive corun calls causing stack overflow
2266 if(auto t = w._wsq.pop(); t) {
2267 _invoke(w, t);
2268 continue;
2269 }
2270
2271 // local queue empty: steal from others until stop_predicate or stolen.
2272 // stop is set by the inner loop condition so when predicate becomes true
2273 // the outer loop exits immediately without calling stop_predicate again.
2274 size_t num_steals = 0;
2275 size_t vtm = w._sticky_victim;
2276
2277 while(!(stop = stop_predicate())) {
2278
2279 auto t = (vtm < _workers.size())
2280 ? _workers[vtm]._wsq.steal()
2281 : _buffers[vtm - _workers.size()].queue.steal();
2282
2283 if(t) {
2284 // STOLEN: invoke task then return to outer loop to re-check
2285 // local queue and stop_predicate
2286 _invoke(w, t);
2287 w._sticky_victim = vtm;
2288 break;
2289 }
2290
2291 // pick a new victim excluding self
2292 vtm = w._rdgen() % (MAX_VICTIM - 1);
2293 if(vtm >= w._id) vtm++;
2294
2295 if(++num_steals > MAX_STEALS) {
2296 // unlike _explore_task we cannot sleep here — the calling worker
2297 // is blocked inside a task and must keep making progress to avoid
2298 // deadlock. yield to let other threads run and make progress.
2299 std::this_thread::yield();
2300 }
2301 }
2302 }
2303}*/
2304
2305// Function: _corun_until
2306 template <typename P>
2307 void Executor::_corun_until(Worker& w, P&& stop_predicate) {
2308
2309 const size_t MAX_VICTIM = num_queues();
2310 const size_t MAX_STEALS = ((MAX_VICTIM + 1) << 1);
2311
2312 exploit:
2313
2314 while(!stop_predicate()) {
2315
2316 // here we don't do while-loop to drain out the local queue as it can
2317 // potentially enter a very deep recursive corun, cuasing stack overflow
2318 if(auto t = w._wsq.pop(); t) {
2319 _invoke(w, t);
2320 }
2321 else {
2322 size_t num_steals = 0;
2323 size_t vtm = w._sticky_victim;
2324
2325 explore:
2326
2327 t = (vtm < _workers.size())
2328 ? _workers[vtm]._wsq.steal()
2329 : _buffers[vtm-_workers.size()].queue.steal();
2330
2331 if(t) {
2332 _invoke(w, t);
2333 w._sticky_victim = vtm;
2334 goto exploit;
2335 }
2336 else if(!stop_predicate()) {
2337 if(++num_steals > MAX_STEALS) {
2338 std::this_thread::yield();
2339 }
2340 vtm = w._rdgen() % MAX_VICTIM;
2341 goto explore;
2342 }
2343 else {
2344 break;
2345 }
2346 }
2347 }
2348}
2349
2350// Function: corun
2351 template <typename T>
2352 void Executor::corun(T& target) {
2353
2354 Worker* w = this_worker();
2355 if(w == nullptr) {
2356 TF_THROW("corun must be called by a worker of the executor");
2357 }
2358
2359 NodeBase anchor;
2360 _corun_graph(*w, retrieve_graph(target), nullptr, &anchor);
2361}
2362
2363// Procedure: _corun_graph
2364 inline void Executor::_corun_graph(Worker& w, Graph& g, Topology* tpg, NodeBase* p) {
2365
2366 // empty graph
2367 if(g.empty()) {
2368 return;
2369 }
2370
2371 // anchor this parent as the blocking point
2372 {
2373 ExplicitAnchorGuard anchor(p);
2374 _schedule_graph(w, g, tpg, p);
2375 _corun_until(w, [p] () -> bool {
2376 return p->_join_counter.load(std::memory_order_acquire) == 0; }
2377 );
2378 }
2379
2380 // rethrow the exception to the caller
2381 p->_rethrow_exception();
2382}
2383
2384// Procedure: _increment_topology
2385 inline void Executor::_increment_topology() {
2386 _num_topologies.fetch_add(1, std::memory_order_relaxed);
2387}
2388
2389// Procedure: _decrement_topology
2390 inline void Executor::_decrement_topology() {
2391 if(_num_topologies.fetch_sub(1, std::memory_order_acq_rel) == 1) {
2392 _num_topologies.notify_all();
2393 }
2394}
2395
2396// Procedure: wait_for_all
2397 inline void Executor::wait_for_all() {
2398 size_t n = _num_topologies.load(std::memory_order_acquire);
2399 while(n != 0) {
2400 _num_topologies.wait(n, std::memory_order_acquire);
2401 n = _num_topologies.load(std::memory_order_acquire);
2402 }
2403}
2404
2405// Function: _schedule_graph
2406 inline void Executor::_schedule_graph(
2407 Worker& worker, Graph& graph, Topology* tpg, NodeBase* parent
2408) {
2409 size_t num_srcs = _set_up_graph(graph, tpg, parent);
2410 parent->_join_counter.fetch_add(num_srcs, std::memory_order_relaxed);
2411 _bulk_schedule(worker, graph.begin(), num_srcs);
2412}
2413
2414// Function: _set_up_topology
2415 inline void Executor::_set_up_topology(Worker* w, Topology* tpg) {
2416 // ---- under taskflow lock ----
2417 auto& g = tpg->_taskflow._graph;
2418 size_t num_srcs = _set_up_graph(g, tpg, tpg);
2419 tpg->_join_counter.store(num_srcs, std::memory_order_relaxed);
2420 w ? _bulk_schedule(*w, g.begin(), num_srcs) : _bulk_schedule(g.begin(), num_srcs);
2421}
2422
2423// Function: _set_up_graph
2424 inline size_t Executor::_set_up_graph(Graph& graph, Topology* tpg, NodeBase* parent) {
2425
2426 auto first = graph.begin();
2427 auto last = graph.end();
2428 auto send = first;
2429 for(; first != last; ++first) {
2430
2431 auto node = *first;
2432 node->_topology = tpg;
2433 node->_parent = parent;
2434 node->_nstate = NSTATE::NONE;
2435 node->_estate.store(ESTATE::NONE, std::memory_order_relaxed);
2436 node->_set_up_join_counter();
2437 node->_exception_ptr = nullptr;
2438
2439 // move source to the first partition
2440 // root, root, root, v1, v2, v3, v4, ...
2441 if(node->num_predecessors() == 0) {
2442 std::iter_swap(send++, first);
2443 }
2444 }
2445 return send - graph.begin();
2446}
2447
2448// Function: _tear_down_topology
2449 inline void Executor::_tear_down_topology(Worker& worker, Topology* tpg, Node*& cache) {
2450
2451 auto &f = tpg->_taskflow;
2452
2453 //assert(&tpg == &(f._topologies.front()));
2454
2455 // case 1: we still need to run the topology again
2456 //if(!tpg->_exception_ptr && !tpg->cancelled() && !tpg->predicate()) {
2457 if(!tpg->cancelled() && !tpg->_predicate()) {
2458 //assert(tpg->_join_counter == 0);
2459 //std::lock_guard<std::mutex> lock(f._mutex);
2460 _schedule_graph(worker, tpg->_taskflow._graph, tpg, tpg);
2461 }
2462 // case 2: the final run of this topology
2463 else {
2464
2465 // invoke the callback after each run
2466 tpg->_on_finish();
2467
2468 // there is another topologies to run
2469 if(std::unique_lock<std::mutex> lock(f._mutex); f._topologies.size()>1) {
2470
2471 auto fetched_tpg {std::move(f._topologies.front())};
2472 //assert(fetched_tpg.get() == tpg);
2473
2474 f._topologies.pop();
2475 tpg = f._topologies.front().get();
2476
2477 lock.unlock();
2478
2479 // Soon after we carry out the promise, the associate taskflow may got destroyed
2480 // from the user side, and we should never tough it again.
2481 fetched_tpg->_carry_out_promise();
2482
2483 // decrement the topology
2484 _decrement_topology();
2485
2486 _schedule_graph(worker, tpg->_taskflow._graph, tpg, tpg);
2487 }
2488 else {
2489 //assert(f._topologies.size() == 1);
2490
2491 auto fetched_tpg {std::move(f._topologies.front())};
2492 //assert(fetched_tpg.get() == tpg);
2493
2494 f._topologies.pop();
2495
2496 lock.unlock();
2497
2498 // Soon after we carry out the promise, the associate taskflow may got destroyed
2499 // from the user side, and we should never tough it again.
2500 fetched_tpg->_carry_out_promise();
2501
2502 _decrement_topology();
2503
2504 // remove the parent that owns the moved taskflow so the storage can be freed
2505 if(auto parent = fetched_tpg->_parent; parent) {
2506 //auto state = parent->_nstate;
2507 if(parent->_join_counter.fetch_sub(1, std::memory_order_acq_rel) == 1) {
2508 // this async is spawned from a preempted parent, so we need to resume it
2509 //if(state & NSTATE::PREEMPTED) {
2510 _update_cache(worker, cache, static_cast<Node*>(parent));
2511 //}
2512 }
2513 }
2514 }
2515 }
2516}
2517
2518// ############################################################################
2519// Forward Declaration: Subflow
2520// ############################################################################
2521
2522 inline void Subflow::join() {
2523
2524 if(!joinable()) {
2525 TF_THROW("subflow already joined");
2526 }
2527
2528 _executor._corun_graph(_worker, _graph, _node->_topology, _node);
2529
2530 // join here since corun graph may throw exception
2531 _node->_nstate |= NSTATE::JOINED_SUBFLOW;
2532}
2533
2534#endif
2535
2536
2537
2538
2539} // end of namespace tf -----------------------------------------------------
tf::AsyncTask
class to hold a dependent asynchronous task with shared ownership
Definition async_task.hpp:45
tf::Executor
class to create an executor
Definition executor.hpp:62
tf::Executor::dependent_async
auto dependent_async(F &&func, Tasks &&... tasks)
runs the given function asynchronously when the given predecessors finish
tf::Executor::silent_async
void silent_async(P &&params, F &&func)
similar to tf::Executor::async but does not return a future object
tf::Executor::silent_dependent_async
tf::AsyncTask silent_dependent_async(F &&func, Tasks &&... tasks)
runs the given function asynchronously when the given predecessors finish
tf::Executor::run_until
tf::Future< void > run_until(Taskflow &taskflow, P &&pred)
runs a taskflow multiple times until the predicate becomes true
tf::Executor::corun_until
void corun_until(P &&predicate)
keeps running the work-stealing loop until the predicate returns true
tf::Executor::remove_observer
void remove_observer(std::shared_ptr< Observer > observer)
removes an observer from the executor
tf::Executor::run
tf::Future< void > run(Taskflow &&taskflow)
runs a moved taskflow once
tf::Executor::run
tf::Future< void > run(Taskflow &taskflow)
runs a taskflow once
tf::Executor::num_waiters
size_t num_waiters() const noexcept
queries the number of workers that are in the waiting loop
tf::Executor::run
tf::Future< void > run(Taskflow &&taskflow, C &&callable)
runs a moved taskflow once and invoke a callback upon completion
tf::Executor::~Executor
~Executor()
destructs the executor
tf::Executor::this_worker_id
int this_worker_id() const
queries the id of the caller thread within this executor
tf::Executor::num_queues
size_t num_queues() const noexcept
queries the number of work-stealing queues used by the executor
tf::Executor::run_n
tf::Future< void > run_n(Taskflow &taskflow, size_t N)
runs a taskflow for N times
tf::Executor::num_topologies
size_t num_topologies() const
queries the number of running topologies at the time of this call
tf::Executor::Executor
Executor(size_t N=std::thread::hardware_concurrency(), std::shared_ptr< WorkerInterface > wif=nullptr)
constructs the executor with N worker threads
tf::Executor::task_group
TaskGroup task_group()
creates a task group that executes a collection of asynchronous tasks
Definition task_group.hpp:863
tf::Executor::corun
void corun(T &target)
runs a target graph and waits until it completes using an internal worker of this executor
tf::Executor::num_workers
size_t num_workers() const noexcept
queries the number of worker threads
tf::Executor::run_until
tf::Future< void > run_until(Taskflow &&taskflow, P &&pred)
runs a moved taskflow and keeps running it until the predicate becomes true
tf::Executor::wait_for_all
void wait_for_all()
waits for all tasks to complete
tf::Executor::run_n
tf::Future< void > run_n(Taskflow &taskflow, size_t N, C &&callable)
runs a taskflow for N times and then invokes a callback
tf::Executor::run
tf::Future< void > run(Taskflow &taskflow, C &&callable)
runs a taskflow once and invoke a callback upon completion
tf::Executor::run_n
tf::Future< void > run_n(Taskflow &&taskflow, size_t N)
runs a moved taskflow for N times
tf::Executor::run_n
tf::Future< void > run_n(Taskflow &&taskflow, size_t N, C &&callable)
runs a moved taskflow for N times and then invokes a callback
tf::Executor::run_until
tf::Future< void > run_until(Taskflow &taskflow, P &&pred, C &&callable)
runs a taskflow multiple times until the predicate becomes true and then invokes the callback
tf::Executor::this_worker
Worker * this_worker()
queries pointer to the calling worker if it belongs to this executor, otherwise returns nullptr
tf::Executor::async
auto async(P &&params, F &&func)
creates a parameterized asynchronous task to run the given function
tf::Executor::make_observer
std::shared_ptr< Observer > make_observer(ArgsT &&... args)
constructs an observer to inspect the activities of worker threads
tf::Executor::num_observers
size_t num_observers() const noexcept
queries the number of observers
tf::FlowBuilder
class to build a task dependency graph
Definition flow_builder.hpp:114
tf::Future
class to access the result of an execution
Definition taskflow.hpp:630
tf::NonblockingNotifier
class to create a non-blocking notifier
Definition nonblocking_notifier.hpp:84
tf::NonblockingNotifier::num_waiters
size_t num_waiters() const
returns the number of committed waiters
Definition nonblocking_notifier.hpp:176
tf::NonblockingNotifier::cancel_wait
void cancel_wait(size_t wid)
cancels a previously prepared wait operation
Definition nonblocking_notifier.hpp:356
tf::NonblockingNotifier::prepare_wait
void prepare_wait(size_t wid)
prepares the calling thread to enter the waiting set
Definition nonblocking_notifier.hpp:212
tf::NonblockingNotifier::commit_wait
void commit_wait(size_t wid)
commits a previously prepared wait operation
Definition nonblocking_notifier.hpp:235
tf::NonblockingNotifier::notify_one
void notify_one()
notifies one waiter from the waiting set
Definition nonblocking_notifier.hpp:389
tf::NonblockingNotifier::notify_n
void notify_n(size_t N)
notifies up to N waiters from the waiting set
Definition nonblocking_notifier.hpp:473
tf::NonblockingNotifier::notify_all
void notify_all()
notifies all waiter from the waiting set
Definition nonblocking_notifier.hpp:437
tf::Runtime
class to create a runtime task
Definition runtime.hpp:47
tf::Subflow
class to construct a subflow graph from the execution of a dynamic task
Definition flow_builder.hpp:1913
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:2012
tf::TaskGroup
class to create a task group from a task
Definition task_group.hpp:61
tf::Taskflow
class to create a taskflow object
Definition taskflow.hpp:64
tf::UnboundedWSQ
class to create a lock-free unbounded work-stealing queue
Definition wsq.hpp:105
tf::Worker
class to create a worker in an executor
Definition worker.hpp:55
tf::AsyncTaskHandleLike
concept to check if a type is a tf::AsyncTask
Definition async_task.hpp:314
tf::TaskParamsLike
concept that determines if a type is a task parameter type
Definition graph.hpp:202
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::has_env
bool has_env(const std::string &str)
checks whether an environment variable is defined
Definition os.hpp:310