asio：如何将对象从一个 io 上下文传递到另一个

Question

我试图更好地理解异步 asio 的工作原理。

我有以下代码，我在套接字上调用 async_read 以读取接下来的 10 个字节的数据。

struct SocketReader {
    void do_read_body()
    {
        asio::async_read(socket_,
            asio::buffer(msg_, 10),
            [this](asio::error_code ec, std::size_t length)
            {
                if (!ec)
                {
                    //messages_to_work_on.emplace_back(msg_); // <-- I'm trying to send this msg_ instance to another io_context
                    do_read_body(); // call again
                }
                else
                {
                    socket_.close();
                }
            });
    }
std::vector<uint8_t> msg_;
asio::tcp::socket _socket;
}

这些读取是在他自己的 std::thread 中运行的 io_context 内完成的，我在队列中收集从套接字读取的所有消息。到目前为止一切顺利。

我还有另一个“工人”类，它只是根据队列中可用的内容执行一些工作：

struct Worker
{
    asio::io_context& io_context_;
    std::deque< std::vector<uint8_t> > queue;
    Worker(asio::io_context& io_context)
        : io_context_(io_context) {
        asio::post(io_context_, [this]() {doWork();});
    }
    void doWork() {
        if (!queue.empty())
        {
            // do some work with front()
            queue.pop_front();
        }
        asio::post(io_context_, [this]() {doWork();});
    }
};

那个也在他自己的io_context中执行，在他自己的线程中运行。所以socket线程和worker线程之间是有并发的。

将从套接字接收的数据发布到工人类的正确方法是什么？ 我在想我应该能够从套接字完成处理程序中调用，例如：

asio::post(worker_io_context, [this]() {worker.queue.push_back(msg_)});

这样，我至少可以确定没有同时使用工作队列。 但我不确定是否允许我从一个 io_context 发布到另一个，以及我是否不会以这种方式创建另一个竞争条件。 我也不太明白我的消息的内存应该放在哪里，尤其是从一个 io_context 到另一个的传输“中间”。是否需要按值传递消息（因为 this.msg_ 可以在执行后处理程序之前修改）？

谢谢！

Answer 1

我应该能够从套接字完成处理程序中调用， 类似：

asio::post(worker_io_context, [this]() {worker.queue.push_back(msg_)});

当然。

这样，我至少可以确定没有同时使用工作队列。 但我不确定我是否可以从一个 io_context 发布到另一个，

io_context 不是魔法。它们基本上是协作任务队列。

如果我不会以这种方式创建另一个竞争条件。

我不会坐在这里不看你的代码就做出裁决（我可能不想读完所有代码），但让我再说一遍：io_context 不是魔法。您可以根据线程、任务和资源以您已经知道的方式推理它们。

我也不太明白我的消息的内存应该放在哪里，尤其是从一个 io_context 到另一个的传输“中间”。是否需要我按值传递消息（因为 this.msg_ 可以在执行后处理程序之前修改）？

是的。确实。类似的东西

post(worker_io_context, [this, msg=std::move(msg_)]() {worker.queue.push_back(std::move(msg)); });

如果移动并不便宜，则可以选择使用引用计数的智能指针（如 shared_ptr）。如果您实际上在线程之间共享所有权，请考虑将其设为 smartpointer<T const>。

---

淋浴思想：也许你可以不用“工人”队列。由于您正在转向反应器式异步（使用 Asio），因此您可能专注于对任务进行排队，而不是数据。不这样做的原因包括您想要优先排队、负载平衡/背压等。[原则上所有这些都可以使用自定义执行器来实现，但我会坚持在这样做之前我所知道的。]

Answer 2

注意：您不需要额外的 io_context。
如果您必须进行长时间运行的计算，您可以编写自己的 async_xyz 函数并像其他异步函数一样使用它。这个想法是将工作发布到提升线程池以在那里进行计算，然后在工作完成时调用完成处理程序。这是一个使用 boost 线程池对密码进行耗时散列的示例。

template <boost::asio::completion_token_for<void (std::string)> CompletionToken>
auto
async_hash (boost::asio::thread_pool &pool, boost::asio::io_context &io_context, std::string const &password, CompletionToken &&token)
{
  return boost::asio::async_initiate<CompletionToken, void (std::string)> (
      [&] (auto completion_handler, std::string const &passwordToHash) {
        auto io_eq = boost::asio::prefer (io_context.get_executor (), boost::asio::execution::outstanding_work.tracked);
        boost::asio::post (pool, [&, io_eq = std::move (io_eq), completion_handler = std::move (completion_handler), passwordToHash] () mutable {
          auto hashedPw = pw_to_hash (passwordToHash);
          boost::asio::post (io_eq, [hashedPw = std::move (hashedPw), completion_handler = std::move (completion_handler)] () mutable { completion_handler (hashedPw); });
        });
      },
      token, password);
}

然后调用它：

auto hashedPw = co_await async_hash (pool, io_context, createAccountObject.password, boost::asio::use_awaitable);

看来你不使用协程 ts 所以我认为你必须做这样的事情

async_hash (pool, io_context, createAccountObject.password, /*your lambda here*/);

Answer 3

您不需要多个 io_context，即使对于多线程应用程序，您也可以只使用一个 io_context。你想让你的 SocketReader 成为一个共享指针，每次发生读取时，添加一个计数。我假设接受器、套接字创建和 some_io_context.run() 部分已完成。我会这样做：

class SocketReader 
   : public std::enable_shared_from_this<SocketReader> // we need this! 
{
public:
    // Constructor
    SocketReader(io_context& ctx) 
    : ctx_(ctx)
    {
    }
    
    // read
    void do_read_body()
    {
        auto self(this->shared_from_this()); // we need this!
        asio::async_read(socket_,
            asio::buffer(msg_, 10),
            [this](asio::error_code ec, std::size_t length)
            {
                if (!ec)
                {
                    // later edit - @sehe is spot on -> is better to move it
                    asio::post(ctx_, [this, msg=std::move(msg_)]() { // do the long work });
                    do_read_body();
                }
                else
                {
                    socket_.shutdown(asio::tcp::socket::shutdown_send, ec);
                    socket_.close();
                }
            });
    }
    
private:
    
    io_context& ctx_; // l.e. ctx_ must be a reference
    vector<uint8_t> msg_;
    asio::tcp::socket _socket;
};

...
// somewhere in the code
auto s_reader = std::make_shared<SocketReader>(some_io_context);
s_reader->do_read_body();