11
\$\begingroup\$

I have developed a threadpool using the standard c++11 features and am looking for feedback.

Right now I think the implementation is pretty solid, but as I recently learned about multithreading I'm not sure if I got a solid grasp on it.

Code

Threadpool.h

#pragma once

#include <thread>
#include <future>
#include <condition_variable>
#include <mutex>
#include <list>
#include <memory>
#include <cassert>
#include <algorithm>

namespace threadpool {

  template <class Result> class TaskHandler;
  class Threadpool {
  public:
    // Defines how the tasks will be ordered by default
    enum TaskOrdering { 
      FIFO, // First In First OUT
      LIFO  // Last In First OUT
    };

    // Priority with which a task will be queued
    enum TaskPriority { 
      MAX,      // The task will be the next one to execute
      DEFAULT,  // FIFO or LIFO
      MIN       // The task will be put past all stored tasks
    };


    /**
     * @brief Create a threadpool holding a certain number of threads
     * 
     * @param int number of workers
     * @param ordering default ordering function for tasks
     *                 FIFO = First In First Out
     *                 LIFO = Last In First Out
     */
    explicit Threadpool(unsigned int nWorkers, TaskOrdering ordering = FIFO) :
      mOrdering(ordering),
      mThreadsToKill(0),
      mNWorkers(nWorkers),
      mPoolDestroyed(false)
    {
      for (unsigned int i = 0; i < nWorkers; ++i) {
        mWorkers.emplace_back(&Threadpool::workerFunction, this);
      }
    }

    /**
     * @brief Join all worker threads, leaving all remaining tasks undone
     */
    ~Threadpool() {
      {
        std::lock_guard<std::mutex> lock(mTasksMutex);
        mPoolDestroyed = true;
        // cancel possible thread suicides
        mThreadsToKill = 0;
      }
      mTasksCond.notify_all();
      for (auto it = std::begin(mWorkers); it != std::end(mWorkers); ++it) {
        it->join();
      }
    }

    /**
     * @brief Queue a task and return a handle to it
     * 
     * @param fn Callable
     * @param args Arguments for fn
     * @tparam Priority
     * @return Handle to the task
     */
    template <TaskPriority Priority = DEFAULT, class FN, class... Args>
    auto queueAndHandleTask(FN&& fn, Args&&... args)
      -> TaskHandler<typename std::result_of<FN(Args...)>::type>
    {
      using return_type = typename std::result_of<FN(Args...)>::type;
      auto task = std::make_shared<std::packaged_task<return_type()>>(
        std::bind(std::forward<FN>(fn), std::forward<Args>(args)...)
      );

      queueTask<Priority>([task](){ (*task)(); });

      return TaskHandler<return_type>(task);
    }

    /**
     * @brief Queue a task
     * 
     * @param fn Callable
     * @param args Arguments for fn
     * @tparam Priority 
     */
    template <TaskPriority Priority = DEFAULT, class FN, class... Args>
    void queueTask(FN&& fn, Args&&... args) {
      {
        std::lock_guard<std::mutex> lock(mTasksMutex);
        auto insertPosition = getInsertPosition(Priority, mOrdering);
        mTasks.emplace(insertPosition, std::bind(std::forward<FN>(fn), std::forward<Args>(args)...));
      }
      mTasksCond.notify_one();
    }

    /**
     * @brief Changes the default task ordering for the threadpool
     * @param newOrdering 
     */
    void setDefaultOrdering(TaskOrdering newOrdering) { mOrdering = newOrdering; }

    /**
     * @brief Change the number of threads managed by the threadpool
     * @param int new thread number
     */
    void resize(unsigned int newWorkers) {
      int workerDiff = newWorkers - mNWorkers;
      mNWorkers = newWorkers;

      if (workerDiff > 0) { // add new threads
        std::lock_guard<std::mutex> lock(mTasksMutex);
        for (int i = 0; i < workerDiff; ++i) {
          mWorkers.emplace_back(&Threadpool::workerFunction, this);
        }
      }
      else if (workerDiff < 0) { // reduce threads
        {
          std::lock_guard<std::mutex> lock(mTasksMutex);
          // dont assign in case resize is called while there are worker threads left to kill
          mThreadsToKill += -workerDiff;
        }
        // 'mThreadsToKill' not used on purpose to prevent data races
        // when threads get notified while there are threads to notify yet
        for (int i = 0; i < -workerDiff; ++i) {
          mTasksCond.notify_one();
        }
      }
    }

    /**
     * @brief Blocks calling thread until there are no tasks left to execute
     */
    void waitForTasks() {
      std::unique_lock<std::mutex> lock(mWaitMutex);
      if (mTasks.size() > 0) {
        mWaitCond.wait(lock, [this] { return mTasks.size() == 0; });
      }
    }

  private:
    Threadpool(const Threadpool&) = delete;
    enum WorkerAction {
      WAIT,
      WORK,
      EXIT,
      SUICIDE
    };

    WorkerAction getNextAction() {
      if (mPoolDestroyed) return EXIT;
      if (mThreadsToKill > 0) return SUICIDE;
      if (!mTasks.empty()) return WORK;
      return WAIT;
    }

    std::list<std::function<void()>>::const_iterator
    getInsertPosition(TaskPriority priority, TaskOrdering ordering) {
      if (priority == MAX) return mTasks.cbegin();
      if (priority == MIN) return mTasks.cend();
      return (ordering == LIFO) ? mTasks.cbegin() : mTasks.cend();
    }

    void workerFunction() {
      WorkerAction nextAction;
      while (true) {
        std::function<void()> task;
        {
          std::unique_lock<std::mutex> lock(mTasksMutex);
          nextAction = getNextAction();
          if (nextAction == WAIT) {
            mTasksCond.wait(lock, [this, &nextAction] {return (nextAction = getNextAction()) != WAIT; });
          }
          if (nextAction == EXIT) {
            break;
          }
          else if (nextAction == SUICIDE) {
            assert(mThreadsToKill > 0);
            --mThreadsToKill;
            auto threadIt = std::find_if(std::begin(mWorkers), std::end(mWorkers), [](const std::thread& thread) {
              return thread.get_id() == std::this_thread::get_id();
            });
            assert(threadIt == std::begin(mWorkers) || threadIt != std::end(mWorkers));
            threadIt->detach();
            mWorkers.erase(threadIt);
            break;
          }
          else if (nextAction == WORK) {
            // Lock in case some thread is waiting for all tasks to complete
            std::lock_guard<std::mutex> lock(mWaitMutex);
            task = std::move(mTasks.front());
            mTasks.pop_front();
          }
        }
        assert(nextAction == WORK);
        // The task itself is not performed under the lock
        task();
        // Notify in case some thread is waiting for all tasks to complete
        mWaitCond.notify_one();
      }
    }

    // The following resources are shared between main thread and worker threads,
    // so they need to have its access synchronized:
    //  - mWorkers
    //  - mTasks
    //  - mThreadsToKill
    //  - mNWorkers;
    //  - nPoolDestroyed;

    // list used to prevent iterator invalidation
    // when adding or removing threads
    std::list<std::thread> mWorkers;

    std::list<std::function<void()>> mTasks;

    // sync threads with the task list 
    std::condition_variable mTasksCond;
    std::mutex mTasksMutex;

    // allow a thread to wait for all remaining tasks to complete
    std::condition_variable mWaitCond;
    std::mutex mWaitMutex;

    TaskOrdering mOrdering;
    // holds the number of threads that need to die as a result of calling
    // Threadpool::resize
    unsigned int mThreadsToKill;
    // cache worker count to prevent data races in case Threadpool::resize 
    // is called multiple times quickly
    unsigned int mNWorkers;
    bool mPoolDestroyed;
  };

  /**
  * @brief Handles a task state and allows to get the task result
  *
  * @tparam Result type of the task result
  */
  template <class Result>
  class TaskHandler {
    // Users are not allow to construct new tasks handlers
    template <Threadpool::TaskPriority Priority, class FN, class... Args>
    friend TaskHandler<typename std::result_of<FN(Args...)>::type>
    Threadpool::queueAndHandleTask(FN&&, Args&&...);

  public:
    TaskHandler(TaskHandler&& other) :
      mTaskHandle(std::move(other.mTaskHandle)),
      mResult(std::move(other.mResult))
    {}

    TaskHandler& operator=(TaskHandler&& other) {
      mTaskHandle = std::move(other.mTaskHandle);
      mResult = std::move(other.mResult);
      return *this;
    }

    /**
    * @brief Returns wheter tha task has completed
    * @return bool
    */
    bool resultReady() const {
      return mResult.wait_for(std::chrono::seconds(0)) == std::future_status::ready;
    }

    /**
    * @brief Get the task result
    * @details Blocks the calling thread until the result is ready
    *          Users are responsible for calling this method only once
    * @return The value the task returned
    */
    Result get() { return mResult.get(); }

    /**
    * @brief Blocks the calling thread until the result is ready
    */
    void wait() { mResult.wait(); }

  private:
    explicit TaskHandler(const std::shared_ptr<std::packaged_task<Result()>>& task) :
      mTaskHandle(task),
      mResult(task->get_future())
    {}

    TaskHandler(const TaskHandler&) = delete;

    std::shared_ptr <std::packaged_task<Result()>> mTaskHandle;
    std::future<Result> mResult;
  };
}

Examples - main.cpp

// Threadpool.cpp : Defines the entry point for the console application.
//

#include "stdafx.h"
#include "Threadpool.h"
#include <iostream>
#include <memory>

using namespace threadpool;

struct SampleFunctor {
  SampleFunctor(int a) : data(a) {}

  int operator()() { return data; }
  void expensiveMethod() {
    std::this_thread::sleep_for(std::chrono::seconds(5));
    ++data;
  }

  int data;
};

std::unique_ptr<SampleFunctor> makeFunctor(int data) {
  return std::make_unique<SampleFunctor>(data);
}

class SomeManager {
public:
  static SomeManager& getInstance()
  {
    static SomeManager instance;
    std::this_thread::sleep_for(std::chrono::seconds(3));
    return instance;
  }
private:
  SomeManager() {}
};

int main()
{
  Threadpool threadpool(4);

  {
    // Queue free functions
    auto handler = threadpool.queueAndHandleTask(makeFunctor, 4);
    auto result = handler.get();
    std::cout << "Result is " << result->data << "\n";
  }

  // Queue a lambda
  threadpool.queueTask([]() {
    int counter = 0;
    while (counter < 1000) ++counter;
  });
  
  {
    // Queue functors
    auto handler = threadpool.queueAndHandleTask(SampleFunctor{ 20 });
    auto result = handler.get();
    std::cout << "Functor example: " << result << "\n";
  }

  {
    // Queue member methods
    // You can handle void tasks too!
    SampleFunctor f{ 20 };
    auto handler = threadpool.queueAndHandleTask(&SampleFunctor::expensiveMethod, &f);
    // note that this blocks the calling thread
    handler.wait();
    // you can use handler.get() too (if you dont assign it to anything)
    std::cout << "Expensive method done! Data should be 21 -> " << f.data << "\n";
  }

  {
    // Queue static class methods
    auto handler = threadpool.queueAndHandleTask(&SomeManager::getInstance);
    SomeManager& managerSingleton = handler.get();
    std::cout << "Our singleton is located at " << &managerSingleton << "\n";
  }
  return 0;
}

Extended usage

Additional features:

  • Adding/removing threads in runtime
  • Getting tasks results and waiting for them to be ready
  • Task order and priority

You can find the complete documentation in the github repo.

Questions

My main concerns are:

  • Does the interface seem easy to use?
  • Is there some edge-case the threadpool doesn't handle?
  • Did I follow some anti-pattern I don't know of?
\$\endgroup\$
  • 1
    \$\begingroup\$ The TaskHandler cannot stop a task. An interrupt method could e.g. set a TLS atomic value to true, so a worker function could check whether it's interrupted and return. \$\endgroup\$ – Dmitry T. Jun 5 at 15:39

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.