CODEWITHSUNDEEP
c++multithreadingboostthreadpool
Portisch
Portisch

Reputation: 49

Boost, create thread pool before io_service.post

I successfully was testing an example about boost io_service:

for(x = 0; x < loops; x++)
{
    // Add work to ioService.
    for (i = 0; i < number_of_threads; i++)
    {   
        ioService.post(boost::bind(worker_task, data, pre_data[i]));
    }

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

    // threads in the threadpool will be completed and can be joined.
    threadpool.join_all();
}

This will loop several times and it take a little bit long because every time the threads are created for each loop.

Is there a way to create all needed threads. Then post in the loop the work for each thread. After the work it is needed to wait until all threads have finished their work!

Something like this:

// start/create threads
for (i = 0; i < number_of_threads; i++)
{   
    threadpool.create_thread(boost::bind(
        &boost::asio::io_service::run, &ioService));      
}

for(x = 0; x < loops; x++)
{
    // Add work to ioService.
    for (i = 0; i < number_of_threads; i++)
    {   
        ioService.post(boost::bind(worker_task, data, pre_data[i]));
    }

    // threads in the threadpool will be completed and can be joined.
    threadpool.join_all();
}

Upvotes: 0

Views: 806

Answers (2)

Portisch
Portisch

Reputation: 49

The loop wasn't realy usefull.

Here is a better shwoing how it works. I getting data in a callback:

void worker_task(uint8_t * data, uint32_t len)
{
    uint32_t pos = 0;
    while(pos < len)
    {
        pos += process_data(data + pos);
    }
}

void callback_f(uint8_t *data, uint32_t len)
{

    //split data into parts
    uint32_t number_of_data_per_thread = len / number_of_threads;

    // Add work to ioService.
    uint32_t x = 0;
    for (i = 0; i < number_of_threads; i++)
    {   
         ioService.post(boost::bind(worker_task, data + x, number_of_data_per_thread));
         x += number_of_data_per_thread ;
    }

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

    // threads in the threadpool will be completed and can be joined.
    threadpool.join_all();
}

So this callback get called very fast from the host application (media stream). If the len what is comming in is big enough the threadpool makes sense. This is because the working time is higher than the init time of the threads and start running.

If the len of the data is small the advantage of the threadpool is getting lost because the init and starting of the threads takes more time then the processing of the data.

May question is now if it is possible to have the threads already running and waiting for data. If the callback get called push the data to the threads and wait for their finish. The number of threads is constant (CPU count). And as because it is a callback from a host application the pointer to data is only valid while being in the callback function. This is why I have to wait until all threads have finished work.

The thread can starting working immediately after getting data even before other threads are getting started. There is no sync problem because every thread have its own memory area of the data.

Upvotes: 0

ComicSansMS
ComicSansMS

Reputation: 54669

The problem here is that your worker threads will finish immediately after creation, since there is no work to be done. io_service::run() will just return right away, so unless you manage to sneak in one of the post-calls before all worker threads have had an opportunity to call run(), they will all finish right away.

Two ways to fix this:

  • Use a barrier to stop the workers from calling run() right away. Only unblock them once the work has been posted.
  • Use an io_service::work object to prevent run from returning. You can destroy the work object once you posted everything (and must do so before attempting to join the workers again).

Upvotes: 2

Related Questions