boost :: asio async_receive_from线程之间共享的UDP端点？

Question

Boost asio特别允许多个线程在io_service上调用run()方法.这似乎是创建多线程UDP服务器的好方法.然而,我遇到了一个障碍,我正在努力寻找答案.

查看典型的async_receive_from调用:

m_socket->async_receive_from(
        boost::asio::buffer(m_recv_buffer),
        m_remote_endpoint,
        boost::bind(
            &udp_server::handle_receive,
            this,
            boost::asio::placeholders::error,
            boost::asio::placeholders::bytes_transferred));

远程端点和消息缓冲区不会传递给处理程序,但是处于更高的作用域级别(在我的示例中为成员变量).UDP消息到达时处理UDP消息的代码如下所示:

void dns_server::handle_receive(const boost::system::error_code &error, std::size_t size)
{
    // process message
    blah(m_recv_buffer, size);

    // send something back
    respond(m_remote_endpoint);
}

如果有多个线程在运行,那么同步如何工作？在线程之间共享一个端点和接收缓冲区意味着在同时消息到达的情况下,asio在另一个线程中调用处理程序之前等待处理程序在单个线程内完成.这似乎否定了允许多个线程首先调用run的观点.

如果我想获得并发服务的请求,看起来我需要将工作包和端点的副本一起交给一个单独的线程,允许处理程序方法立即返回,以便asio可以继续并通过另一个与另一个调用run()的线程并行的消息.

这看起来有点令人讨厌.我在这里错过了什么？

Answer 1

在线程之间共享单个端点和接收缓冲区意味着 asio 等待处理程序在单个线程内完成

如果您的意思是“使用单个线程运行服务时”，那么这是正确的。

否则，情况并非如此。相反，当您同时调用单个服务对象（即套接字，而不是 io_service）的操作时，Asio 只是说行为是“未定义的”。

这似乎否定了允许多个线程首先调用 run 的意义。

除非处理需要相当长的时间。

Timer.5 示例介绍的第一段似乎很好地阐述了您的主题。

会议

要分离特定于请求的数据（缓冲区和端点），您需要一些会话的概念。Asio 中的一种流行机制是bound shared_ptrs或 shared-from-this 会话类（boost bind 支持直接绑定到 boost::shared_ptr 实例）。

斯特兰德

为了避免对m_socket您的成员进行并发、不同步的访问，可以添加锁或使用strand上面链接的 Timer.5 示例中记录的方法。

演示

在这里供您欣赏的是Daytime.6异步 UDP 日间服务器，经过修改以与许多服务 IO 线程一起工作。

请注意，从逻辑上讲，仍然只有一个 IO 线程 (the strand)，因此我们不会违反套接字类记录的线程安全性。

但是，与官方示例不同，响应可能会无序排队，具体取决于 中实际处理所花费的时间udp_session::handle_request。

请注意

• 一个udp_session类来保存每个请求的缓冲区和远程端点
• 一个线程池，能够在多个内核上扩展实际处理（而不是 IO）的负载。

#include <ctime>
#include <iostream>
#include <string>
#include <boost/array.hpp>
#include <boost/bind.hpp>
#include <boost/shared_ptr.hpp>
#include <boost/enable_shared_from_this.hpp>
#include <boost/make_shared.hpp>
#include <boost/asio.hpp>
#include <boost/thread.hpp>

using namespace boost;
using asio::ip::udp;
using system::error_code;

std::string make_daytime_string()
{
    using namespace std; // For time_t, time and ctime;
    time_t now = time(0);
    return ctime(&now);
}

class udp_server; // forward declaration

struct udp_session : enable_shared_from_this<udp_session> {

    udp_session(udp_server* server) : server_(server) {}

    void handle_request(const error_code& error);

    void handle_sent(const error_code& ec, std::size_t) {
        // here response has been sent
        if (ec) {
            std::cout << "Error sending response to " << remote_endpoint_ << ": " << ec.message() << "\n";
        }
    }

    udp::endpoint remote_endpoint_;
    array<char, 100> recv_buffer_;
    std::string message;
    udp_server* server_;
};

class udp_server
{
    typedef shared_ptr<udp_session> shared_session;
  public:
    udp_server(asio::io_service& io_service)
        : socket_(io_service, udp::endpoint(udp::v4(), 1313)), 
          strand_(io_service)
    {
        receive_session();
    }

  private:
    void receive_session()
    {
        // our session to hold the buffer + endpoint
        auto session = make_shared<udp_session>(this);

        socket_.async_receive_from(
                asio::buffer(session->recv_buffer_), 
                session->remote_endpoint_,
                strand_.wrap(
                    bind(&udp_server::handle_receive, this,
                        session, // keep-alive of buffer/endpoint
                        asio::placeholders::error,
                        asio::placeholders::bytes_transferred)));
    }

    void handle_receive(shared_session session, const error_code& ec, std::size_t /*bytes_transferred*/) {
        // now, handle the current session on any available pool thread
        socket_.get_io_service().post(bind(&udp_session::handle_request, session, ec));

        // immediately accept new datagrams
        receive_session();
    }

    void enqueue_response(shared_session const& session) {
        socket_.async_send_to(asio::buffer(session->message), session->remote_endpoint_,
                strand_.wrap(bind(&udp_session::handle_sent, 
                        session, // keep-alive of buffer/endpoint
                        asio::placeholders::error,
                        asio::placeholders::bytes_transferred)));
    }

    udp::socket  socket_;
    asio::strand strand_;

    friend struct udp_session;
};

void udp_session::handle_request(const error_code& error)
{
    if (!error || error == asio::error::message_size)
    {
        message = make_daytime_string(); // let's assume this might be slow

        // let the server coordinate actual IO
        server_->enqueue_response(shared_from_this());
    }
}

int main()
{
    try {
        asio::io_service io_service;
        udp_server server(io_service);

        thread_group group;
        for (unsigned i = 0; i < thread::hardware_concurrency(); ++i)
            group.create_thread(bind(&asio::io_service::run, ref(io_service)));

        group.join_all();
    }
    catch (std::exception& e) {
        std::cerr << e.what() << std::endl;
    }
}

结束语

有趣的是，在大多数情况下，您会看到单线程版本的性能也一样好，而且没有理由使设计复杂化。

或者，io_service如果这确实是 CPU 密集型部分，您可以使用专用于 IO的单线程并使用老式工作池来执行请求的后台处理。首先，这简化了设计，其次这可能会提高 IO 任务的吞吐量，因为不再需要协调发布在链上的任务。