How to detect multiprocessing.Pipe is full?

user_6543850 0
Viewed 5423

Problem description: I am doing multiprocessing in Python and using multiprocessing.Pipe() to communicate between processes. I have been searching a lot but still couldn't find a way to detect whether the Pipe is full or not. For example below, the writePipe process keeps putting the number into 2 different Pipes (Odd & Even) and the readPipe process continuously read from these 2 Pipes. However, the speed of read from Odd Pipe is much faster so the Even Pipe will be full. At that moment, the writePipe process will be blocked while the readPipe process still keep waiting to read from Odd Pipe which causes deadlock.

My question: Is there any way we can detect the Pipe is full so we can stop putting number into full Pipe while still running and put number into the Pipe which still has spaces? 

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()
python-multiprocessing ipc python-3.x python
2 Answers

You can use the select function from the select module to implement a test for whether an output pipe is full. 

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))
user_35240470
Related