6
\$\begingroup\$

With the new additions in c++11 and c++17 I wanted to create a simple implementation of thread pool.

I would like your opinion on:

  • Thread safety
  • API
  • performace
  • and general code quality

I also would like to know if it good idea to have wait_until_empty method. Without id I probably could have avoided using a mutex.

#ifndef WORKER_POOL_H
#define WORKER_POOL_H

#include <../cpp11-on-multicore/common/sema.h>

#include <atomic>
#include <cassert>
#include <condition_variable>
#include <functional>
#include <future>
#include <memory>
#include <mutex>
#include <optional>
#include <queue>
#include <thread>
#include <vector>

#if __cplusplus < 201703l
#error "Compile using c++17 or later"
#endif

/**
 * Simplistic implementation of thread pool
 * using C++17.
 */
class worker_pool {
private:
  /**
   * Inner class that represents individual workers.
   */
  class worker {
  private:
    worker_pool *wp;
    long id;

  public:
    worker(worker_pool *_wp, long _id) : wp(_wp), id(_id){};

    /**
     * Main worker loop.
     */
    void operator()() {
      // work until asked to stop
      while (!wp->stop.load()) {
        auto t = wp->fetch();
        // when asked to stop workers will wake up
        // and recieve a nullopt
        if (t.has_value())
          t.value()();
      }
    };
  };

  std::vector<std::thread> workers;
  std::queue<std::function<void(void)>> job_queue;
  // access control for the queue
  std::mutex queue_mutex;
  Semaphore queue_sem;

  // this is used to notify that queue has been emptied
  std::condition_variable cv_empty;

  // stop indicates that we were asked to stop but workers are not terminated
  // yet
  std::atomic<bool> stop;
  // term means that workers are terminated
  std::atomic<bool> term;

  /**
   * Thread safe job fetchind
   */
  std::optional<std::function<void(void)>> fetch() {
    queue_sem.wait();
    std::unique_lock l(queue_mutex);
    // return nothing if asked to stop
    if (stop.load())
      return std::nullopt;
    auto res = std::move(job_queue.front());
    // if we happen to have emptied the queue notify everyone who is waiting
    job_queue.pop();
    if (job_queue.empty())
      cv_empty.notify_all();
    return std::move(res);
  };

public:
  /**
   * Initializing worker pool with n workers.
   * By default the number of workers is equal to number
   * of cores on the machine.
   */
  worker_pool(long tcount = std::thread::hardware_concurrency())
      : queue_sem(0), stop(false), term(false) {
    assert(tcount > 0);
    for (long i = 0; i < tcount; i++) {
      workers.push_back(std::thread(worker(this, i)));
    }
  }

  /**
   * Terminate all workers before getting destroyed
   */
  ~worker_pool() { terminate(); }

  /**
   * No-copy and no-move
   */
  worker_pool(worker_pool const &) = delete;
  worker_pool &operator=(worker_pool const &) = delete;
  worker_pool(worker_pool &&) = delete;
  worker_pool &operator=(worker_pool &&) = delete;

  /**
   * Thread-safe job submition. Accepts any callable and
   * returns a future.
   */
  template <typename F, typename... Args>
  auto submit(F &&f, Args &&... args) -> std::future<decltype(f(args...))> {
    std::lock_guard l(queue_mutex);
    // Wrapping callable with arguments into a packaged task
    auto func = std::bind(std::forward<F>(f), std::forward<Args>(args)...);
    auto task_ptr =
        std::make_shared<std::packaged_task<decltype(f(args...))()>>(func);
    // Wrapping packaged task into a simple lambda for convenience
    job_queue.push([task_ptr] { (*task_ptr)(); });
    queue_sem.signal();
    return task_ptr->get_future();
  }

  /**
   * Terminate will stop all workers ignoring any remaining jobs.
   */
  void terminate() {
    // do nothing if already terminated
    if (term.load())
      return;
    stop.store(true);
    // wakeup all workers
    queue_sem.signal(workers.size());
    // wait for each worker to terminate
    for (size_t i = 0; i < workers.capacity(); i++) {
      if (workers[i].joinable())
        workers[i].join();
    }
    term.store(true);
  }

  /**
   * Check how many jobs remain in the queue
   */
  long jobs_remaining() {
    std::lock_guard l(queue_mutex);
    return job_queue.size();
  }

  /**
   * This function will block until all
   * the jobs in the queue have been processed
   */
  void wait_until_empty() {
    std::unique_lock l(queue_mutex);
    while (!(job_queue.empty() || stop.load()))
      cv_empty.wait(l, [&] { return job_queue.empty() || stop.load(); });
  }

  /**
   * Check if there was a demand to stop.
   * Note: there may be still workers running.
   */
  bool stopped() { return stop.load(); }

  /**
   * Check if workers have been terminated
   */
  bool terminated() { return term.load(); }
};

#endif // WORKER_POOL_H

Edit: updated code, added an assert to contructor, removed a mutex, typos

\$\endgroup\$
-1
\$\begingroup\$
#include <../cpp11-on-multicore/common/sema.h>

You didn't show this file, so I cannot comment on it nor the use of the semaphore.

#if __cplusplus < 201703l
#error "Compile using c++17 or later"
#endif

What's the point of this? If some feature isn't supported, the compiler will say so.

class worker_pool {
private:

I would put the public interface on top, since that is what's relevant to the largest audience.

  class worker {

This can just be a lambda function. id is unused.

      // work until asked to stop
      while (!wp->stop.load()) {
        auto t = wp->fetch();
        // when asked to stop workers will wake up
        // and recieve a nullopt
        if (t.has_value())
          t.value()();
      }

You check for the stop condition twice in each iteration. Accessing the atomic can be quite expensive when done frequently from multiple threads. Instead, you can simply write:

      // work until asked to stop
      while (true) {
        auto t = wp->fetch();
        // when asked to stop workers will wake up
        // and recieve a nullopt
        if (!t.has_value())
          break;
        t.value()();
      }

.

  // stop indicates that we were asked to stop but workers are not terminated
  // yet
  std::atomic<bool> stop;
  // term means that workers are terminated
  std::atomic<bool> term;

This allows for an invalid state (term && !stop). You can represent this better with an enum { Running, Stopping, Stopped }.

  std::optional<std::function<void(void)>> fetch() {
    queue_sem.wait();
    std::unique_lock l(queue_mutex);
    // return nothing if asked to stop
    if (stop.load())
      return std::nullopt;
    auto res = std::move(job_queue.front());
    // if we happen to have emptied the queue notify everyone who is waiting
    job_queue.pop();
    if (job_queue.empty())
      cv_empty.notify_all();
    return std::move(res);
  };

There's no need to return an optional here. function can already be empty.

You can check for the stop condition before locking the mutex. It's best to unlock the mutex as soon as possible; in this case, before notifying the CV.

Finally, there shouldn't be a semicolon after a function body.

    for (long i = 0; i < tcount; i++) {
      workers.push_back(std::thread(worker(this, i)));
    }

You can write workers.emplace_back(worker(this, i));.

  /**
   * Terminate all workers before getting destroyed
   */
  ~worker_pool() { terminate(); }

When dealing with threads, "terminate" usually means killing a thread, which should almost always be avoided, as it leaves the program in an unpredictable (if not undefined) state. Better just use the term "stop" instead. Since the only way to start the threads in a worker pool is to instantiate one, it would make sense if the only way to stop them is by destroying that instance.

  worker_pool(worker_pool const &) = delete;
  worker_pool &operator=(worker_pool const &) = delete;
  worker_pool(worker_pool &&) = delete;
  worker_pool &operator=(worker_pool &&) = delete;

There's no need to explicitly declare the move c'tor/assignment operator deleted here.

  template <typename F, typename... Args>
  auto submit(F &&f, Args &&... args) -> std::future<decltype(f(args...))> {
    std::lock_guard l(queue_mutex);
    // Wrapping callable with arguments into a packaged task
    auto func = std::bind(std::forward<F>(f), std::forward<Args>(args)...);
    auto task_ptr =
        std::make_shared<std::packaged_task<decltype(f(args...))()>>(func);
    // Wrapping packaged task into a simple lambda for convenience
    job_queue.push([task_ptr] { (*task_ptr)(); });
    queue_sem.signal();
    return task_ptr->get_future();
  }

Note that in the return value declaration you do not forward the arguments, which could yield a different overload resolution (and therefore a different return type) than when the function is actually called.

I don't think the bind will work well here when you pass a temporary as an argument to this function, since it will not store a copy of the argument. When the function gets called on the worker thread, the argument has been destroyed.

  void terminate() {
    // do nothing if already terminated
    if (term.load())
      return;
    stop.store(true);
    // wakeup all workers
    queue_sem.signal(workers.size());
    // wait for each worker to terminate
    for (size_t i = 0; i < workers.capacity(); i++) {
      if (workers[i].joinable())
        workers[i].join();
    }
    term.store(true);
  }

You should compare the index to the size of workers, not its capacity.

This function has a race condition. When it's called from two threads, the second call may begin before term has been updated. Then, the two threads may call join on the same thread instance.

  /**
   * Check how many jobs remain in the queue
   */
  long jobs_remaining() {
    std::lock_guard l(queue_mutex);
    return job_queue.size();
  }

I don't see any real use for this function.

  /**
   * This function will block until all
   * the jobs in the queue have been processed
   */
  void wait_until_empty() {
    std::unique_lock l(queue_mutex);
    while (!(job_queue.empty() || stop.load()))
      cv_empty.wait(l, [&] { return job_queue.empty() || stop.load(); });
  }

The comment is wrong. This function only waits for all jobs to be taken off the queue. The function may return while the last jobs are still being processed. That makes the whole function pointless to begin with. I would remove it.

  /**
   * Check if there was a demand to stop.
   * Note: there may be still workers running.
   */
  bool stopped() { return stop.load(); }

  /**
   * Check if workers have been terminated
   */
  bool terminated() { return term.load(); }

These functions also seem useless.

\$\endgroup\$

Your Answer

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

Not the answer you're looking for? Browse other questions tagged or ask your own question.