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
});

---

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

不发送回调到写函数，它将返回一个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?? */
}

我注意到我没有任何等待响应的意思，因为它是在另一个异步循环上。我能够想象出一个我想要的系统，但我不知道如何将我自己的心智模型转化为具有协同程序的io。

// 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);
}
}
}

据我所知，"承诺类型"我在这里写的一个例子在boost中不存在。没有这种类型，我真的很纠结我的指挥系统如何存在。我需要为这种系统编写自己的协程类型吗?是否有一种方法，我可以摆脱使用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);
}
);
}