Using Boost threads and io_service to create a threadpool

Question

I have looked around Stack Overflow and there have been a few really good answers on this, (my code is actually based on this answer here) but for some reason I am getting weird behavior - in that thread_func should be called ls1 times, but it is only running between 0 and 2 times before the threads exit. It seems like ioService.stop() is cutting off queued jobs before they are completed, but from what I understand this should not happen. Here is the relevant code snippet:

boost::asio::io_service ioService;
boost::asio::io_service::work work(ioService);

boost::thread_group threadpool;

for (unsigned t = 0; t < num_threads; t++)
{   
    threadpool.create_thread(boost::bind(&boost::asio::io_service::run, &ioService));
}   

//Iterate over the dimensions of the matrices
for (unsigned i = 0; i < ls1; i++)
{   
    ioService.post(boost::bind(&thread_func,i, rs1, rs2, ls2, arr, left_arr, &result));
}   

ioService.stop();
threadpool.join_all();

Any help would be greatly appreciated, thanks!

Answer 1

io_service::stop() causes all invocations of run() or run_one() to return as soon as possible. It does not remove any outstanding handlers that are already queued into the io_service. When io_service::stop() is invoked, the threads in threadpool will return as soon as possible, causing each thread of execution to be complete.

As io_service::post() will return immediately after requesting that the io_service invoke the handler, it is non-deterministic as to how many posted handlers will be invoked by threads in threadpool before the io_service is stopped.

If you wish for thread_func to be invoked ls1 times, then one simple alternative is to reorganize the code so that work is added to the io_service before the threadpool services it, and then the application lets the io_service run to completion.

boost::asio::io_service ioService;

// Add work to ioService.
for (unsigned i = 0; i < ls1; i++)
{   
  ioService.post(boost::bind(
      &thread_func,i, rs1, rs2, ls2, arr, left_arr, &result));
}   

// Now that the ioService has work, use a pool of threads to service it.
boost::thread_group threadpool;    
for (unsigned t = 0; t < num_threads; t++)
{   
  threadpool.create_thread(boost::bind(
      &boost::asio::io_service::run, &ioService));
}   

// Once all work has been completed (thread_func invoked ls1 times), the
// threads in the threadpool will be completed and can be joined.
threadpool.join_all();

Answer 2

If you're expecting thread_func to be called ls1 times, then you should wait until it is actually called ls1 times before stopping io_service. As written, stop() may be called before any of the threads had a chance to have been scheduled even.

There are many ways to wait for that condition. For example you could use a condition variable:

#include <boost/asio.hpp>
#include <boost/thread.hpp>
unsigned num_threads = 10, ls1=11;
int result = 0;
boost::mutex m;
boost::condition_variable cv;
void thread_func(unsigned , int* result) {
    /* do stuff */
    {
        boost::lock_guard<boost::mutex> lk(m);
        ++*result;
    }
    cv.notify_one();
}
int main()
{
    boost::asio::io_service ioService;
    boost::asio::io_service::work work(ioService);
    boost::thread_group threadpool;
    for (unsigned t = 0; t < num_threads; t++)
        threadpool.create_thread(boost::bind(&boost::asio::io_service::run,
                                             &ioService));
    for (unsigned i = 0; i < ls1; i++)
        ioService.post(boost::bind(&thread_func,i,&result));

    {
        boost::unique_lock<boost::mutex> lk(m);
        cv.wait(lk, []{return result == ls1; });
    }
    ioService.stop();
    threadpool.join_all();
    std::cout << "result = " << result << '\n';
}