A full version of the library can be downloaded at http://www.boost.org/.
-cpp-taskflow-*
+taskflow-*
--------------
-Contains the cpp-taskflow project licensed under the MIT license.
+Contains the taskflow project licensed under the MIT license.
A full version of the project project can be downloaded at
-https://cpp-taskflow.github.io/.
+https://taskflow.github.io/.
muparser_*
"Always use the bundled taskflow header library instead of an external one."
OFF)
-SET(TASKFLOW_FOLDER "${CMAKE_SOURCE_DIR}/bundled/taskflow-2.4.0")
+SET(TASKFLOW_FOLDER "${CMAKE_SOURCE_DIR}/bundled/taskflow-2.5.0")
#
-MIT License
+Taskflow LICENSE
Copyright (c) 2018-2020 T.-W. Huang, C.-X. Lin, G. Guo, and M. Wong
-# Cpp-Taskflow <img align="right" width="10%" src="image/cpp-taskflow_logo.png">
+# Taskflow <img align="right" width="10%" src="image/taskflow_logo.png">
-[](https://app.codacy.com/app/tsung-wei-huang/cpp-taskflow?utm_source=github.com&utm_medium=referral&utm_content=cpp-taskflow/cpp-taskflow&utm_campaign=Badge_Grade_Dashboard)
-[](https://travis-ci.com/cpp-taskflow/cpp-taskflow)
-[](https://ci.appveyor.com/project/TsungWeiHuang/cpp-taskflow)
+[](https://app.codacy.com/gh/taskflow/taskflow?utm_source=github.com&utm_medium=referral&utm_content=taskflow/taskflow&utm_campaign=Badge_Grade_Dashboard)
+[](https://travis-ci.com/taskflow/taskflow)
+[](https://ci.appveyor.com/project/tsung-wei-huang/taskflow)
[][wiki]
-[](https://cpp-taskflow.github.io/tfprof/)
+[](https://taskflow.github.io/tfprof/)
[](https://arxiv.org/abs/2004.10908v2)
-Cpp-Taskflow helps you quickly write parallel programs with high performance scalability
-and simultaneous high productivity.
+Taskflow helps you quickly write parallel tasks programs in modern C++
-# Why Cpp-Taskflow?
+:exclamation: Starting from [v2.5.0](https://github.com/taskflow/taskflow/releases/tag/2.5.0), we have renamed cpp-taskflow to ***taskflow*** to broaden its support and future application scopes. The core codebase remains *unchanged*. You may only need to [change the remote URL](https://help.github.com/en/github/using-git/changing-a-remotes-url) to this new repository. Thank your for the support!
-Cpp-Taskflow is faster, more expressive, and easier for drop-in integration
+# Why Taskflow?
+
+Taskflow is faster, more expressive, and easier for drop-in integration
than many of existing task programming frameworks
in handling complex parallel workloads.
-Cpp-Taskflow lets you quickly implement task decomposition strategies
+Taskflow lets you quickly implement task decomposition strategies
that incorporate both regular and irregular compute patterns,
together with an efficient *work-stealing* scheduler to optimize your multithreaded performance.
-| [Static Tasking](#get-started-with-cpp-taskflow) | [Dynamic Tasking](#dynamic-tasking) |
+| [Static Tasking](#get-started-with-taskflow) | [Dynamic Tasking](#dynamic-tasking) |
| :------------: | :-------------: |
|  | <img align="right" src="image/dynamic_graph.svg" width="100%"> |
-Cpp-Taskflow supports conditional tasking for you to make rapid control-flow decisions
+Taskflow supports conditional tasking for you to make rapid control-flow decisions
across dependent tasks to implement cycles and conditions that were otherwise difficult to do
with existing tools.
| :-----------------: |
|  |
-Cpp-Taskflow is composable. You can create large parallel graphs through
+Taskflow is composable. You can create large parallel graphs through
composition of modular and reusable blocks that are easier to optimize
at an individual scope.
| :---------------: |
||
-Cpp-Taskflow supports heterogeneous tasking for you to
+Taskflow supports heterogeneous tasking for you to
accelerate a wide range of scientific computing applications
by harnessing the power of CPU-GPU collaborative computing.
| :-----------------: |
|  |
+
+Taskflow provides visualization and tooling needed for profiling Taskflow programs.
+
+| [Taskflow Profiler](https://taskflow.github.io/tfprof) |
+| :-----------------: |
+|  |
+
We are committed to support trustworthy developments for both academic and industrial research projects
-in parallel computing. Check out [Who is Using Cpp-Taskflow](#who-is-using-cpp-taskflow) and what our users say:
+in parallel computing. Check out [Who is Using Taskflow](#who-is-using-taskflow) and what our users say:
-+ *"Cpp-Taskflow is the cleanest Task API I've ever seen." [Damien Hocking @Corelium Inc](http://coreliuminc.com)*
-+ *"Cpp-Taskflow has a very simple and elegant tasking interface. The performance also scales very well." [Glen Fraser][totalgee]*
-+ *"Cpp-Taskflow lets me handle parallel processing in a smart way." [Hayabusa @Cpp-Learning](https://cpp-learning.com/cpp-taskflow/)*
-+ *"Cpp-Taskflow improves the throughput of our graph engine in just a few hours of coding." [Jean-Michaël @KDAB](https://ossia.io/)*
++ *"Taskflow is the cleanest Task API I've ever seen." [Damien Hocking @Corelium Inc](http://coreliuminc.com)*
++ *"Taskflow has a very simple and elegant tasking interface. The performance also scales very well." [Glen Fraser][totalgee]*
++ *"Taskflow lets me handle parallel processing in a smart way." [Hayabusa @Learning](https://cpp-learning.com/cpp-taskflow/)*
++ *"Taskflow improves the throughput of our graph engine in just a few hours of coding." [Jean-Michaël @KDAB](https://ossia.io/)*
+ *"Best poster award for open-source parallel programming library." [Cpp Conference 2018][Cpp Conference 2018]*
+ *"Second Prize of Open-source Software Competition." [ACM Multimedia Conference 2019](https://tsung-wei-huang.github.io/img/mm19-ossc-award.jpg)*
See a quick [presentation][Presentation] and
-visit the [documentation][wiki] to learn more about Cpp-Taskflow.
+visit the [documentation][wiki] to learn more about Taskflow.
Technical details can be referred to our [arXiv paper](https://arxiv.org/abs/2004.10908v2).
# Table of Contents
-* [Get Started with Cpp-Taskflow](#get-started-with-cpp-taskflow)
+* [Get Started with Taskflow](#get-started-with-taskflow)
* [Create a Taskflow Application](#create-a-taskflow-application)
* [Step 1: Create a Taskflow](#step-1-create-a-taskflow)
* [Step 2: Define Task Dependencies](#step-2-define-task-dependencies)
* [Step 3: Execute a Taskflow](#step-3-execute-a-taskflow)
* [Dynamic Tasking](#dynamic-tasking)
* [Conditional Tasking](#conditional-tasking)
- * [Step 1: Create a Condition Task](#step-1-create-a-condition-task)
- * [Step 2: Scheduling Rules for Condition Tasks](#step-2-scheduling-rules-for-condition-tasks)
* [Composable Tasking](#composable-tasking)
* [Concurrent CPU-GPU Tasking](#concurrent-cpu-gpu-tasking)
* [Step 1: Create a cudaFlow](#step-1-create-a-cudaflow)
* [Step 2: Compile and Execute a cudaFlow](#step-2-compile-and-execute-a-cudaflow)
* [Visualize a Taskflow Graph](#visualize-a-taskflow-graph)
-* [Monitor Thread Activities](#monitor-thread-activities)
* [API Reference](#api-reference)
* [System Requirements](#system-requirements)
* [Compile Unit Tests, Examples, and Benchmarks](#compile-unit-tests-examples-and-benchmarks)
-* [Who is Using Cpp-Taskflow?](#who-is-using-cpp-taskflow)
+* [Who is Using Taskflow?](#who-is-using-taskflow)
-# Get Started with Cpp-Taskflow
+# Get Started with Taskflow
-The following example [simple.cpp](./examples/simple.cpp) shows the basic Cpp-Taskflow API
+The following example [simple.cpp](./examples/simple.cpp) shows the basic Taskflow API
you need in most applications.
```cpp
-#include <taskflow/taskflow.hpp> // Cpp-Taskflow is header-only
+#include <taskflow/taskflow.hpp> // Taskflow is header-only
int main(){
# Create a Taskflow Application
-Cpp-Taskflow defines a very expressive API to create task dependency graphs.
+Taskflow defines a very expressive API to create task dependency graphs.
Most applications are developed through the following three steps:
## Step 1: Create a Taskflow
# Conditional Tasking
-Taskflow supports *conditional tasking* for users to implement *dynamic* and *cyclic* control flows.
-You can create highly versatile and efficient parallel patterns through condition tasks.
-
-## Step 1: Create a Condition Task
-
+Taskflow supports *conditional tasking* for users to implement *general* control flow with cycles and conditionals.
A *condition task* evalutes a set of instructions and returns an integer index
of the next immediate successor to execute.
The index is defined with respect to the order of its successor construction.
executor.run(tf).wait();
```
-If the return value from `cond` is 0, it loops back to itself, or otherwise to `stop`.
-Cpp-Taskflow terms the preceding link from a condition task a *weak dependency*
-(dashed lines above).
-Others are *strong depedency* (solid lines above).
-
-
-## Step 2: Scheduling Rules for Condition Tasks
-
-When you submit a taskflow to an executor,
-the scheduler starts with tasks of *zero dependency* (both weak and strong dependencies)
-and continues to execute successive tasks whenever *strong dependencies* are met.
-However,
-the scheduler skips this rule for a condition task and jumps directly to its successor
-indexed by the return value.
-
-
-
-It is users' responsibility to ensure a taskflow is properly conditioned.
-Top things to avoid include no source tasks to start with and task race.
-The figure shows common pitfalls and their remedies.
-In the risky scenario, task X may not be raced if P and M is exclusively
-branching to X.
-
-
-
-
-A good practice for avoiding mistakes of conditional tasking is to infer the execution flow of your graphs based on our scheduling rules.
-Make sure there is no task race.
-
<div align="right"><b><a href="#table-of-contents">[↑]</a></b></div>
# Concurrent CPU-GPU Tasking
-Cpp-Taskflow enables concurrent CPU-GPU tasking by leveraging
+Taskflow enables concurrent CPU-GPU tasking by leveraging
[Nvidia CUDA Toolkit][cuda-toolkit].
You can harness the power of CPU-GPU collaborative computing
to implement heterogeneous decomposition algorithms.
-# Monitor Thread Activities
-
-Cpp-Taskflow Profiler ([TFProf](https://github.com/cpp-taskflow/tfprof))
-provides the visualization and tooling needed for profiling cpp-taskflow programs.
-
-<p align="center">
- <a href="https://cpp-taskflow.github.io/tfprof/">
- <img width="100%" src="image/tfprof.png">
- </a>
-</p>
-
-<div align="right"><b><a href="#table-of-contents">[↑]</a></b></div>
-
# API Reference
The official [documentation][wiki] explains a complete list of
-Cpp-Taskflow API.
+Taskflow API.
Here, we highlight commonly used methods.
## Taskflow API
The class `tf::Taskflow` is the main place to create a task dependency graph.
-The table below summarizes a list of commonly used methods.
-
-| Method | Argument | Return | Description |
-| -------- | --------- | ------- | ----------- |
-| emplace | callables | tasks | creates a task with a given callable(s) |
-| placeholder | none | task | inserts a node without any work; work can be assigned later |
-| parallel_for | beg, end, callable, chunk | task pair | concurrently applies the callable chunk by chunk to the result of dereferencing every iterator in the range |
-| parallel_for | beg, end, step, callable, chunk | task pair | concurrently applies the callable chunk by chunk to an index-based range with a step size |
-| num_workers | none | size | queries the number of working threads in the pool |
-| dump | ostream | none | dumps the taskflow to an output stream in GraphViz format |
### *emplace/placeholder*
You can use `emplace` to create a task from a target callable.
```cpp
-tf::Task task = tf.emplace([] () { std::cout << "my task\n"; });
+tf::Task task = taskflow.emplace([] () { std::cout << "my task\n"; });
```
When a task cannot be determined beforehand, you can create a placeholder and assign the callable later.
```cpp
-tf::Task A = tf.emplace([](){});
-tf::Task B = tf.placeholder();
+tf::Task A = taskflow.emplace([](){});
+tf::Task B = taskflow.placeholder();
A.precede(B);
B.work([](){ /* do something */ });
```
```cpp
auto v = {'A', 'B', 'C', 'D'};
-auto [S, T] = tf.parallel_for(
+auto [S, T] = taskflow.parallel_for(
v.begin(), // iterator to the beginning
v.end(), // iterator to the end
[] (int i) {
```cpp
auto v = {'A', 'B', 'C', 'D'};
-auto [S, T] = tf.parallel_for(
+auto [S, T] = taskflow.parallel_for(
v.begin(), // iterator to the beginning
v.end(), // iterator to the end
[] (int i) {
```cpp
// [0, 11) with a step size of 2
-auto [S, T] = tf.parallel_for(
+auto [S, T] = taskflow.parallel_for(
0, 11, 2,
[] (int i) {
std::cout << "parallel_for on index " << i << std::endl;
Each time you create a task, the taskflow object adds a node to the present task dependency graph
and return a *task handle* to you.
-A task handle is a lightweight object that defines a set of methods for users to
-access and modify the attributes of the associated task.
-The table below summarizes a list of commonly used methods.
-
-| Method | Argument | Return | Description |
-| -------------- | ----------- | ------ | ----------- |
-| name | string | self | assigns a human-readable name to the task |
-| work | callable | self | assigns a work of a callable object to the task |
-| precede | task list | self | enables this task to run *before* the given tasks |
-| succeed | task list | self | enables this task to run *after* the given tasks |
-| num_dependents | none | size | returns the number of dependents (inputs) of this task |
-| num_successors | none | size | returns the number of successors (outputs) of this task |
-| empty | none | bool | returns true if the task points to a graph node or false otherwise |
-| has_work | none | bool | returns true if the task points to a graph node with a callable assigned |
+You can access or modify the attributes of the associated task node.
### *name*
A.work([] () { std::cout << "hello world!"; });
```
-### *precede*
+### *precede/succeed*
-The method `precede` lets you add a preceding link from self to other tasks.
+The method `precede/succedd` lets you add a preceding/succeeding link between tasks.
<img align="right" width="30%" src="image/broadcast.svg">
### *empty/has_work*
-A task is empty is it is not associated with any graph node.
+A task is empty if it is not associated with any graph node.
```cpp
tf::Task task; // assert(task.empty());
## Executor API
-The class `tf::Executor` is used for execution of one or multiple taskflow objects.
-The table below summarizes a list of commonly used methods.
-
-| Method | Argument | Return | Description |
-| --------- | -------------- | ------------- | ------------------------ |
-| run | taskflow | future | runs the taskflow once |
-| run_n | taskflow, N | future | runs the taskflow N times |
-| run_until | taskflow, binary predicate | future | keeps running the taskflow until the predicate becomes true |
-| wait_for_all | none | none | blocks until all running tasks finish |
-| make_observer | arguments to forward to user-derived constructor | pointer to the observer | creates an observer to monitor the thread activities of the executor |
+The class `tf::Executor` is used for executing one or multiple taskflow objects.
### *run/run_n/run_until*
-The run series are non-blocking call to execute a taskflow graph.
+The run series are *thread-safe* and *non-blocking* calls to execute a taskflow.
Issuing multiple runs on the same taskflow will automatically synchronize
to a sequential chain of executions.
# System Requirements
-To use the latest [Cpp-Taskflow](https://github.com/cpp-taskflow/cpp-taskflow/archive/master.zip), you only need a [C++14][C++14] compiler.
+To use the latest [Taskflow](https://github.com/taskflow/taskflow/archive/master.zip), you only need a [C++14][C++14] compiler.
+ GNU C++ Compiler at least v5.0 with -std=c++14
+ Clang C++ Compiler at least v4.0 with -std=c++14
-+ Microsoft Visual Studio at least v15.7 (MSVC++ 19.14); see [vcpkg guide](https://github.com/cpp-taskflow/cpp-taskflow/issues/143)
++ Microsoft Visual Studio at least v15.7 (MSVC++ 19.14); see [vcpkg guide](https://github.com/taskflow/taskflow/issues/143)
+ AppleClang Xode Version at least v8
+ Nvidia CUDA Toolkit and Compiler ([nvcc][nvcc]) at least v10.0 with -std=c++14
-Cpp-Taskflow works on Linux, Windows, and Mac OS X. See the [C++ compiler support](https://en.cppreference.com/w/cpp/compiler_support) status.
+Taskflow works on Linux, Windows, and Mac OS X. See the [C++ compiler support](https://en.cppreference.com/w/cpp/compiler_support) status.
<div align="right"><b><a href="#table-of-contents">[↑]</a></b></div>
# Compile Unit Tests, Examples, and Benchmarks
-Cpp-Taskflow uses [CMake](https://cmake.org/) to build examples and unit tests.
+Taskflow uses [CMake](https://cmake.org/) to build examples and unit tests.
We recommend using out-of-source build.
```bash
## Examples
-The folder `examples/` contains several examples and is a great place to learn to use Cpp-Taskflow.
+The folder `examples/` contains several examples and is a great place to learn to use Taskflow.
| Example | Description |
| ------- | ----------- |
<div align="right"><b><a href="#table-of-contents">[↑]</a></b></div>
-# Who is Using Cpp-Taskflow?
+# Who is Using Taskflow?
-Cpp-Taskflow is being used in both industry and academic projects to scale up existing workloads
+Taskflow is being used in both industry and academic projects to scale up existing workloads
that incorporate complex task dependencies.
- [OpenTimer][OpenTimer]: A High-performance Timing Analysis Tool for Very Large Scale Integration (VLSI) Systems
- [Heteroflow](https://github.com/Heteroflow/Heteroflow): A Modern C++ Parallel CPU-GPU Task Programming Library
- [OpenPhySyn](https://github.com/The-OpenROAD-Project/OpenPhySyn): A plugin-based physical synthesis optimization kit as part of the OpenRoad flow
- [OSSIA](https://ossia.io/): Open-source Software System for Interactive Applications
+- [deal.II](https://github.com/dealii/dealii): A C++ software library to support the creation of finite element code
-[More...](https://github.com/search?q=cpp-taskflow&type=Code)
+[More...](https://github.com/search?q=taskflow&type=Code)
<div align="right"><b><a href="#table-of-contents">[↑]</a></b></div>
# Contributors
-Cpp-Taskflow is being actively developed and contributed by the
-[these people](https://github.com/cpp-taskflow/cpp-taskflow/graphs/contributors).
+Taskflow is being actively developed and contributed by the
+[these people](https://github.com/taskflow/taskflow/graphs/contributors).
Meanwhile, we appreciate the support from many organizations for our developments.
# License
-Cpp-Taskflow is licensed under the [MIT License](./LICENSE).
+Taskflow is licensed under the [MIT License](./LICENSE).
* * *
[Tsung-Wei Huang]: https://tsung-wei-huang.github.io/
[Chun-Xun Lin]: https://github.com/clin99
[Martin Wong]: https://ece.illinois.edu/directory/profile/mdfwong
-[Andreas Olofsson]: https://github.com/aolofsson
-[Gitter]: https://gitter.im/cpp-taskflow/Lobby
[Gitter badge]: ./image/gitter_badge.svg
-[GitHub releases]: https://github.com/coo-taskflow/cpp-taskflow/releases
-[GitHub issues]: https://github.com/cpp-taskflow/cpp-taskflow/issues
-[GitHub insights]: https://github.com/cpp-taskflow/cpp-taskflow/pulse
-[GitHub pull requests]: https://github.com/cpp-taskflow/cpp-taskflow/pulls
-[GitHub contributors]: https://github.com/cpp-taskflow/cpp-taskflow/graphs/contributors
+[GitHub releases]: https://github.com/taskflow/taskflow/releases
+[GitHub issues]: https://github.com/taskflow/taskflow/issues
+[GitHub insights]: https://github.com/taskflow/taskflow/pulse
+[GitHub pull requests]: https://github.com/taskflow/taskflow/pulls
+[GitHub contributors]: https://github.com/taskflow/taskflow/graphs/contributors
[GraphViz]: https://www.graphviz.org/
[AwesomeGraphViz]: https://dreampuf.github.io/GraphvizOnline/
[OpenMP Tasking]: https://www.openmp.org/spec-html/5.0/openmpsu99.html
[UIUC]: https://illinois.edu/
[CSL]: https://csl.illinois.edu/
[UofU]: https://www.utah.edu/
-[wiki]: https://cpp-taskflow.github.io/cpp-taskflow/index.html
-[release notes]: https://cpp-taskflow.github.io/cpp-taskflow/Releases.html
+[wiki]: https://taskflow.github.io/taskflow/index.html
+[release notes]: https://taskflow.github.io/taskflow/Releases.html
[PayMe]: https://www.paypal.me/twhuang/10
[C++17]: https://en.wikipedia.org/wiki/C%2B%2B17
[C++14]: https://en.wikipedia.org/wiki/C%2B%2B14
[NovusCore]: https://github.com/novuscore/NovusCore
[SA-PCB]: https://github.com/choltz95/SA-PCB
-[Presentation]: https://cpp-taskflow.github.io/
+[Presentation]: https://taskflow.github.io/
[chrome://tracing]: chrome://tracing
@brief execution interface for running a taskflow graph
-An executor object manages a set of worker threads and implements
+An executor object manages a set of worker threads and internalements
an efficient work-stealing scheduling algorithm to run a taskflow.
*/
class Executor {
+ friend class Subflow;
+
struct Worker {
size_t id;
size_t victim;
Notifier::Waiter* waiter;
std::mt19937 rdgen { std::random_device{}() };
TaskQueue<Node*> wsq[NUM_DOMAINS];
- Node* cache {nullptr};
};
struct PerThread {
void _worker_loop(Worker&);
void _exploit_task(Worker&, Node*&);
void _explore_task(Worker&, Node*&);
- void _schedule(Node*, bool);
+ void _schedule(Node*);
void _schedule(PassiveVector<Node*>&);
void _invoke(Worker&, Node*);
void _invoke_static_work(Worker&, Node*);
- void _invoke_dynamic_work(Worker&, Node*, bool&);
+ void _invoke_dynamic_work(Worker&, Node*);
+ void _invoke_dynamic_work_internal(Worker&, Node*, Graph&, bool);
+ void _invoke_dynamic_work_external(Graph&, Node*);
void _invoke_condition_work(Worker&, Node*);
- void _invoke_module_work(Worker&, Node*, bool&);
+ void _invoke_module_work(Worker&, Node*);
#ifdef TF_ENABLE_CUDA
void _invoke_cudaflow_work(Worker&, Node*);
- void _invoke_cudaflow_work_impl(Worker&, Node*);
+ void _invoke_cudaflow_work_internal(Worker&, Node*);
#endif
void _set_up_topology(Topology*);
- void _tear_down_topology(Topology**);
+ void _tear_down_topology(Topology*);
void _increment_topology();
void _decrement_topology();
void _decrement_topology_and_notify();
// Procedure: _exploit_task
inline void Executor::_exploit_task(Worker& w, Node*& t) {
- assert(!w.cache);
-
if(t) {
const auto d = w.domain;
if(_num_actives[d].fetch_add(1) == 0 && _num_thieves[d] == 0) {
_notifier[d].notify(false);
}
-
- auto tpg = t->_topology;
- auto par = t->_parent;
- auto exe = size_t{1};
do {
- _invoke(w, t);
- if(w.cache) {
- t = w.cache;
- w.cache = nullptr;
- }
- else {
- t = w.wsq[d].pop();
- if(t) {
- // We only increment the counter when poping task from wsq
- // (NOT including cache!)
- if(t->_parent == par) {
- exe++;
- }
- // joined subflow
- else {
- if(par == nullptr) {
- // still have tasks so the topology join counter can't be zero
- t->_topology->_join_counter.fetch_sub(exe);
- }
- else {
- auto ret = par->_join_counter.fetch_sub(exe);
- if(ret == exe) {
- if(par->domain() == d) {
- w.wsq[d].push(par);
- }
- else {
- _schedule(par, false);
- }
- }
- }
- exe = 1;
- par = t->_parent;
- }
- }
- else {
- // If no more local tasks!
- if(par == nullptr) {
- if(tpg->_join_counter.fetch_sub(exe) == exe) {
- // TODO: Store tpg in local variable not in w
- _tear_down_topology(&tpg);
- if(tpg != nullptr) {
- t = w.wsq[d].pop();
- if(t) {
- exe = 1;
- }
- }
- }
- }
- else {
- if(par->_join_counter.fetch_sub(exe) == exe) {
- if(par->domain() == d) {
- t = par;
- par = par->_parent;
- exe = 1;
- }
- else {
- _schedule(par, false);
- }
- }
- }
+ _invoke(w, t);
+
+ if(t->_parent == nullptr) {
+ if(t->_topology->_join_counter.fetch_sub(1) == 1) {
+ _tear_down_topology(t->_topology);
}
}
+ else { // joined subflow
+ t->_parent->_join_counter.fetch_sub(1);
+ }
+
+ t = w.wsq[d].pop();
+
} while(t);
--_num_actives[d];
// Procedure: _schedule
// The main procedure to schedule a give task node.
// Each task node has two types of tasks - regular and subflow.
-inline void Executor::_schedule(Node* node, bool bypass_hint) {
+inline void Executor::_schedule(Node* node) {
//assert(_workers.size() != 0);
auto worker = _per_thread().worker;
if(worker != nullptr && worker->executor == this) {
- if(bypass_hint) {
- assert(!worker->cache);
- worker->cache = node;
- }
- else {
- worker->wsq[d].push(node);
- if(worker->domain != d) {
- if(_num_actives[d] == 0 && _num_thieves[d] == 0) {
- _notifier[d].notify(false);
- }
+ worker->wsq[d].push(node);
+ if(worker->domain != d) {
+ if(_num_actives[d] == 0 && _num_thieves[d] == 0) {
+ _notifier[d].notify(false);
}
}
return;
// module task
case Node::MODULE_WORK: {
- bool first_time = !node->_has_state(Node::SPAWNED);
- bool emptiness = false;
- _invoke_module_work(worker, node, emptiness);
- if(first_time && !emptiness) {
- return;
- }
+ _invoke_module_work(worker, node);
}
break;
// dynamic task
case Node::DYNAMIC_WORK: {
- // Need to create a subflow if it is the first time entering here
- if(!node->_has_state(Node::SPAWNED)) {
- bool join = false;
- _invoke_dynamic_work(worker, node, join);
- if(join) {
- return;
- }
- }
+ _invoke_dynamic_work(worker, node);
}
break;
default:
break;
}
-
// We MUST recover the dependency since subflow may have
// a condition node to go back (cyclic).
node->_join_counter = node->num_dependents();
}
- node->_unset_state(Node::SPAWNED);
-
// At this point, the node storage might be destructed.
- Node* cache {nullptr};
-
for(size_t i=0; i<num_successors; ++i) {
if(--(node->_successors[i]->_join_counter) == 0) {
- if(node->_successors[i]->domain() != worker.domain) {
- c.fetch_add(1);
- _schedule(node->_successors[i], false);
- }
- else {
- if(cache) {
- c.fetch_add(1);
- _schedule(cache, false);
- }
- cache = node->_successors[i];
- }
+ c.fetch_add(1);
+ _schedule(node->_successors[i]);
}
}
-
- if(cache) {
- _schedule(cache, true);
- }
}
// Procedure: _observer_prologue
}
// Procedure: _invoke_dynamic_work
-inline void Executor::_invoke_dynamic_work(Worker& worker, Node* node, bool& join) {
+inline void Executor::_invoke_dynamic_work(Worker& w, Node* node) {
- _observer_prologue(worker, node);
-
- auto& subgraph = nstd::get<Node::DynamicWork>(node->_handle).subgraph;
+ _observer_prologue(w, node);
+
+ auto& handle = nstd::get<Node::DynamicWork>(node->_handle);
+
+ handle.subgraph.clear();
+
+ Subflow fb(*this, node, handle.subgraph);
+
+ handle.work(fb);
+
+ if(!fb._joined) {
+ _invoke_dynamic_work_internal(w, node, handle.subgraph, fb._detach);
+ }
+
+ // TODO
+ _observer_epilogue(w, node);
+}
+
+// Procedure: _invoke_dynamic_work_external
+inline void Executor::_invoke_dynamic_work_external(Graph& g, Node* p) {
- subgraph.clear();
- Subflow fb(subgraph);
+ auto worker = _per_thread().worker;
- nstd::get<Node::DynamicWork>(node->_handle).work(fb);
+ assert(worker && worker->executor == this);
+
+ _invoke_dynamic_work_internal(*worker, p, g, false);
+}
- node->_set_state(Node::SPAWNED);
+// Procedure: _invoke_dynamic_work_internal
+inline void Executor::_invoke_dynamic_work_internal(Worker& w, Node* p, Graph& g, bool d) {
- if(!subgraph.empty()) {
+ if(!g.empty()) {
PassiveVector<Node*> src;
- for(auto n : subgraph._nodes) {
+ for(auto n : g._nodes) {
- n->_topology = node->_topology;
+ n->_topology = p->_topology;
n->_set_up_join_counter();
+ n->_parent = d ? nullptr : p;
- if(!fb.detached()) {
- n->_parent = node;
- }
-
if(n->num_dependents() == 0) {
src.push_back(n);
}
}
-
- join = fb.joined();
-
- if(!join) { // Detach mode
- node->_topology->_join_counter.fetch_add(src.size());
+
+ // detach here
+ if(d) {
+ p->_topology->_join_counter.fetch_add(src.size());
+ _schedule(src);
}
- else { // Join mode (spawned nodes need second-round execution
- node->_join_counter.fetch_add(src.size());
+ // join here
+ else {
+ p->_join_counter.fetch_add(src.size());
+ _schedule(src);
+ Node* t = nullptr;
- node->_parent ? node->_parent->_join_counter.fetch_add(1) :
- node->_topology->_join_counter.fetch_add(1);
- }
+ do {
+ t = w.wsq[w.domain].pop();
+
+ if(t) {
+ _invoke(w, t);
+ t->_parent ? t->_parent->_join_counter.fetch_sub(1) :
+ t->_topology->_join_counter.fetch_sub(1);
+ }
- _schedule(src);
+ } while(p->_join_counter != 0);
+ }
}
-
- _observer_epilogue(worker, node);
}
// Procedure: _invoke_condition_work
auto s = node->_successors[id];
s->_join_counter.store(0);
- if(s->domain() == worker.domain) {
- _schedule(s, true);
- }
- else {
- node->_parent ? node->_parent->_join_counter.fetch_add(1) :
- node->_topology->_join_counter.fetch_add(1);
- _schedule(s, false);
- }
+ node->_parent ? node->_parent->_join_counter.fetch_add(1) :
+ node->_topology->_join_counter.fetch_add(1);
+ _schedule(s);
+
+
+ //if(s->domain() == worker.domain) {
+ // _schedule(s, true);
+ //}
+ //else {
+ // node->_parent ? node->_parent->_join_counter.fetch_add(1) :
+ // node->_topology->_join_counter.fetch_add(1);
+ // _schedule(s, false);
+ //}
}
_observer_epilogue(worker, node);
// Procedure: _invoke_cudaflow_work
inline void Executor::_invoke_cudaflow_work(Worker& worker, Node* node) {
_observer_prologue(worker, node);
- _invoke_cudaflow_work_impl(worker, node);
+ _invoke_cudaflow_work_internal(worker, node);
_observer_epilogue(worker, node);
}
-// Procedure: _invoke_cudaflow_work_impl
-inline void Executor::_invoke_cudaflow_work_impl(Worker& w, Node* node) {
+// Procedure: _invoke_cudaflow_work_internal
+inline void Executor::_invoke_cudaflow_work_internal(Worker& w, Node* node) {
assert(w.domain == node->domain());
#endif
// Procedure: _invoke_module_work
-inline void Executor::_invoke_module_work(Worker& worker, Node* node, bool& ept) {
+inline void Executor::_invoke_module_work(Worker& w, Node* node) {
- // second time to enter this context
- if(node->_has_state(Node::SPAWNED)) {
- return;
- }
-
- _observer_prologue(worker, node);
+ _observer_prologue(w, node);
- // first time to enter this context
- node->_set_state(Node::SPAWNED);
-
auto module = nstd::get<Node::ModuleWork>(node->_handle).module;
-
- if(module->empty()) {
- ept = true;
- return;
- }
-
- PassiveVector<Node*> src;
-
- for(auto n: module->_graph._nodes) {
-
- n->_topology = node->_topology;
- n->_parent = node;
- n->_set_up_join_counter();
-
- if(n->num_dependents() == 0) {
- src.push_back(n);
- }
- }
-
- node->_join_counter.fetch_add(src.size());
-
- if(node->_parent == nullptr) {
- node->_topology->_join_counter.fetch_add(1);
- }
- else {
- node->_parent->_join_counter.fetch_add(1);
- }
- // src can't be empty (banned outside)
- _schedule(src);
+ _invoke_dynamic_work_internal(w, node, module->_graph, false);
- _observer_epilogue(worker, node);
+ _observer_epilogue(w, node);
}
// Function: run
}
// Function: _tear_down_topology
-inline void Executor::_tear_down_topology(Topology** tpg) {
+inline void Executor::_tear_down_topology(Topology* tpg) {
- auto &f = (*tpg)->_taskflow;
+ auto &f = tpg->_taskflow;
//assert(&tpg == &(f._topologies.front()));
// case 1: we still need to run the topology again
- if(! (*tpg)->_pred() ) {
+ if(! tpg->_pred() ) {
//tpg->_recover_num_sinks();
- assert((*tpg)->_join_counter == 0);
- (*tpg)->_join_counter = (*tpg)->_sources.size();
+ assert(tpg->_join_counter == 0);
+ tpg->_join_counter = tpg->_sources.size();
- _schedule((*tpg)->_sources);
+ _schedule(tpg->_sources);
}
// case 2: the final run of this topology
else {
- if((*tpg)->_call != nullptr) {
- (*tpg)->_call();
+ if(tpg->_call != nullptr) {
+ tpg->_call();
}
f._mtx.lock();
// If there is another run (interleave between lock)
if(f._topologies.size() > 1) {
- assert((*tpg)->_join_counter == 0);
+ assert(tpg->_join_counter == 0);
// Set the promise
- (*tpg)->_promise.set_value();
+ tpg->_promise.set_value();
f._topologies.pop_front();
f._mtx.unlock();
// decrement the topology but since this is not the last we don't notify
_decrement_topology();
- *tpg = &(f._topologies.front());
+ tpg = &(f._topologies.front());
- _set_up_topology(*tpg);
- _schedule((*tpg)->_sources);
+ _set_up_topology(tpg);
+ _schedule(tpg->_sources);
//f._topologies.front()._bind(f._graph);
//*tpg = &(f._topologies.front());
// Need to back up the promise first here becuz taskflow might be
// destroy before taskflow leaves
- auto p {std::move((*tpg)->_promise)};
+ auto p {std::move(tpg->_promise)};
f._topologies.pop_front();
p.set_value();
_decrement_topology_and_notify();
-
- // Reset topology so caller can stop execution
- *tpg = nullptr;
}
}
}
_topology_cv.wait(lock, [&](){ return _num_topologies == 0; });
}
+// ----------------------------------------------------------------------------
+// Cyclic Dependency
+// ----------------------------------------------------------------------------
+
+inline void Subflow::join() {
+
+ if(_joined) {
+ TF_THROW("subflow already joined");
+ }
+
+ _executor._invoke_dynamic_work_external(_graph, _parent);
+ _joined = true;
+}
+
} // end of namespace tf -----------------------------------------------------
+
+
+
+
+
+
+
+
+
*/
class FlowBuilder {
- friend class Task;
+ friend class Executor;
public:
- /**
- @brief constructs a flow builder object
-
- @param graph a task dependency graph to manipulate
- */
- FlowBuilder(Graph& graph);
-
/**
@brief creates a static task from a given callable object
/**
@brief constructs a task dependency graph of range-based parallel_for
- The task dependency graph applies a callable object
- to the dereferencing of every iterator
- in the range [beg, end) chunk by chunk.
+ The task dependency graph applies the callable object
+ @p callable to each object obtained by dereferencing
+ every iterator in the range [beg, end). The range
+ is split into chunks of size @p chunk, where each of them
+ is processed by one Task.
+
+ The callable needs to accept a single argument, the object in the range.
@tparam I input iterator type
@tparam C callable type
*/
void succeed(std::initializer_list<Task> others, Task A);
- private:
-
+ protected:
+
+ /**
+ @brief constructs a flow builder with a graph
+ */
+ FlowBuilder(Graph& graph);
+
+ /**
+ @brief associated graph object
+ */
Graph& _graph;
+ private:
+
template <typename L>
void _linearize(L&);
};
@brief building methods of a subflow graph in dynamic tasking
+By default, a subflow automatically joins its parent node. You may explicitly
+join or detach a subflow by calling Subflow::join or Subflow::detach.
+
*/
class Subflow : public FlowBuilder {
+ friend class Executor;
+
public:
- /**
- @brief constructs a subflow builder object
- */
- template <typename... Args>
- Subflow(Args&&... args);
-
/**
@brief enables the subflow to join its parent task
+
+ Performs an immediate action to join the subflow. Once the subflow is joined,
+ it is considered finished and you may not apply any other actions to it.
*/
void join();
/**
@brief enables the subflow to detach from its parent task
+
+ A joined subflow cannot be detached. The subflow will be detached upon leaving
+ its execution context.
*/
void detach();
bool detached() const;
/**
- @brief queries if the subflow will join its parent task
+ @brief queries if the subflow is joinable
+
+ When a subflow is joined, it becomes not joinable.
*/
- bool joined() const;
+ bool joinable() const;
private:
+
+ Subflow(Executor&, Node*, Graph&);
+
+ Executor& _executor;
+ Node* _parent;
- bool _detached {false};
+ bool _joined {false};
+ bool _detach {false};
};
// Constructor
-template <typename... Args>
-Subflow::Subflow(Args&&... args) :
- FlowBuilder {std::forward<Args>(args)...} {
-}
-
-// Procedure: join
-inline void Subflow::join() {
- _detached = false;
+Subflow::Subflow(Executor& executor, Node* parent, Graph& graph) :
+ FlowBuilder {graph},
+ _executor {executor},
+ _parent {parent} {
}
// Procedure: detach
inline void Subflow::detach() {
- _detached = true;
+ if(_joined) {
+ TF_THROW("subflow already joined");
+ }
+ _detach = true;
}
// Function: detached
inline bool Subflow::detached() const {
- return _detached;
+ return _detach;
}
// Function: joined
-inline bool Subflow::joined() const {
- return !_detached;
+inline bool Subflow::joinable() const {
+ return !_joined;
}
TF_ENABLE_POOLABLE_ON_THIS;
// state bit flag
- constexpr static int SPAWNED = 0x1;
- constexpr static int BRANCH = 0x2;
+ constexpr static int BRANCH = 0x1;
// static work handle
struct StaticWork {
struct Timeline {
std::chrono::time_point<std::chrono::steady_clock> origin;
std::vector<std::vector<Segment>> segments;
+ std::vector<std::stack<std::chrono::time_point<std::chrono::steady_clock>>> stacks;
};
public:
// Procedure: set_up
inline void ChromeTracingObserver::set_up(size_t num_workers) {
-
_timeline.segments.resize(num_workers);
+ _timeline.stacks.resize(num_workers);
for(size_t w=0; w<num_workers; ++w) {
_timeline.segments[w].reserve(32);
}
// Procedure: on_entry
-inline void ChromeTracingObserver::on_entry(size_t w, TaskView tv) {
- _timeline.segments[w].emplace_back(
- tv.name(), std::chrono::steady_clock::now()
- );
+inline void ChromeTracingObserver::on_entry(size_t w, TaskView) {
+ _timeline.stacks[w].push(std::chrono::steady_clock::now());
}
// Procedure: on_exit
-inline void ChromeTracingObserver::on_exit(size_t w, TaskView) {
- assert(_timeline.segments[w].size() > 0);
- _timeline.segments[w].back().end = std::chrono::steady_clock::now();
+inline void ChromeTracingObserver::on_exit(size_t w, TaskView tv) {
+ assert(!_timeline.stacks[w].empty());
+
+ auto beg = _timeline.stacks[w].top();
+ _timeline.stacks[w].pop();
+
+ _timeline.segments[w].emplace_back(
+ tv.name(), beg, std::chrono::steady_clock::now()
+ );
}
// Function: clear
inline void ChromeTracingObserver::clear() {
for(size_t w=0; w<_timeline.segments.size(); ++w) {
_timeline.segments[w].clear();
+ while(!_timeline.stacks[w].empty()) {
+ _timeline.stacks[w].pop();
+ }
}
}
struct Timeline {
std::chrono::time_point<std::chrono::steady_clock> origin;
std::vector<std::vector<Segment>> segments;
+ std::vector<std::stack<std::chrono::time_point<std::chrono::steady_clock>>> stacks;
};
public:
inline void TFProfObserver::set_up(size_t num_workers) {
_timeline.segments.resize(num_workers);
+ _timeline.stacks.resize(num_workers);
for(size_t w=0; w<num_workers; ++w) {
_timeline.segments[w].reserve(32);
}
// Procedure: on_entry
-inline void TFProfObserver::on_entry(size_t w, TaskView tv) {
- _timeline.segments[w].emplace_back(
- tv.name(), tv.type(), std::chrono::steady_clock::now()
- );
+inline void TFProfObserver::on_entry(size_t w, TaskView) {
+ _timeline.stacks[w].push(std::chrono::steady_clock::now());
}
// Procedure: on_exit
-inline void TFProfObserver::on_exit(size_t w, TaskView) {
- assert(_timeline.segments[w].size() > 0);
- _timeline.segments[w].back().end = std::chrono::steady_clock::now();
+inline void TFProfObserver::on_exit(size_t w, TaskView tv) {
+ assert(!_timeline.stacks[w].empty());
+
+ auto beg = _timeline.stacks[w].top();
+ _timeline.stacks[w].pop();
+
+ _timeline.segments[w].emplace_back(
+ tv.name(), tv.type(), beg, std::chrono::steady_clock::now()
+ );
}
// Function: clear
inline void TFProfObserver::clear() {
for(size_t w=0; w<_timeline.segments.size(); ++w) {
_timeline.segments[w].clear();
+ while(!_timeline.stacks[w].empty()) {
+ _timeline.stacks[w].pop();
+ }
}
}
#define TF_ENABLE_POOLABLE_ON_THIS \
template <typename T, size_t S> friend class ObjectPool; \
- void* _object_pool_block;
+ void* _object_pool_block
// Class: ObjectPool
//
#include <iostream>
#include <fstream>
#include <mutex>
-#include <deque>
+#include <stack>
+#include <queue>
#include <vector>
#include <algorithm>
#include <memory>
projects / dealii.git / commitdiff