boost::asio - 服务器之间的冲突，双向通信

Question

我正在尝试使用 ASIO 编写一个允许服务器充当客户端的应用程序。例如：

我有 3 台服务器需要相互通信。在与网络中的其他服务器通信时，它们需要能够充当客户端。所有 3 台服务器都可以通过 unix 域套接字或带有 SSL 的 TCP/IP 服务请求。

以下是数据的流动方式：

1) 独立客户端连接到服务器 A（通过 unix 域套接字）并向其发送请求。

2) 服务器尝试响应请求，但如果不能，它会启动到服务器 B 的 TCP/IP 连接（现在服务器 A 充当服务器 B 的客户端）并将请求转发给它。服务器还会“污染”数据包，告诉服务器 B 不要将消息转发到另一台服务器，这样就不会创建无限循环。

3) 如果服务器 B 可以处理请求，服务器 B 会响应服务器 A。

4) 如果服务器 B 可以处理请求，服务器 A 将响应返回给独立客户端。

5) 如果服务器 B 无法处理请求，服务器 A 会尝试联系服务器 C、服务器 D、服务器 E 等

这行得通...直到拥有自己的独立客户端的服务器 B 尝试联系服务器 A，同时服务器 A 尝试联系服务器 B。它创建了一个冲突，两个服务器将无限期地等待获得响应从另一个。使用截止时间计时器可以避免无限期等待，但这并不能解决问题。

这样做的正确方法是什么？

编辑：我将 Server 拆分为 2 个类（Server 和 PeerProxy），在单独的线程中运行，但我仍然遇到死锁。

这就是我所做的。我将 Unix 侦听器和 TCP 侦听器拆分为 Server 和 PeerProxy 类。服务器有自己的io_service，PeerProxy也有自己的。当服务器启动时，它还会启动在第二个线程中运行的 PeerProxy（因此它不会阻止服务器的执行）。现在的数据流是这样的：

独立客户端 -> 服务器 A（无法回答）-> PeerProxy B -> 服务器 B（得到答案） -> PeerProxy B -> 服务器 A -> 独立客户端

同样的问题，当服务器 B 的独立客户端在服务器 A 进入 PeerProxy B 的同时进入 PeerProxy A 时出现死锁。

Answer 1

您应该在服务器中异步处理每个请求，即将处理分成单独的执行线程。这样服务器就可以保持响应，即它们可以在与其他客户端或服务器通信时对新请求做出反应。

因此，在您的情况下，当两个客户端 1 和 2 向服务器 A 和 B 发送请求时，只有另一台服务器可以回答（或不回答），这两个服务器可能如下所示：

Server A:                                   Server B:
Thread 0  | Thread 1     | Thread 2         Thread 0  | Thread 1     | Thread 2

listen...                                   listen...
-> req 1                                    -> req 2
listen... | handle req 1                    listen... | handle req 2
listen... | forward to B                    listen... | forward to A
-> req B  | wait...                         -> req A  | wait...
listen... | wait...        | handle req B   listen... | wait...      | reject req A  
listen... | -> B: rejected | answer req B   listen... | wait...       
listen... | forward to C                    listen... | -> A: answer       
listen... | -> C: answer                    listen... | req 2 done       
listen... | req 1 done                      listen...
listen...                                   listen...

这里，每个服务器的线程 0 除了侦听传入请求并旋转处理这些请求的其他线程之外没有其他用途。其他线程每个都处理一个请求，要么回答它，要么将它转发给所有服务器，或者如果它已被“污染”，则拒绝它。

注意：这些线程不一定是真正的线程对象。它们可以是 ASIO 异步*调用序列或某些线程框架（如 TBB）中的轻量级任务。

更新：我将为您发布一些 sceleton 伪代码，我将如何使用 Boost.Asio 实现服务器。为此，我想介绍一个我发现对理解 Asio 的执行很有用的概念：您可以将其视为状态机，其中 async_* 调用是状态转换，而处理程序是状态。通常，每个处理程序执行都有一个async_*-call，这意味着您从一种状态转到另一种状态。如果您在处理程序中有多个后续async_*-call，则意味着处理程序正在产生辅助执行线程。如果处理程序没有调用任何async_* 函数，则相应的执行线程结束。

现在开始实施。

Thread 0 就像典型的 Asio 教程所示，创建一个套接字并监听传入的连接。唯一的事情是，在每个新的客户端连接上都会产生一个新的执行线程（读取：处理程序序列）：

accept_handler(client connection) {
  async_read(client_request, request_handler) //spawn new thread of execution
  async_accept(next client connection, accept_handler) //transition to accept_handler
}

线程 N：以 request_handler 开头：

request_handler(client_request) {
  if canProcess 
    async_send_answer(client, done_handler) //transition to done_handler
  else  //relay request to first server on list
    async_connect(first server on list, connect_handler) //transition to connect_handler
}  

done_handler 通常会记录成功的答案并且不调用另一个async_* 函数，这意味着与客户端的连接将被关闭并且执行线程结束。

将请求发送到其他服务器的处理程序序列是典型的连接-发送-接收-断开序列：

connect_handler          -- async_send(request) ---------> send_handler
send_handler             -- async_read(answer) ----------> read_handler
read_handler (no answer) -- async_connect(next server) --> connect_handler

如果从其中一台服务器接收到答案或列表中没有更多服务器，则该循环结束：

read_handler (answer ok)       -- async_send_answer(client) --> done_handler
read_handler (no more servers) -- async_send_fail(client) ----> done_handler

Answer 2

这是一个简单的竞争条件。您希望实现某种原子锁变量或信号量或标志，因此如果一台服务器即将向另一台服务器发送请求，它将从那一刻起拒绝来自其他服务器的任何传入请求。我可能会使用 std::atomic 来实现它。