asio::io_service.run() exits despite work object

Question

I have a dispatcher class (stripped implementation below) which creates an I/O service and allows the start worker threads which will execute io_service handlers. These threads should stay active until I tell them to stop. For that I use an io_service::work object. However, these threads do not stay active, despite the presence of the work object. Any idea on how to use io_service::work correctly?

This is a minimal working example that shows the problem (DispatcherClass given underneath):

#include <boost/asio.hpp>
#include <boost/thread.hpp>

void myDispatcherMainThread(DispatcherClass &dispatcher)
{
    std::cout << "Dispatcher Started\n";
    dispatcher.AddCtrlCHandling();
    if (dispatcher.Run(2))
        std::cout << "Dispatcher is about to finish (expected).\n";
    else
        std::cout << "Dispatcher is about to finish (unexpected).\n";
}


int main(int argc, char* argv[]) {
    DispatcherClass dispatcher;
    boost::thread maindispatcherthread(myDispatcherMainThread,boost::ref(dispatcher));
    boost::this_thread::sleep(boost::posix_time::seconds(5));
    if (dispatcher.get_io_service().stopped())
        std::cout << "Dispatcher should not have finished!";

    dispatcher.cleanStop();
    maindispatcherthread.join();
}

This is the dispatcher class (put here to make this post readable):

class DispatcherClass : private boost::noncopyable {
public:
    DispatcherClass():
      _ioService(),
      _pKeepWorking(new boost::asio::io_service::work( _ioService )),
      _expectingdispatchend(false){}

    boost::asio::io_service &get_io_service() { return _ioService; }

    // Start the dispatcher threads
    bool Run(unsigned int numThreads = -1)
    {
        // Recover after a clean stop
        if (!_pKeepWorking)
            _pKeepWorking.reset(new boost::asio::io_service::work( _ioService ));

        // Reset after clean or forced stop
        if (_ioService.stopped())
        {
            _expectingdispatchend = false;
            _ioService.reset();
        }

        // Start message dispatch threads
        boost::thread_group workerThreads;
        for (unsigned int i = 0; i < ((numThreads == (unsigned int)-1) ? (boost::thread::hardware_concurrency()) : numThreads); ++i)
            workerThreads.create_thread(boost::bind(&DispatcherClass::WorkerThread, this));

        // Wait until all threads finish
        workerThreads.join_all();

        return _expectingdispatchend;
    }

    // Stop the dispatcher threads
    void Stop()
    {
        _expectingdispatchend = true;
        _ioService.stop();
    }

    void cleanStop()
    {
        _expectingdispatchend = true;
        _pKeepWorking.reset();
    }

    // Stop the dispatcher threads on CTRL-C
    void AddCtrlCHandling()
    {
        boost::asio::signal_set sig_set(_ioService, SIGTERM, SIGINT);
        sig_set.async_wait(boost::bind(&boost::asio::io_service::stop, boost::ref(_ioService)));
    }

private:
    bool _expectingdispatchend;

    boost::asio::io_service _ioService;
    boost::shared_ptr< boost::asio::io_service::work > _pKeepWorking;

    void WorkerThread()
    {
        while (true) {
        try
        {
            boost::system::error_code ec;
            _ioService.run(ec);
            break;
        }
        catch (const std::exception &) {}
        }
    }
};

Answer 1

The io_service::stop() is being invoked within the cancelled SignalHandler, causing all invocations of io_service::run() to return as soon as possible.

void DispatcherClass::AddCtrlCHandling()
{
    boost::asio::signal_set sig_set(...);
    sig_set.async_wait(boost::bind(
        &boost::asio::io_service::stop, boost::ref(_ioService)));
}

In DispatcherClass::AddCtrlCHandling(), the sig_set is an automatic variable with a type of boost::asio::signal_set with its lifetime ending once the function returns. In the signal_set destructor, all outstanding asynchronous wait operations on the signal_set will be completed with an error code of boost::asio::error::operation_aborted. Hence, a SignalHandler that will invoke io_service::stop() is queued into the io_service and invoked by one of the worker threads. To resolve this, consider extending the signal_set's lifetime by making sig_set a member variable of DispatcherClass.

Answer 2

To narrow it down: Your threads instantly terminate after starting:

 boost::system::error_code ec;
 _ioService.run(ec); // <----- blocks until the work is complete
 break;

It seems to me that there is no work to be done, so run returns almost instantly (showing system:0 for the error code, which seems to be OK).

I'm not familiar with asio's API but I guess that the underlying problem is that _ioService is not properly set up before you spawn your threads.