[Boost-users] Problem with throwing exceptions in async handler functions of ASIO library

Subject: [Boost-users] Problem with throwing exceptions in async handler functions of ASIO library
From: J.W.F. Thirion (derik_at_[hidden])
Date: 2009-05-28 12:57:11

Dear Boost community,

We're having a problem using the ASIO class with multiple IO service
threads and throwing exceptions inside the handle_xxx functions, as used
by the async_xxx functions of the ASIO library.

It seems that when any error is thrown in a normal/non-handler function,
the try/catch block of the main program is able to catch the error.
However, when the handler functions are called (I would assume by the io
service, now running in a separate thread), the main function's
try/catch blocks never receive the execeptions, and a std::terminate()
is called automatically.

To demonstrate this, we've taken the HTTP Server 2 example from the ASIO
examples page and reduced its complexity. Below are the sources of the
example.

Basically, to duplicate the problem, run the main executable and then in
another terminal, simply "telnet 127.0.0.1 60000" to see the
std::terminate() being called, even though there are try/catch blocks in
the main function (main.cpp). Here is an output:

$ ./main 
terminate called after throwing an instance of 'int'
Aborted
$

We've also taken the HTTP Server 2 example "as is" and added a "throw
123;" at the top of handle_accept in server.cpp and a "catch (...)"
block in posix_main.cpp. The same issue occurred here, which indicates
that the exceptions are not propagated through all the way (maybe
because the io service runs in a separate thread).

Does anyone have a suggestion on how we can throw exceptions inside the
handler functions of the async functions used? We'd like to handle all
our errors with custom exception classes (e.g. ExceptionAccept is thrown
if there is an error detected in handle_accept, which can then be
handled inside the main function of main.cpp).

We've compiled the sources below by typing:

g++ -I/usr/local/boost/include/boost-1_38 -L/usr/local/boost/lib
-lboost_thread-gcc41-mt-1_38 -lboost_system-gcc41-mt-1_38
io_service_pool.cpp server.cpp main.cpp -o main

where /usr/local/boost is the local boost installation path.

Many thanks in advance!

Kind regards,
Derik Thirion

main.cpp
--------
#include <iostream>
#include <string>
#include <boost/asio.hpp>
#include <boost/thread.hpp>
#include <boost/bind.hpp>
#include <boost/lexical_cast.hpp>
#include "server.hpp"

#include <pthread.h>
#include <signal.h>

int main(int argc, char* argv[])
{
  try
  {
    // Block all signals for background thread.
    sigset_t new_mask;
    sigfillset(&new_mask);
    sigset_t old_mask;
    pthread_sigmask(SIG_BLOCK, &new_mask, &old_mask);

        server s("127.0.0.1", "60000", 2); // 2 threads
    boost::thread t(boost::bind(&server::run, &s)); // 2 threads

    // Restore previous signals.
    pthread_sigmask(SIG_SETMASK, &old_mask, 0);

    // Wait for signal indicating time to shut down.
    sigset_t wait_mask;
    sigemptyset(&wait_mask);
    sigaddset(&wait_mask, SIGINT);
    sigaddset(&wait_mask, SIGQUIT);
    sigaddset(&wait_mask, SIGTERM);
    pthread_sigmask(SIG_BLOCK, &wait_mask, 0);
    int sig = 0;
    sigwait(&wait_mask, &sig);

    s.stop();
    t.join();
  }
  catch (std::exception& e)
  {
    std::cerr << "exception: " << e.what() << "\n";
  }
  catch (...)
  {
    std::cerr << "all exception" << "\n";
  }

  return 0;
}

server.hpp
----------
#ifndef SERVER_HPP
#define SERVER_HPP

#include <boost/asio.hpp>
#include <string>
#include <vector>
#include <boost/noncopyable.hpp>
#include <boost/shared_ptr.hpp>
#include "io_service_pool.hpp"

/// The top-level class of the HTTP server.
class server
  : private boost::noncopyable
{
public:
  /// Construct the server to listen on the specified TCP address and
port, and
  /// serve up files from the given directory.
  explicit server(const std::string& address, const std::string& port,
std::size_t io_service_pool_size);

  /// Run the server's io_service loop.
  void run();

  /// Stop the server.
  void stop();

private:
  /// Handle completion of an asynchronous accept operation.
  void handle_accept(const boost::system::error_code& e);

  /// The pool of io_service objects used to perform asynchronous
operations.
  io_service_pool io_service_pool_;

  /// Acceptor used to listen for incoming connections.
  boost::asio::ip::tcp::acceptor acceptor_;
};

#endif // SERVER_HPP

server.cpp
----------
#include "server.hpp"
#include <boost/bind.hpp>

server::server(const std::string& address, const std::string& port,
std::size_t io_service_pool_size)
  : io_service_pool_(io_service_pool_size),
    acceptor_(io_service_pool_.get_io_service())
{
  // Open the acceptor with the option to reuse the address (i.e.
SO_REUSEADDR).
  boost::asio::ip::tcp::resolver resolver(acceptor_.io_service());
  boost::asio::ip::tcp::resolver::query query(address, port);
  boost::asio::ip::tcp::endpoint endpoint = *resolver.resolve(query);
  acceptor_.open(endpoint.protocol());

acceptor_.set_option(boost::asio::ip::tcp::acceptor::reuse_address(true));
  acceptor_.bind(endpoint);
  acceptor_.listen();
  acceptor_.async_accept(*(new
boost::asio::ip::tcp::socket(io_service_pool_.get_io_service())),
      boost::bind(&server::handle_accept, this,
        boost::asio::placeholders::error));
}

void server::run()
{
  io_service_pool_.run();
}

void server::stop()
{
  io_service_pool_.stop();
}

void server::handle_accept(const boost::system::error_code& e)
{
        throw 123;  // this one never gets caught in main
                    // instead std::terminate() gets called
}

io_service_pool.cpp
-------------------
#include <stdexcept>
#include <boost/thread.hpp>
#include <boost/bind.hpp>
#include <boost/shared_ptr.hpp>
#include "io_service_pool.hpp"

io_service_pool::io_service_pool(std::size_t pool_size)
  : next_io_service_(0)
{
  if (pool_size == 0)
    throw std::runtime_error("io_service_pool size is 0");

  // Give all the io_services work to do so that their run() functions
will not
  // exit until they are explicitly stopped.
  for (std::size_t i = 0; i < pool_size; ++i)
  {
    io_service_ptr io_service(new boost::asio::io_service);
    work_ptr work(new boost::asio::io_service::work(*io_service));
    io_services_.push_back(io_service);
    work_.push_back(work);
  }
}

void io_service_pool::run()
{
  // Create a pool of threads to run all of the io_services.
  std::vector<boost::shared_ptr<boost::thread> > threads;
  for (std::size_t i = 0; i < io_services_.size(); ++i)
  {
    boost::shared_ptr<boost::thread> thread(new boost::thread(
          boost::bind(&boost::asio::io_service::run, io_services_[i])));
    threads.push_back(thread);
  }

  // Wait for all threads in the pool to exit.
  for (std::size_t i = 0; i < threads.size(); ++i)
    threads[i]->join();
}

void io_service_pool::stop()
{
  // Explicitly stop all io_services.
  for (std::size_t i = 0; i < io_services_.size(); ++i)
    io_services_[i]->stop();
}

boost::asio::io_service& io_service_pool::get_io_service()
{
  // Use a round-robin scheme to choose the next io_service to use.
  boost::asio::io_service& io_service = *io_services_[next_io_service_];
  ++next_io_service_;
  if (next_io_service_ == io_services_.size())
    next_io_service_ = 0;
  return io_service;
}

io_service_pool.hpp:
--------------------
#ifndef IO_SERVICE_POOL_HPP
#define IO_SERVICE_POOL_HPP

#include <boost/asio.hpp>
#include <vector>
#include <boost/noncopyable.hpp>
#include <boost/shared_ptr.hpp>

/// A pool of io_service objects.
class io_service_pool
  : private boost::noncopyable
{
public:
  /// Construct the io_service pool.
  explicit io_service_pool(std::size_t pool_size);

  /// Run all io_service objects in the pool.
  void run();

  /// Stop all io_service objects in the pool.
  void stop();

  /// Get an io_service to use.
  boost::asio::io_service& get_io_service();

private:
  typedef boost::shared_ptr<boost::asio::io_service> io_service_ptr;
  typedef boost::shared_ptr<boost::asio::io_service::work> work_ptr;

  /// The pool of io_services.
  std::vector<io_service_ptr> io_services_;

  /// The work that keeps the io_services running.
  std::vector<work_ptr> work_;

  /// The next io_service to use for a connection.
  std::size_t next_io_service_;
};

#endif // IO_SERVICE_POOL_HPP