updated work stealing queue

tsung-wei-huang · tsung-wei-huang · commit 1b6735b6440b · 2018-12-06T12:39:29.000-06:00
diff --git a/example/simple.cpp b/example/simple.cpp
@@ -6,15 +6,8 @@
 #include <taskflow/taskflow.hpp>  // the only include you need
 #include <new>
 
-struct keep_apart {
-  alignas(64) std::atomic<int> cat;
-  alignas(64) std::atomic<int> dog;
-};
-
 int main(){
 
-  std::cout << sizeof(keep_apart);
-
   tf::Taskflow tf;
   
   auto [A, B, C, D] = tf.silent_emplace(   //  the taskflow graph
diff --git a/taskflow/threadpool/privatized_threadpool.hpp b/taskflow/threadpool/privatized_threadpool.hpp
@@ -1,3 +1,5 @@
+// 2018/12/06 - this class is left for probation
+//
 // 2018/11/28 - modified by Chun-Xun Lin
 // 
 // Added the method batch to insert a vector of tasks.
diff --git a/taskflow/threadpool/speculative_threadpool.hpp b/taskflow/threadpool/speculative_threadpool.hpp
@@ -241,7 +241,10 @@ void SpeculativeThreadpool<Closure>::emplace(ArgsT&&... args) {
 
 template <typename Closure>
 void SpeculativeThreadpool<Closure>::batch(std::vector<Closure>&& tasks){
-  size_t consumed {0};
+
+  if(tasks.empty()) {
+    return;
+  }
 
   //no worker thread available
   if(num_workers() == 0){
@@ -250,6 +253,8 @@ void SpeculativeThreadpool<Closure>::batch(std::vector<Closure>&& tasks){
     }
     return;
   }
+  
+  size_t consumed {0};
 
   // speculation
   if(std::this_thread::get_id() != _owner){
diff --git a/taskflow/threadpool/workstealing_threadpool.hpp b/taskflow/threadpool/workstealing_threadpool.hpp
@@ -227,6 +227,7 @@ class WorkStealingThreadpool {
   struct Worker {
     std::condition_variable cv;
     WorkStealingQueue<Closure> queue;
+    std::optional<Closure> cache;
     bool exit  {false};
     bool ready {false};
     uint64_t seed;
@@ -256,7 +257,6 @@ class WorkStealingThreadpool {
     mutable std::mutex _mutex;
 
     std::vector<Worker> _workers;
-    //std::vector<Worker*> _idlers;
     std::vector<std::thread> _threads;
 
     std::unordered_map<std::thread::id, unsigned> _worker_maps;
@@ -366,18 +366,27 @@ void WorkStealingThreadpool<Closure>::_spawn(unsigned N) {
           lock.lock();
           if(_queue.empty()) {
             w.ready = false;
-            //_idlers.push_back(&w);
             _idlers.push(&w);
             while(!w.ready && !w.exit) {
               w.cv.wait(lock);
             }
           }
           lock.unlock();
+
+          if(w.cache) {
+            std::swap(t, w.cache);
+          }
         }
 
         while(t) {
           (*t)();
-          t = w.queue.pop();
+          if(w.cache) {
+            t = std::move(*w.cache);
+            w.cache = std::nullopt;
+          }
+          else {
+            t = w.queue.pop();
+          }
         }
       } // End of while ------------------------------------------------------ 
 
@@ -409,6 +418,7 @@ void WorkStealingThreadpool<Closure>::_balance_load(unsigned me) {
     if(auto idler = _idlers.steal(); idler) {
       (*idler)->ready = true;
       (*idler)->victim_hint = me;
+      (*idler)->cache = _workers[me].queue.steal();
       (*idler)->cv.notify_one();
       factor += load_balancing_factor;
     }
@@ -473,7 +483,12 @@ void WorkStealingThreadpool<Closure>::emplace(ArgsT&&... args){
 
       unsigned me = itr->second;
 
-      _workers[me].queue.push(Closure{std::forward<ArgsT>(args)...});
+      if(_workers[me].cache) {
+        _workers[me].queue.push(Closure{std::forward<ArgsT>(args)...});
+      }
+      else {
+        _workers[me].cache.emplace(std::forward<ArgsT>(args)...);
+      }
       
       // load balancing
       _balance_load(me);
@@ -484,7 +499,7 @@ void WorkStealingThreadpool<Closure>::emplace(ArgsT&&... args){
 
   if(auto idler = _idlers.steal(); idler) {
     (*idler)->ready = true;
-    (*idler)->queue.push(Closure{std::forward<ArgsT>(args)...});
+    (*idler)->cache.emplace(std::forward<ArgsT>(args)...);
     (*idler)->cv.notify_one(); 
   }
   else {
@@ -497,6 +512,10 @@ void WorkStealingThreadpool<Closure>::emplace(ArgsT&&... args){
 template <typename Closure>
 void WorkStealingThreadpool<Closure>::batch(std::vector<Closure>&& tasks) {
 
+  if(tasks.empty()) {
+    return;
+  }
+
   //no worker thread available
   if(num_workers() == 0){
     for(auto &t: tasks){
@@ -512,9 +531,15 @@ void WorkStealingThreadpool<Closure>::batch(std::vector<Closure>&& tasks) {
     if(auto itr = _worker_maps.find(tid); itr != _worker_maps.end()){
 
       unsigned me = itr->second;
+      
+      size_t i = 0;
 
-      for(auto& t : tasks) {
-        _workers[me].queue.push(std::move(t));
+      if(!_workers[me].cache) {
+        _workers[me].cache = std::move(tasks[i++]);
+      }
+
+      for(; i<tasks.size(); ++i) {
+        _workers[me].queue.push(std::move(tasks[i]));
       }
       
       // load balancing 
@@ -535,11 +560,10 @@ void WorkStealingThreadpool<Closure>::batch(std::vector<Closure>&& tasks) {
   while(!_queue.empty()) {
     if(auto idler = _idlers.steal(); idler) {
       (*idler)->ready = true;
+      (*idler)->cache = _queue.steal();
       (*idler)->cv.notify_one();
     }
-    else {
-      break;
-    }
+    else break;
   }
 } 
 

Original file line number	Diff line number	Diff line change
`@@ -1,3 +1,5 @@`
	`1`	`+// 2018/12/06 - this class is left for probation`
	`2`	`+//`
`1`	`3`	`// 2018/11/28 - modified by Chun-Xun Lin`
`2`	`4`	`//`
`3`	`5`	`// Added the method batch to insert a vector of tasks.`