A work-stealing multiqueue threadpool. Each worker thread has a thread_local task queue to reduce thread contention. More...

#include <flow_multi_queue_thread_pool.h>

Public Member Functions
	MultiQueueThreadPool (std::size_t threadCount=std::thread::hardware_concurrency() - 1)
std::size_t	poolSize () const
template<typename Callable, typename ... Args>
std::future< std::invoke_result_t< Callable, Args &&... > >	submit (Callable callable, Args &&... args)
	Submit a task to the thread-local queue. If the task is submit by an external thread outside of the thread pool, then it submits to the pool queue shared by all threads instead.
void	runPendingTask ()
	Try run a task from the local queue. If the local queue is empty, it tries to run a task from the pool queue. If the pool queue is empty, it tries to steal the work from other thread's local queue. If all queues are empty, then it yields the thread. This helps resolves thread deadlock due to dependency waiting.

Private Member Functions
void	runTasks (std::stop_token token, std::size_t id)

Private Attributes
flow::Vector< std::jthread >	threads_
flow::Vector< LocalQueue >	localQueues_
ThreadSafeQueue	poolQueue_

Static Private Attributes
static thread_local std::size_t	threadId = 0
static thread_local LocalQueue *	localQueue = nullptr

Detailed Description

template<typename ThreadSafeQueue = ConcurrentQueue<ThreadTask>, typename LocalQueue = WorkStealingQueue<ThreadTask>>
class flow::MultiQueueThreadPool< ThreadSafeQueue, LocalQueue >

A work-stealing multiqueue threadpool. Each worker thread has a thread_local task queue to reduce thread contention.

Definition at line 15 of file flow_multi_queue_thread_pool.h.

Constructor & Destructor Documentation

◆ MultiQueueThreadPool()

template<typename ThreadSafeQueue = ConcurrentQueue<ThreadTask>, typename LocalQueue = WorkStealingQueue<ThreadTask>>

flow::MultiQueueThreadPool< ThreadSafeQueue, LocalQueue >::MultiQueueThreadPool ( std::size_t threadCount = std::thread::hardware_concurrency() - 1 )

inlineexplicit

Definition at line 24 of file flow_multi_queue_thread_pool.h.

      : localQueues_(threadCount) {
 
      threads_.reserve(threadCount);
      for (std::size_t i = 0; i < threadCount; ++i) {
        threads_.emplaceBack(std::bind(&MultiQueueThreadPool::runTasks, this, std::placeholders::_1, i));
      }
    }

References localQueues_, runTasks(), and threads_.

Member Function Documentation

◆ poolSize()

template<typename ThreadSafeQueue = ConcurrentQueue<ThreadTask>, typename LocalQueue = WorkStealingQueue<ThreadTask>>

std::size_t flow::MultiQueueThreadPool< ThreadSafeQueue, LocalQueue >::poolSize ( ) const

inline

Returns: The number of worker threads.

Definition at line 34 of file flow_multi_queue_thread_pool.h.

                               {
      return threads_.size();
    }

References threads_.

Referenced by runPendingTask().

◆ runPendingTask()

template<typename ThreadSafeQueue = ConcurrentQueue<ThreadTask>, typename LocalQueue = WorkStealingQueue<ThreadTask>>

void flow::MultiQueueThreadPool< ThreadSafeQueue, LocalQueue >::runPendingTask ( )

inline

Try run a task from the local queue. If the local queue is empty, it tries to run a task from the pool queue. If the pool queue is empty, it tries to steal the work from other thread's local queue. If all queues are empty, then it yields the thread. This helps resolves thread deadlock due to dependency waiting.

Definition at line 63 of file flow_multi_queue_thread_pool.h.

                          {
 
      // Pop from local queue.
      if (localQueue) {
        if (auto task = localQueue->tryPop(); task) {
          task->execute();
          return;
        }
      }
 
      // Pop from the pool queue.
      if (auto task = poolQueue_.tryPop(); task) {
        task->execute();
        return;
      }
 
      // Steal from other queue.
      std::size_t threadCount = poolSize();
      for (std::size_t i = 0; i < threadCount; ++i) {
        std::size_t id = (threadId + 1 + i) % threadCount;
        if (auto task = localQueues_[id].trySteal(); task) {
          task->execute();
          return;
        }
      }
 
      std::this_thread::yield();
    }

References localQueue, localQueues_, poolQueue_, poolSize(), and threadId.

Referenced by runTasks().

◆ runTasks()

template<typename ThreadSafeQueue = ConcurrentQueue<ThreadTask>, typename LocalQueue = WorkStealingQueue<ThreadTask>>

void flow::MultiQueueThreadPool< ThreadSafeQueue, LocalQueue >::runTasks	(	std::stop_token	token,
		std::size_t	id )

inlineprivate

Definition at line 93 of file flow_multi_queue_thread_pool.h.

                                                     {
      threadId = id;
      localQueue = &localQueues_[id];
 
      while (!token.stop_requested()) {
        runPendingTask();
      }
    }

References localQueue, localQueues_, runPendingTask(), and threadId.

Referenced by MultiQueueThreadPool().

◆ submit()

template<typename ThreadSafeQueue = ConcurrentQueue<ThreadTask>, typename LocalQueue = WorkStealingQueue<ThreadTask>>

template<typename Callable, typename ... Args>

std::future< std::invoke_result_t< Callable, Args &&... > > flow::MultiQueueThreadPool< ThreadSafeQueue, LocalQueue >::submit	(	Callable	callable,
		Args &&...	args )

inline

Submit a task to the thread-local queue. If the task is submit by an external thread outside of the thread pool, then it submits to the pool queue shared by all threads instead.

Parameters

callable	The callable.
...args	The callable arguments.

Definition at line 44 of file flow_multi_queue_thread_pool.h.

                                                                                                 {
      using ReturnType = std::invoke_result_t<Callable, Args&&...>;
      std::packaged_task<ReturnType()> task(std::bind(std::move(callable), std::forward<Args>(args)...));
      auto result = task.get_future();
 
      if (localQueue) {
        localQueue->push(ThreadTask(std::move(task)));
      } else {
        poolQueue_.push(ThreadTask(std::move(task)));
      }
      
      return result;
    }

References localQueue, and poolQueue_.

Member Data Documentation

◆ localQueue

template<typename ThreadSafeQueue = ConcurrentQueue<ThreadTask>, typename LocalQueue = WorkStealingQueue<ThreadTask>>

thread_local LocalQueue* flow::MultiQueueThreadPool< ThreadSafeQueue, LocalQueue >::localQueue = nullptr

inlinestaticprivate

Definition at line 21 of file flow_multi_queue_thread_pool.h.

Referenced by runPendingTask(), runTasks(), and submit().

◆ localQueues_

template<typename ThreadSafeQueue = ConcurrentQueue<ThreadTask>, typename LocalQueue = WorkStealingQueue<ThreadTask>>

flow::Vector<LocalQueue> flow::MultiQueueThreadPool< ThreadSafeQueue, LocalQueue >::localQueues_

private

Definition at line 17 of file flow_multi_queue_thread_pool.h.

Referenced by MultiQueueThreadPool(), runPendingTask(), and runTasks().

◆ poolQueue_

template<typename ThreadSafeQueue = ConcurrentQueue<ThreadTask>, typename LocalQueue = WorkStealingQueue<ThreadTask>>

ThreadSafeQueue flow::MultiQueueThreadPool< ThreadSafeQueue, LocalQueue >::poolQueue_

private

Definition at line 18 of file flow_multi_queue_thread_pool.h.

Referenced by runPendingTask(), and submit().

◆ threadId

template<typename ThreadSafeQueue = ConcurrentQueue<ThreadTask>, typename LocalQueue = WorkStealingQueue<ThreadTask>>

thread_local std::size_t flow::MultiQueueThreadPool< ThreadSafeQueue, LocalQueue >::threadId = 0

inlinestaticprivate

Definition at line 20 of file flow_multi_queue_thread_pool.h.

Referenced by runPendingTask(), and runTasks().

◆ threads_

template<typename ThreadSafeQueue = ConcurrentQueue<ThreadTask>, typename LocalQueue = WorkStealingQueue<ThreadTask>>

flow::Vector<std::jthread> flow::MultiQueueThreadPool< ThreadSafeQueue, LocalQueue >::threads_

private

Definition at line 16 of file flow_multi_queue_thread_pool.h.

Referenced by MultiQueueThreadPool(), and poolSize().

The documentation for this class was generated from the following file:

flow_multi_queue_thread_pool.h

Public Member Functions

Private Member Functions

Private Attributes

Static Private Attributes

Detailed Description

Constructor & Destructor Documentation

◆ MultiQueueThreadPool()

Member Function Documentation

◆ poolSize()

◆ runPendingTask()

◆ runTasks()

◆ submit()

Member Data Documentation

◆ localQueue

◆ localQueues_

◆ poolQueue_

◆ threadId

◆ threads_