Boost :: asio - 如何中断被阻塞的tcp服务器线程？

Question

我正在研究一个多线程应用程序,其中一个线程充当从客户端接收命令的tcp服务器.该线程使用Boost套接字和acceptor等待客户端连接,从客户端接收命令,将命令传递给应用程序的其余部分,然后再次等待.这是代码:

void ServerThreadFunc()
{
    using boost::asio::ip::tcp;
    boost::asio::io_service io_service;
    tcp::acceptor acceptor(io_service, tcp::endpoint(tcp::v4(), port_no));

    for (;;)
    {
        //  listen for command connection
        tcp::socket socket(io_service);
        acceptor.accept(socket);

        //  connected; receive command
        boost::array<char,256> msg_buf;
        socket.receive(boost::asio::buffer(msg_buf));

        //  do something with received bytes here
    }
}

这个线程大部分时间都在调用时被阻塞acceptor.accept().此时,线程仅在应用程序退出时终止.不幸的是,这会在main()返回后导致崩溃 - 我相信因为线程在单例被销毁后尝试访问应用程序的日志单例.(当我到这里时,就像那个,诚实的guv.)

当应用程序退出时,如何干净地关闭此线程？我已经读过可以通过从另一个线程关闭套接字来中断原始套接字上的阻塞accept()调用,但这似乎不适用于Boost套接字.我已经尝试使用Boost异步tcp echo服务器示例将服务器逻辑转换为异步i/o ,但这似乎只是为阻塞调用交换acceptor::accept()阻塞调用io_service::run(),所以我留下了同样的问题:一个阻塞打电话我不能打断.有任何想法吗？

Answer 1

简而言之,有两种选择:

• 将代码更改为异步(acceptor::async_accept()和async_read),在事件循环中运行io_service::run(),并取消通过io_service::stop().
• 使用较低级别的机制强制阻止调用中断,例如信号.

我会推荐第一个选项,因为它更可能是便携式的,更容易维护.要理解的重要概念是,io_service::run()只要有待处理的工作,就可以使用唯一的块.当io_service::stop()被调用时,它会尝试使阻塞的所有线程io_service::run(),以尽快恢复; 它不会中断同步操作,如acceptor::accept()和socket::receive(),即使同步操作事件循环中调用.重要的是要注意这io_service::stop()是一个非阻塞调用,因此与被阻塞的线程的同步io_service::run()必须使用另一个机制,例如thread::join().

这是一个将运行10秒并侦听端口8080的示例:

#include <boost/asio.hpp>
#include <boost/bind.hpp>
#include <boost/shared_ptr.hpp>
#include <boost/thread.hpp>
#include <iostream>

void StartAccept( boost::asio::ip::tcp::acceptor& );

void ServerThreadFunc( boost::asio::io_service& io_service )
{
  using boost::asio::ip::tcp;
  tcp::acceptor acceptor( io_service, tcp::endpoint( tcp::v4(), 8080 ) );

  // Add a job to start accepting connections.
  StartAccept( acceptor );

  // Process event loop.
  io_service.run();

  std::cout << "Server thread exiting." << std::endl;
}

void HandleAccept( const boost::system::error_code& error,
                   boost::shared_ptr< boost::asio::ip::tcp::socket > socket,
                   boost::asio::ip::tcp::acceptor& acceptor )
{
  // If there was an error, then do not add any more jobs to the service.
  if ( error )
  {
    std::cout << "Error accepting connection: " << error.message() 
              << std::endl;
    return;
  }

  // Otherwise, the socket is good to use.
  std::cout << "Doing things with socket..." << std::endl;

  // Perform async operations on the socket.

  // Done using the socket, so start accepting another connection.  This
  // will add a job to the service, preventing io_service::run() from
  // returning.
  std::cout << "Done using socket, ready for another connection." 
            << std::endl;
  StartAccept( acceptor );
};

void StartAccept( boost::asio::ip::tcp::acceptor& acceptor )
{
  using boost::asio::ip::tcp;
  boost::shared_ptr< tcp::socket > socket(
                                new tcp::socket( acceptor.get_io_service() ) );

  // Add an accept call to the service.  This will prevent io_service::run()
  // from returning.
  std::cout << "Waiting on connection" << std::endl;
  acceptor.async_accept( *socket,
    boost::bind( HandleAccept,
      boost::asio::placeholders::error,
      socket,
      boost::ref( acceptor ) ) );
}

int main()
{
  using boost::asio::ip::tcp;

  // Create io service.
  boost::asio::io_service io_service;

  // Create server thread that will start accepting connections.
  boost::thread server_thread( ServerThreadFunc, boost::ref( io_service ) );

  // Sleep for 10 seconds, then shutdown the server.
  std::cout << "Stopping service in 10 seconds..." << std::endl;
  boost::this_thread::sleep( boost::posix_time::seconds( 10 ) );
  std::cout << "Stopping service now!" << std::endl;

  // Stopping the io_service is a non-blocking call.  The threads that are
  // blocked on io_service::run() will try to return as soon as possible, but
  // they may still be in the middle of a handler.  Thus, perform a join on 
  // the server thread to guarantee a block occurs.
  io_service.stop();

  std::cout << "Waiting on server thread..." << std::endl;
  server_thread.join();
  std::cout << "Done waiting on server thread." << std::endl;

  return 0;
}

在跑步时,我打开了两个连接.这是输出:

Stopping service in 10 seconds...
Waiting on connection
Doing things with socket...
Done using socket, ready for another connection.
Waiting on connection
Doing things with socket...
Done using socket, ready for another connection.
Waiting on connection
Stopping service now!
Waiting on server thread...
Server thread exiting.
Done waiting on server thread.