Can I use asyncio to read from and write to a multiprocessing.Pipe?
aiopipe
seems to do what you want! It can be used with the builtin multiprocessing
module, and provides a similar API to the regular blocking pipes.
multiprocessing.Pipe
uses the high level multiprocessing.Connection
module that pickles and unpickles Python objects and transmits additional bytes under the hood. If you wanted to read data from one of these pipes using loop.connect_read_pipe()
, you would have to re-implement all of this yourself.
The easiest way to read from a multiprocessing.Pipe
without blocking the event loop would be to use loop.add_reader()
. Consider the following example:
import asyncio
import multiprocessing
def main():
read, write = multiprocessing.Pipe(duplex=False)
writer_process = multiprocessing.Process(target=writer, args=(write,))
writer_process.start()
asyncio.get_event_loop().run_until_complete(reader(read))
async def reader(read):
data_available = asyncio.Event()
asyncio.get_event_loop().add_reader(read.fileno(), data_available.set)
if not read.poll():
await data_available.wait()
print(read.recv())
data_available.clear()
def writer(write):
write.send('Hello World')
if __name__ == '__main__':
main()
Pipes created using the lower-level os.pipe
don't add anything extra the way that pipes from multiprocessing.Pipe
do. As a result, we can use os.pipe
with loop.connect_read_pipe()
, without re-implementing any sort of inner-workings. Here is an example:
import asyncio
import multiprocessing
import os
def main():
read, write = os.pipe()
writer_process = multiprocessing.Process(target=writer, args=(write,))
writer_process.start()
asyncio.get_event_loop().run_until_complete(reader(read))
async def reader(read):
pipe = os.fdopen(read, mode='r')
loop = asyncio.get_event_loop()
stream_reader = asyncio.StreamReader()
def protocol_factory():
return asyncio.StreamReaderProtocol(stream_reader)
transport, _ = await loop.connect_read_pipe(protocol_factory, pipe)
print(await stream_reader.readline())
transport.close()
def writer(write):
os.write(write, b'Hello World\n')
if __name__ == '__main__':
main()
This code helped me figure out to use loop.connect_read_pipe
.