二、基本数据类型

1.类Node

类Node最重要的是保存了当前节点的执行函数，所有前继节点的指针、后续节点的指针，子图（如果有的话），每个前继节点完成后需要修改的计数器（当所有的前继节点都完成时，当前节点就可以执行了）。
class Node {
  //static节点的执行函数是返回值和参数都是void的函数
  using StaticWork  = std::function<void()>;
  //dynamic节点的执行函数是返回值为void、参数为Subflow的函数，
  //在代码中通过模板元编程判断参数是否是Subflow来区分静态节点还是动态节点
  using DynamicWork = std::function<void(Subflow&)>;
  //节点的status
  constexpr static int SPAWNED = 0x1;
  constexpr static int SUBTASK = 0x2;
  public:
    Node() = default;
   //可以通过构造函数传入节点的执行函数，可以不传入（default构造函数），后续再set
    template <typename C>
    Node(C&&);
    //传入当前节点的一个后续节点
    void precede(Node&);
  private:
    std::string _name;
    //用了C++17的std::variant函数，可以认为就是union类型，未设置、static work、dynamic work
    std::variant<std::monostate, StaticWork, DynamicWork> _work;
    //所有的后续节点
    tf::PassiveVector<Node*> _successors;
    //所有的前继节点
    tf::PassiveVector<Node*> _dependents;
    //子图，正如github上的文档所说，每个节点可以在执行期动态的创建子图
    //这里用了C++17的std::optional函数，多了”未设置“状态，好处是如果未设置不会调用Graph的构造函数
    std::optional<Graph> _subgraph;
    //若有值，表示此节点实际一个Taskflow，通过组合成为了一个Node
    Taskflow* _module {nullptr};
    //以下两个是弱引用
    Topology* _topology {nullptr};
    Taskflow* _module {nullptr};
    int _status {0};
    //前继节点的个数，在调度执行过程会多线程修改，所以是std::atomic类型。
    std::atomic<int> _num_dependents {0};
重点说下precede函数的实现 
inline void Node::precede(Node& v) {
  //1.将v添加到当前节点的后续节点vector中
  _successors.push_back(&v);
  //2.将当前节点添加到v的前继节点vector中
  v._dependents.push_back(this);
  //3.将v的前继节点的计数+1，调度执行时每个前继节点完成时-1，等于0时即可执行当前节点
  v._num_dependents.fetch_add(1, std::memory_order_relaxed);
2.类Task
 
类Task很简单，就是Node指针的包装，弱引用一个Node对象指针，并不管理其内存。提供了一些public函数的实现也是直接调用类Node对应的函数。根据注释，用类Task而不是直接使用Node指针的目的，是为了防止使用方直接操作内部存储数据。 
class Task {
  public:
    template <typename C>
    Task& work(C&& callable);
    template <typename... Ts>
    Task& precede(Ts&&... tasks);
    Task& precede(std::vector<Task>& tasks);
    Task& precede(std::initializer_list<Task> tasks);
    template <typename... Ts>
    Task& succeed(Ts&&... tasks);
    Task& succeed(std::vector<Task>& tasks);
    Task& succeed(std::initializer_list<Task> tasks);
    template <typename... Ts>
    Task& gather(Ts&&... tasks);
    Task& gather(std::vector<Task>& tasks);
    Task& gather(std::initializer_list<Task> tasks);
  private:
    //弱引用，不负责管理其内存
    Node* _node {nullptr};
3.类Graph
 
类Graph负责管理Node对象内存。 
class Graph {
  public:
    void clear();
    bool empty() const;
    size_t size() const;
    template <typename C>
    Node& emplace_back(C&&); 
    Node& emplace_back();
    std::vector<std::unique_ptr<Node>>& nodes();
    const std::vector<std::unique_ptr<Node>>& nodes() const;
  private:
    //强引用，负责Node对象内存的管理
    std::vector<std::unique_ptr<Node>> _nodes;
重点看下emplace_back函数的实现： 
template <typename C>
Node& Graph::emplace_back(C&& c) {
  //1.创建一个新的Node节点
  //2.传入c作为新节点的执行函数
  //3.将新节点指针插入到vector中
  //4.返回新节点的引用
  _nodes.push_back(std::make_unique<Node>(std::forward<C>(c)));
  return *(_nodes.back());
inline Node& Graph::emplace_back() {
  //同上，但没有传执行函数，后续再set
  _nodes.push_back(std::make_unique<Node>());
  return *(_nodes.back());
4. 类Topology
 
class Topology {
  public:
    //Taskflow是弱引用，P是判断函数、传给_pred，C是回调函数、传给_call
    template <typename P, typename C>
    Topology(Taskflow&, P&&, C&&);
  private:
    Taskflow& _taskflow;
    //Executor的各种run函数返回的就是这个promise对应的future
    //可以异步的等待taskflow执行完毕
    std::promise<void> _promise;
    //DAG的所有入口节点指针（即graph中所有前继节点为0的节点）
    PassiveVector<Node*> _sources;
     //DAG的所有出口节点的个数（出口只需要知道个数就行了）
    int _cached_num_sinks {0};
    //DAG的出口节点的完成计数器（当所有的出口都执行完毕，整个DAG就执行完毕了），多线程写，因此为 std::atomic类型
    std::atomic<int> _num_sinks {0};
    //判断函数，当DAG执行完毕后执行这个函数，若返回false，则再执行一遍
    std::function<bool()> _pred;
    //回调函数，taskflow执行完成后执行此回调函数
    std::function<void()> _call;
    void _bind(Graph& g);
    void _recover_num_sinks();
重点看下_bind函数： 
inline void Topology::_bind(Graph& g) {
  _num_sinks = 0;
  _sources.clear();
  //遍历graph中的每个node，找到所有的入口节点，计算所有的出口节点个数
  for(auto& node : g.nodes()) {
    node->_topology = this;
    if(node->num_dependents() == 0) {
      _sources.push_back(node.get());
    if(node->num_successors() == 0) {
      _num_sinks++;
  _cached_num_sinks = _num_sinks;
三、流数据类型
 
1.类FlowBuilder
 
FlowBuilder是Taskflow的父类，因此先看下FlowBuilder。 
class FlowBuilder {
    FlowBuilder(Graph& graph);
    //插入一个执行函数，实现中将在graph中创建一个新节点，参数callable作为新节点的执行函数
    template <typename C>
    Task emplace(C&& callable);
    //同上，区别是插入多个执行函数
    template <typename... C, std::enable_if_t<(sizeof...(C)>1), void>* = nullptr>
    auto emplace(C&&... callables);
    //对一个容器，从迭代器[beg,end)进行分片，并行度为p（如果为0则设为CPU核数）分片，分片中对每个成员执行callable操作。创建一个占位的开始节点和结束节点（都没有实际的执行函数），返回的就是pair<开始节点，结束节点>
    template <typename I, typename C>
    std::pair<Task, Task> parallel_for(I beg, I end, C&& callable, size_t partitions = 0);
    //同上，多了一个step参数
    template <typename I, typename C, std::enable_if_t<std::is_arithmetic_v<I>, void>* = nullptr >
    std::pair<Task, Task> parallel_for(I beg, I end, I step, C&& callable, size_t partitions = 0);
    //并行执行reduce操作（bop），先分片，每个分片并行执行bop，在对每个分片的结果执行bop。例如求容器中的min或max，可以用这种方法提高效率。
    template <typename I, typename T, typename B>
    std::pair<Task, Task> reduce(I beg, I end, T& result, B&& bop);
    template <typename I, typename T>
    std::pair<Task, Task> reduce_min(I beg, I end, T& result);
    template <typename I, typename T>
    std::pair<Task, Task> reduce_max(I beg, I end, T& result);
   //基本同上，区别在于执行bop之前先对容器中的每个元素执行uop操作。
    template <typename I, typename T, typename B, typename U>
    std::pair<Task, Task> transform_reduce(I beg, I end, T& result, B&& bop, U&& uop);
    template <typename I, typename T, typename B, typename P, typename U>
    std::pair<Task, Task> transform_reduce(I beg, I end, T& result, B&& bop1, P&& bop2, U&& uop);
    //创建一个空的task，无执行体
    Task placeholder();
   //A是B的前继节点
    void precede(Task A, Task B);
    //vector中的第i个task是第i+1个task的前继节点，即后续会串行执行
    void linearize(std::vector<Task>& tasks);
    void linearize(std::initializer_list<Task> tasks);
   //A是所有others的前继节点
    void broadcast(Task A, std::vector<Task>& others);
    void broadcast(Task A, std::initializer_list<Task> others);
    // A是所有others的后续节点
    void gather(std::vector<Task>& others, Task A);
    void gather(std::initializer_list<Task> others, Task A);
  private:
    Graph& _graph;
2.类Taskflow
 
类Taskflow继承自类FlowBuilder，大部分功能FlowBuilder都已经实现了，看下不一样的： 
class Taskflow : public FlowBuilder {
public:
    //Taskflow之间是可以组合的，一个Taskflow作为其他的Taskflow的其中一个节点
    tf::Task composed_of(Taskflow& taskflow);
  private:
    std::string _name;
    //Taskflow保存Graph，而FlowBuilder只是弱引用Graph
    Graph _graph;
    std::mutex _mtx;
   //这里很有意思，Graph只有一个而Topology是一个列表，也就是说一个taskflow可以执行多个拓扑结构，而节点却是可以复用的
    std::list<Topology> _topologies;
类Executor是负责执行taskflow的类。 
class Executor {
  //每个工作线程处理一个Worker
  struct Worker {
    std::mt19937 rdgen { std::random_device{}() };
    //这个queue需要理解：说明每个工作线程有自己单独的queue
    WorkStealingQueue<Node*> queue;
    std::optional<Node*> cache;
    //作为thread local data在每个工作线程中都有一个副本
  struct PerThread {
    Executor* pool {nullptr}; 
    int worker_id  {-1};  //这个id作为索引来访问对应的Worker
  public:
    explicit Executor(unsigned n = std::thread::hardware_concurrency());
    ~Executor();
    std::future<void> run(Taskflow& taskflow);
    template<typename C>
    std::future<void> run(Taskflow& taskflow, C&& callable);
    std::future<void> run_n(Taskflow& taskflow, size_t N);
    template<typename C>
    std::future<void> run_n(Taskflow& taskflow, size_t N, C&& callable);
    template<typename P>
    std::future<void> run_until(Taskflow& taskflow, P&& pred);
    //上面几个run函数最终都是调用此函数，下文重点看下此函数
    template<typename P, typename C>
    std::future<void> run_until(Taskflow& taskflow, P&& pred, C&& callable);
    void wait_for_all();
    size_t num_workers() const;
    template<typename Observer, typename... Args>
    Observer* make_observer(Args&&... args);
    void remove_observer();
  private:
    std::condition_variable _topology_cv;
    std::mutex _topology_mutex;
    std::mutex _queue_mutex;
    unsigned _num_topologies {0};
    // scheduler field
    std::vector<Worker> _workers;//每个工作现场固定访问其中一个
    std::vector<Notifier::Waiter> _waiters;
    std::vector<std::thread> _threads;//N个工作线程
    //这个queue需要理解：这是调用线程（不是工作线程）的queue。也就是说有N+1个queue
    WorkStealingQueue<Node*> _queue;
    std::atomic<size_t> _num_actives {0};
    std::atomic<size_t> _num_thieves {0};
    std::atomic<bool>   _done        {0};
    Notifier _notifier;
    std::unique_ptr<ExecutorObserverInterface> _observer;
Executor在构造函数中做了这几件事：
 1.创建了N个工作线程，N为并发度、默认为CPU核数
 2.工作线程中初始化tls数据PerThread
 3.然后就是一个死循环： 
std::optional<Node*> t;
while(1) {
        // i是工作线程的索引；执行task
        _exploit_task(i, t);
        // 等待可用的task
        if(_wait_for_task(i, t) == false) {
          break;
注意上面先执行_exploit_task，再执行_wait_for_task，而通常确实相反的操作。这里这么做的原因是，执行到_wait_for_task说明当前工作线程自己的queue已经全部执行完、为空了。 
_wait_for_task中用了一个Eigen中Notifier类，相对于通常实现生产者-消费者模式直接使用信号量性能更好，理解上可以当做信号量理解。 
_wait_for_task原理：
 1.生成[0，N)之间的随机数作为索引，尝试从对应的工作线程的queue中偷一个task（如果随机到本线程id，则到调用线程queue中偷）。偷到了则进行下一步，偷不到则继续随机，直到随机了Y（Y=100）次还是没偷到，也进行下一步。
 2.偷到了则返回，没偷到则继续下一步
 3.再次尝试从调用线程queue中偷任务，偷到了则返回
 4.阻塞等待 
可以看到，在阻塞等待前通过work stealing提高线程使用率。 
_exploit_task原理：
 1.对于取到的任务（实际是Node*），进行执行：
 A 对于static节点，执行该组合节点
 B 对于dynamic节点，执行节点的node->_work，并根据_subgraph创建子图的拓扑结构，开始调度这个子图拓扑结构
 2. 对于当前节点的所有后继节点的_num_dependents - 1，如果减到0，则加入调度队列
 3. 如果当前节点是一个最终节点，那么对拓扑结构的_num_sinks - 1，如果减到0，则到tear_down逻辑
 4. 执行拓扑结构的_pred函数（如果有的话），如果返回false，则再执行一遍打当前拓扑结构，否则：
 A 调用回调函数_call
 B 设置_promise
 C 执行拓扑结构队列中的下一个拓扑结构，全部都执行完成后设置 
五、一些有趣的代码
 
cpp-taskflow中有一些很有趣的代码：
 1.Work Stealing Queue
 2.来源于Eigen中的Notifier类 
这块再单独写文档。 
                    一、简介cpp-taskflow 源码：https://github.com/cpp-taskflow/cpp-taskflow（后面简称taskflow）taskflow一个写的比较好的基于task有向无环图（DAG）的并行调度的框架，之所以说写的比较好，个人觉得有几点原因：1.是一个兼具学术研究和工业使用的项目，并非一个玩具2.现代C++开发，风格简洁（源码要求编译器支持C++1...
 Taskflow帮助您使用现代C ++快速编写并行和异构任务程序
为什么选择任务流？
 与许多现有任务编程框架相比，任务流在处理复杂的并行工作负载方面更快，更富表现力，并且更易于进行嵌入式集成。
 使用Taskflow，您可以快速实施结合常规和不规则计算模式的任务分解策略，以及有效的工作窃取调度程序，以优化您的多线程性能。
 Taskflow支持有条件的任务，使您能够跨相关任务做出快速的控制流决策，以实现原本很难用现有工具完成的周期和条件。
 任务流是可组合的。 您可以通过组合模块化和可重复使用的块来创建大型并行图，这些块在单个范围内更易于优化。
 Taskflow通过利用CPU-GPU协作计算的功能来支持异构任务，以加速各种科学计算应用程序。
 Taskflow提供了概要分析Taskflow程序所需的可视化和工具。
 我们致力于为并行计算领域的学术研究和工业研究项目提供可靠的
				Cpp-Taskflow一个快速的C ++仅标头库，可帮助您快速构建具有复杂任务依赖关系的并行程序。
为什么选择Cpp-Taskflow？
Cpp-Taskflow可让您使用现代程序快速构建并行依赖关系图Taskflow Taskflow可帮助您在现代C ++中快速编写并行任务程序:red_exclamation_mark:从v2.5.0开始，我们将cpp-taskflow重命名为Taskflow以扩大其支持范围和将来的应用范围。
核心代码库保持不变。
您可能只需要将远程URL更改为此新存储库。
感谢您的支持！
为什么选择任务流？
与许多现有的任务编程框架相比，任务流在处理复杂问题时更快，更富表现力，并且更易于进行嵌入式集成。
				TaskFlow主要由OpenStack社区在维护，致力于为一组任务的执行提供高效细粒度的控制。在Cinder组件源码中，创建卷等耗时比较长且容易出错的操作便利用到了taskflow，同样的，Neutron组件也使用它控制创建网络的复杂操作。
一、taskflow的特性
TaskFlow作为OpenStack的通用流程引擎，能让作业执行变得容易，同时满足三种特性。
1.独立性
不相关的原...
				TaskFlow 是一个以高度可用，易于理解和声明试方式来执行[作业，任务，流程]的库，可与OpenStack和其他项目一起使用。
taskflow库在oslo项目中是一个实现比较复杂的项目，要弄清楚其实现原理，首先需要对其中的相关概念有所了解。所以，本文首先总结了taskflow中常用的一些基本概念，这些概念主要包括如下几个：
Atom类是taskflow的最小单位，task...
				免费的跨平台SSH和SFTP工具： 
https://www.putty.org/ -- Windows 
https://www.chiark.greenend.org.uk/~sgtatham/putty/latest.html 
https://www.poderosa-terminal.com/ -- Windows/Mac 
https://filezilla-project.org/...
tf::Task A = taskflow.emplace([] () {}).name("A");
tf::Task B = taskflow.emplace([] () {}).name("B");
tf::Task C = taskflow.emplace([] () {}).name("C");
tf::Task D = taskflow.emplace([] () {}).name("
 Flow使用指南之九中介绍了如何调用Remote TaskFlow, 但那是通过TaskFlow Call Activity调用的，如何通过程序调用一个Bounded TaskFlow呢？
首先我们必须知道TaskFlow的URL的格式，