vreThreadPool.h

Go to the documentation of this file.

/*******************************************************************************
** Copyright (c) 2025 MAK Technologies
** All rights reserved.
*******************************************************************************/
 
#pragma once
 
#include "vreUtil/export.h"
 
#include <functional>
#include <thread>
#include <atomic>
#include <vector>
#include <memory>
#include <future>
#include <mutex>
#include <queue>
#include <chrono>
 
namespace makVre
{
//! \brief A thread pool class for managing and executing asynchronous tasks
//!
//! This class provides a pool of worker threads that can execute jobs
//! asynchronously. It manages thread creation, job submission, and
//! thread synchronization. Jobs can be submitted as functions with or
//! without arguments, and the results can be retrieved using futures.
class UTIL_DLL DtVreThreadPool
{
public:
   //! \brief Function type for jobs to be executed by the thread pool
   using JobType = std::function<void()>;
 
public:
   //! \brief Constructor that initializes the thread pool with a specified number of threads
   //!
   //! \param threads The number of worker threads to create in the pool
   explicit DtVreThreadPool(std::size_t threads);
 
   //! \brief Destructor that shuts down the thread pool
   //!
   //! Calls shutdown(true) to ensure all threads are properly joined
   ~DtVreThreadPool();
 
   //! \brief Copy constructor (deleted)
   DtVreThreadPool(const DtVreThreadPool&) = delete;
 
   //! \brief Move constructor (deleted)
   DtVreThreadPool(DtVreThreadPool&&) = delete;
 
   //! \brief Copy assignment operator (deleted)
   DtVreThreadPool& operator=(const DtVreThreadPool&) = delete;
 
   //! \brief Move assignment operator (deleted)
   DtVreThreadPool& operator=(DtVreThreadPool&&) = delete;
 
   //! \brief Shuts down the thread pool
   //!
   //! Stops all worker threads and optionally waits for them to complete
   //! their current tasks.
   //!
   //! \param join If true, waits for all threads to finish; if false, detaches them
   void shutdown(bool join = true);
 
   //! \brief Returns the number of threads in the pool
   //!
   //! \return The total number of threads managed by this pool
   std::size_t threadCount() const;
 
   //! \brief Resizes the thread pool to a new number of threads
   //!
   //! Shuts down the current pool and creates a new one with the specified number of threads
   //!
   //! \param nThreads The new number of threads to use
   void resize(std::size_t nThreads);
 
   //! \brief Returns the number of currently active (working) threads
   //!
   //! \return The number of threads currently executing jobs
   std::size_t activeCount() const;
 
   //! \brief Submits a job with arguments to the thread pool
   //!
   //! \param f The function to execute
   //! \param args The arguments to pass to the function
   //! \return A future that can be used to retrieve the result of the function
   template <typename F, typename... Args, typename R = typename std::result_of<F(Args...)>::type>
   std::future<R> submitJob(F&& f, Args&&... args);
 
   //! \brief Submits a job without arguments to the thread pool
   //!
   //! \param f The function to execute
   //! \return A future that can be used to retrieve the result of the function
   template <typename F, typename R = typename std::result_of<F()>::type>
   std::future<R> submitJob(F&& f);
 
   //! \brief Checks if a future is ready (completed)
   //!
   //! \param f The future to check
   //! \return true if the future is ready, false otherwise
   template <typename R>
   static bool isFutureReady(std::future<R> const& f);
 
   //! \brief Gets the status of a future
   //!
   //! \param f The future to check
   //! \return The status of the future (ready, timeout, or deferred)
   template <typename R>
   static std::future_status getFutureStatus(std::future<R> const& f);
 
   //! \brief Checks if a shared future is ready (completed)
   //!
   //! \param f The shared future to check
   //! \return true if the shared future is ready, false otherwise
   template <typename R>
   static bool isSharedFutureReady(std::shared_future<R> const& f);
 
   //! \brief Gets the status of a shared future
   //!
   //! \param f The shared future to check
   //! \return The status of the shared future (ready, timeout, or deferred)
   template <typename R>
   static std::future_status getSharedFutureStatus(std::shared_future<R> const& f);
 
protected:
   //! \brief Gets the next job from the queue
   //!
   //! Waits until a job is available or the thread pool is stopped
   //!
   //! \return The next job to execute, or an empty function if the pool is stopped
   JobType getNextJob();
 
   //! \brief Empties the job queue
   //!
   //! Removes all pending jobs from the queue
   void emptyQueue();
 
   //! \brief Creates worker threads for the pool
   //!
   //! \param nThreads The number of threads to create
   void createThreads(std::size_t nThreads);
 
protected:
   //! \brief Collection of worker threads
   std::vector<std::unique_ptr<std::thread>> myThreads;
 
   //! \brief Queue of pending jobs
   std::queue<JobType> myJobQueue;
 
   //! \brief Flag indicating if the thread pool should stop
   std::atomic<bool> myStop;
 
   //! \brief Counter for the number of active threads
   std::atomic<std::size_t> myActiveThreads;
 
   //! \brief Mutex for synchronizing access to the job queue
   std::mutex myMutex;
 
   //! \brief Condition variable for notifying threads of new jobs
   std::condition_variable myNotifier;
};
 
 
template <typename F, typename R>
std::future<R> DtVreThreadPool::submitJob(F&& f)
{
   auto job = std::make_shared<std::packaged_task<R()>>(std::forward<F>(f));
 
   {
      std::unique_lock<std::mutex> lock(myMutex);
      myJobQueue.emplace(
         [job]()
         {
            (*job)();
         });
   }
 
   myNotifier.notify_one();
 
   return job->get_future();
}
 
template <typename F, typename... Args, typename R>
std::future<R> DtVreThreadPool::submitJob(F&& f, Args&&... args)
{
   auto job = std::make_shared<std::packaged_task<R()>>(std::bind(std::forward<F>(f), std::forward<Args>(args)...));
 
   {
      std::unique_lock<std::mutex> lock(myMutex);
      myJobQueue.emplace(
         [job]()
         {
            (*job)();
         });
   }
 
   myNotifier.notify_one();
 
   return job->get_future();
}
 
template <typename R>
bool DtVreThreadPool::isFutureReady(std::future<R> const& f)
{
   return f.wait_for(std::chrono::seconds(0)) == std::future_status::ready;
}
 
template <typename R>
std::future_status DtVreThreadPool::getFutureStatus(std::future<R> const& f)
{
   return f.wait_for(std::chrono::seconds(0));
}
 
template <typename R>
bool DtVreThreadPool::isSharedFutureReady(std::shared_future<R> const& f)
{
   return f.wait_for(std::chrono::seconds(0)) == std::future_status::ready;
}
 
template <typename R>
std::future_status DtVreThreadPool::getSharedFutureStatus(std::shared_future<R> const& f)
{
   return f.wait_for(std::chrono::seconds(0));
}
} // namespace makVre