使用管道进行多处理 - 应用程序总是在大约 5 分钟后停止

Question

我正在开发一个简单的应用程序来测试 Python 中的管道。应用程序启动一个新进程，作为 "server"。 "server" 启动另外两个工作进程，向 "server" 发送一些数据当 "server" 收到消息时，它应该更新其状态，将其发送给所有工作人员以及主要过程。但是应用程序总是在收到 278 条消息后停止。我究竟做错了什么？这是代码：

class Server ( multiprocessing.Process ):

    def __init__( self, connection ):
        super( Server, self ).__init__()
        self.conn = connection

    def run( self ):
        state = []
        conn1a, conn1b = multiprocessing.Pipe()
        conn2a, conn2b = multiprocessing.Pipe()
        p1 = multiprocessing.Process( target=worker, args=(conn1b, 2, 'p1', 1 ) )
        p2 = multiprocessing.Process( target=worker, args=(conn2b, 3, 'p2', -1 ) )
        p1.start()
        p2.start()
        counter = 0
        while 1:
            if conn1a.poll() or conn2a.poll():
                if conn1a.poll():
                    state = conn1a.recv()
                if conn2a.poll():
                    state = conn2a.recv()
                self.conn.send( state )
                conn1a.send( state )
                conn2a.send( state )
                counter += 1
                print "counter", counter
        p1.join()
        p2.join()

def worker( conn, seconds, name, multiplier ):
    x = 0
    conn.send( [name, x*multiplier] )
    while 1:
        if conn.poll():
            x += seconds
            time.sleep( seconds )
            conn.send( [name, x*multiplier] )

if __name__ == '__main__':
    conn1, conn2 = multiprocessing.Pipe( False )
    p = Server( conn2 )
    p.start()
    while 1:
        print "received", conn1.recv()
    p.join()

应用程序的输出如下：

enter codstarting process p2
received ['p2', 0]
counter 1
starting process p1
received ['p1', 0]
counter 2
received ['p1', 2]
counter 3
...
counter 277
received ['p1', 332]
counter 278
received ['p2', -333]

Answer 1

您忘记从工人中的管道读取。管道被填满。

    if conn.poll():
        conn.recv()  # consume messages!
        x += seconds

使用管道进行多处理 - 应用程序总是在大约 5 分钟后停止

Multiprocessing using Pipes - application always stops after about 5 minutes

python

pipe

python-multiprocessing