When using asyncio to read from pipe, if process on the other side of pipe crashes read operation hangs indefinitely.

Example:


import asyncio
import contextlib
import os
import signal
import time

prx, ctx = os.pipe()

read_transport = None
read_stream = None

async def _connect_read(loop):
    global read_transport
    global read_stream
    stream_reader = asyncio.StreamReader()
    def protocol_factory():
        return asyncio.StreamReaderProtocol(stream_reader)
    rpipe = os.fdopen(prx, mode='r')

    transport, _ = await loop.connect_read_pipe(protocol_factory, rpipe)
    read_transport = transport
    read_stream = stream_reader

def close():
    read_transport.close()

@contextlib.asynccontextmanager
async def connect():
    try:
        loop = asyncio.get_event_loop()
        await _connect_read(loop)
        yield
    finally:
        close()

cpid = os.fork()
if cpid == 0:
    os.kill(os.getpid(), signal.SIGKILL)
    os.write(ctx, b'A')
    time.sleep(10.0)
else:
    async def read_from_child():
        async with connect():
            input = await read_stream.read(1)
            print('Parent received: ', input)

    asyncio.run(read_from_child())

@kormang this is an expected behaviour, this is a problem even for the OS level, just because it is impossible to know when the reader needs to stop waiting, the best option here is to implement some timeout mechanism.

@jaswdr Hm, this is was somewhat unexpected for me, since when reading directly from pipe, and EOF is sent.

Timeout was somewhat awkward in my case, since I don't know when other process will start sending, and how long it would take. On the other hand, I use asyncio loop, and can do this asynchronously, so I get notified when child process dies, by other means, and close the stream. So there are plenty of possible workarounds, but I'm not sure it is impossible to solve the problem on the library level yet. It would take more digging into implementation from my side, however.

In the example script, I believe you need to close the write end of the pipe in the parent after forking:

cpid = os.fork()
if cpid == 0:
    # Write to pipe (child)
else:
    # Parent
    os.close(ctx)
    # Read from pipe

This is the same with synchronous code: os.read(prx, 1) also hangs. You only get EOF when nothing has the write end open any more. All the asyncio machinery doesn't really make any difference to this.

For a similar reason, the code writing (the child, in this case) should close the read end of the pipe after forking. If the parent goes away but the child still has the read end open, then trying to write to the pipe can hang (if the buffer is already full). If the child has closed the read end, trying to write will give you a BrokenPipeError.

Thanks, takluyver!

You are right. Synchronous code that I was comparing it to had os.close(ctx), but I forgot to add it in the async example, so I thought it was a bug.

Closing this issue.

stage: resolved