Round Robin Queue Servicing support

Changes from inkooboo#24 are incorporated. Conditional variables instead spin-lock are kept as well.

Author: yvoinov

include/thread_pool/thread_pool.hpp

@@ -32,6 +32,7 @@ static bool v_affinity = false;	/* Default: disabled */
 
 template <typename Task, template<typename> class Queue>
 class ThreadPoolImpl;
+
 using ThreadPool = ThreadPoolImpl<FixedFunction<void(), 128>,
                                   MPMCBoundedQueue>;
 
@@ -45,6 +46,9 @@ using ThreadPool = ThreadPoolImpl<FixedFunction<void(), 128>,
  */
 template <typename Task, template<typename> class Queue>
 class ThreadPoolImpl {
+
+using WorkerVector = std::vector<std::unique_ptr<Worker<Task, Queue>>>;
+
 public:
     /**
      * @brief ThreadPool Construct and start new thread pool.
@@ -90,7 +94,7 @@ class ThreadPoolImpl {
 private:
     Worker<Task, Queue>& getWorker();
 
-    std::vector<std::unique_ptr<Worker<Task, Queue>>> m_workers;
+    WorkerVector m_workers;
     std::atomic<std::size_t> m_next_worker;
 
     #if defined __sun__ || defined __linux__ || defined __FreeBSD__
@@ -129,9 +133,6 @@ inline ThreadPoolImpl<Task, Queue>::ThreadPoolImpl(
 
     for(std::size_t i = 0; i < m_workers.size(); ++i)
     {
-        Worker<Task, Queue>* steal_donor =
-                                m_workers[(i + 1) % m_workers.size()].get();
-
 	#if defined __sun__ || defined __linux__ || defined __FreeBSD__
         if (v_affinity) {
             if (v_cpu > v_cpu_max)
@@ -160,7 +161,7 @@ inline ThreadPoolImpl<Task, Queue>::ThreadPoolImpl(
         }
 	#endif
 
-        m_workers[i]->start(i, steal_donor);
+        m_workers[i]->start(i, &m_workers);
     }
 }
 
@@ -195,7 +196,7 @@ template <typename Task, template<typename> class Queue>
 template <typename Handler>
 inline bool ThreadPoolImpl<Task, Queue>::tryPost(Handler&& handler)
 {
-    return getWorker().post(std::forward<Handler>(handler));
+    return getWorker().tryPost(std::forward<Handler>(handler));
 }
 
 template <typename Task, template<typename> class Queue>

include/thread_pool/worker.hpp

@@ -4,6 +4,7 @@
 #include <thread>
 #include <condition_variable>
 #include <mutex>
+#include <limits>
 
 namespace tp
 {
@@ -17,6 +18,8 @@ namespace tp
 template <typename Task, template<typename> class Queue>
 class Worker
 {
+    using WorkerVector = std::vector<std::unique_ptr<Worker<Task, Queue>>>;
+
 public:
     /**
      * @brief Worker Constructor.
@@ -37,9 +40,9 @@ class Worker
     /**
      * @brief start Create the executing thread and start tasks execution.
      * @param id Worker ID.
-     * @param steal_donor Sibling worker to steal task from it.
+     * @param workers Sibling workers for performing round robin work stealing.
      */
-    void start(std::size_t id, Worker* steal_donor);
+    void start(std::size_t id, WorkerVector* workers);
 
     /**
      * @brief stop Stop all worker's thread and stealing activity.
@@ -48,19 +51,19 @@ class Worker
     void stop();
 
     /**
-     * @brief post Post task to queue.
+     * @brief tryPost Post task to queue.
      * @param handler Handler to be executed in executing thread.
      * @return true on success.
      */
     template <typename Handler>
-    bool post(Handler&& handler);
+    bool tryPost(Handler&& handler);
 
     /**
-     * @brief steal Steal one task from this worker queue.
-     * @param task Place for stealed task to be stored.
+     * @brief tryGetLocalTask Get one task from this worker queue.
+     * @param task Place for the obtained task to be stored.
      * @return true on success.
      */
-    bool steal(Task& task);
+    bool tryGetLocalTask(Task& task);
 
     /**
      * @brief getWorkerIdForCurrentThread Return worker ID associated with
@@ -70,16 +73,24 @@ class Worker
     static std::size_t getWorkerIdForCurrentThread();
 
 private:
+    /**
+    * @brief tryRoundRobinSteal Try stealing a thread from sibling workers in a round-robin fashion.
+    * @param task Place for the obtained task to be stored.
+    * @param workers Sibling workers for performing round robin work stealing.
+    */
+    bool tryRoundRobinSteal(Task& task, WorkerVector* workers);
+
     /**
      * @brief threadFunc Executing thread function.
      * @param id Worker ID to be associated with this thread.
-     * @param steal_donor Sibling worker to steal task from it.
+     * @param workers Sibling workers for performing round robin work stealing.
      */
-    void threadFunc(std::size_t id, Worker* steal_donor);
+    void threadFunc(size_t id, WorkerVector* workers);
 
     Queue<Task> m_queue;
     std::atomic<bool> m_running_flag;
     std::thread m_thread;
+    std::size_t m_next_donor;
     std::mutex m_conditional_mutex;
     std::condition_variable m_conditional_lock;
 };
@@ -91,7 +102,7 @@ namespace detail
 {
     inline std::size_t* thread_id()
     {
-        static thread_local std::size_t tss_id = -1u;
+        static thread_local std::size_t tss_id = std::numeric_limits<std::size_t>::max();
         return &tss_id;
     }
 }
@@ -100,6 +111,7 @@ template <typename Task, template<typename> class Queue>
 inline Worker<Task, Queue>::Worker(std::size_t queue_size)
     : m_queue(queue_size)
     , m_running_flag(true)
+    , m_next_donor(0) // Initialized in threadFunc.
 {
 }
 
@@ -132,9 +144,9 @@ inline void Worker<Task, Queue>::stop()
 }
 
 template <typename Task, template<typename> class Queue>
-inline void Worker<Task, Queue>::start(std::size_t id, Worker* steal_donor)
+inline void Worker<Task, Queue>::start(std::size_t id, WorkerVector* workers)
 {
-    m_thread = std::thread(&Worker<Task, Queue>::threadFunc, this, id, steal_donor);
+    m_thread = std::thread(&Worker<Task, Queue>::threadFunc, this, id, workers);
 }
 
 template <typename Task, template<typename> class Queue>
@@ -145,36 +157,58 @@ inline std::size_t Worker<Task, Queue>::getWorkerIdForCurrentThread()
 
 template <typename Task, template<typename> class Queue>
 template <typename Handler>
-inline bool Worker<Task, Queue>::post(Handler&& handler)
+inline bool Worker<Task, Queue>::tryPost(Handler&& handler)
 {
     m_conditional_lock.notify_all();
     return m_queue.push(std::forward<Handler>(handler));
 }
 
 template <typename Task, template<typename> class Queue>
-inline bool Worker<Task, Queue>::steal(Task& task)
+inline bool Worker<Task, Queue>::tryGetLocalTask(Task& task)
 {
     return m_queue.pop(task);
 }
 
 template <typename Task, template<typename> class Queue>
-inline void Worker<Task, Queue>::threadFunc(std::size_t id, Worker* steal_donor)
+inline bool Worker<Task, Queue>::tryRoundRobinSteal(Task& task, WorkerVector* workers)
+{
+    auto starting_index = m_next_donor;
+    // Iterate once through the worker ring, checking for queued work items on each thread.
+    do
+    {
+        // Don't steal from local queue.
+        if (m_next_donor != *detail::thread_id() && workers->at(m_next_donor)->tryGetLocalTask(task))
+        {
+            // Increment before returning so that m_next_donor always points to the worker that has gone the longest
+            // without a steal attempt. This helps enforce fairness in the stealing.
+            ++m_next_donor %= workers->size();
+            return true;
+        }
+        ++m_next_donor %= workers->size();
+    } while (m_next_donor != starting_index);
+    return false;
+}
+
+template <typename Task, template<typename> class Queue>
+inline void Worker<Task, Queue>::threadFunc(size_t id, WorkerVector* workers)
 {
     *detail::thread_id() = id;
+    m_next_donor = ++id % workers->size();
 
     Task handler;
 
     while (m_running_flag.load(std::memory_order_relaxed))
     {
-        if (m_queue.pop(handler) || steal_donor->steal(handler))
+        // Prioritize local queue, then try stealing from sibling workers.
+        if (tryGetLocalTask(handler) || tryRoundRobinSteal(handler, workers))
         {
             try
             {
                 handler();
             }
             catch(...)
             {
-                // suppress all exceptions
+                // Suppress all exceptions.
             }
         }
         else