python - 如何检测多处理。管道已满？

Question

问题描述:我在 Python 中进行多处理并使用 multiprocessing.Pipe() 在进程之间进行通信。我一直在搜索，但仍然找不到检测管道是否已满的方法。例如下面的例子，writePipe 进程不断地将数字放入 2 个不同的管道(奇数和偶数)中，而 readPipe 进程不断地从这 2 个管道中读取。但是，从奇数管道读取的速度要快得多，因此偶数管道会被填满。此时，writePipe 进程将被阻塞，而 readPipe 进程仍在等待从 Odd Pipe 读取，从而导致死锁。

我的问题:有什么方法可以检测到管道已满，这样我们就可以在仍在运行的同时停止将数字放入完整的管道并将数字放入仍有空格的管道中？

from multiprocessing import Process, Pipe


def writePipe(sendNumberOdd, sendNumberEven):
    i = 0
    while True:
        if i % 2 == 0:
            sendNumberEven.send(i)
        else:
            sendNumberOdd.send(i)
        i += 1

def readPipe(recvNumberOdd, recvNumberEven):
    countEven = 0
    while True:
        countEven += 1
        print(countEven, recvNumberEven.recv())

        countOdd = 0
        while countOdd < 50:
            countOdd += 1
            print (countOdd, recvNumberOdd.recv())



if __name__ == '__main__':
    recvNumberOdd, sendNumberOdd = Pipe(duplex=False)
    recvNumberEven, sendNumberEven = Pipe(duplex=False)

    write = Process(target=writePipe, args=(sendNumberOdd, sendNumberEven))
    read = Process(target=readPipe, args=(recvNumberOdd, recvNumberEven))
    write.start()
    read.start()

    sendNumberOdd.close()
    sendNumberEven.close()

Answer 1

您可以使用 select 来自 select 的函数模块来实现输出管道是否已满的测试。

import select
import multiprocessing.connection as mpc


def pipe_full(conn):
    r, w, x = select.select([], [conn], [], 0.0)
    return 0 == len(w)


i, o = mpc.Pipe(duplex=False)

n = 0
while not pipe_full(o):
    o.send(n)
    n += 1

print('{} items fit.'.format(n))