| 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
11
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#ifdef TF_ENABLE_TASK_PRIORITY
1683 if(cache) {
1684 if(node->priority() < cache->priority()) {
1685 std::swap(cache, node);
1686 }
1687 _schedule(worker, node);
1688 } else {
1689 cache = node;
1690 }
1691#else
1692 if(cache) {
1693 _schedule(worker, cache);
1694 }
1695 cache = node;
1696#endif
1697}
1698
1699// Function: _bulk_update_cache
1700 template <size_t N>
1701TF_FORCE_INLINE void Executor::_bulk_update_cache(
1702 Worker& worker, Node*& cache, Node* node, std::array<Node*, N>& array, size_t& n
1703) {
1704 // experimental results show no benefit of using bulk_update_cache
1705 if(cache) {
1706 array[n++] = cache;
1707 if(n == N) {
1708 _bulk_schedule(worker, array, n);
1709 n = 0;
1710 }
1711 }
1712 cache = node;
1713}
1714
1715// Procedure: _invoke
1716 inline void Executor::_invoke(Worker& worker, Node* node) {
1717
1718 #define TF_INVOKE_CONTINUATION() \
1719 if (cache) { \
1720 node = cache; \
1721 goto begin_invoke; \
1722 }
1723
1724 begin_invoke:
1725
1726 Node* cache {nullptr};
1727
1728 // if this is the second invoke due to preemption, directly jump to invoke task
1729 if(node->_nstate & NSTATE::PREEMPTED) {
1730 goto invoke_task;
1731 }
1732
1733 // If the work has been cancelled, there is no need to continue.
1734 // Here, we do tear_down_invoke since async tasks may also get cancelled where
1735 // we need to recycle the node.
1736 if(node->_is_parent_cancelled()) {
1737 _tear_down_invoke(worker, node, cache);
1738 TF_INVOKE_CONTINUATION();
1739 return;
1740 }
1741
1742 // if acquiring semaphore(s) exists, acquire them first
1743 if(node->_semaphores && !node->_semaphores->to_acquire.empty()) {
1744 SmallVector<Node*> waiters;
1745 if(!node->_acquire_all(waiters)) {
1746 _bulk_schedule(worker, waiters.begin(), waiters.size());
1747 return;
1748 }
1749 }
1750
1751 invoke_task:
1752
1753 SmallVector<int> conds;
1754
1755 // switch is faster than nested if-else due to jump table
1756 switch(node->_handle.index()) {
1757 // static task
1758 case Node::STATIC:{
1759 _invoke_static_task(worker, node);
1760 }
1761 break;
1762
1763 // runtime task
1764 case Node::RUNTIME:{
1765 if(_invoke_runtime_task(worker, node)) {
1766 return;
1767 }
1768 }
1769 break;
1770
1771 // non-preemptive runtime task
1772 case Node::NONPREEMPTIVE_RUNTIME:{
1773 _invoke_nonpreemptive_runtime_task(worker, node);
1774 }
1775 break;
1776
1777 // subflow task
1778 case Node::SUBFLOW: {
1779 if(_invoke_subflow_task(worker, node)) {
1780 return;
1781 }
1782 }
1783 break;
1784
1785 // condition task
1786 case Node::CONDITION: {
1787 _invoke_condition_task(worker, node, conds);
1788 }
1789 break;
1790
1791 // multi-condition task
1792 case Node::MULTI_CONDITION: {
1793 _invoke_multi_condition_task(worker, node, conds);
1794 }
1795 break;
1796
1797 // module task
1798 case Node::MODULE: {
1799 if(_invoke_module_task(worker, node)) {
1800 return;
1801 }
1802 }
1803 break;
1804
1805 // adopted module task
1806 case Node::ADOPTED_MODULE: {
1807 if(_invoke_adopted_module_task(worker, node)) {
1808 return;
1809 }
1810 }
1811 break;
1812
1813 // async task
1814 case Node::ASYNC: {
1815 if(_invoke_async_task(worker, node)) {
1816 return;
1817 }
1818 _tear_down_async(worker, node, cache);
1819 TF_INVOKE_CONTINUATION();
1820 return;
1821 }
1822 break;
1823
1824 // dependent async task
1825 case Node::DEPENDENT_ASYNC: {
1826 if(_invoke_dependent_async_task(worker, node)) {
1827 return;
1828 }
1829 _tear_down_dependent_async(worker, node, cache);
1830 TF_INVOKE_CONTINUATION();
1831 return;
1832 }
1833 break;
1834
1835 // monostate (placeholder)
1836 default:
1837 break;
1838 }
1839
1840 // if releasing semaphores exist, release them
1841 if(node->_semaphores && !node->_semaphores->to_release.empty()) {
1842 SmallVector<Node*> waiters;
1843 node->_release_all(waiters);
1844 _bulk_schedule(worker, waiters.begin(), waiters.size());
1845 }
1846
1847 // Reset the join counter with strong dependencies to support cycles.
1848 // + We must do this before scheduling the successors to avoid race
1849 // condition on _predecessors.
1850 // + We must use fetch_add instead of direct assigning
1851 // because the user-level call on "invoke" may explicitly schedule
1852 // this task again (e.g., pipeline) which can access the join_counter.
1853 node->_join_counter.fetch_add(
1854 node->_nstate & NSTATE::STRONG_DEPENDENCIES_MASK, std::memory_order_relaxed
1855 );
1856
1857 // Invoke the task based on the corresponding type
1858 switch(node->_handle.index()) {
1859
1860 // condition and multi-condition tasks
1861 case Node::CONDITION:
1862 case Node::MULTI_CONDITION: {
1863 for(auto cond : conds) {
1864 if(cond >= 0 && static_cast<size_t>(cond) < node->_num_successors) {
1865 auto s = node->_edges[cond];
1866 // zeroing the join counter for invariant
1867 s->_join_counter.store(0, std::memory_order_relaxed);
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 // non-condition task
1876 default: {
1877 for(size_t i=0; i<node->_num_successors; ++i) {
1878 if(auto s = node->_edges[i]; s->_join_counter.fetch_sub(1, std::memory_order_acq_rel) == 1) {
1879 node->_parent->_join_counter.fetch_add(1, std::memory_order_relaxed);
1880 _update_cache(worker, cache, s);
1881 }
1882 }
1883 }
1884 break;
1885 }
1886
1887 // clean up the node after execution
1888 _tear_down_nonasync(worker, node, cache);
1889 TF_INVOKE_CONTINUATION();
1890}
1891
1892// Procedure: _tear_down_nonasync
1893 inline void Executor::_tear_down_nonasync(Worker& worker, Node* node, Node*& cache) {
1894
1895 // we must check parent first before subtracting the join counter,
1896 // or it can introduce data race
1897 if(auto parent = node->_parent; parent == node->_topology) {
1898 if(parent->_join_counter.fetch_sub(1, std::memory_order_acq_rel) == 1) {
1899 _tear_down_topology(worker, node->_topology, cache);
1900 }
1901 }
1902 else {
1903 // needs to fetch every data before join counter becomes zero at which
1904 // the node may be deleted
1905 auto state = parent->_nstate;
1906 if(parent->_join_counter.fetch_sub(1, std::memory_order_acq_rel) == 1) {
1907 // this task is spawned from a preempted parent, so we need to resume it
1908 if(state & NSTATE::PREEMPTED) {
1909 _update_cache(worker, cache, static_cast<Node*>(parent));
1910 }
1911 }
1912 }
1913}
1914
1915// Procedure: _tear_down_invoke
1916 inline void Executor::_tear_down_invoke(Worker& worker, Node* node, Node*& cache) {
1917 switch(node->_handle.index()) {
1918 case Node::ASYNC:
1919 _tear_down_async(worker, node, cache);
1920 break;
1921
1922 case Node::DEPENDENT_ASYNC:
1923 _tear_down_dependent_async(worker, node, cache);
1924 break;
1925
1926 default:
1927 _tear_down_nonasync(worker, node, cache);
1928 break;
1929 }
1930}
1931
1932// Procedure: _observer_prologue
1933 inline void Executor::_observer_prologue(Worker& worker, Node* node) {
1934 for(auto& observer : _observers) {
1935 observer->on_entry(WorkerView(worker), TaskView(*node));
1936 }
1937}
1938
1939// Procedure: _observer_epilogue
1940 inline void Executor::_observer_epilogue(Worker& worker, Node* node) {
1941 for(auto& observer : _observers) {
1942 observer->on_exit(WorkerView(worker), TaskView(*node));
1943 }
1944}
1945
1946// Procedure: _process_exception
1947 inline void Executor::_process_exception(Worker&, Node* node) {
1948
1949 // Finds the anchor and mark the entire path with exception,
1950 // so recursive tasks can be cancelled properly.
1951 // Since exception can come from asynchronous task (with runtime), the node itself can be anchored.
1952 NodeBase* ea = node; // explicit anchor
1953 NodeBase* ia = nullptr; // implicit anchor
1954
1955 while(ea && (ea->_estate.load(std::memory_order_relaxed) & ESTATE::EXPLICITLY_ANCHORED) == 0) {
1956 ea->_estate.fetch_or(ESTATE::EXCEPTION, std::memory_order_relaxed);
1957 // we only want the inner-most implicit anchor
1958 if(ia == nullptr && (ea->_nstate & NSTATE::IMPLICITLY_ANCHORED)) {
1959 ia = ea;
1960 }
1961 ea = ea->_parent;
1962 }
1963
1964 // flag used to ensure execution is caught in a thread-safe manner
1965 constexpr static auto flag = ESTATE::EXCEPTION | ESTATE::CAUGHT;
1966
1967 // The exception occurs under a blocking call (e.g., corun, join).
1968 if(ea) {
1969 // multiple tasks may throw, and we only take the first thrown exception
1970 if((ea->_estate.fetch_or(flag, std::memory_order_relaxed) & ESTATE::CAUGHT) == 0) {
1971 ea->_exception_ptr = std::current_exception();
1972 return;
1973 }
1974 }
1975 // Implicit anchor has the lowest priority
1976 else if(ia){
1977 if((ia->_estate.fetch_or(flag, std::memory_order_relaxed) & ESTATE::CAUGHT) == 0) {
1978 ia->_exception_ptr = std::current_exception();
1979 return;
1980 }
1981 }
1982
1983 // For now, we simply store the exception in this node; this can happen in an
1984 // execution that does not have any external control to capture the exception,
1985 // such as silent async task without any parent.
1986 node->_exception_ptr = std::current_exception();
1987}
1988
1989// Procedure: _invoke_static_task
1990 inline void Executor::_invoke_static_task(Worker& worker, Node* node) {
1991 _observer_prologue(worker, node);
1992 TF_EXECUTOR_EXCEPTION_HANDLER(worker, node, {
1993 std::get_if<Node::Static>(&node->_handle)->work();
1994 });
1995 _observer_epilogue(worker, node);
1996}
1997
1998// Procedure: _invoke_subflow_task
1999 inline bool Executor::_invoke_subflow_task(Worker& worker, Node* node) {
2000
2001 auto& h = *std::get_if<Node::Subflow>(&node->_handle);
2002 auto& g = h.subgraph;
2003
2004 if((node->_nstate & NSTATE::PREEMPTED) == 0) {
2005
2006 // set up the subflow
2007 Subflow sf(*this, worker, node, g);
2008
2009 // invoke the subflow callable
2010 _observer_prologue(worker, node);
2011 TF_EXECUTOR_EXCEPTION_HANDLER(worker, node, {
2012 h.work(sf);
2013 });
2014 _observer_epilogue(worker, node);
2015
2016 // spawn the subflow if it is joinable and its graph is non-empty
2017 // implicit join is faster than Subflow::join as it does not involve corun
2018 if(sf.joinable() && !g.empty()) {
2019
2020 // signal the executor to preempt this node
2021 node->_nstate |= NSTATE::PREEMPTED;
2022
2023 // set up and schedule the graph
2024 _schedule_graph(worker, g, node->_topology, node);
2025 return true;
2026 }
2027 }
2028 else {
2029 node->_nstate &= ~NSTATE::PREEMPTED;
2030 }
2031
2032 // The subflow has finished or joined.
2033 // By default, we clear the subflow storage as applications can perform recursive
2034 // subflow tasking which accumulates a huge amount of memory overhead, hampering
2035 // the performance.
2036 if((node->_nstate & NSTATE::RETAIN_SUBFLOW) == 0) {
2037 g.clear();
2038 }
2039
2040 return false;
2041}
2042
2043// Procedure: _invoke_condition_task
2044 inline void Executor::_invoke_condition_task(
2045 Worker& worker, Node* node, SmallVector<int>& conds
2046) {
2047 _observer_prologue(worker, node);
2048 TF_EXECUTOR_EXCEPTION_HANDLER(worker, node, {
2049 auto& work = std::get_if<Node::Condition>(&node->_handle)->work;
2050 conds = { work() };
2051 });
2052 _observer_epilogue(worker, node);
2053}
2054
2055// Procedure: _invoke_multi_condition_task
2056 inline void Executor::_invoke_multi_condition_task(
2057 Worker& worker, Node* node, SmallVector<int>& conds
2058) {
2059 _observer_prologue(worker, node);
2060 TF_EXECUTOR_EXCEPTION_HANDLER(worker, node, {
2061 conds = std::get_if<Node::MultiCondition>(&node->_handle)->work();
2062 });
2063 _observer_epilogue(worker, node);
2064}
2065
2066// Procedure: _invoke_module_task
2067 inline bool Executor::_invoke_module_task(Worker& w, Node* node) {
2068 return _invoke_module_task_impl(w, node, std::get_if<Node::Module>(&node->_handle)->graph);
2069}
2070
2071// Procedure: _invoke_adopted_module_task
2072 inline bool Executor::_invoke_adopted_module_task(Worker& w, Node* node) {
2073 return _invoke_module_task_impl(w, node, std::get_if<Node::AdoptedModule>(&node->_handle)->graph);
2074}
2075
2076// Procedure: _invoke_module_task_impl
2077 inline bool Executor::_invoke_module_task_impl(Worker& w, Node* node, Graph& graph) {
2078
2079 // No need to do anything for empty graph
2080 if(graph.empty()) {
2081 return false;
2082 }
2083
2084 // first entry - not spawned yet
2085 if((node->_nstate & NSTATE::PREEMPTED) == 0) {
2086 // signal the executor to preempt this node
2087 node->_nstate |= NSTATE::PREEMPTED;
2088 _schedule_graph(w, graph, node->_topology, node);
2089 return true;
2090 }
2091
2092 // second entry - already spawned
2093 node->_nstate &= ~NSTATE::PREEMPTED;
2094
2095 return false;
2096}
2097
2098
2099// Procedure: _invoke_async_task
2100 inline bool Executor::_invoke_async_task(Worker& worker, Node* node) {
2101 auto& work = std::get_if<Node::Async>(&node->_handle)->work;
2102 switch(work.index()) {
2103 // void()
2104 case 0:
2105 _observer_prologue(worker, node);
2106 TF_EXECUTOR_EXCEPTION_HANDLER(worker, node, {
2107 std::get_if<0>(&work)->operator()();
2108 });
2109 _observer_epilogue(worker, node);
2110 break;
2111
2112 // void(Runtime&)
2113 case 1:
2114 if(_invoke_runtime_task_impl(worker, node, *std::get_if<1>(&work))) {
2115 return true;
2116 }
2117 break;
2118
2119 // void(Runtime&, bool)
2120 case 2:
2121 if(_invoke_runtime_task_impl(worker, node, *std::get_if<2>(&work))) {
2122 return true;
2123 }
2124 break;
2125 }
2126
2127 return false;
2128}
2129
2130// Procedure: _invoke_dependent_async_task
2131 inline bool Executor::_invoke_dependent_async_task(Worker& worker, Node* node) {
2132 auto& work = std::get_if<Node::DependentAsync>(&node->_handle)->work;
2133 switch(work.index()) {
2134 // void()
2135 case 0:
2136 _observer_prologue(worker, node);
2137 TF_EXECUTOR_EXCEPTION_HANDLER(worker, node, {
2138 std::get_if<0>(&work)->operator()();
2139 });
2140 _observer_epilogue(worker, node);
2141 break;
2142
2143 // void(Runtime&) - silent async
2144 case 1:
2145 if(_invoke_runtime_task_impl(worker, node, *std::get_if<1>(&work))) {
2146 return true;
2147 }
2148 break;
2149
2150 // void(Runtime&, bool) - async
2151 case 2:
2152 if(_invoke_runtime_task_impl(worker, node, *std::get_if<2>(&work))) {
2153 return true;
2154 }
2155 break;
2156 }
2157 return false;
2158}
2159
2160// Function: run
2161 inline tf::Future<void> Executor::run(Taskflow& f) {
2162 return run_n(f, 1, [](){});
2163}
2164
2165// Function: run
2166 inline tf::Future<void> Executor::run(Taskflow&& f) {
2167 return run_n(std::move(f), 1, [](){});
2168}
2169
2170// Function: run
2171 template <typename C>
2172tf::Future<void> Executor::run(Taskflow& f, C&& c) {
2173 return run_n(f, 1, std::forward<C>(c));
2174}
2175
2176// Function: run
2177 template <typename C>
2178tf::Future<void> Executor::run(Taskflow&& f, C&& c) {
2179 return run_n(std::move(f), 1, std::forward<C>(c));
2180}
2181
2182// Function: run_n
2183 inline tf::Future<void> Executor::run_n(Taskflow& f, size_t repeat) {
2184 return run_n(f, repeat, [](){});
2185}
2186
2187// Function: run_n
2188 inline tf::Future<void> Executor::run_n(Taskflow&& f, size_t repeat) {
2189 return run_n(std::move(f), repeat, [](){});
2190}
2191
2192// Function: run_n
2193 template <typename C>
2194tf::Future<void> Executor::run_n(Taskflow& f, size_t repeat, C&& c) {
2195 return run_until(
2196 f, [repeat]() mutable { return repeat-- == 0; }, std::forward<C>(c)
2197 );
2198}
2199
2200// Function: run_n
2201 template <typename C>
2202tf::Future<void> Executor::run_n(Taskflow&& f, size_t repeat, C&& c) {
2203 return run_until(
2204 std::move(f), [repeat]() mutable { return repeat-- == 0; }, std::forward<C>(c)
2205 );
2206}
2207
2208// Function: run_until
2209 template<typename P>
2210tf::Future<void> Executor::run_until(Taskflow& f, P&& pred) {
2211 return run_until(f, std::forward<P>(pred), [](){});
2212}
2213
2214// Function: run_until
2215 template<typename P>
2216tf::Future<void> Executor::run_until(Taskflow&& f, P&& pred) {
2217 return run_until(std::move(f), std::forward<P>(pred), [](){});
2218}
2219
2220// Function: run_until
2221 template <typename P, typename C>
2222tf::Future<void> Executor::run_until(Taskflow& f, P&& p, C&& c) {
2223
2224 // No need to create a real topology but returns an dummy future for invariant.
2225 if(f.empty() || p()) {
2226 c();
2227 std::promise<void> promise;
2228 promise.set_value();
2229 return tf::Future<void>(promise.get_future());
2230 }
2231
2232 _increment_topology();
2233
2234 // create a topology for this run
2235 auto t = std::make_shared<Topology>(f, std::forward<P>(p), std::forward<C>(c));
2236 //auto t = std::make_shared<DerivedTopology<P, C>>(f, std::forward<P>(p), std::forward<C>(c));
2237
2238 // need to create future before the topology got torn down quickly
2239 tf::Future<void> future(t->_promise.get_future(), t);
2240
2241 // modifying topology needs to be protected under the lock
2242 if(f._fetch_enqueue(t) == 0) {
2243 _set_up_topology(this_worker(), t.get());
2244 }
2245
2246 return future;
2247}
2248
2249
2250
2251// Function: corun_until
2252 template <typename P>
2253 void Executor::corun_until(P&& predicate) {
2254
2255 Worker* w = this_worker();
2256 if(w == nullptr) {
2257 TF_THROW("corun_until must be called by a worker of the executor");
2258 }
2259
2260 _corun_until(*w, std::forward<P>(predicate));
2261}
2262
2263/*
2264// Function: _corun_until
2265 template <typename P>
2266 void Executor::_corun_until(Worker& w, P&& stop_predicate) {
2267
2268 const size_t MAX_VICTIM = num_queues();
2269 const size_t MAX_STEALS = ((MAX_VICTIM + 1) << 1);
2270
2271 bool stop = false;
2272
2273 while(!stop && !(stop = stop_predicate())) {
2274
2275 // try local queue first — only one task at a time to avoid deep
2276 // recursive corun calls causing stack overflow
2277 if(auto t = w._wsq.pop(); t) {
2278 _invoke(w, t);
2279 continue;
2280 }
2281
2282 // local queue empty: steal from others until stop_predicate or stolen.
2283 // stop is set by the inner loop condition so when predicate becomes true
2284 // the outer loop exits immediately without calling stop_predicate again.
2285 size_t num_steals = 0;
2286 size_t vtm = w._sticky_victim;
2287
2288 while(!(stop = stop_predicate())) {
2289
2290 auto t = (vtm < _workers.size())
2291 ? _workers[vtm]._wsq.steal()
2292 : _buffers[vtm - _workers.size()].queue.steal();
2293
2294 if(t) {
2295 // STOLEN: invoke task then return to outer loop to re-check
2296 // local queue and stop_predicate
2297 _invoke(w, t);
2298 w._sticky_victim = vtm;
2299 break;
2300 }
2301
2302 // pick a new victim excluding self
2303 vtm = w._rdgen() % (MAX_VICTIM - 1);
2304 if(vtm >= w._id) vtm++;
2305
2306 if(++num_steals > MAX_STEALS) {
2307 // unlike _explore_task we cannot sleep here — the calling worker
2308 // is blocked inside a task and must keep making progress to avoid
2309 // deadlock. yield to let other threads run and make progress.
2310 std::this_thread::yield();
2311 }
2312 }
2313 }
2314}*/
2315
2316// Function: _corun_until
2317 template <typename P>
2318 void Executor::_corun_until(Worker& w, P&& stop_predicate) {
2319
2320 const size_t MAX_VICTIM = num_queues();
2321 const size_t MAX_STEALS = ((MAX_VICTIM + 1) << 1);
2322
2323 exploit:
2324
2325 while(!stop_predicate()) {
2326
2327 // here we don't do while-loop to drain out the local queue as it can
2328 // potentially enter a very deep recursive corun, cuasing stack overflow
2329 if(auto t = w._wsq.pop(); t) {
2330 _invoke(w, t);
2331 }
2332 else {
2333 size_t num_steals = 0;
2334 size_t vtm = w._sticky_victim;
2335
2336 explore:
2337
2338 t = (vtm < _workers.size())
2339 ? _workers[vtm]._wsq.steal()
2340 : _buffers[vtm-_workers.size()].queue.steal();
2341
2342 if(t) {
2343 _invoke(w, t);
2344 w._sticky_victim = vtm;
2345 goto exploit;
2346 }
2347 else if(!stop_predicate()) {
2348 if(++num_steals > MAX_STEALS) {
2349 std::this_thread::yield();
2350 }
2351 vtm = w._rdgen() % MAX_VICTIM;
2352 goto explore;
2353 }
2354 else {
2355 break;
2356 }
2357 }
2358 }
2359}
2360
2361// Function: corun
2362 template <typename T>
2363 void Executor::corun(T& target) {
2364
2365 Worker* w = this_worker();
2366 if(w == nullptr) {
2367 TF_THROW("corun must be called by a worker of the executor");
2368 }
2369
2370 NodeBase anchor;
2371 _corun_graph(*w, retrieve_graph(target), nullptr, &anchor);
2372}
2373
2374// Procedure: _corun_graph
2375 inline void Executor::_corun_graph(Worker& w, Graph& g, Topology* tpg, NodeBase* p) {
2376
2377 // empty graph
2378 if(g.empty()) {
2379 return;
2380 }
2381
2382 // anchor this parent as the blocking point
2383 {
2384 ExplicitAnchorGuard anchor(p);
2385 _schedule_graph(w, g, tpg, p);
2386 _corun_until(w, [p] () -> bool {
2387 return p->_join_counter.load(std::memory_order_acquire) == 0; }
2388 );
2389 }
2390
2391 // rethrow the exception to the caller
2392 p->_rethrow_exception();
2393}
2394
2395// Procedure: _increment_topology
2396 inline void Executor::_increment_topology() {
2397 _num_topologies.fetch_add(1, std::memory_order_relaxed);
2398}
2399
2400// Procedure: _decrement_topology
2401 inline void Executor::_decrement_topology() {
2402 if(_num_topologies.fetch_sub(1, std::memory_order_acq_rel) == 1) {
2403 _num_topologies.notify_all();
2404 }
2405}
2406
2407// Procedure: wait_for_all
2408 inline void Executor::wait_for_all() {
2409 size_t n = _num_topologies.load(std::memory_order_acquire);
2410 while(n != 0) {
2411 _num_topologies.wait(n, std::memory_order_acquire);
2412 n = _num_topologies.load(std::memory_order_acquire);
2413 }
2414}
2415
2416// Function: _schedule_graph
2417 inline void Executor::_schedule_graph(
2418 Worker& worker, Graph& graph, Topology* tpg, NodeBase* parent
2419) {
2420 size_t num_srcs = _set_up_graph(graph, tpg, parent);
2421 parent->_join_counter.fetch_add(num_srcs, std::memory_order_relaxed);
2422 _bulk_schedule(worker, graph.begin(), num_srcs);
2423}
2424
2425// Function: _set_up_topology
2426 inline void Executor::_set_up_topology(Worker* w, Topology* tpg) {
2427 // ---- under taskflow lock ----
2428 auto& g = tpg->_taskflow._graph;
2429 size_t num_srcs = _set_up_graph(g, tpg, tpg);
2430 tpg->_join_counter.store(num_srcs, std::memory_order_relaxed);
2431 w ? _bulk_schedule(*w, g.begin(), num_srcs) : _bulk_schedule(g.begin(), num_srcs);
2432}
2433
2434// Function: _set_up_graph
2435 inline size_t Executor::_set_up_graph(Graph& graph, Topology* tpg, NodeBase* parent) {
2436
2437 auto first = graph.begin();
2438 auto last = graph.end();
2439 auto send = first;
2440 for(; first != last; ++first) {
2441
2442 auto node = *first;
2443 node->_topology = tpg;
2444 node->_parent = parent;
2445#ifdef TF_ENABLE_TASK_PRIORITY
2446 node->_nstate &= NSTATE::PRIORITY_MASK;
2447#else
2448 node->_nstate &= NSTATE::NONE;
2449#endif
2450 node->_estate.store(ESTATE::NONE, std::memory_order_relaxed);
2451 node->_set_up_join_counter();
2452 node->_exception_ptr = nullptr;
2453
2454 // move source to the first partition
2455 // root, root, root, v1, v2, v3, v4, ...
2456 if(node->num_predecessors() == 0) {
2457 std::iter_swap(send++, first);
2458 }
2459 }
2460 return send - graph.begin();
2461}
2462
2463// Function: _tear_down_topology
2464 inline void Executor::_tear_down_topology(Worker& worker, Topology* tpg, Node*& cache) {
2465
2466 auto &f = tpg->_taskflow;
2467
2468 //assert(&tpg == &(f._topologies.front()));
2469
2470 // case 1: we still need to run the topology again
2471 //if(!tpg->_exception_ptr && !tpg->cancelled() && !tpg->predicate()) {
2472 if(!tpg->cancelled() && !tpg->_predicate()) {
2473 //assert(tpg->_join_counter == 0);
2474 //std::lock_guard<std::mutex> lock(f._mutex);
2475 _schedule_graph(worker, tpg->_taskflow._graph, tpg, tpg);
2476 }
2477 // case 2: the final run of this topology
2478 else {
2479
2480 // invoke the callback after each run
2481 tpg->_on_finish();
2482
2483 // there is another topologies to run
2484 if(std::unique_lock<std::mutex> lock(f._mutex); f._topologies.size()>1) {
2485
2486 auto fetched_tpg {std::move(f._topologies.front())};
2487 //assert(fetched_tpg.get() == tpg);
2488
2489 f._topologies.pop();
2490 tpg = f._topologies.front().get();
2491
2492 lock.unlock();
2493
2494 // Soon after we carry out the promise, the associate taskflow may got destroyed
2495 // from the user side, and we should never tough it again.
2496 fetched_tpg->_carry_out_promise();
2497
2498 // decrement the topology
2499 _decrement_topology();
2500
2501 _schedule_graph(worker, tpg->_taskflow._graph, tpg, tpg);
2502 }
2503 else {
2504 //assert(f._topologies.size() == 1);
2505
2506 auto fetched_tpg {std::move(f._topologies.front())};
2507 //assert(fetched_tpg.get() == tpg);
2508
2509 f._topologies.pop();
2510
2511 lock.unlock();
2512
2513 // Soon after we carry out the promise, the associate taskflow may got destroyed
2514 // from the user side, and we should never tough it again.
2515 fetched_tpg->_carry_out_promise();
2516
2517 _decrement_topology();
2518
2519 // remove the parent that owns the moved taskflow so the storage can be freed
2520 if(auto parent = fetched_tpg->_parent; parent) {
2521 //auto state = parent->_nstate;
2522 if(parent->_join_counter.fetch_sub(1, std::memory_order_acq_rel) == 1) {
2523 // this async is spawned from a preempted parent, so we need to resume it
2524 //if(state & NSTATE::PREEMPTED) {
2525 _update_cache(worker, cache, static_cast<Node*>(parent));
2526 //}
2527 }
2528 }
2529 }
2530 }
2531}
2532
2533// ############################################################################
2534// Forward Declaration: Subflow
2535// ############################################################################
2536
2537 inline void Subflow::join() {
2538
2539 if(!joinable()) {
2540 TF_THROW("subflow already joined");
2541 }
2542
2543 _executor._corun_graph(_worker, _graph, _node->_topology, _node);
2544
2545 // join here since corun graph may throw exception
2546 _node->_nstate |= NSTATE::JOINED_SUBFLOW;
2547}
2548
2549#endif
2550
2551
2552
2553
2554} // end of namespace tf -----------------------------------------------------
tf::AsyncTask
class to hold a dependent asynchronous task with shared ownership
Definition async_task.hpp:45
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::Future
class to access the result of an execution
Definition taskflow.hpp:640
tf::Graph
class to create a graph object
Definition graph.hpp:47
tf::SmallVector
class to define a vector optimized for small array
Definition small_vector.hpp:931
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::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 to check if a type is a task parameter
Definition graph.hpp:228
tf
taskflow namespace
Definition small_vector.hpp:20
tf::DefaultNotifier
NonblockingNotifier DefaultNotifier
the default notifier type used by Taskflow
Definition worker.hpp:38
tf::retrieve_graph
Graph & retrieve_graph(T &target)
retrieves a reference to the underlying tf::Graph from an object
Definition graph.hpp:1108
tf::has_env
bool has_env(const std::string &str)
checks whether an environment variable is defined
Definition os.hpp:310