python - 什么可以使 connection.send() 阻塞？ (来自 conn1、conn2 = multiprocessing.Pipe() )

Question

我正在调试从 2 个传感器收集信息的应用程序:一个网络摄像头和一个麦克风。

总体架构非常简单:

• 主进程通过管道向子进程(每个子进程一个)发送消息(开始、停止、get_data)。
• 子进程收集数据并将其发送给主进程

子进程和主进程无限循环处理命令(主进程来自用户，子进程来自主进程)。

它在全局范围内有效，但我无法停止子进程。

我已经记录了代码，它似乎发生了两件事:

1. “停止”消息已发送但未通过管道。
2. 子进程继续发送数据和 conn.send(data) block 。

此行为显然与连接状态有关，因为不发送任何内容的子进程没有此行为。尽管如此，我还是看不出如何调试/修改似乎合理的当前架构。

那么，是什么导致了这种阻​​塞行为以及如何避免呢？

这是为子进程中无限循环的每次迭代执行的代码:

    def do(self):
        while self.cnx.poll():
            msg = self.cnx.recv()
            self.queue.append(msg)
        #==
        if not self.queue:
            func_name = 'default_action'
            self.queue.append([func_name, ])
        #==
        msg             = self.queue.pop()
        func_name, args = msg[0], msg[1:]
        #==
        res = self.target.__getattribute__(func_name)(*args)
        #==
        running = func_name != 'stop'
        #==
        if res and self.send:
            assert running
            self.output_queue.append(res[0])
        if self.output_queue and running:
            self.cnx.send(self.output_queue.popleft())
        #==
        return running

update : 似乎不能在两端同时写入 Pipe。如果将上述代码的最后几行更改为:

        if self.output_queue and running:
            if not self.cnx.poll(): 
                self.cnx.send(self.output_queue.popleft())

这个问题仍然悬而未决，因为 Pipe 默认情况下被记录为全双工并且根本没有记录这种行为。我一定是误会了什么。请赐教!

更新 2:需要说明的是，在这种情况下，连接不会关闭。描述事件的顺序:

• 主进程发送一条消息(“停止”)(它在发送消息之前清空连接)
• 主进程进入一个(无限)循环，该循环在子进程终止时停止。
• 同时，子进程在发送过程中被阻塞，永远不会收到消息。

Answer 1

全双工multiprocessing.Pipe 实现为socketpair()。调用 .send 可以在与套接字通话时由于所有正常原因而阻塞。根据您的描述，我认为您的 Pipe 的阅读器可能已经停止阅读并且数据已经在内核的缓冲区中累积到您的 .send 阻塞的程度.

如果您显式地.close 接收方，当您尝试.发送。如果您的接收连接超出范围，这可能会自动发生。您可以通过更小心地不要将引用(直接或间接)存储到接收方来解决问题，以便在该线程消失时将其释放。

阻塞 .send 的简单演示:

import multiprocessing

a, b = multiprocessing.Pipe()

while True:
    print "send!"
    a.send("hello world")

现在请注意，一段时间后它停止打印“发送!”