std::condition_variable – notify once but wait thread wakened twice

Question

Here's a simple C++ thread pool implementation. It's an altered version orginated from https://github.com/progschj/ThreadPool.

#ifndef __THREAD_POOL_H__
#define __THREAD_POOL_H__

#include <vector>
#include <queue>
#include <memory>
#include <thread>
#include <chrono>
#include <mutex>
#include <condition_variable>
#include <future>
#include <functional>
#include <stdexcept>

namespace ThreadPool {

class FixedThreadPool {
 public:
  FixedThreadPool(size_t);

  template<class F, class... Args>
  auto Submit(F&& f, Args&&... args)
      -> std::future<typename std::result_of<F(Args...)>::type>;

  template<class F, class... Args>
  void Execute(F&& f, Args&&... args);

  ~FixedThreadPool();

  void AwaitTermination();

  void Stop();

 private:
  void ThreadWorker();

  // need to keep track of threads so we can join them
  std::vector<std::thread> workers;

  // the task queue
  std::queue< std::function<void()> > tasks;

  // synchronization
  std::mutex worker_mutex;
  std::mutex queue_mutex;
  std::condition_variable condition;

  // stop flag
  bool stop_;

  // thread size
  int thread_size_;
};

// Constructor does nothing. Threads are created when new task submitted.
FixedThreadPool::FixedThreadPool(size_t num_threads): 
    stop_(false),
    thread_size_(num_threads) {}

// Destructor joins all threads
FixedThreadPool::~FixedThreadPool() {
  //std::this_thread::sleep_for(std::chrono::seconds(5));
  for(std::thread &worker: workers) {
    if (worker.joinable()) {
      worker.join();
    }
  }
}

// Thread worker
void FixedThreadPool::ThreadWorker() {
  std::function<void()> task;
  while (1) {
    {
      std::unique_lock<std::mutex> lock(this->queue_mutex);
      this->condition.wait(lock,
                     [this]() { return this->stop_ || !this->tasks.empty(); });
      printf("wakeeeeeened\n");
      if (this->stop_ && this->tasks.empty()) {
        printf("returning ...\n");
        return;
      }
      task = std::move(this->tasks.front());
      this->tasks.pop();
    }
    task();
  }
}

// Add new work item to the pool
template<class F, class... Args>
auto FixedThreadPool::Submit(F&& f, Args&&... args)
    -> std::future<typename std::result_of<F(Args...)>::type >
{
  {
    std::unique_lock<std::mutex> lock(this->worker_mutex);
    if (workers.size() < thread_size_) {
      workers.emplace_back(std::thread(&FixedThreadPool::ThreadWorker, this));
    }
  }

  using return_type = typename std::result_of<F(Args...)>::type;

  auto task = std::make_shared< std::packaged_task<return_type()> >(
      std::bind(std::forward<F>(f), std::forward<Args>(args)...)
  );

  std::future<return_type> res = task->get_future();
  {
    std::unique_lock<std::mutex> lock(queue_mutex);
    if(stop_) {
      throw std::runtime_error("ThreadPool has been shutdown.");
    }
    tasks.emplace([task]() { (*task)(); });
  }
  condition.notify_one();
  return res;
}

// Execute new task without returning std::future object.
template<class F, class... Args>
void FixedThreadPool::Execute(F&& f, Args&&... args) {
  Submit(std::forward<F>(f), std::forward<Args>(args)...);
}

// Blocks and wait for all previously submitted tasks to be completed.
void FixedThreadPool::AwaitTermination() {
  for(std::thread &worker: workers) {
    if (worker.joinable()) {
      worker.join();
    }
  }
}

// Shut down the threadpool. This method does not wait for previously submitted
// tasks to be completed.
void FixedThreadPool::Stop() {
  printf("Stopping ...\n");
  {
    std::unique_lock<std::mutex> lock(queue_mutex);
    stop_ = true;
  }
}


} // namespace ThreadPool

#endif /* __THREAD_POOL_H__ */

and the test main.cpp:

#include <iostream>
#include <vector>
#include <chrono>
#include <exception>

#include "ThreadPool.h"

int main(int argc, char** argv) {
  ThreadPool::FixedThreadPool pool(3);

  pool.Execute([]() {
      std::cout << "hello world" << std::endl;
    }
  );
  pool.Stop();
  pool.AwaitTermination();
  std::cout << "All tasks complted." << std::endl;

  return 0;
}

I have a bug in this test program. Only one task is submitted to threadpool, but I got working thread being wakened up twice:

>>./test 
Stopping ...
wakeeeeeened
hello world
wakeeeeeened
returning ...
All tasks complted.

I think the problem is in FixedThreadPool::ThreadWorker() itself. The working thread continuously wait on a conditional variable to get new tasks. The function FixedThreadPool::Submit() adds a new task to queue and call condition.nofity_one() to waken a working thread.

But I can't figure out how the working thread can be wakened twice. I have only one task submitted and exactly one working thread in this test.

Answer 1

Converting comments into answer:

condition_variable::wait(lock, pred) is equivalent to while(!pred()) wait(lock);. If pred() returns true then no wait actually takes place and the call returns immediately.

Your first wake is from the notify_one() call; the second "wake" is because the second wait() call happens to execute after the Stop() call, so your predicate returns true and wait() returns immediately without waiting.

It should be obvious that you got (un)lucky here: if the second wait() call took place before the Stop() call, then your worker thread will be stuck waiting forever (in the absence of spurious wakeups), and so will your main thread.

Also, get rid of __THREAD_POOL_H__. Burn those double underscores to the ground.