refactored privatized_threadpool

tsung-wei-huang · tsung-wei-huang · commit 18eba9f55182 · 2018-10-05T18:34:49.000-05:00
diff --git a/CMakeLists.txt b/CMakeLists.txt
@@ -141,7 +141,7 @@ add_test(WorkerQueue.TriThread  ${TF_UTEST_DIR}/threadpool -tc=WorkerQueue.TriTh
 add_test(simple_threadpool      ${TF_UTEST_DIR}/threadpool -tc=SimpleThreadpool)
 add_test(proactive_threadpool   ${TF_UTEST_DIR}/threadpool -tc=ProactiveThreadpool)
 add_test(speculative_threadpool ${TF_UTEST_DIR}/threadpool -tc=SpeculativeThreadpool)
-add_test(privatized_threadpool ${TF_UTEST_DIR}/threadpool -tc=PrivatizedThreadpool)
+add_test(privatized_threadpool  ${TF_UTEST_DIR}/threadpool -tc=PrivatizedThreadpool)
 
 ## threadpool_cxx14 unittest (contributed by Glen Fraser)
 add_executable(threadpool_cxx14_tmp unittest/threadpool_cxx14.cpp)
diff --git a/doc/faq.md b/doc/faq.md
@@ -9,16 +9,16 @@ If you cannot find a solution here, please post an issue [here][Github issues].
 
 ---
 
-## General Questions
+# General Questions
 
-### Q: How do I use Cpp-Taskflow in my projects?
+## Q: How do I use Cpp-Taskflow in my projects?
 
 Cpp-Taskflow is a header-only library with zero dependencies. 
 The only thing you need is a [C++17][C++17] compiler.
 To use Cpp-Taskflow, simply drop the folder 
 [taskflow](../taskflow) to your project and include [taskflow.hpp](../taskflow/taskflow.hpp).
 
-### Q: What is the difference between static tasking and dynamic tasking?
+## Q: What is the difference between static tasking and dynamic tasking?
 
 Static tasking refers to those tasks created before execution,
 while dynamic tasking refers to those tasks created during the execution of static tasks
@@ -30,21 +30,21 @@ Dynamic tasks created by the same task node are grouped together to a subflow.
 | ![](../image/static_graph.png) | ![](../image/dynamic_graph.png) |
 
 
-### Q: How many tasks can Cpp-Taskflow handle?
+## Q: How many tasks can Cpp-Taskflow handle?
 
 Cpp-Taskflow is a very lightweight and efficient tasking library.
 It has been applied in many academic and industry projects to scale up their existing workload.
 A research project [OpenTimer][OpenTimer] has used Cpp-Taskflow to deal with hundreds of millions of tasks.
 
-### Q: What are the differences between Cpp-Taskflow and other tasking libraries?
+## Q: What are the differences between Cpp-Taskflow and other tasking libraries?
 
 From our humble opinion, Cpp-Taskflow is superior in its tasking API, interface, and performance.
 In most cases, users can quickly master Cpp-Taskflow to create large and complex dependency graphs
 in just a few minutes.
 The performance scales very well and is comparable to hard-coded multi-threading.
 Of course, the judge is always left for users -:)
 
-### Q: What is the weird hex value, like 0x7fc39d402ab0, in the dumped graph?
+## Q: What is the weird hex value, like 0x7fc39d402ab0, in the dumped graph?
 
 Each task has a method `name(const std::string&)` for user to assign a human readable string
 to ease the debugging process. 
@@ -53,14 +53,14 @@ its address value in the memory is used instead.
 
 ---
 
-## Compilation Issues
+# Compilation Issues
 
-### Q: I can't get Cpp-Taskflow compiled in my project!
+## Q: I can't get Cpp-Taskflow compiled in my project!
 
 Please make sure your compile supports the latest version of [C++17][C++17]. 
 Make sure your project meets the System Requirements described at [README][README].
 
-### Q: Clang can't compile due to the use of std::variant.
+## Q: Clang can't compile due to the use of std::variant.
 
 Cpp-Taskflow uses `std::variant` to enable uniform interface between static and dynamic tasking.
 However it has been reported in 
@@ -81,9 +81,9 @@ For clang users, you will need to use this patch in `taskflow.hpp` as follows:
 
 ---
 
-## Programming Questions
+# Programming Questions
 
-### Q: What is the difference between Cpp-Taskflow threads and workers?
+## Q: What is the difference between Cpp-Taskflow threads and workers?
 
 The master thread owns the thread pool and can spawn workers to run tasks 
 or shutdown the pool. 
@@ -96,18 +96,19 @@ tf::Taskflow(N);    // N workers, N+1 threads in the program.
 
 If there is no worker threads in the pool, the master thread will do all the works by itself.
 
-### Q: Is taskflow thread-safe?
+## Q: Is taskflow thread-safe?
+
 No, the taskflow object is not thread-safe. You can't create tasks from multiple threads
 at the same time.
 
-### Q: My program hangs and never returns after dispatching a taskflow graph. What's wrong?
+## Q: My program hangs and never returns after dispatching a taskflow graph. What's wrong?
 
 When the program hangs forever it is very likely your taskflow graph has a cycle.
 Try the `dump` method to debug the graph before dispatching your taskflow graph.
 If there is no cycle, make sure you are using `future.get()` in the right way, 
 i.e., not blocking your control flow.
 
-### Q: In the following example where B spawns a joined subflow of two tasks B1 and B2, do they run concurrently with task A?
+## Q: In the following example where B spawns a joined subflow of two tasks B1 and B2, do they run concurrently with task A?
 
 <p>
 <img src="../image/dynamic_graph.png" width="60%">
@@ -119,7 +120,7 @@ This graph may looks strange because B seems to run twice!
 However, Cpp-Taskflow will schedule B only once to create its subflow.
 Whether this subflow joins or detaches from B only affects the future object returned from B.
 
-### Q: How can I parallelize multiple runs on the same function with different arguments?
+## Q: How can I parallelize multiple runs on the same function with different arguments?
 
 Many people have been asking how to apply Taskflow's `parallel_for` method
 to parallelize a sequential loop over an index sequence.
@@ -142,7 +143,7 @@ for(int i=0; i<N; ++i) {
 tf.wait_for_all();
 ```
 
-### Q: What is a Task?
+## Q: What is a Task?
 
 A `Task` is a lightweight handle associated with an internal graph node
 for users to build task dependencies.
diff --git a/taskflow/taskflow.hpp b/taskflow/taskflow.hpp
@@ -1095,6 +1095,7 @@ class Taskflow : public FlowBuilder {
     //SimpleThreadpool<Closure> _executor;
     //ProactiveThreadpool<Closure> _executor;
     SpeculativeThreadpool<Closure> _executor;
+    //PrivatizedThreadpool<Closure> _executor;
 
     Graph _graph;
 
diff --git a/taskflow/threadpool/privatized_threadpool.hpp b/taskflow/threadpool/privatized_threadpool.hpp
@@ -1,14 +1,21 @@
+// 2018/10/05 - modified by Chun-Xun
+//   - adopted the new threadpool model   
+//
 // 2018/09/21 - modified by Tsung-Wei and Chun-Xun
 //   - refactored the code
 //  
-// TODO:
-//   - Problems can occur when external threads insert tasks during spawn.
-//
 // 2018/09/12 - created by Tsung-Wei Huang and Chun-Xun Lin
 //
 // Implemented PrivatizedThreadpool using the data structre inspired
 // Eigen CXX/Threadpool.
 
+// TODO
+// - double check whether we can use std::forward<Args>(args)... in enqueue/dequeue
+// - can we replace lock with CAS
+// - refactored the WorkQueue class ...
+// - add more example to threadpool and use std::future to mimic the control flow
+// - atomic add problem (extremely slow)
+
 #pragma once
 
 #include <iostream>
@@ -30,10 +37,13 @@ namespace tf {
 
 // ---------------------------------------------------------------------------- 
 // Privatized queue of worker. The lock-free queue is inspired by 
-//   http://www.1024cores.net/home/lock-free-algorithms/queues/bounded-mpmc-queue
+// http://www.1024cores.net/home/lock-free-algorithms/queues/bounded-mpmc-queue
 template<typename T, size_t buffer_size>
 class PrivatizedTaskQueue {
-public:
+
+  public:
+
+  // TODO replaced the new with std::array
   PrivatizedTaskQueue()
     : buffer_(new cell_t [buffer_size])
     , buffer_mask_(buffer_size - 1)
@@ -106,7 +116,8 @@ class PrivatizedTaskQueue {
            dequeue_pos_.load(std::memory_order_relaxed);
   };
 
-private:
+  private:
+  
   struct cell_t {
     std::atomic<size_t>   sequence_;
     T                     data_;
@@ -123,12 +134,10 @@ class PrivatizedTaskQueue {
   cacheline_pad_t         pad2_;
   std::atomic<size_t>     dequeue_pos_;
   cacheline_pad_t         pad3_;
-
-  PrivatizedTaskQueue(PrivatizedTaskQueue const&);
-  void operator = (PrivatizedTaskQueue const&);
 }; 
 
 
+// ----------------------------------------------------------------------------
 
 // Class: PrivatizedThreadpool
 template <typename Task>
@@ -159,13 +168,12 @@ class PrivatizedThreadpool {
     const std::thread::id _owner {std::this_thread::get_id()};
 
     mutable std::mutex _mutex;
-    std::condition_variable _empty_cv;
 
     std::vector<Task> _tasks;
     std::vector<std::thread> _threads;
+    std::vector<Worker> _workers;
     
     std::unordered_map<std::thread::id, size_t> _worker_maps;
-    std::vector<Worker> _workers;
 
     size_t _num_idlers {0};
     size_t _next_queue {0};
@@ -254,22 +262,15 @@ size_t PrivatizedThreadpool<Task>::num_workers() const {
 template <typename Task>
 void PrivatizedThreadpool<Task>::_shutdown(){
 
-  if(!is_owner()){
-    throw std::runtime_error("Worker thread cannot shut down the pool");
-  }
-
-  if(_threads.empty()) {
-    return;
-  }
+  assert(is_owner());
 
   {
-    std::scoped_lock<std::mutex> lock(_mutex);
-    // Notify workers to exit
+    std::scoped_lock lock(_mutex);
     for(auto& w : _workers){
       w.exit = true;
       w.cv.notify_one();
     }
-  } // Release lock
+  } 
 
   for(auto& t : _threads){
     t.join();
@@ -284,9 +285,7 @@ void PrivatizedThreadpool<Task>::_shutdown(){
 template <typename Task>
 void PrivatizedThreadpool<Task>::_spawn(unsigned N) {
 
-  if(! is_owner()){
-    throw std::runtime_error("Worker thread cannot spawn threads");
-  }
+  assert(is_owner());
 
   // Lock to synchronize all workers before creating _worker_mapss
   std::scoped_lock lock(_mutex);
@@ -329,10 +328,15 @@ void PrivatizedThreadpool<Task>::_spawn(unsigned N) {
           lock.unlock();
         } // End of pop_front 
 
-        while(t){
+        while(t) {
           (*t)();
-          std::swap(t, w.cache);
-          w.cache = std::nullopt;
+          if(w.cache) {
+            t = std::move(w.cache);
+            w.cache = std::nullopt;
+          }
+          else {
+            t = std::nullopt;
+          }
         }
       } // End of while ------------------------------------------------------
     });     
@@ -353,15 +357,17 @@ void PrivatizedThreadpool<Task>::emplace(ArgsT&&... args){
   }
 
   Task t {std::forward<ArgsT>(args)...};
-
+  
+  // caller is not the owner
   if(auto tid = std::this_thread::get_id(); tid != _owner){
+    // the caller is the worker of the threadpool
     if(auto itr = _worker_maps.find(tid); itr != _worker_maps.end()){
       if(!_workers[itr->second].cache.has_value()){
         _workers[itr->second].cache = std::move(t);
         return ;
       }
       if(!_workers[itr->second].queue.enqueue(t)){
-        std::scoped_lock<std::mutex> lock(_mutex);       
+        std::scoped_lock lock(_mutex);       
         _tasks.push_back(std::move(t));
       }
       return ;
