C++11 ThreadPool implementation

Question

A new version of this code based on the provided advice can be found here. Latest improvement is found here.

---

I made a simple ThreadPool implementation in C++ using only atomics ( neither std::mutex nor std::condition_variable were used).

The idea is to have a threads_ready increasing to threads.size() until all threads are finished, and then back to 0 when all threads are ready to execute again. This was the simplest I could think of, that actually works in the given scenario.

#include <iostream>
#include <functional>
#include <vector>
#include <thread>
#include <mutex>
#include <atomic>
#include <iomanip>
#include <cassert>
#include <numeric>
using std::vector;
using std::function;
using std::thread;
using std::mutex;
using std::unique_lock;
using std::atomic;
class ThreadPool{
public:
 ThreadPool(int number_of_threads, function<void(vector<double>&, int, int)> function)
 : target_buffer(nullptr)
 , worker_function(function)
 , threads()
 , threads_ready(0)
 , state(IDLE_VALUE)
 {
 for(int i = 0; i < number_of_threads; ++i)
 threads.push_back(thread(&ThreadPool::worker, this, i));
 }
 ~ThreadPool(){
 while(0 < threads_ready.load()); /* Wait until all pending operations finish */
 state.store(END_VALUE);
 while(0 < threads.size()){
 if(true == threads.back().joinable())
 threads.back().join();
 threads.pop_back();
 }
 }
 void start_and_block(vector<double>& buffer){
 /* initialize, start.. */
 target_buffer = &buffer;
 state.store(START_VALUE);
 /* wait until the work is done */
 while(threads.size() > threads_ready.load());
 state.store(IDLE_VALUE);
 /* wait until the threads are available again */
 while(0 < threads_ready.load());
 }
private:
 static const int IDLE_VALUE = 0;
 static const int START_VALUE = 1;
 static const int END_VALUE = 2;
 vector<double>* target_buffer;
 function<void(vector<double>&, int, int)> worker_function; /* buffer, start, length */
 vector<thread> threads;
 atomic<int> threads_ready;
 atomic<int> state; /* 1 = start; 0 = don't start */
 void worker(int thread_index){
 int tmp_num;
 while(END_VALUE != state){ /* Until the pool is stopped */
 while(START_VALUE == state){ /* Wait until start signal is provided */
 worker_function(
 (*target_buffer),
 (thread_index * (target_buffer->size()/threads.size())),
 (target_buffer->size()/threads.size())
 );/* do the work */
 tmp_num = threads_ready.load(); /* signal that work is done! */
 while(!threads_ready.compare_exchange_weak(tmp_num, (tmp_num + 1u)))
 tmp_num = threads_ready.load(); /* increase "done counter" */
 while(START_VALUE == state.load()); /* Wait until the other threads are finished */
 tmp_num = threads_ready.load(); /* Signal that thread is finished */
 while(!threads_ready.compare_exchange_weak(tmp_num, (tmp_num - 1u)))
 tmp_num = threads_ready.load(); /* decrease "done counter" */
 } /*while(START_VALUE == state)*/
 } /*while(END_VALUE != state)*/
 }
};
static int result = 0;
static mutex cout_mutex;
void worker(vector<double> buffer, int start, int length){
 double sum = 0;
 for(int i = 0; i < length; ++i){
 sum += buffer[i];
 }
 std::lock_guard<mutex> my_lock(cout_mutex);
 std::cout << "Partial sum: " << std::setw(4) << sum << " \t\t |" << "\r";
 result += sum;
}
int main(int argc, char** agrs){
 ThreadPool pool(5,&worker);
 result = 0;
 for(int i = 0; i< 10000; ++i){
 vector<double> test_buffer(500, rand()%10);
 result = 0;
 pool.start_and_block(test_buffer);
 std::cout << "result["<< i << "]: " << std::setw(4) << result << " \t\t |" << "\r";
 assert(std::accumulate(test_buffer.begin(),test_buffer.end(), 0) == result);
 }
 std::cout << "All assertions passed! |"<< std::endl;
 return 0;
}

According to the assertions, the behavior is as expected, but is there any dangers/wrong behavior here that I missed? Is there any way to improve this? How might this put up with real-world usage?

• Toby Speight: I wouldn't call this a thread pool - that would imply that calling code could come along and find a running thread to do its work. Here, we just have a gang of threads that start, do work, and finish.
• Toby Speight: Incorporating advice from an answer into the question violates the question-and-answer nature of this site. You could post improved code as a new question, as an answer, or as a link to an external site - as described in I improved my code based on the reviews. What next?. I have rolled back the edit, so the answers make sense again.
• Dávid Tóth: Thank you! I'll keep this in mind next time.

Answer 1

start_and_block() is mis-named - call it start_and_busy_wait(). Busy-waiting is inefficient. Instead of avoiding std::condition_variable, embrace it; use it in the slow path instead of while(threads.size() > threads_ready.load()); and while(0 < threads_ready.load());.

Also, it's redundant to compare against literal bools - if(true == threads.back().joinable()) should be simply if(threads.back().joinable()).

• Dávid Tóth: That is not true, the threads also need to be finished and ready, for another potential ̇start_and_block` call. If the slow path doesn't wait until threads_ready is back to 0 there are all kinds of synchronisation problems.
• Dávid Tóth: Is the only way to eliminate busy-waiting condition variables?
• Dávid Tóth: Well the reason was the lack of understanding, but I think a conditional variable could be used in the slow path instead of the while cycles, where the predicate would be the same as the condition in the while cycle.
• Toby Speight: Yes, that's right.
• Dávid Tóth: I've made an improved version: codereview.stackexchange.com/questions/263059/…

Answer 2

Why is the state an int rather than an enumerated type? Your definitions of constants IDLE_VALUE etc. looks like it ought to be an enumerated type. Is this used with values other than those special three?

while(0 < threads_ready.load());
Busy waiting is inefficient. At worst, you may end up with the thread doing the work being pre-empted to run the loop that's waiting for it to finish! Better to use a semaphore to wait for all threads to complete.

 : target_buffer(nullptr)
 , worker_function(function)
 , threads()
 , threads_ready(0)
 , state(IDLE_VALUE)

Only one of those are initialized with a parameter to the constructor. The rest can use inline immediate initializers on those members for clearer code and maintenance benefits.

if(true == threads.back().joinable())
Testing against true is just silly. Where does it stop? After all, operator== is also returning a bool so should you write true == (true == threads.back().joinable()) ? Bools are bool. Your test is if(threads.back().joinable()).

Your "buffer" is not const so I thought it was going to be for output. It's not. That's the work input being split up. Buffer is a misleading name for this. You should use iterators rather than a collection plus indexes to indicate a group of elements to work on.

• Dávid Tóth: "The rest can use inline immediate initializers on those members for clearer code and maintenance benefits. " - Does that mean, to simply omit them from the constructor, for the members to be zero-initialized?
• Dávid Tóth: The Busy waiting is eliminated in the question linked to this one, and indeed you are right, thank you!
• Dávid Tóth: You are also right about the enumeration, I'll keep this in mind with the next rework, although I only have a vague outline of how I might be using iterators rather than a collection..
• JDługosz: @DavidTóth each worker is given a pair (begin and end) of iterators to define the input elements assigned to it. Are you familiar with the standard library algorithms?
• Dávid Tóth: I'd much rather we discuss this with the next version I am about to post, it will definately be more specific.