Boost awaitable：写入套接字并等待特定响应

Question

问题可能有点复杂。我会尽力解释最好的情况以及我想用什么工具来解决我的问题。

我正在编写一个套接字应用程序，它可能会写入套接字并期望得到响应。该协议以一种简单的方式实现了这一点：每个请求都有一个“命令 id”，它将被转发回响应中，这样我们就可以拥有对该特定请求做出反应的代码。

为简单起见，我们假设所有通信都是使用套接字中的 json 完成的。

首先，让我们假设这个会话类型：

using json = /* assume any json lib */;

struct socket_session {
    auto write(json data) -> boost::awaitable<void>;
    auto read() -> boost::awaitable<json>;

private:
    boost::asio::ip::tcp::socket socket;
};

通常，我会使用一个（非常）大致像这样的回调系统。

using command_it_t = std::uint32_t;

// global incrementing command id
command_it_t command_id = 0;

// All callbacks associated with commands
std::unordered_map<command_id_t, std::function<void(json)>> callbacks;

void write_command_to_socket(
    boost::io_context& ioc,
    socket_session& session,
    json command,
    std::function<void(json)> callback
) {
    boost::co_spawn(ioc, session->write(command), asio::detached);
    callbacks.emplace(command_id++, callback);
}

// ... somewhere in the read loop, we call this:
void call_command(json response) {
    if (auto const& command_id = response["command"]; command_id.is_integer()) {
        if (auto const it = callbacks.find(command_id_t{command_id}); it != callbacks.end()) {
            // We found the callback for this command, call it!
            auto const& [id, callback] = *it;
            callback(response["payload"]);
            callbacks.erase(it);
        }
    }
}

它会这样使用：

write_command_to_socket(ioc, session, json_request, [](json response) {
    // do stuff
});

---

随着我开始越来越多地将协程用于异步代码，我注意到这是在这种系统中使用协程的绝佳机会。

它不会向 write 函数发送回调，而是返回一个包含响应负载的 boost::awaitable<json>，我想象它有点像这样：

auto const json_response = co_await write_command_to_socket(session, json_request);

好的，问题来了

所以第一步就是像这样转换我的代码：

void write_command_to_socket(socket_session& session, json command) {
    co_await session->write(command);
    co_return /* response data from the read loop?? */
}

我注意到我没有任何意思等待响应，因为它在另一个异步循环中。我能够想象一个看起来像我想要的系统，但我不知道如何将我自己的心智模型转换为带有协程的 asio。

// Type from my mental model: an async promise
template<typename T>
struct promise {
    auto get_value() -> boost::awaitable<T>;
    auto write_value(T value);
};

// Instead of callbacks, my mental model needs promises structured in a similar way:
std::unordered_map<command_id_t, promise<json>> promises;

void write_command_to_socket(socket_session& session, json command) {
    auto const [it, inserted] = promises.emplace(session_id++, promise<json>{});
    auto const [id, promise] = *it;

    co_await session->write(command);

    // Here we awaits until the reader loop sets the value
    auto const response_json = co_await promise.get_value();
    co_return response_json;
}

// ... somewhere in the read loop

void call_command(json response) {
    if (auto const& command_id = response["command"]; command_id.is_integer()) {
        if(auto const it = promises.find(command_id_t{command_id}); it != promises.end()) {
            auto const& [id, promise] = *it;

            // Effectively calls the write_command_to_socket coroutine to continue
            promise.write_value(response["payload"]);
            promise.erase(it);
        }
    }
}

据我所知，我在这里作为示例编写的“promise 类型”在 boost.js 中不存在。没有那种类型，我真的很挣扎我的指挥系统如何存在。我需要为这种系统编写自己的协程类型吗？有没有办法让我摆脱使用 boost 的协程类型？

Answer 1

正如我所说，对于 asio，“承诺类型”不存在。相反，Asio 使用延续处理程序，这是一种回调，实际上可以调用回调或恢复协程。

要创建这样的继续处理程序，必须首先启动异步操作。如果需要，异步操作可以由另一个恢复，或者由许多异步操作组成。这是通过asio::async_initiate 函数完成的，该函数采用一些参数来保护延续的形式：

// the completion token type can be a callback,
// could be `asio::use_awaitable_t const&` or even `asio::detached_t const&`
return asio::async_initiate<CompletionToken, void(json)>(
    [self = shared_from_this()](auto&& handler) {
        // HERE! `handler` is a callable that resumes the coroutine!
        // We can register it somewhere
        callbacks.emplace(command_id, std::forward<decltype(handler)>(handler));
    }
);

要恢复异步操作，您只需调用延续处理程序：

void call_command(json response) {
    if (auto const& command_id = response["command"]; command_id.is_integer()) {
        if (auto const it = callbacks.find(command_id_t{command_id}); it != callbacks.end()) {
            // We found the continuation handler for this command, call it!
            // It resumes the coroutine with the json as its result
            auto const& [id, callback] = *it;
            callback(response["payload"]);
            callbacks.erase(it);
        }
    }
}

这是系统的其余部分，它的外观（非常粗略）：

using command_it_t = std::uint32_t;

// global incrementing command id
command_it_t command_id = 0;

// All callbacks associated with commands
std::unordered_map<command_id_t, moveable_function<void(json)>> callbacks;

void write_command_to_socket(
    boost::io_context& ioc,
    socket_session session,
    json command
) -> boost::asio::awaitable<json> {
    return asio::async_initiate<boost::asio::use_awaitable_t<> const&, void(json)>(
        [&ioc, session](auto&& handler) {
            callbacks.emplace(command_id, std::forward<decltype(handler)>(handler));
            boost::asio::co_spawn(ioc, session.write(command), asio::detached);
        }
    );
}