From 95e7822a74f20895addd33b36ba7e62999a3366b Mon Sep 17 00:00:00 2001
From: clin99 <artiferlly@gmail.com>
Date: Fri, 1 Mar 2019 21:57:15 -0600
Subject: [PATCH 01/12] Add execution model

---
 example/pipeline.cpp              |  60 +++++++
 taskflow/graph/basic_taskflow.hpp | 267 +++++++++++++++++++++++++++++-
 taskflow/graph/graph.hpp          |  17 +-
 taskflow/graph/topology.hpp       |   3 +
 4 files changed, 343 insertions(+), 4 deletions(-)
 create mode 100644 example/pipeline.cpp

diff --git a/example/pipeline.cpp b/example/pipeline.cpp
new file mode 100644
index 000000000..119acdc78
--- /dev/null
+++ b/example/pipeline.cpp
@@ -0,0 +1,60 @@
+// A simple example to capture the following task dependencies.
+//
+// TaskA---->TaskB---->TaskD
+// TaskA---->TaskC---->TaskD
+
+#include <taskflow/taskflow.hpp>  // the only include you need
+
+int main(){
+
+  tf::Taskflow tf;
+  tf::Framework f;
+
+  std::mutex mtx;
+  
+  auto [A, B, C] = f.emplace( 
+    [&] () { 
+      std::scoped_lock lock(mtx);
+      std::puts("TaskA");
+    },     
+    [&] (auto &subflow) { 
+      std::scoped_lock lock(mtx);
+      std::puts("TaskB");
+      subflow.emplace([&](){
+       std::scoped_lock lock(mtx);
+       std::puts("TaskB1");
+      });
+      subflow.emplace([&](){
+       std::scoped_lock lock(mtx);
+       std::puts("TaskB2");
+      });
+      subflow.detach();
+    },     
+    [&] () { 
+      std::scoped_lock lock(mtx);
+      std::puts("TaskC");
+    }
+  ); 
+
+  A.name("A");
+  B.name("B");
+  C.name("C");
+   
+  // Linear   
+  A.precede(B);  
+  B.precede(C);  
+
+
+  tf.pipeline_until(f, [iter = 3]() mutable { 
+    std::cout << "iter = " << iter << std::endl;
+    return iter -- == 0; }, [](){}
+  ).get();
+
+  //std::cout << f.dump() << std::endl;
+  //tf.wait_for_all();  // block until finished
+
+  return 0;
+}
+
+
+
diff --git a/taskflow/graph/basic_taskflow.hpp b/taskflow/graph/basic_taskflow.hpp
index 2be4ad8b3..e46aca0fc 100644
--- a/taskflow/graph/basic_taskflow.hpp
+++ b/taskflow/graph/basic_taskflow.hpp
@@ -49,7 +49,12 @@ class BasicTaskflow : public FlowBuilder {
 
     Closure& operator = (const Closure&) = default;
     
-    void operator ()() const;
+    void operator ()() ;
+
+    void normal_mode() ;
+    void pipeline_mode() ;
+
+    bool execute_pipeline_task(Graph&);
 
     BasicTaskflow* taskflow {nullptr};
     Node*          node     {nullptr};
@@ -237,6 +242,11 @@ class BasicTaskflow : public FlowBuilder {
     template<typename P, typename C>
     std::shared_future<void> run_until(Framework& framework, P&& predicate, C&& callable);
 
+
+    template<typename P, typename C>
+    std::shared_future<void> pipeline_until(Framework& framework, P&& predicate, C&& callable);
+
+
   private:
     
     Graph _graph;
@@ -247,6 +257,7 @@ class BasicTaskflow : public FlowBuilder {
 
     void _schedule(Node&);
     void _schedule(PassiveVector<Node*>&);
+
 };
 
 // ============================================================================
@@ -374,6 +385,65 @@ std::shared_future<void> BasicTaskflow<E>::run_until(Framework& f, P&& predicate
   return tpg._future;
 }
 
+
+// Function: pipeline_until
+template <template <typename...> typename E>
+template <typename P, typename C>
+std::shared_future<void> BasicTaskflow<E>::pipeline_until(Framework& f, P&& predicate, C&& c) {
+  if(std::invoke(predicate)) {
+    return std::async(std::launch::deferred, [](){}).share();
+  }
+
+  // create a topology for this run
+  auto &tpg = _topologies.emplace_back(f, std::forward<P>(predicate));
+
+  // Iterative execution to avoid stack overflow
+  if(num_workers() == 0) {
+
+    // Clear last execution data & Build precedence between nodes and target
+    tpg._bind(f._graph);
+
+    do {
+      _schedule(tpg._sources);
+      tpg._recover_num_sinks();
+    } while(!std::invoke(tpg._predicate));
+
+    std::invoke(c);
+    tpg._promise.set_value();
+
+    return tpg._future;  
+  }
+
+  // Multi-threaded execution.
+  std::scoped_lock lock(f._mtx);
+
+  f._topologies.push_back(&tpg);
+
+  tf::Node* predicate_node = new Node();
+  predicate_node->_topology = &tpg;
+  predicate_node->_num_dependents = -1;
+  tpg._work = [predicate_node, &f, c{std::move(c)}]() {
+    printf("Leave\n");
+    delete predicate_node;
+    std::invoke(c);
+    auto tpg = f._topologies.front();
+    f._topologies.pop_front();
+    tpg->_promise.set_value();
+  };
+
+  // Clear last execution data & Build precedence between nodes and target
+  tpg._bind(f._graph);
+  //for(auto &s: tpg._sources) {
+  //  predicate_node->precede(*s);
+  //}
+  predicate_node->set_pipeline();
+
+  _schedule(*predicate_node);
+
+  return tpg._future;
+}
+
+
 // Constructor
 template <template <typename...> typename E>
 BasicTaskflow<E>::Closure::Closure(BasicTaskflow& t, Node& n) : 
@@ -382,7 +452,134 @@ BasicTaskflow<E>::Closure::Closure(BasicTaskflow& t, Node& n) :
 
 // Operator ()
 template <template <typename...> typename E>
-void BasicTaskflow<E>::Closure::operator () () const {
+void BasicTaskflow<E>::Closure::operator () () {
+   if(node->is_pipeline()) {
+    pipeline_mode();
+  }
+  else {
+    normal_mode();
+  }
+}
+
+// Pipeline mode
+template <template <typename...> typename E>
+void BasicTaskflow<E>::Closure::pipeline_mode() {
+  bool last_pipe {false};
+  const auto num_successors = node->num_successors();
+ 
+  do {
+    if(node->_num_dependents == -1) {
+      if(node->_cur_pipeline == node->_topology->_num_pipeline) {
+        return ;
+      }
+      else {
+        auto stop = std::invoke(node->_topology->_predicate); 
+        if(!stop) {
+          node->_topology->_num_pipeline ++;
+        }
+
+        for(auto &s: node->_topology->_sources) {
+          //if(--s->_num_dependents == 0) {
+          //  s->_num_dependents = s->_dependents.size(); 
+            if(s->_num_run.fetch_add(1) == 0) {
+              s->set_pipeline();
+              s->_num_dependents = 0;
+              taskflow->_schedule(*s);
+            }
+          //}
+        }                                          
+        node->_cur_pipeline ++;
+        //node->_num_run = 10;
+        continue;
+      }
+    }
+    else {
+
+      node->_num_dependents = 0;
+
+      //if(node->_subgraph.has_value()) {
+      //  //node->_subgraph.reset();
+      //  node->unset_spawned();
+      //}
+
+      if(!node->is_spawned()) {
+        Graph subgraph;
+        if(node->_subgraph.has_value() && !node->_subgraph->empty()) {
+          subgraph.splice(subgraph.begin(), node->_subgraph.value());
+          node->_subgraph.reset();
+        }
+        //if(auto joined = execute_pipeline_node(); joined) {
+        //std::cout << "To execute node ==========  " << node->_name << std::endl;
+        //if(auto joined = execute_node(); joined) {
+        if(auto joined = execute_pipeline_task(subgraph); joined) {
+          std::puts("Join mode");
+          // Recover must be done before scheduling the subgraph, 
+          // otherwise u will have race condition 
+
+          ////node->_round.fetch_sub(1);
+          //if(!backup.empty()) {
+          //  if(!node->_subgraph.has_value()) {
+          //    node->_subgraph.emplace();
+          //  }
+          //  node->_subgraph->splice(node->_subgraph->begin(), backup);
+          //}
+          ////assert(node->_round > 0);
+          return ;
+        }
+        if(!subgraph.empty()) {
+          if(!node->_subgraph.has_value()) {
+            node->_subgraph.emplace();
+          }
+          node->_subgraph.value().splice(node->_subgraph.value().begin(), subgraph);
+        }
+      }
+
+      node->unset_spawned();
+
+      if(++node->_cur_pipeline == node->_topology->_num_pipeline) {
+        last_pipe = true;
+        node->_num_dependents = node->_dependents.size();
+        node->_cur_pipeline = 0;
+        node->_num_run = 0;
+        node->unset_pipeline();
+      }
+
+      for(size_t i=0; i<node->num_successors(); ++i) {
+        //if(--(node->_successors[i]->_num_dependents) == 0) {
+        //  node->_successors[i]->_num_dependents = node->_successors[i]->_dependents.size();
+          if(node->_successors[i]->_num_run.fetch_add(1) == 0) {
+            //std::cout << "Schedule " << node->_successors[i]->_name << std::endl; 
+            node->_successors[i]->set_pipeline();
+            node->_successors[i]->_num_dependents = 0;
+            taskflow->_schedule(*(node->_successors[i]));
+          }
+        //}
+      }
+      //if(node->num_successors() == 0) {
+      if(num_successors == 0) {
+        //if(++node->_cur == node->_topology->_num_run) {
+        if(last_pipe) {
+         //if(node->num_successors() == 0) {
+          if(--(node->_topology->_num_sinks) == 0) {
+            if(node->_topology->_work != nullptr) {
+              printf("Final set value\n");
+              //std::puts("Final set value");
+              std::invoke(node->_topology->_work);
+            }
+          }
+          //no_check = true;
+        }
+      }
+
+    }  // End of else
+
+  } while(!last_pipe && node->_num_run.fetch_sub(1) != 1);
+}
+
+
+// Normal mode
+template <template <typename...> typename E>
+void BasicTaskflow<E>::Closure::normal_mode() {
 
   // Here we need to fetch the num_successors first to avoid the invalid memory
   // access caused by topology clear.
@@ -477,6 +674,72 @@ void BasicTaskflow<E>::Closure::operator () () const {
   }
 }
 
+
+template <template <typename...> typename E>
+bool BasicTaskflow<E>::Closure::execute_pipeline_task(Graph& subgraph) {
+  // regular node type
+  // The default node work type. We only need to execute the callback if any.
+  if(auto index=node->_work.index(); index == 0) {
+    if(auto &f = std::get<StaticWork>(node->_work); f != nullptr){
+      std::invoke(f);
+    }
+  }
+  // subflow node type 
+  else {
+    
+    // Clear the subgraph before the task execution
+    if(!node->is_spawned()) {
+      node->_subgraph.emplace();
+    }
+   
+    SubflowBuilder fb(*(node->_subgraph));
+
+    std::invoke(std::get<DynamicWork>(node->_work), fb);
+    
+    // Need to create a subflow if first time & subgraph is not empty 
+    if(!node->is_spawned()) {
+      node->set_spawned();
+      if(!node->_subgraph->empty()) {
+        // For storing the source nodes
+        PassiveVector<Node*> src; 
+        for(auto& n : *(node->_subgraph)) {
+          n._topology = node->_topology;
+          n.set_subtask();
+          if(n.num_successors() == 0) {
+            if(fb.detached()) {
+              node->_topology->_num_sinks ++;
+            }
+            else {
+              n.precede(*node);
+            }
+          }
+          if(n.num_dependents() == 0) {
+            src.push_back(&n);
+          }
+        }
+
+        if(!fb.detached()) {
+          if(!subgraph.empty()) {
+            if(!node->_subgraph.has_value()) {
+              node->_subgraph.emplace();
+            }
+            node->_subgraph->splice(node->_subgraph->begin(), subgraph);
+          }
+        }
+
+        taskflow->_schedule(src);
+
+        if(!fb.detached()) {
+          // Return true only when joining with parent
+          return true;
+        }
+      }
+    }
+  } // End of DynamicWork -----------------------------------------------------
+  
+  return false;
+}
+
 // ============================================================================
 // BasicTaskflow Method Definitions
 // ============================================================================
diff --git a/taskflow/graph/graph.hpp b/taskflow/graph/graph.hpp
index 2441e3633..0286303dc 100644
--- a/taskflow/graph/graph.hpp
+++ b/taskflow/graph/graph.hpp
@@ -35,6 +35,7 @@ class Node {
 
   constexpr static int SPAWNED = 0x1;
   constexpr static int SUBTASK = 0x2;
+  constexpr static int PIPELINE = 0x4;
 
   public:
 
@@ -57,8 +58,16 @@ class Node {
     // Status-related functions
     bool is_spawned() const { return _status & SPAWNED; }
     bool is_subtask() const { return _status & SUBTASK; }
-    void set_spawned()  { _status |= SPAWNED; }
-    void set_subtask()  { _status |= SUBTASK; }
+    bool is_pipeline() const { return _status & PIPELINE; }
+
+    void set_spawned()   { _status |= SPAWNED; }
+    void set_subtask()   { _status |= SUBTASK; }
+    void set_pipeline()  { _status |= PIPELINE; }
+
+    void unset_spawned()   { _status &= ~SPAWNED; }
+    void unset_subtask()   { _status &= ~SUBTASK; }
+    void unset_pipeline()  { _status &= ~PIPELINE; }
+
     void clear_status() { _status = 0; }
 
   private:
@@ -76,6 +85,10 @@ class Node {
     Topology* _topology;
 
     int _status {0};
+
+    // Pipeline 
+    std::atomic<unsigned> _num_run {0};
+    unsigned _cur_pipeline {0};
 };
 
 // Constructor
diff --git a/taskflow/graph/topology.hpp b/taskflow/graph/topology.hpp
index 8a4e09d1d..7775422e0 100644
--- a/taskflow/graph/topology.hpp
+++ b/taskflow/graph/topology.hpp
@@ -53,6 +53,9 @@ class Topology {
 
     void _bind(Graph& g);
     void _recover_num_sinks();
+
+    // Pipeline
+    std::atomic<unsigned> _num_pipeline {1};
 };
 
 // Constructor

From 6b25c875926b46d6034ae25ed4eeb65b29347115 Mon Sep 17 00:00:00 2001
From: clin99 <artiferlly@gmail.com>
Date: Fri, 1 Mar 2019 23:11:07 -0600
Subject: [PATCH 02/12] refactor

---
 taskflow/graph/basic_taskflow.hpp | 81 ++++++++++++++++++-------------
 1 file changed, 47 insertions(+), 34 deletions(-)

diff --git a/taskflow/graph/basic_taskflow.hpp b/taskflow/graph/basic_taskflow.hpp
index e46aca0fc..5b579eeac 100644
--- a/taskflow/graph/basic_taskflow.hpp
+++ b/taskflow/graph/basic_taskflow.hpp
@@ -419,12 +419,13 @@ std::shared_future<void> BasicTaskflow<E>::pipeline_until(Framework& f, P&& pred
 
   f._topologies.push_back(&tpg);
 
-  tf::Node* predicate_node = new Node();
-  predicate_node->_topology = &tpg;
-  predicate_node->_num_dependents = -1;
-  tpg._work = [predicate_node, &f, c{std::move(c)}]() {
+  //tf::Node* predicate_node = new Node();
+  //predicate_node->_topology = &tpg;
+  //predicate_node->_num_dependents = -1;
+  //tpg._work = [predicate_node, &f, c{std::move(c)}]() {
+  tpg._work = [&f, c{std::move(c)}]() {
     printf("Leave\n");
-    delete predicate_node;
+    //delete predicate_node;
     std::invoke(c);
     auto tpg = f._topologies.front();
     f._topologies.pop_front();
@@ -436,9 +437,12 @@ std::shared_future<void> BasicTaskflow<E>::pipeline_until(Framework& f, P&& pred
   //for(auto &s: tpg._sources) {
   //  predicate_node->precede(*s);
   //}
-  predicate_node->set_pipeline();
+  //predicate_node->set_pipeline();
 
-  _schedule(*predicate_node);
+  //_schedule(*predicate_node);
+  
+  tpg._sources.front()->set_pipeline();
+  _schedule(*(tpg._sources.front()));
 
   return tpg._future;
 }
@@ -468,33 +472,35 @@ void BasicTaskflow<E>::Closure::pipeline_mode() {
   const auto num_successors = node->num_successors();
  
   do {
-    if(node->_num_dependents == -1) {
-      if(node->_cur_pipeline == node->_topology->_num_pipeline) {
-        return ;
-      }
-      else {
-        auto stop = std::invoke(node->_topology->_predicate); 
-        if(!stop) {
-          node->_topology->_num_pipeline ++;
-        }
-
-        for(auto &s: node->_topology->_sources) {
-          //if(--s->_num_dependents == 0) {
-          //  s->_num_dependents = s->_dependents.size(); 
-            if(s->_num_run.fetch_add(1) == 0) {
-              s->set_pipeline();
-              s->_num_dependents = 0;
-              taskflow->_schedule(*s);
-            }
-          //}
-        }                                          
-        node->_cur_pipeline ++;
-        //node->_num_run = 10;
-        continue;
-      }
-    }
-    else {
+    ////if(node->_num_dependents == -1) {
+    //if(node->_dependents.size() == 0) {
+    //  if(node->_cur_pipeline == node->_topology->_num_pipeline) {
+    //    return ;
+    //  }
+    //  else {
+    //    auto stop = std::invoke(node->_topology->_predicate); 
+    //    if(!stop) {
+    //      node->_topology->_num_pipeline ++;
+    //    }
+
+    //    for(auto &s: node->_topology->_sources) {
+    //      //if(--s->_num_dependents == 0) {
+    //      //  s->_num_dependents = s->_dependents.size(); 
+    //        if(s->_num_run.fetch_add(1) == 0) {
+    //          s->set_pipeline();
+    //          s->_num_dependents = 0;
+    //          taskflow->_schedule(*s);
+    //        }
+    //      //}
+    //    }                                          
+    //    node->_cur_pipeline ++;
+    //    //node->_num_run = 10;
+    //    continue;
+    //  }
+    //}
+    //else {
 
+      
       node->_num_dependents = 0;
 
       //if(node->_subgraph.has_value()) {
@@ -503,6 +509,13 @@ void BasicTaskflow<E>::Closure::pipeline_mode() {
       //}
 
       if(!node->is_spawned()) {
+        if(node->_dependents.size() == 0) {
+          auto stop = std::invoke(node->_topology->_predicate); 
+          if(!stop) {
+            node->_topology->_num_pipeline ++;
+          }
+        }
+
         Graph subgraph;
         if(node->_subgraph.has_value() && !node->_subgraph->empty()) {
           subgraph.splice(subgraph.begin(), node->_subgraph.value());
@@ -571,7 +584,7 @@ void BasicTaskflow<E>::Closure::pipeline_mode() {
         }
       }
 
-    }  // End of else
+    //}  // End of else
 
   } while(!last_pipe && node->_num_run.fetch_sub(1) != 1);
 }

From 6394c2fa26ae86cfc62714737a3818a8e6817d3b Mon Sep 17 00:00:00 2001
From: clin99 <artiferlly@gmail.com>
Date: Sat, 2 Mar 2019 10:20:25 -0600
Subject: [PATCH 03/12] update

---
 example/long_pipeline.cpp         | 49 ++++++++++++++++++++++++++++++
 taskflow/graph/basic_taskflow.hpp | 50 +++++++++++++++++++++++++------
 2 files changed, 90 insertions(+), 9 deletions(-)
 create mode 100644 example/long_pipeline.cpp

diff --git a/example/long_pipeline.cpp b/example/long_pipeline.cpp
new file mode 100644
index 000000000..33842a8aa
--- /dev/null
+++ b/example/long_pipeline.cpp
@@ -0,0 +1,49 @@
+// A simple example to capture the following task dependencies.
+//
+// TaskA---->TaskB---->TaskD
+// TaskA---->TaskC---->TaskD
+
+#include <taskflow/taskflow.hpp>  // the only include you need
+
+int main(){
+
+  tf::Taskflow tf;
+  tf::Framework f;
+
+  constexpr size_t pipeline_length {1000000};
+  std::vector<tf::Task> tasks;
+  int counter {1};
+  
+  for(size_t i=0; i<pipeline_length; i++) {
+    tasks.emplace_back(
+      f.emplace( [&, i=i+1]() {
+          assert(counter % pipeline_length == i % pipeline_length);
+          counter ++;
+        }
+      )
+    );
+  }
+
+  for(size_t i=0; i<pipeline_length-1; i++) {
+    tasks[i].precede(tasks[i+1]);
+  }
+
+  auto fu1 = tf.pipeline_until(f, [iter = 1]() mutable { 
+    return iter -- == 0; }, [&](){ assert(counter == pipeline_length+1); counter = 1; }
+  );
+
+  auto fu2 = tf.pipeline_until(f, [iter = 1]() mutable { 
+    return iter -- == 0; }, [&](){ assert(counter == pipeline_length+1); counter = 1; }
+  );
+
+  //fu1.get();
+  fu2.get();
+
+  //std::cout << f.dump() << std::endl;
+  //tf.wait_for_all();  // block until finished
+
+  return 0;
+}
+
+
+
diff --git a/taskflow/graph/basic_taskflow.hpp b/taskflow/graph/basic_taskflow.hpp
index 5b579eeac..7bc36133c 100644
--- a/taskflow/graph/basic_taskflow.hpp
+++ b/taskflow/graph/basic_taskflow.hpp
@@ -419,30 +419,62 @@ std::shared_future<void> BasicTaskflow<E>::pipeline_until(Framework& f, P&& pred
 
   f._topologies.push_back(&tpg);
 
+  bool run_now = (f._topologies.size() == 1);
+  if(run_now) {
+    // Clear last execution data & Build precedence between nodes and target
+    tpg._bind(f._graph);
+    assert(tpg._sources.size() == 1);
+  }
+
   //tf::Node* predicate_node = new Node();
   //predicate_node->_topology = &tpg;
   //predicate_node->_num_dependents = -1;
   //tpg._work = [predicate_node, &f, c{std::move(c)}]() {
-  tpg._work = [&f, c{std::move(c)}]() {
+  tpg._work = [&, c = std::forward<C>(c)]() mutable {
     printf("Leave\n");
     //delete predicate_node;
     std::invoke(c);
-    auto tpg = f._topologies.front();
-    f._topologies.pop_front();
-    tpg->_promise.set_value();
+
+    f._mtx.lock();
+    //auto tpg = f._topologies.front();
+    //f._topologies.pop_front();
+    //tpg->_promise.set_value();
+
+    // If there is another run (interleave between lock)
+    if(f._topologies.size() > 1) {
+      // Set the promise
+      f._topologies.front()->_promise.set_value();
+      f._topologies.pop_front();
+      f._topologies.front()->_bind(f._graph);
+      f._mtx.unlock();
+      f._topologies.front()->_sources.front()->set_pipeline();
+      _schedule(f._topologies.front()->_sources);
+    }
+    else {
+      assert(f._topologies.size() == 1);
+      // Need to back up the promise first here becuz framework might be 
+      // destroy before taskflow leaves
+      auto &p = f._topologies.front()->_promise; 
+      f._topologies.pop_front();
+      f._mtx.unlock();
+     
+      // We set the promise in the end in case framework leaves before taskflow
+      p.set_value();
+    }
+
   };
 
-  // Clear last execution data & Build precedence between nodes and target
-  tpg._bind(f._graph);
   //for(auto &s: tpg._sources) {
   //  predicate_node->precede(*s);
   //}
   //predicate_node->set_pipeline();
 
   //_schedule(*predicate_node);
-  
-  tpg._sources.front()->set_pipeline();
-  _schedule(*(tpg._sources.front()));
+ 
+  if(run_now) { 
+    tpg._sources.front()->set_pipeline();
+    _schedule(*(tpg._sources.front()));
+  }
 
   return tpg._future;
 }

From 351fba9fc2319eac697d797af7712308c94deb60 Mon Sep 17 00:00:00 2001
From: clin99 <artiferlly@gmail.com>
Date: Sat, 2 Mar 2019 13:37:53 -0600
Subject: [PATCH 04/12] update

---
 taskflow/graph/basic_taskflow.hpp | 113 ++++++++++++++++++++++++++++++
 taskflow/graph/framework.hpp      |  97 ++++++++++++++++++++++++-
 taskflow/graph/graph.hpp          |   4 ++
 taskflow/graph/topology.hpp       |  20 +++++-
 4 files changed, 232 insertions(+), 2 deletions(-)

diff --git a/taskflow/graph/basic_taskflow.hpp b/taskflow/graph/basic_taskflow.hpp
index 7bc36133c..fcc0f29e7 100644
--- a/taskflow/graph/basic_taskflow.hpp
+++ b/taskflow/graph/basic_taskflow.hpp
@@ -247,6 +247,10 @@ class BasicTaskflow : public FlowBuilder {
     std::shared_future<void> pipeline_until(Framework& framework, P&& predicate, C&& callable);
 
 
+    template<typename P, typename C>
+    std::shared_future<void> run_until(WorkGroup& workgroup, P&& predicate, C&& callable);
+
+
   private:
     
     Graph _graph;
@@ -386,6 +390,115 @@ std::shared_future<void> BasicTaskflow<E>::run_until(Framework& f, P&& predicate
 }
 
 
+
+// Function: run_until
+template <template <typename...> typename E>
+template <typename P, typename C>
+std::shared_future<void> BasicTaskflow<E>::run_until(WorkGroup& wg, P&& predicate, C&& c) {
+  if(std::invoke(predicate)) {
+    return std::async(std::launch::deferred, [](){}).share();
+  }
+
+  // create a topology for this run
+  auto &tpg = _topologies.emplace_back(wg, std::forward<P>(predicate));
+
+
+  // Multi-threaded execution.
+  std::scoped_lock lock(wg._mtx);
+
+  wg._topologies.push_back(&tpg);
+
+  bool run_now = (wg._topologies.size() == 1);
+
+  if(run_now) {
+    tpg._bind(wg._graph);
+  }
+
+  for(auto &p: wg._pairs) {
+    std::get<Node*>(p)->_work = [&, tf=this](auto& subflow) mutable {
+      
+      std::cout << std::get<1>(p)->name() << std::endl;
+
+      auto sink = subflow.placeholder();
+
+      PassiveVector<Node*> src;
+      PassiveVector<Node*> tgt;
+
+      for(auto &n: std::get<tf::Framework*>(p)->_graph) {
+        n._topology = &tpg;
+        if(n.num_dependents() == 0) {
+          src.push_back(&n);
+        }
+        if(n.num_successors() == 0) {
+          n.precede(*(sink._node));
+          tgt.push_back(&n);
+        }
+      }
+
+      sink.work([tgt{std::move(tgt)}](){
+        std::puts("=======>  Clear");
+        //for(auto& t: tgt) {
+        //  t->_successors.clear();
+        //}
+      });
+      subflow.join();
+
+      //subflow.emplace(
+      //  [tf=tf, src{std::move(src)}]() {
+      //    //tf->_schedule(src);
+      //  }
+      //).precede(sink);
+
+      tf->_schedule(src);
+    };
+  }
+
+  tpg._work = [&wg, c=std::forward<C>(c), this] () mutable {
+      
+    // case 1: we still need to run the topology again
+    if(!std::invoke(wg._topologies.front()->_predicate)) {
+      wg._topologies.front()->_recover_num_sinks();
+      _schedule(wg._topologies.front()->_sources); 
+    }
+    // case 2: the final run of this topology
+    else {
+      std::invoke(c);
+
+      wg._mtx.lock();
+
+      // If there is another run (interleave between lock)
+      if(wg._topologies.size() > 1) {
+
+        // Set the promise
+        wg._topologies.front()->_promise.set_value();
+        wg._topologies.pop_front();
+        wg._topologies.front()->_bind(wg._graph);
+        wg._mtx.unlock();
+        _schedule(wg._topologies.front()->_sources);
+      }
+      else {
+        assert(wg._topologies.size() == 1);
+        // Need to back up the promise first here becuz framework might be 
+        // destroy before taskflow leaves
+        auto &p = wg._topologies.front()->_promise; 
+        wg._topologies.pop_front();
+        wg._mtx.unlock();
+       
+        // We set the promise in the end in case framework leaves before taskflow
+        p.set_value();
+      }
+    }
+  };
+
+  if(run_now) {
+    _schedule(tpg._sources);
+  }
+
+  return tpg._future;
+}
+
+
+
 // Function: pipeline_until
 template <template <typename...> typename E>
 template <typename P, typename C>
diff --git a/taskflow/graph/framework.hpp b/taskflow/graph/framework.hpp
index 1c6349ffd..6e3cb74f3 100644
--- a/taskflow/graph/framework.hpp
+++ b/taskflow/graph/framework.hpp
@@ -4,6 +4,8 @@
 
 namespace tf {
 
+class WorkGroup;
+
 /**
 @class Framework 
 
@@ -20,6 +22,7 @@ class Framework : public FlowBuilder {
   friend class BasicTaskflow;
   
   friend class Topology;
+  friend class WorkGroup;
 
   public:
 
@@ -50,7 +53,13 @@ class Framework : public FlowBuilder {
     */
     size_t num_nodes() const;
 
+    auto& name(const std::string&) ;
+
+    const std::string& name() const ;
+
   private:
+
+    std::string _name;
     
     Graph _graph;
 
@@ -67,6 +76,19 @@ inline Framework::~Framework() {
   assert(_topologies.empty());
 }
 
+// Function: name
+inline auto& Framework::name(const std::string &name) {
+  _name = name;
+  return *this;
+}
+
+// Function: name
+inline const std::string& Framework::name() const {
+  return _name;
+}
+
+
+
 // Function: num_noces
 inline size_t Framework::num_nodes() const {
   return _graph.size();
@@ -74,7 +96,7 @@ inline size_t Framework::num_nodes() const {
 
 // Procedure: dump
 inline void Framework::dump(std::ostream& os) const {
-  os << "digraph Framework {\n";
+  os << "digraph " << _name << " Framework {\n";
   for(const auto& n: _graph) {
     n.dump(os);
   }
@@ -88,5 +110,78 @@ inline std::string Framework::dump() const {
   return os.str();
 }
 
+
+
+class WorkGroup : public FlowBuilder {
+
+  friend class Topology;
+  
+  template <template<typename...> typename E>  
+  friend class BasicTaskflow;
+
+  public:
+
+    WorkGroup();
+
+    virtual ~WorkGroup();
+
+    using FlowBuilder::silent_emplace;
+    using FlowBuilder::emplace;
+    tf::Task emplace(Framework& framework);
+
+    void dump(std::ostream& ostream) const;
+    
+    std::string dump() const;
+
+
+  protected:
+
+    Graph _graph;
+
+  private:
+
+    std::vector<std::pair<Node*, Framework*>> _pairs;
+
+    std::mutex _mtx;
+    std::list<Topology*> _topologies;
+    Node* _last_target {nullptr};   
+
+    size_t _now_iteration;
+};
+
+// Constructor
+inline WorkGroup::WorkGroup() : FlowBuilder{_graph} {
+}
+
+// Destructor
+inline WorkGroup::~WorkGroup() {
+  assert(_topologies.empty());
+}
+
+
+inline tf::Task WorkGroup::emplace(Framework& framework) {
+  auto task = placeholder();
+  _graph.back().set_workgroup();
+  _pairs.emplace_back(&(_graph.back()), &framework);
+  return task;
+}
+
+// Procedure: dump
+inline void WorkGroup::dump(std::ostream& os) const {
+  os << "digraph WorkGroup {\n";
+  for(const auto& n: _graph) {
+    n.dump(os);
+  }
+  os << "}\n";
+}
+
+// Function: dump
+inline std::string WorkGroup::dump() const { 
+  std::ostringstream os;
+  dump(os);
+  return os.str();
+}
+
+
 }  // end of namespace tf. ---------------------------------------------------
 
diff --git a/taskflow/graph/graph.hpp b/taskflow/graph/graph.hpp
index 0286303dc..d965031bf 100644
--- a/taskflow/graph/graph.hpp
+++ b/taskflow/graph/graph.hpp
@@ -36,6 +36,7 @@ class Node {
   constexpr static int SPAWNED = 0x1;
   constexpr static int SUBTASK = 0x2;
   constexpr static int PIPELINE = 0x4;
+  constexpr static int WORKGROUP = 0x8;
 
   public:
 
@@ -59,14 +60,17 @@ class Node {
     bool is_spawned() const { return _status & SPAWNED; }
     bool is_subtask() const { return _status & SUBTASK; }
     bool is_pipeline() const { return _status & PIPELINE; }
+    bool is_workgroup() const { return _status & WORKGROUP; }
 
     void set_spawned()   { _status |= SPAWNED; }
     void set_subtask()   { _status |= SUBTASK; }
     void set_pipeline()  { _status |= PIPELINE; }
+    void set_workgroup()  { _status |= WORKGROUP; }
 
     void unset_spawned()   { _status &= ~SPAWNED; }
     void unset_subtask()   { _status &= ~SUBTASK; }
     void unset_pipeline()  { _status &= ~PIPELINE; }
+    void unset_workgroup()  { _status &= ~WORKGROUP; }
 
     void clear_status() { _status = 0; }
 
diff --git a/taskflow/graph/topology.hpp b/taskflow/graph/topology.hpp
index 7775422e0..005932d20 100644
--- a/taskflow/graph/topology.hpp
+++ b/taskflow/graph/topology.hpp
@@ -13,6 +13,7 @@ class Topology {
   friend class BasicTaskflow; 
 
   friend class Framework;
+  friend class WorkGroup;
   
   // TODO: make graph/framework handle uniform
   //struct GraphHandle {
@@ -34,12 +35,15 @@ class Topology {
     template <typename P>
     Topology(Framework&, P&&);
 
+    template <typename P>
+    Topology(WorkGroup&, P&&);
+
     std::string dump() const;
     void dump(std::ostream&) const;
 
   private:
 
-    std::variant<Graph, Framework*> _handle;
+    std::variant<Graph, Framework*, WorkGroup*> _handle;
 
     std::promise<void> _promise;
     std::shared_future<void> _future {_promise.get_future().share()};
@@ -58,6 +62,15 @@ class Topology {
     std::atomic<unsigned> _num_pipeline {1};
 };
 
+
+// Constructor
+template <typename P>
+inline Topology::Topology(WorkGroup& w, P&& p): 
+  _handle    {&w}, 
+  _predicate {std::forward<P>(p)} {
+}
+
+
 // Constructor
 template <typename P>
 inline Topology::Topology(Framework& f, P&& p): 
@@ -125,6 +138,11 @@ inline void Topology::dump(std::ostream& os) const {
       for(const auto& node : framework->_graph) {
         node.dump(os);
       }
+    },
+    [&] (const WorkGroup* workgroup) {
+      for(const auto& node : workgroup->_graph) {
+        node.dump(os);
+      }
     }
   }, _handle);
 

From e6b413fad695e71dc2c702c28ea4f19a25336430 Mon Sep 17 00:00:00 2001
From: clin99 <artiferlly@gmail.com>
Date: Sat, 2 Mar 2019 15:25:49 -0600
Subject: [PATCH 05/12] update

---
 example/workgroup.cpp             | 46 +++++++++++++++++++++++++++++++
 taskflow/graph/basic_taskflow.hpp | 17 ++++++------
 taskflow/graph/flow_builder.hpp   |  2 ++
 3 files changed, 57 insertions(+), 8 deletions(-)
 create mode 100644 example/workgroup.cpp

diff --git a/example/workgroup.cpp b/example/workgroup.cpp
new file mode 100644
index 000000000..1a8cedb83
--- /dev/null
+++ b/example/workgroup.cpp
@@ -0,0 +1,46 @@
+// A simple example to capture the following task dependencies.
+//
+// TaskA---->TaskB---->TaskD
+// TaskA---->TaskC---->TaskD
+
+#include <taskflow/taskflow.hpp>  // the only include you need
+
+int main(){
+
+  tf::Taskflow tf;
+
+  tf::Framework f1;
+  f1.name("F1")
+    .emplace(
+    [&](){ std::puts("TaskA"); },
+    [&](){ std::puts("TaskB"); },
+    [&](){ std::puts("TaskC"); }
+  );
+
+  tf::Framework f2;
+  f2.name("F2")
+    .emplace(
+    [&](){ std::puts("  TaskD"); },
+    [&](auto &subflow){ 
+      std::puts("  TaskE"); 
+      auto [E1, E2] = subflow.emplace(
+        [](){ std::puts("    Task E1"); },
+        [](){ std::puts("    Task E2"); }
+      );
+    }
+  );
+
+
+  tf::WorkGroup wg;
+  auto F1 = wg.emplace(f1);
+  auto F2 = wg.emplace(f2);
+  F1.precede(F2);
+  wg.emplace([](){ std::cout << "Glue task\n"; });
+
+  tf.run_until(wg, [iter = 4] () mutable { std::puts("\n"); return iter-- == 0; }, [](){}).get();
+
+  return 0;
+}
+
+
+
diff --git a/taskflow/graph/basic_taskflow.hpp b/taskflow/graph/basic_taskflow.hpp
index fcc0f29e7..d4ec3c579 100644
--- a/taskflow/graph/basic_taskflow.hpp
+++ b/taskflow/graph/basic_taskflow.hpp
@@ -416,8 +416,9 @@ std::shared_future<void> BasicTaskflow<E>::run_until(WorkGroup& wg, P&& predicat
 
   for(auto &p: wg._pairs) {
     std::get<Node*>(p)->_work = [&, tf=this](auto& subflow) mutable {
+      if(!subflow.empty()) return;
       
-      std::cout << std::get<1>(p)->name() << std::endl;
+      //std::cout << "Framework name: " << std::get<1>(p)->name() << std::endl;
 
       auto sink = subflow.placeholder();
 
@@ -436,10 +437,10 @@ std::shared_future<void> BasicTaskflow<E>::run_until(WorkGroup& wg, P&& predicat
       }
 
       sink.work([tgt{std::move(tgt)}](){
-        std::puts("=======>  Clear");
-        //for(auto& t: tgt) {
-        //  t->_successors.clear();
-        //}
+        //std::puts("=======>  Clear");
+        for(auto& t: tgt) {
+          t->_successors.clear();
+        }
       });
       subflow.join();
 
@@ -468,7 +469,6 @@ std::shared_future<void> BasicTaskflow<E>::run_until(WorkGroup& wg, P&& predicat
 
       // If there is another run (interleave between lock)
       if(wg._topologies.size() > 1) {
-
         // Set the promise
         wg._topologies.front()->_promise.set_value();
         wg._topologies.pop_front();
@@ -602,7 +602,8 @@ BasicTaskflow<E>::Closure::Closure(BasicTaskflow& t, Node& n) :
 // Operator ()
 template <template <typename...> typename E>
 void BasicTaskflow<E>::Closure::operator () () {
-   if(node->is_pipeline()) {
+  if(node->is_pipeline()) {
+    assert(false);
     pipeline_mode();
   }
   else {
@@ -821,7 +822,7 @@ void BasicTaskflow<E>::Closure::normal_mode() {
     if(--(node->_topology->_num_sinks) == 0) {
 
       // This is the last executing node 
-      bool is_framework = node->_topology->_handle.index() == 1;
+      bool is_framework = node->_topology->_handle.index() != 0;
       if(node->_topology->_work != nullptr) {
         std::invoke(node->_topology->_work);
       }
diff --git a/taskflow/graph/flow_builder.hpp b/taskflow/graph/flow_builder.hpp
index 1236aec2a..8af84cfc2 100644
--- a/taskflow/graph/flow_builder.hpp
+++ b/taskflow/graph/flow_builder.hpp
@@ -264,6 +264,8 @@ class FlowBuilder {
     @param A task A
     */
     void gather(std::initializer_list<Task> others, Task A);
+
+    bool empty() const { return _graph.empty(); }
     
   private:
 

From 21ca35a6c8e7bccc3a22651b6a48fba47dfe7337 Mon Sep 17 00:00:00 2001
From: clin99 <artiferlly@gmail.com>
Date: Sat, 2 Mar 2019 15:29:45 -0600
Subject: [PATCH 06/12] update

---
 taskflow/graph/basic_taskflow.hpp | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/taskflow/graph/basic_taskflow.hpp b/taskflow/graph/basic_taskflow.hpp
index d4ec3c579..6db11442d 100644
--- a/taskflow/graph/basic_taskflow.hpp
+++ b/taskflow/graph/basic_taskflow.hpp
@@ -417,10 +417,11 @@ std::shared_future<void> BasicTaskflow<E>::run_until(WorkGroup& wg, P&& predicat
   for(auto &p: wg._pairs) {
     std::get<Node*>(p)->_work = [&, tf=this](auto& subflow) mutable {
       if(!subflow.empty()) return;
-      
+
       //std::cout << "Framework name: " << std::get<1>(p)->name() << std::endl;
 
       auto sink = subflow.placeholder();
+      sink._node->set_subtask();
 
       PassiveVector<Node*> src;
       PassiveVector<Node*> tgt;

From 1681eac444387fb95907fb6fcd6db56e4feb78df Mon Sep 17 00:00:00 2001
From: clin99 <artiferlly@gmail.com>
Date: Sat, 2 Mar 2019 15:31:58 -0600
Subject: [PATCH 07/12] update

---
 example/workgroup.cpp | 4 ++++
 1 file changed, 4 insertions(+)

diff --git a/example/workgroup.cpp b/example/workgroup.cpp
index 1a8cedb83..9c36e1639 100644
--- a/example/workgroup.cpp
+++ b/example/workgroup.cpp
@@ -39,6 +39,10 @@ int main(){
 
   tf.run_until(wg, [iter = 4] () mutable { std::puts("\n"); return iter-- == 0; }, [](){}).get();
 
+  tf.run_n(f1, 3u).get();
+  std::puts("\n");
+  tf.run_n(f2, 3u).get();
+
   return 0;
 }
 

From 241d6464662a8d557259a912714eeafa8f3b0bc1 Mon Sep 17 00:00:00 2001
From: clin99 <artiferlly@gmail.com>
Date: Sat, 2 Mar 2019 16:55:18 -0600
Subject: [PATCH 08/12] update

---
 example/workgroup.cpp             | 7 +++++--
 taskflow/graph/basic_taskflow.hpp | 1 -
 taskflow/graph/framework.hpp      | 3 +++
 3 files changed, 8 insertions(+), 3 deletions(-)

diff --git a/example/workgroup.cpp b/example/workgroup.cpp
index 9c36e1639..d2b9da635 100644
--- a/example/workgroup.cpp
+++ b/example/workgroup.cpp
@@ -35,9 +35,12 @@ int main(){
   auto F1 = wg.emplace(f1);
   auto F2 = wg.emplace(f2);
   F1.precede(F2);
-  wg.emplace([](){ std::cout << "Glue task\n"; });
+  wg.emplace([](){ std::cout << "Glue task\n"; }).name("Glue task");
 
-  tf.run_until(wg, [iter = 4] () mutable { std::puts("\n"); return iter-- == 0; }, [](){}).get();
+  tf.run_until(wg, [iter = 1] () mutable { std::puts("\n"); return iter-- == 0; }, [](){}).get();
+
+  std::cout << wg.dump() << std::endl;
+  exit(1);
 
   tf.run_n(f1, 3u).get();
   std::puts("\n");
diff --git a/taskflow/graph/basic_taskflow.hpp b/taskflow/graph/basic_taskflow.hpp
index 6db11442d..05b6eee58 100644
--- a/taskflow/graph/basic_taskflow.hpp
+++ b/taskflow/graph/basic_taskflow.hpp
@@ -402,7 +402,6 @@ std::shared_future<void> BasicTaskflow<E>::run_until(WorkGroup& wg, P&& predicat
   // create a topology for this run
   auto &tpg = _topologies.emplace_back(wg, std::forward<P>(predicate));
 
-
   // Multi-threaded execution.
   std::scoped_lock lock(wg._mtx);
 
diff --git a/taskflow/graph/framework.hpp b/taskflow/graph/framework.hpp
index 6e3cb74f3..cdc01fd54 100644
--- a/taskflow/graph/framework.hpp
+++ b/taskflow/graph/framework.hpp
@@ -163,6 +163,9 @@ inline tf::Task WorkGroup::emplace(Framework& framework) {
   auto task = placeholder();
   _graph.back().set_workgroup();
   _pairs.emplace_back(&(_graph.back()), &framework);
+  if(!framework.name().empty()) {
+    task.name(framework.name());
+  }
   return task;
 }
 

From 60e52157dd508b9a3e4d232dddbf45d61ce13dd3 Mon Sep 17 00:00:00 2001
From: clin99 <34017491+clin99@users.noreply.github.com>
Date: Sun, 3 Mar 2019 11:22:12 -0600
Subject: [PATCH 09/12] Create pipeline_prof.cpp

---
 example/pipeline_prof.cpp | 268 ++++++++++++++++++++++++++++++++++++++
 1 file changed, 268 insertions(+)
 create mode 100644 example/pipeline_prof.cpp

diff --git a/example/pipeline_prof.cpp b/example/pipeline_prof.cpp
new file mode 100644
index 000000000..20ec2cbb9
--- /dev/null
+++ b/example/pipeline_prof.cpp
@@ -0,0 +1,268 @@
+// A simple example to capture the following task dependencies.
+//
+// TaskA---->TaskB---->TaskD
+// TaskA---->TaskC---->TaskD
+
+
+#include <cstring>
+#include <cstdlib>
+#include <cstdio>
+#include <cctype>
+#include <chrono>
+#include <taskflow/taskflow.hpp>  // the only include you need 
+
+// https://software.intel.com/en-us/forums/intel-moderncode-for-parallel-architectures/topic/297118
+#define compiler_fence() __asm__ __volatile__ ("" : : : "memory")
+
+// load with 'consume' (data-dependent) memory ordering 
+template<typename T> 
+T load_consume(T const* addr) { 
+	// hardware fence is implicit on x86 
+	T v = *const_cast<T const volatile*>(addr); 
+  compiler_fence();
+	//__memory_barrier(); // compiler fence 
+	return v; 
+} 
+
+// store with 'release' memory ordering 
+template<typename T> 
+void store_release(T* addr, T v) { 
+	// hardware fence is implicit on x86 
+	//__memory_barrier(); // compiler fence 
+  compiler_fence();
+	*const_cast<T volatile*>(addr) = v; 
+} 
+
+// cache line size on modern x86 processors (in bytes) 
+size_t const cache_line_size = 64; 
+// single-producer/single-consumer queue 
+template<typename T> 
+class spsc_queue { 
+	public: 
+		spsc_queue() { 
+			node* n = new node; 
+			n->next_ = 0; 
+			tail_ = head_ = first_= tail_copy_ = n; 
+		} 
+
+		~spsc_queue() {  
+			node* n = first_; 
+			do 
+			{ 
+				node* next = n->next_; 
+				delete n; 
+				n = next; 
+			} while (n); 
+		} 
+
+		void enqueue(T v) { 
+			node* n = alloc_node(); 
+			n->next_ = 0; 
+			n->value_ = v; 
+			store_release(&head_->next_, n); 
+			head_ = n; 
+		} 
+
+		// returns 'false' if queue is empty 
+		bool dequeue(T& v) { 
+			if (load_consume(&tail_->next_)) { 
+				v = tail_->next_->value_; 
+				store_release(&tail_, tail_->next_); 
+				return true; 
+			} 
+			else { 
+				return false; 
+			} 
+		} 
+
+	private: 
+		// internal node structure 
+		struct node { 
+			node* next_; 
+			T value_; 
+		}; 
+
+		// consumer part 
+		// accessed mainly by consumer, infrequently be producer 
+		node* tail_; // tail of the queue 
+
+		// delimiter between consumer part and producer part, 
+		// so that they situated on different cache lines 
+		char cache_line_pad_ [cache_line_size]; 
+
+		// producer part 
+		// accessed only by producer 
+		node* head_; // head of the queue 
+		node* first_; // last unused node (tail of node cache) 
+		node* tail_copy_; // helper (points somewhere between first_ and tail_) 
+
+		node* alloc_node() { 
+			// first tries to allocate node from internal node cache, 
+			// if attempt fails, allocates node via ::operator new() 
+
+			if (first_ != tail_copy_) { 
+				node* n = first_; 
+				first_ = first_->next_; 
+				return n; 
+			} 
+			tail_copy_ = load_consume(&tail_); 
+			if (first_ != tail_copy_) { 
+				node* n = first_; 
+				first_ = first_->next_; 
+				return n; 
+			} 
+			node* n = new node; 
+			return n; 
+		} 
+
+		spsc_queue(spsc_queue const&); 
+		spsc_queue& operator = (spsc_queue const&); 
+}; 
+
+
+class TextSlice {
+    //! Pointer to one past last character in sequence
+    char* logical_end;
+    //! Pointer to one past last available byte in sequence.
+    char* physical_end;
+  public:
+    //! Allocate a TextSlice object that can hold up to max_size characters.
+    static TextSlice* allocate( size_t max_size ) {
+      // +1 leaves room for a terminating null character.
+      TextSlice* t = static_cast<TextSlice*>(std::malloc(sizeof(TextSlice)+max_size+1));
+      t->logical_end = t->begin();
+      t->physical_end = t->begin()+max_size;
+      return t;
+    }
+    //! Free a TextSlice object 
+    void free() {
+      std::free((char*)this);
+        //tbb::tbb_allocator<char>().deallocate((char*)this,sizeof(TextSlice)+(physical_end-begin())+1);
+    } 
+    //! Pointer to beginning of sequence
+    char* begin() {return (char*)(this+1);}
+    //! Pointer to one past last character in sequence
+    char* end() {return logical_end;}
+    void end(char c) { *logical_end = c; }
+    //! Length of sequence
+    size_t size() const {return logical_end-(char*)(this+1);}
+    //! Maximum number of characters that can be appended to sequence
+    size_t avail() const {return physical_end-logical_end;}
+    //! Append sequence [first,last) to this sequence.
+    void append( char* first, char* last ) {
+      memcpy( logical_end, first, last-first );
+      logical_end += last-first;
+    }
+    //! Set end() to given value.
+    void set_end( char* p ) {logical_end=p;}
+};
+
+
+constexpr size_t MAX_CHAR_PER_INPUT_SLICE = 4000;
+const std::string InputFileName = "input.txt";
+const std::string OutputFileName = "output.txt";
+
+int main(){
+
+  auto t1 = std::chrono::high_resolution_clock::now(); 
+
+  tf::Taskflow tf(4);
+  tf::Framework f;
+
+  spsc_queue<TextSlice*> input_queue;
+  spsc_queue<TextSlice*> output_queue;
+
+  FILE* input_file = fopen( InputFileName.c_str(), "r" );
+  FILE* output_file = fopen( OutputFileName.c_str(), "w" );
+  
+  auto [transform, output] = f.emplace( 
+    [&] () { 
+      TextSlice *input;
+      assert(input_queue.dequeue(input));
+
+      input->end('\0');
+      char* p = input->begin();
+      TextSlice& out = *TextSlice::allocate( 2*MAX_CHAR_PER_INPUT_SLICE );
+      char* q = out.begin();
+      for(;;) {
+        while( p<(*input).end() && !isdigit(*p) ) { 
+          *q++ = *p++; 
+        }
+        if( p==(*input).end() ) {
+          break;
+        }
+        long x = strtol( p, &p, 10 );
+        // Note: no overflow checking is needed here, as we have twice the 
+        // input string length, but the square of a non-negative integer n 
+        // cannot have more than twice as many digits as n.
+        long y = x*x; 
+        sprintf(q,"%ld",y);
+        q = strchr(q,0);
+      }
+      out.set_end(q);
+      (*input).free();
+
+      output_queue.enqueue(&out);
+    },     
+    [&] () { 
+      TextSlice* out;
+      assert(output_queue.dequeue(out));
+
+      size_t n = fwrite( out->begin(), 1, out->size(), output_file );
+      if( n!=out->size() ) {
+          fprintf(stderr,"Can't write into file '%s'\n", OutputFileName.c_str());
+          exit(1);
+      }
+      out->free();
+    }
+  ); 
+
+  //transform.name("transform");
+  //output.name("output");
+  transform.name("2");
+  output.name("3");
+
+   
+  // Linear   
+  transform.precede(output);  
+
+  TextSlice* next_slice = TextSlice::allocate( MAX_CHAR_PER_INPUT_SLICE );
+
+
+  tf.pipeline_until(f, [&]() mutable { 
+      // Read characters into space that is available in the next slice.
+      size_t m = next_slice->avail();
+      size_t n = fread( next_slice->end(), 1, m, input_file );
+      if( !n && next_slice->size()==0 ) {
+        // No more characters to process 
+        std::puts("All Done");
+        return true;
+      } 
+      else {
+        // Have more characters to process.
+        TextSlice& t = *next_slice;
+        next_slice = TextSlice::allocate( MAX_CHAR_PER_INPUT_SLICE );
+        char* p = t.end()+n;
+        if( n==m ) {
+          // Might have read partial number.  If so, transfer characters of partial number to next slice.
+          while( p>t.begin() && isdigit(p[-1]) ) {
+            --p;
+          }
+          next_slice->append( p, t.end()+n );
+        }
+        t.set_end(p);
+        input_queue.enqueue(&t);
+        return false;
+        //return &t;
+      }
+    }, 
+    [](){}
+  ).get();
+
+  auto t2 = std::chrono::high_resolution_clock::now();
+  std::cout << std::chrono::duration_cast<std::chrono::microseconds>(t2 - t1).count()/1000000.0 << std::endl;
+
+  return 0;
+}
+
+

From 486355308c485759a6bab22a58dc5b2bfb402ee1 Mon Sep 17 00:00:00 2001
From: clin99 <34017491+clin99@users.noreply.github.com>
Date: Sun, 3 Mar 2019 14:18:04 -0600
Subject: [PATCH 10/12] Update pipeline_prof.cpp

---
 example/pipeline_prof.cpp | 103 +++++++++++++++++++++++++++++++++++++-
 1 file changed, 102 insertions(+), 1 deletion(-)

diff --git a/example/pipeline_prof.cpp b/example/pipeline_prof.cpp
index 20ec2cbb9..8d4882556 100644
--- a/example/pipeline_prof.cpp
+++ b/example/pipeline_prof.cpp
@@ -162,8 +162,106 @@ constexpr size_t MAX_CHAR_PER_INPUT_SLICE = 4000;
 const std::string InputFileName = "input.txt";
 const std::string OutputFileName = "output.txt";
 
+
+bool input_task(TextSlice **next_slice, spsc_queue<TextSlice*>& input_queue, FILE* input_file) {
+  size_t m = (*next_slice)->avail();
+  size_t n = fread( (*next_slice)->end(), 1, m, input_file );
+  if( !n && (*next_slice)->size()==0 ) {
+    // No more characters to process 
+    return true;
+  } 
+  else {
+    // Have more characters to process.
+    TextSlice& t = **next_slice;
+    *next_slice = TextSlice::allocate( MAX_CHAR_PER_INPUT_SLICE );
+    char* p = t.end()+n;
+    if( n==m ) {
+      // Might have read partial number.  If so, transfer characters of partial number to next slice.
+      while( p>t.begin() && isdigit(p[-1]) ) {
+        --p;
+      }
+      (*next_slice)->append( p, t.end()+n );
+    }
+    t.set_end(p);
+    input_queue.enqueue(&t);
+    return false;
+  }
+}
+
+
+void output_task(TextSlice* input, spsc_queue<TextSlice*>& output_queue) {
+  input->end('\0');
+  char* p = input->begin();
+  TextSlice& out = *TextSlice::allocate( 2*MAX_CHAR_PER_INPUT_SLICE );
+  char* q = out.begin();
+  for(;;) {
+    while( p<(*input).end() && !isdigit(*p) ) { 
+      *q++ = *p++; 
+    }
+    if( p==(*input).end() ) {
+      break;
+    }
+    long x = strtol( p, &p, 10 );
+    // Note: no overflow checking is needed here, as we have twice the 
+    // input string length, but the square of a non-negative integer n 
+    // cannot have more than twice as many digits as n.
+    long y = x*x; 
+    sprintf(q,"%ld",y);
+    q = strchr(q,0);
+  }
+  out.set_end(q);
+  (*input).free();
+
+  output_queue.enqueue(&out);
+}
+
+
+void sequential() {
+  auto t1 = std::chrono::high_resolution_clock::now(); 
+  spsc_queue<TextSlice*> input_queue;
+  spsc_queue<TextSlice*> output_queue;
+  FILE* input_file = fopen( InputFileName.c_str(), "r" );
+  FILE* output_file = fopen( OutputFileName.c_str(), "w" );
+
+  size_t num_inputs {0};
+  TextSlice* next_slice = TextSlice::allocate( MAX_CHAR_PER_INPUT_SLICE );
+  while(!input_task(&next_slice, input_queue, input_file)) {
+    num_inputs ++;
+  }
+
+  auto t2 = std::chrono::high_resolution_clock::now(); 
+
+  for(size_t i=0; i<num_inputs; i++) {
+    TextSlice *slice;
+    assert(input_queue.dequeue(slice));
+    output_task(slice, output_queue);
+  }
+
+  auto t3 = std::chrono::high_resolution_clock::now(); 
+
+  for(size_t i=0; i<num_inputs; i++) {
+    TextSlice* out;
+    assert(output_queue.dequeue(out));
+
+    size_t n = fwrite( out->begin(), 1, out->size(), output_file );
+    if( n!=out->size() ) {
+      fprintf(stderr,"Can't write into file '%s'\n", OutputFileName.c_str());
+      exit(1);
+    }
+    out->free();
+  }
+  auto t4 = std::chrono::high_resolution_clock::now();  
+
+  std::cout << std::chrono::duration_cast<std::chrono::microseconds>(t3 - t2).count()/1000000.0 << std::endl;
+  std::cout << std::chrono::duration_cast<std::chrono::microseconds>(t4 - t1).count()/1000000.0 << std::endl;
+  exit(1);
+}
+
 int main(){
 
+  sequential();
+
+
   auto t1 = std::chrono::high_resolution_clock::now(); 
 
   tf::Taskflow tf(4);
@@ -177,6 +275,7 @@ int main(){
   
   auto [transform, output] = f.emplace( 
     [&] () { 
+
       TextSlice *input;
       assert(input_queue.dequeue(input));
 
@@ -205,6 +304,7 @@ int main(){
       output_queue.enqueue(&out);
     },     
     [&] () { 
+
       TextSlice* out;
       assert(output_queue.dequeue(out));
 
@@ -228,7 +328,6 @@ int main(){
 
   TextSlice* next_slice = TextSlice::allocate( MAX_CHAR_PER_INPUT_SLICE );
 
-
   tf.pipeline_until(f, [&]() mutable { 
       // Read characters into space that is available in the next slice.
       size_t m = next_slice->avail();
@@ -262,6 +361,8 @@ int main(){
   auto t2 = std::chrono::high_resolution_clock::now();
   std::cout << std::chrono::duration_cast<std::chrono::microseconds>(t2 - t1).count()/1000000.0 << std::endl;
 
+  std::fclose(input_file);
+  std::fclose(output_file);
   return 0;
 }
 

From ae039fa767c55f696c67b704664711915695df3d Mon Sep 17 00:00:00 2001
From: clin99 <34017491+clin99@users.noreply.github.com>
Date: Mon, 4 Mar 2019 09:54:07 -0600
Subject: [PATCH 11/12] Create batch_pipeline.cpp

---
 example/batch_pipeline.cpp | 411 +++++++++++++++++++++++++++++++++++++
 1 file changed, 411 insertions(+)
 create mode 100644 example/batch_pipeline.cpp

diff --git a/example/batch_pipeline.cpp b/example/batch_pipeline.cpp
new file mode 100644
index 000000000..5d659b023
--- /dev/null
+++ b/example/batch_pipeline.cpp
@@ -0,0 +1,411 @@
+// A simple example to capture the following task dependencies.
+//
+// TaskA---->TaskB---->TaskD
+// TaskA---->TaskC---->TaskD
+
+
+#include <cstring>
+#include <cstdlib>
+#include <cstdio>
+#include <cctype>
+#include <chrono>
+#include <taskflow/taskflow.hpp>  // the only include you need 
+
+// https://software.intel.com/en-us/forums/intel-moderncode-for-parallel-architectures/topic/297118
+#define compiler_fence() __asm__ __volatile__ ("" : : : "memory")
+
+// load with 'consume' (data-dependent) memory ordering 
+template<typename T> 
+T load_consume(T const* addr) { 
+	// hardware fence is implicit on x86 
+	T v = *const_cast<T const volatile*>(addr); 
+  compiler_fence();
+	//__memory_barrier(); // compiler fence 
+	return v; 
+} 
+
+// store with 'release' memory ordering 
+template<typename T> 
+void store_release(T* addr, T v) { 
+	// hardware fence is implicit on x86 
+	//__memory_barrier(); // compiler fence 
+  compiler_fence();
+	*const_cast<T volatile*>(addr) = v; 
+} 
+
+// cache line size on modern x86 processors (in bytes) 
+size_t const cache_line_size = 64; 
+// single-producer/single-consumer queue 
+template<typename T> 
+class spsc_queue { 
+	public: 
+		spsc_queue() { 
+			node* n = new node; 
+			n->next_ = 0; 
+			tail_ = head_ = first_= tail_copy_ = n; 
+		} 
+
+		~spsc_queue() {  
+			node* n = first_; 
+			do 
+			{ 
+				node* next = n->next_; 
+				delete n; 
+				n = next; 
+			} while (n); 
+		} 
+
+		void enqueue(T v) { 
+			node* n = alloc_node(); 
+			n->next_ = 0; 
+			n->value_ = v; 
+			store_release(&head_->next_, n); 
+			head_ = n; 
+		} 
+
+		// returns 'false' if queue is empty 
+		bool dequeue(T& v) { 
+			if (load_consume(&tail_->next_)) { 
+				v = tail_->next_->value_; 
+				store_release(&tail_, tail_->next_); 
+				return true; 
+			} 
+			else { 
+				return false; 
+			} 
+		} 
+
+	private: 
+		// internal node structure 
+		struct node { 
+			node* next_; 
+			T value_; 
+		}; 
+
+		// consumer part 
+		// accessed mainly by consumer, infrequently be producer 
+		node* tail_; // tail of the queue 
+
+		// delimiter between consumer part and producer part, 
+		// so that they situated on different cache lines 
+		char cache_line_pad_ [cache_line_size]; 
+
+		// producer part 
+		// accessed only by producer 
+		node* head_; // head of the queue 
+		node* first_; // last unused node (tail of node cache) 
+		node* tail_copy_; // helper (points somewhere between first_ and tail_) 
+
+		node* alloc_node() { 
+			// first tries to allocate node from internal node cache, 
+			// if attempt fails, allocates node via ::operator new() 
+
+			if (first_ != tail_copy_) { 
+				node* n = first_; 
+				first_ = first_->next_; 
+				return n; 
+			} 
+			tail_copy_ = load_consume(&tail_); 
+			if (first_ != tail_copy_) { 
+				node* n = first_; 
+				first_ = first_->next_; 
+				return n; 
+			} 
+			node* n = new node; 
+			return n; 
+		} 
+
+		spsc_queue(spsc_queue const&); 
+		spsc_queue& operator = (spsc_queue const&); 
+}; 
+
+
+class TextSlice {
+    //! Pointer to one past last character in sequence
+    char* logical_end;
+    //! Pointer to one past last available byte in sequence.
+    char* physical_end;
+  public:
+    //! Allocate a TextSlice object that can hold up to max_size characters.
+    static TextSlice* allocate( size_t max_size ) {
+      // +1 leaves room for a terminating null character.
+      TextSlice* t = static_cast<TextSlice*>(std::malloc(sizeof(TextSlice)+max_size+1));
+      t->logical_end = t->begin();
+      t->physical_end = t->begin()+max_size;
+      return t;
+    }
+    //! Free a TextSlice object 
+    void free() {
+      std::free((char*)this);
+        //tbb::tbb_allocator<char>().deallocate((char*)this,sizeof(TextSlice)+(physical_end-begin())+1);
+    } 
+    //! Pointer to beginning of sequence
+    char* begin() {return (char*)(this+1);}
+    //! Pointer to one past last character in sequence
+    char* end() {return logical_end;}
+    void end(char c) { *logical_end = c; }
+    //! Length of sequence
+    size_t size() const {return logical_end-(char*)(this+1);}
+    //! Maximum number of characters that can be appended to sequence
+    size_t avail() const {return physical_end-logical_end;}
+    //! Append sequence [first,last) to this sequence.
+    void append( char* first, char* last ) {
+      memcpy( logical_end, first, last-first );
+      logical_end += last-first;
+    }
+    //! Set end() to given value.
+    void set_end( char* p ) {logical_end=p;}
+};
+
+
+constexpr size_t MAX_CHAR_PER_INPUT_SLICE = 4000;
+const std::string InputFileName = "input.txt";
+const std::string OutputFileName = "output.txt";
+
+
+bool input_task(TextSlice **next_slice, spsc_queue<TextSlice*>& input_queue, FILE* input_file) {
+  size_t m = (*next_slice)->avail();
+  size_t n = fread( (*next_slice)->end(), 1, m, input_file );
+  if( !n && (*next_slice)->size()==0 ) {
+    // No more characters to process 
+    return true;
+  } 
+  else {
+    // Have more characters to process.
+    TextSlice& t = **next_slice;
+    *next_slice = TextSlice::allocate( MAX_CHAR_PER_INPUT_SLICE );
+    char* p = t.end()+n;
+    if( n==m ) {
+      // Might have read partial number.  If so, transfer characters of partial number to next slice.
+      while( p>t.begin() && isdigit(p[-1]) ) {
+        --p;
+      }
+      (*next_slice)->append( p, t.end()+n );
+    }
+    t.set_end(p);
+    input_queue.enqueue(&t);
+    return false;
+  }
+}
+
+
+void output_task(TextSlice* input, spsc_queue<TextSlice*>& output_queue) {
+  input->end('\0');
+  char* p = input->begin();
+  TextSlice& out = *TextSlice::allocate( 2*MAX_CHAR_PER_INPUT_SLICE );
+  char* q = out.begin();
+  for(;;) {
+    while( p<(*input).end() && !isdigit(*p) ) { 
+      *q++ = *p++; 
+    }
+    if( p==(*input).end() ) {
+      break;
+    }
+    long x = strtol( p, &p, 10 );
+    // Note: no overflow checking is needed here, as we have twice the 
+    // input string length, but the square of a non-negative integer n 
+    // cannot have more than twice as many digits as n.
+    long y = x*x; 
+    sprintf(q,"%ld",y);
+    q = strchr(q,0);
+  }
+  out.set_end(q);
+  (*input).free();
+
+  output_queue.enqueue(&out);
+}
+
+
+void sequential() {
+  auto t1 = std::chrono::high_resolution_clock::now(); 
+  spsc_queue<TextSlice*> input_queue;
+  spsc_queue<TextSlice*> output_queue;
+  FILE* input_file = fopen( InputFileName.c_str(), "r" );
+  FILE* output_file = fopen( OutputFileName.c_str(), "w" );
+
+  size_t num_inputs {0};
+  TextSlice* next_slice = TextSlice::allocate( MAX_CHAR_PER_INPUT_SLICE );
+  while(!input_task(&next_slice, input_queue, input_file)) {
+    num_inputs ++;
+  }
+
+  auto t2 = std::chrono::high_resolution_clock::now(); 
+
+  for(size_t i=0; i<num_inputs; i++) {
+    TextSlice *slice;
+    assert(input_queue.dequeue(slice));
+    output_task(slice, output_queue);
+  }
+
+  auto t3 = std::chrono::high_resolution_clock::now(); 
+
+  for(size_t i=0; i<num_inputs; i++) {
+    TextSlice* out;
+    assert(output_queue.dequeue(out));
+
+    size_t n = fwrite( out->begin(), 1, out->size(), output_file );
+    if( n!=out->size() ) {
+      fprintf(stderr,"Can't write into file '%s'\n", OutputFileName.c_str());
+      exit(1);
+    }
+    out->free();
+  }
+  auto t4 = std::chrono::high_resolution_clock::now();  
+
+  std::cout << std::chrono::duration_cast<std::chrono::microseconds>(t3 - t2).count()/1000000.0 << std::endl;
+  std::cout << std::chrono::duration_cast<std::chrono::microseconds>(t4 - t1).count()/1000000.0 << std::endl;
+  exit(1);
+}
+
+
+TextSlice* output_task(TextSlice* input) {
+  input->end('\0');
+  char* p = input->begin();
+  TextSlice& out = *TextSlice::allocate( 2*MAX_CHAR_PER_INPUT_SLICE );
+  char* q = out.begin();
+  for(;;) {
+    while( p<(*input).end() && !isdigit(*p) ) { 
+      *q++ = *p++; 
+    }
+    if( p==(*input).end() ) {
+      break;
+    }
+    long x = strtol( p, &p, 10 );
+    // Note: no overflow checking is needed here, as we have twice the 
+    // input string length, but the square of a non-negative integer n 
+    // cannot have more than twice as many digits as n.
+    long y = x*x; 
+    sprintf(q,"%ld",y);
+    q = strchr(q,0);
+  }
+  out.set_end(q);
+  (*input).free();
+
+  return &out;
+}
+
+
+int main(){
+
+  //sequential();
+  auto t1 = std::chrono::high_resolution_clock::now(); 
+
+  tf::Taskflow tf(4);
+  tf::Framework f;
+
+  spsc_queue<TextSlice*> input_queue;
+  spsc_queue<TextSlice*> output_queue;
+
+  FILE* input_file = fopen( InputFileName.c_str(), "r" );
+  FILE* output_file = fopen( OutputFileName.c_str(), "w" );
+
+  constexpr size_t batch = 8;
+  std::array<TextSlice*, batch> temp;
+  std::vector<tf::Task> tasks;
+  tasks.resize(batch);
+
+  auto [transform, output] = f.emplace( 
+    [&] (auto &subflow) { 
+      subflow.join();
+
+      size_t total {0};
+      for(size_t i=0 ; i<batch; i++) {
+        TextSlice *input;
+        if(!input_queue.dequeue(input)) {
+          break;
+        }
+        tasks[i] = subflow.emplace([&, input=input, i=i](){
+          temp[i] = output_task(input);
+        });
+        total ++;
+      }
+
+      if(total) {
+        auto sync = subflow.emplace(
+          [&, total=total](){
+            for(size_t i=0; i<total; i++) {
+              output_queue.enqueue(temp[i]);
+            }
+          }
+        );
+        for(size_t i=0; i<total; i++) {
+          tasks[i].precede(sync);
+        }
+      }
+    },     
+    [&] () { 
+      
+      for(size_t i=0; i<batch; i++) {
+        TextSlice* out;
+        if(!output_queue.dequeue(out)) {
+          break;
+        }
+  
+        size_t n = fwrite( out->begin(), 1, out->size(), output_file );
+        if( n!=out->size() ) {
+          fprintf(stderr,"Can't write into file '%s'\n", OutputFileName.c_str());
+          exit(1);
+        }
+        out->free();
+      }
+    }
+  ); 
+
+  transform.name("transform");
+  output.name("output");
+
+  // Linear   
+  transform.precede(output);  
+
+  TextSlice* next_slice = TextSlice::allocate( MAX_CHAR_PER_INPUT_SLICE );
+
+  tf.pipeline_until(f, [&]() mutable { 
+      // Read characters into space that is available in the next slice.
+      size_t m = next_slice->avail();
+      size_t n = fread( next_slice->end(), 1, m, input_file );
+      if( !n && next_slice->size()==0 ) {
+        // No more characters to process 
+        std::puts("All Done");
+        return true;
+      } 
+      else {
+        for(size_t i=0; i<batch ; i++) {
+          // Have more characters to process.
+          TextSlice& t = *next_slice;
+          next_slice = TextSlice::allocate( MAX_CHAR_PER_INPUT_SLICE );
+          char* p = t.end()+n;
+          if( n==m ) {
+            // Might have read partial number.  If so, transfer characters of partial number to next slice.
+            while( p>t.begin() && isdigit(p[-1]) ) {
+              --p;
+            }
+            next_slice->append( p, t.end()+n );
+          }
+          t.set_end(p);
+          input_queue.enqueue(&t);
+
+
+          if(i == batch-1) break;
+
+
+          m = next_slice->avail();
+          n = fread( next_slice->end(), 1, m, input_file );
+
+          if( !n && next_slice->size()==0 ) break;
+        }
+        return false;
+        //return &t;
+      }
+    }, 
+    [](){}
+  ).get();
+
+  auto t2 = std::chrono::high_resolution_clock::now();
+  std::cout << std::chrono::duration_cast<std::chrono::microseconds>(t2 - t1).count()/1000000.0 << std::endl;
+
+  std::fclose(input_file);
+  std::fclose(output_file);
+  return 0;
+}
+
+

From da0429b5908772a6b2de52a72184ea17675074fe Mon Sep 17 00:00:00 2001
From: clin99 <34017491+clin99@users.noreply.github.com>
Date: Mon, 4 Mar 2019 09:54:36 -0600
Subject: [PATCH 12/12] Create subflow_pipeline.cpp

---
 example/subflow_pipeline.cpp | 393 +++++++++++++++++++++++++++++++++++
 1 file changed, 393 insertions(+)
 create mode 100644 example/subflow_pipeline.cpp

diff --git a/example/subflow_pipeline.cpp b/example/subflow_pipeline.cpp
new file mode 100644
index 000000000..2b7af4c15
--- /dev/null
+++ b/example/subflow_pipeline.cpp
@@ -0,0 +1,393 @@
+// A simple example to capture the following task dependencies.
+//
+// TaskA---->TaskB---->TaskD
+// TaskA---->TaskC---->TaskD
+
+
+#include <cstring>
+#include <cstdlib>
+#include <cstdio>
+#include <cctype>
+#include <chrono>
+#include <taskflow/taskflow.hpp>  // the only include you need 
+
+// https://software.intel.com/en-us/forums/intel-moderncode-for-parallel-architectures/topic/297118
+#define compiler_fence() __asm__ __volatile__ ("" : : : "memory")
+
+// load with 'consume' (data-dependent) memory ordering 
+template<typename T> 
+T load_consume(T const* addr) { 
+	// hardware fence is implicit on x86 
+	T v = *const_cast<T const volatile*>(addr); 
+  compiler_fence();
+	//__memory_barrier(); // compiler fence 
+	return v; 
+} 
+
+// store with 'release' memory ordering 
+template<typename T> 
+void store_release(T* addr, T v) { 
+	// hardware fence is implicit on x86 
+	//__memory_barrier(); // compiler fence 
+  compiler_fence();
+	*const_cast<T volatile*>(addr) = v; 
+} 
+
+// cache line size on modern x86 processors (in bytes) 
+size_t const cache_line_size = 64; 
+// single-producer/single-consumer queue 
+template<typename T> 
+class spsc_queue { 
+	public: 
+		spsc_queue() { 
+			node* n = new node; 
+			n->next_ = 0; 
+			tail_ = head_ = first_= tail_copy_ = n; 
+		} 
+
+		~spsc_queue() {  
+			node* n = first_; 
+			do 
+			{ 
+				node* next = n->next_; 
+				delete n; 
+				n = next; 
+			} while (n); 
+		} 
+
+		void enqueue(T v) { 
+			node* n = alloc_node(); 
+			n->next_ = 0; 
+			n->value_ = v; 
+			store_release(&head_->next_, n); 
+			head_ = n; 
+		} 
+
+		// returns 'false' if queue is empty 
+		bool dequeue(T& v) { 
+			if (load_consume(&tail_->next_)) { 
+				v = tail_->next_->value_; 
+				store_release(&tail_, tail_->next_); 
+				return true; 
+			} 
+			else { 
+				return false; 
+			} 
+		} 
+
+	private: 
+		// internal node structure 
+		struct node { 
+			node* next_; 
+			T value_; 
+		}; 
+
+		// consumer part 
+		// accessed mainly by consumer, infrequently be producer 
+		node* tail_; // tail of the queue 
+
+		// delimiter between consumer part and producer part, 
+		// so that they situated on different cache lines 
+		char cache_line_pad_ [cache_line_size]; 
+
+		// producer part 
+		// accessed only by producer 
+		node* head_; // head of the queue 
+		node* first_; // last unused node (tail of node cache) 
+		node* tail_copy_; // helper (points somewhere between first_ and tail_) 
+
+		node* alloc_node() { 
+			// first tries to allocate node from internal node cache, 
+			// if attempt fails, allocates node via ::operator new() 
+
+			if (first_ != tail_copy_) { 
+				node* n = first_; 
+				first_ = first_->next_; 
+				return n; 
+			} 
+			tail_copy_ = load_consume(&tail_); 
+			if (first_ != tail_copy_) { 
+				node* n = first_; 
+				first_ = first_->next_; 
+				return n; 
+			} 
+			node* n = new node; 
+			return n; 
+		} 
+
+		spsc_queue(spsc_queue const&); 
+		spsc_queue& operator = (spsc_queue const&); 
+}; 
+
+
+class TextSlice {
+    //! Pointer to one past last character in sequence
+    char* logical_end;
+    //! Pointer to one past last available byte in sequence.
+    char* physical_end;
+  public:
+    //! Allocate a TextSlice object that can hold up to max_size characters.
+    static TextSlice* allocate( size_t max_size ) {
+      // +1 leaves room for a terminating null character.
+      TextSlice* t = static_cast<TextSlice*>(std::malloc(sizeof(TextSlice)+max_size+1));
+      t->logical_end = t->begin();
+      t->physical_end = t->begin()+max_size;
+      return t;
+    }
+    //! Free a TextSlice object 
+    void free() {
+      std::free((char*)this);
+        //tbb::tbb_allocator<char>().deallocate((char*)this,sizeof(TextSlice)+(physical_end-begin())+1);
+    } 
+    //! Pointer to beginning of sequence
+    char* begin() {return (char*)(this+1);}
+    //! Pointer to one past last character in sequence
+    char* end() {return logical_end;}
+    void end(char c) { *logical_end = c; }
+    //! Length of sequence
+    size_t size() const {return logical_end-(char*)(this+1);}
+    //! Maximum number of characters that can be appended to sequence
+    size_t avail() const {return physical_end-logical_end;}
+    //! Append sequence [first,last) to this sequence.
+    void append( char* first, char* last ) {
+      memcpy( logical_end, first, last-first );
+      logical_end += last-first;
+    }
+    //! Set end() to given value.
+    void set_end( char* p ) {logical_end=p;}
+};
+
+
+constexpr size_t MAX_CHAR_PER_INPUT_SLICE = 4000;
+const std::string InputFileName = "input.txt";
+const std::string OutputFileName = "output.txt";
+
+
+bool input_task(TextSlice **next_slice, spsc_queue<TextSlice*>& input_queue, FILE* input_file) {
+  size_t m = (*next_slice)->avail();
+  size_t n = fread( (*next_slice)->end(), 1, m, input_file );
+  if( !n && (*next_slice)->size()==0 ) {
+    // No more characters to process 
+    return true;
+  } 
+  else {
+    // Have more characters to process.
+    TextSlice& t = **next_slice;
+    *next_slice = TextSlice::allocate( MAX_CHAR_PER_INPUT_SLICE );
+    char* p = t.end()+n;
+    if( n==m ) {
+      // Might have read partial number.  If so, transfer characters of partial number to next slice.
+      while( p>t.begin() && isdigit(p[-1]) ) {
+        --p;
+      }
+      (*next_slice)->append( p, t.end()+n );
+    }
+    t.set_end(p);
+    input_queue.enqueue(&t);
+    return false;
+  }
+}
+
+
+void output_task(TextSlice* input, spsc_queue<TextSlice*>& output_queue) {
+  input->end('\0');
+  char* p = input->begin();
+  TextSlice& out = *TextSlice::allocate( 2*MAX_CHAR_PER_INPUT_SLICE );
+  char* q = out.begin();
+  for(;;) {
+    while( p<(*input).end() && !isdigit(*p) ) { 
+      *q++ = *p++; 
+    }
+    if( p==(*input).end() ) {
+      break;
+    }
+    long x = strtol( p, &p, 10 );
+    // Note: no overflow checking is needed here, as we have twice the 
+    // input string length, but the square of a non-negative integer n 
+    // cannot have more than twice as many digits as n.
+    long y = x*x; 
+    sprintf(q,"%ld",y);
+    q = strchr(q,0);
+  }
+  out.set_end(q);
+  (*input).free();
+
+  output_queue.enqueue(&out);
+}
+
+
+void sequential() {
+  auto t1 = std::chrono::high_resolution_clock::now(); 
+  spsc_queue<TextSlice*> input_queue;
+  spsc_queue<TextSlice*> output_queue;
+  FILE* input_file = fopen( InputFileName.c_str(), "r" );
+  FILE* output_file = fopen( OutputFileName.c_str(), "w" );
+
+  size_t num_inputs {0};
+  TextSlice* next_slice = TextSlice::allocate( MAX_CHAR_PER_INPUT_SLICE );
+  while(!input_task(&next_slice, input_queue, input_file)) {
+    num_inputs ++;
+  }
+
+  auto t2 = std::chrono::high_resolution_clock::now(); 
+
+  for(size_t i=0; i<num_inputs; i++) {
+    TextSlice *slice;
+    assert(input_queue.dequeue(slice));
+    output_task(slice, output_queue);
+  }
+
+  auto t3 = std::chrono::high_resolution_clock::now(); 
+
+  for(size_t i=0; i<num_inputs; i++) {
+    TextSlice* out;
+    assert(output_queue.dequeue(out));
+
+    size_t n = fwrite( out->begin(), 1, out->size(), output_file );
+    if( n!=out->size() ) {
+      fprintf(stderr,"Can't write into file '%s'\n", OutputFileName.c_str());
+      exit(1);
+    }
+    out->free();
+  }
+  auto t4 = std::chrono::high_resolution_clock::now();  
+
+  std::cout << std::chrono::duration_cast<std::chrono::microseconds>(t3 - t2).count()/1000000.0 << std::endl;
+  std::cout << std::chrono::duration_cast<std::chrono::microseconds>(t4 - t1).count()/1000000.0 << std::endl;
+  exit(1);
+}
+
+int main(){
+
+  //sequential();
+
+
+  auto t1 = std::chrono::high_resolution_clock::now(); 
+
+  tf::Taskflow tf(4);
+  tf::Framework f;
+
+  spsc_queue<TextSlice*> input_queue;
+  spsc_queue<TextSlice*> output_queue;
+
+  FILE* input_file = fopen( InputFileName.c_str(), "r" );
+  FILE* output_file = fopen( OutputFileName.c_str(), "w" );
+
+  
+  auto [transform, output] = f.emplace( 
+    [&] (auto &subflow) { 
+      subflow.join();
+
+      for(size_t i=0 ; i<4; i++) {
+        TextSlice *input;
+        if(!input_queue.dequeue(input)) {
+          break;
+        }
+        subflow.emplace([&, input=input](){
+          output_task(input, output_queue);
+        });
+
+        //input->end('\0');
+        //char* p = input->begin();
+        //TextSlice& out = *TextSlice::allocate( 2*MAX_CHAR_PER_INPUT_SLICE );
+        //char* q = out.begin();
+        //for(;;) {
+        //  while( p<(*input).end() && !isdigit(*p) ) { 
+        //    *q++ = *p++; 
+        //  }
+        //  if( p==(*input).end() ) {
+        //    break;
+        //  }
+        //  long x = strtol( p, &p, 10 );
+        //  // Note: no overflow checking is needed here, as we have twice the 
+        //  // input string length, but the square of a non-negative integer n 
+        //  // cannot have more than twice as many digits as n.
+        //  long y = x*x; 
+        //  sprintf(q,"%ld",y);
+        //  q = strchr(q,0);
+        //}
+        //out.set_end(q);
+        //(*input).free();
+
+        //output_queue.enqueue(&out);
+      }
+    },     
+    [&] () { 
+      
+      for(size_t i=0; i<4; i++) {
+        TextSlice* out;
+        if(!output_queue.dequeue(out)) {
+          break;
+        }
+  
+        size_t n = fwrite( out->begin(), 1, out->size(), output_file );
+        if( n!=out->size() ) {
+          fprintf(stderr,"Can't write into file '%s'\n", OutputFileName.c_str());
+          exit(1);
+        }
+        out->free();
+      }
+    }
+  ); 
+
+  //transform.name("transform");
+  //output.name("output");
+  transform.name("2");
+  output.name("3");
+
+   
+  // Linear   
+  transform.precede(output);  
+
+  TextSlice* next_slice = TextSlice::allocate( MAX_CHAR_PER_INPUT_SLICE );
+
+  tf.pipeline_until(f, [&]() mutable { 
+      // Read characters into space that is available in the next slice.
+      size_t m = next_slice->avail();
+      size_t n = fread( next_slice->end(), 1, m, input_file );
+      if( !n && next_slice->size()==0 ) {
+        // No more characters to process 
+        std::puts("All Done");
+        return true;
+      } 
+      else {
+        for(size_t i=0; i<4 ; i++) {
+          // Have more characters to process.
+          TextSlice& t = *next_slice;
+          next_slice = TextSlice::allocate( MAX_CHAR_PER_INPUT_SLICE );
+          char* p = t.end()+n;
+          if( n==m ) {
+            // Might have read partial number.  If so, transfer characters of partial number to next slice.
+            while( p>t.begin() && isdigit(p[-1]) ) {
+              --p;
+            }
+            next_slice->append( p, t.end()+n );
+          }
+          t.set_end(p);
+          input_queue.enqueue(&t);
+
+
+          if(i == 3) break;
+
+
+          m = next_slice->avail();
+          n = fread( next_slice->end(), 1, m, input_file );
+
+          if( !n && next_slice->size()==0 ) break;
+        }
+        return false;
+        //return &t;
+      }
+    }, 
+    [](){}
+  ).get();
+
+  auto t2 = std::chrono::high_resolution_clock::now();
+  std::cout << std::chrono::duration_cast<std::chrono::microseconds>(t2 - t1).count()/1000000.0 << std::endl;
+
+  std::fclose(input_file);
+  std::fclose(output_file);
+  return 0;
+}
+
+