Geek asks and answers

Thread pool using Boost in C ++ is not working properly

I implemented the solution from How to create a thread pool using boost in C ++? , but I had a problem with the io_service :: stop () function, which stops processing my threads further.

In my case, I have 3 threads in my pool, and I'm trying to run 11,000 entries through it. Each of the records is independent of the others, so I just want to speed up the processing by creating parallel runs of each record.

void processRecord (unsigned int i, unsigned int numRecords)
{
    cout << i << "/" << numRecords << endl;

    // do Processing...
}

#define MAX_THREADS 3
unsigned int numRecords=11000

boost::asio::io_service ioService;
boost::thread_group threadPool;

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

for (unsigned int i=0 ; i<MAX_THREADS ; ++i)
{
    threadPool.create_thread (boost::bind (&boost::asio::io_service::run, &ioService));
}

for (unsigned int i=0 ; i<numRecords ; ++i)
{
    ioService.post (boost::bind (processRecord, i, numRecords);
}

// ioService.stop ();          // Was causing ioService to stop
work.reset();                  // Wait for all work to be finished.
threadPool.join_all ();

processAllRecords ();

The problem that I see is that after the loop, when the ioService.post () call starts the processes in the pool, it ends, it calls the ioService.stop () call and stops further processing. This usually happens after only about 400 records have been processed.

, ~ 400 11000 .

++, , . .

: , , , . , ioService.stop() . work.wait(), , .

Edit2: . work.reset().

+3
1

, boost:: asio , scoped_ptr. .

void processRecord (unsigned int i, unsigned int numRecords)
{
    cout << i << "/" << numRecords << endl;

    // do Processing...
}

#define MAX_THREADS 3
unsigned int numRecords=11000

boost::asio::io_service ioService;
boost::thread_group threadPool;

// by using a scoped pointer for the io_service::work
// and enclosing the threading in brackets
// this should run until all the jobs have finished
// and you don't need to call work.reset()  

// added brackets around threading
{
    // made work a boost::scoped_ptr
    boost::scoped_ptr< boost::asio::io_service::work > 
          work ( new boost::asio::io_service(ioService) );

    for (unsigned int i=0 ; i<MAX_THREADS ; ++i)
    {
        threadPool.create_thread (
           boost::bind (&boost::asio::io_service::run, &ioService));
    }

    for (unsigned int i=0 ; i<numRecords ; ++i)
    {
        ioService.post (boost::bind (processRecord, i, numRecords);
    }
}
// now just have to join
threadPool.join_all ();

processAllRecords ();
0

More articles:


All Articles