我目前正在使用 Asio C++ 库，并围绕它编写了一个客户端包装器。我最初的方法是非常基本的，只需要在一个方向上流式传输。要求已更改，我已切换到使用所有异步调用。除asio::async_write(...)外，大多数迁移都很容易。我使用了几种不同的方法，并且不可避免地遇到了每种方法的僵局。
应用程序连续以高容量流式传输数据。我一直远离 strand，因为它们不会阻塞并且可能导致内存问题，尤其是在服务器负载过重时。作业将备份，应用程序堆将无限增长。
因此，我创建了一个阻塞队列，只是为了找出跨回调和/或阻塞事件使用锁会导致未知行为的困难方式。
包装器是一个非常大的类，所以我将尝试在当前状态下解释我的景观，并希望得到一些好的建议：
- 我有一个按固定计划运行的
asio::steady_timer，可将检测信号消息直接推送到阻塞队列中。 - 专用于读取事件并将其推送到阻塞队列的线程
- 专用于使用阻塞队列的线程
例如，在我的队列中，我有一个queue::block()和queue::unblock()，它们只是条件变量/互斥锁的包装器。
std::thread consumer([this]() {
std::string message_buffer;
while (queue.pop(message_buffer)) {
queue.stage_block();
asio::async_write(*socket, asio::buffer(message_buffer), std::bind(&networking::handle_write, this, std::placeholders::_1, std::placeholders::_2));
queue.block();
}
});
void networking::handle_write(const std::error_code& error, size_t bytes_transferred) {
queue.unblock();
}
当套接字备份并且服务器由于当前负载而无法再接受数据时，队列将填满并导致死锁，从不调用handle_write(...)。
另一种方法完全消除了使用者线程，并依赖于handle_write(...)来弹出队列。这样：
void networking::write(const std::string& data) {
if (!queue.closed()) {
std::stringstream stream_buffer;
stream_buffer << data << std::endl;
spdlog::get("console")->debug("pushing to queue {}", queue.size());
queue.push(stream_buffer.str());
if (queue.size() == 1) {
spdlog::get("console")->debug("handle_write: {}", stream_buffer.str());
asio::async_write(*socket, asio::buffer(stream_buffer.str()), std::bind(&networking::handle_write, this, std::placeholders::_1, std::placeholders::_2));
}
}
}
void networking::handle_write(const std::error_code& error, size_t bytes_transferred) {
std::string message;
queue.pop(message);
if (!queue.closed() && !queue.empty()) {
std::string front = queue.front();
asio::async_write(*socket, asio::buffer(queue.front()), std::bind(&networking::handle_write, this, std::placeholders::_1, std::placeholders::_2));
}
}
这也导致了僵局，显然导致了其他种族问题。当我禁用心跳回调时，我绝对没有问题。但是，检测信号是必需的。
我做错了什么？什么是更好的方法？
看来我所有的痛苦都完全来自心跳。在我的异步写入操作的每个变体中禁用检测信号似乎可以解决我的问题，因此这让我相信这可能是使用内置asio::async_wait(...)和asio::steady_timer的结果。
Asio 在内部同步其工作，并在执行下一个作业之前等待作业完成。使用asio::async_wait(...)构造我的心跳功能是我的设计缺陷，因为它在等待挂起作业的同一线程上运行。当心跳等待queue::push(...)时，它与阿西奥造成了僵局。这可以解释为什么在我的第一个示例中从未执行asio::async_write(...)完成处理程序。
解决方案是将心跳放在自己的线程上，并让它独立于 Asio 工作。我仍在使用阻塞队列来同步对asio::async_write(...)的调用，但已修改我的使用者线程以使用std::future和std::promise。这样可以将回调与我的使用者线程完全同步。
std::thread networking::heartbeat_worker() {
return std::thread([&]() {
while (socket_opened) {
spdlog::get("console")->trace("heartbeat pending");
write(heartbeat_message);
spdlog::get("console")->trace("heartbeat sent");
std::unique_lock<std::mutex> lock(mutex);
socket_closed_event.wait_for(lock, std::chrono::milliseconds(heartbeat_interval), [&]() {
return !socket_opened;
});
}
spdlog::get("console")->trace("heartbeat thread exited gracefully");
});
}