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

Question

我正在开发一个多线程应用程序，其中一个线程充当从客户端接收命令的 tcp 服务器。线程使用 Boost 套接字和接受器等待客户端连接，从客户端接收命令，将命令传递给应用程序的其余部分，然后再次等待。代码如下：

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() 返回后导致崩溃——我相信是因为线程在单例被销毁后尝试访问应用程序的日志记录单例。 （老实说，我来的时候就是这样。）

当应用程序退出时，我怎样才能干净地关闭这个线程？我读过原始套接字上的阻塞 accept() 调用可以通过从另一个线程关闭套接字来中断，但这似乎不适用于 Boost 套接字。我尝试使用 Boost asynchronous tcp echo server example 将服务器逻辑转换为异步 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;
}

在运行时，我打开了两个连接。这是输出：

10 秒后停止服务...
等待连接
用socket做事...
使用套接字完成，准备进行另一个连接。
等待连接
用socket做事...
使用套接字完成，准备进行另一个连接。
等待连接
立即停止服务！
正在等待服务器线程...
服务器线程退出。
在服务器线程上等待完成。

Answer 2

当您收到需要退出的事件时，您可以调用acceptor.cancel()，这将取消挂起的接受（错误代码为operation_canceled）。在某些系统上，为了安全起见，您可能还必须close() 接受者。

Answer 3

如果涉及到它，您可以在 localhost 上打开一个到它的临时客户端连接 - 这会唤醒它。您甚至可以向它发送一条特殊消息，以便您可以从 pub 关闭您的服务器 - 应该有一个应用程序：)

Answer 4

只需使用本机句柄和 SHUT_RD 选项调用 shutdown，即可取消现有的接收（接受）操作。

Answer 5

接受的答案并不完全正确。 事实上@JohnYu 回答正确。

使用 ASIO 的阻塞 API 很像使用 ASIO 库封装在其类中的 BSD 套接字 API。

问题是 boost::asio::ip::tcp::acceptor 类不提供 shutdown() 功能，因此您必须使用“旧”套接字 API 来完成。

附加说明：确保在所有使用它的线程退出之前，acceptor、socket 和 io_service 未被删除。在以下代码中，std::shared_ptr 用于保持共享资源处于活动状态，因此ApplicationContext 类的用户可以删除ApplicationContext 对象并避免SEGFAULT 崩溃。

补充说明：注意 boost 文档，有引发异常的重载方法和返回错误代码的方法。原始海报的代码使用 acceptor->accept(socket); 没有 try/catch 会导致程序退出而不是正常的线程例程退出和清理。

解决方案说明如下：

#include <unistd.h> // include ::shutdown() function
// other includes ...

using boost::asio::ip::tcp;
using boost::asio::io_service;

class ApplicationContext {

    // Use shared pointer to extend life of resources afer ApplicationContext is deleted
    // and running threads can still keep using shared resources
    std::shared_ptr<tcp::acceptor> acceptor;
    std::shared_ptr<io_service> ioservice;

    // called `ServerThreadFunc` in question code example
    void AcceptLoopThreadRoutine(int port_no) {
        ioservice = std::make_shared<io_service>();
        acceptor = std::make_shared<tcp::acceptor>(*ioservice, tcp::endpoint(tcp::v4(), port_no));

        try {
            for (;;) {
                // listen for client connection
                tcp::socket socket(*ioservice);
                // Note boost::system::system_error is raised when using this overload
                acceptor->accept(socket);

                // connected receive some data ...
                // // boost::array<char,256> msg_buf;
                // // socket.receive(boost::asio::buffer(msg_buf));
                //  do something with received bytes here
            }
        } catch(std::exception const & exception) {
            // boost::system::system_error here indicates clean exit ;)
        }
    }

    void StopAcceptThread() {
        if(acceptor) {
            // boost::asio::ip::tcp::acceptor does not have shutdown() functionality
            // exposed, so we need to do it with this low-level approach
            int shutdown_status = shutdown(acceptor->native_handle(), SHUT_RDWR);
        }
    }

};

另请注意，使用信号解除阻塞接受线程是非常讨厌的实现，本地主机上的临时客户端连接解除阻塞接受线程非常尴尬。

ASIO 可帮助您通过回调在单线程中完成所有工作。如果您正在混合线程和 ASIO，那么您的设计可能很糟糕。

补充说明：不要混淆shutdown() 和close()。某些系统可能允许您在接受套接字上使用close() 来解除阻塞接受循环，但这不是可移植的。