Asyncio stream process data with pipe between two subprocesses

Question

Hello I can't find a solution for the example here. I found https://stackoverflow.com/a/36666420/6089311 but there is no stream reading.

I'd like to run two subprocesses independent of python program (running in the background). The first subprocess feeds the second process through the pipe and I want to do some processing with the stdout lines by stream way.

The example bellow is blocked but I don't know why:

import asyncio
import os

async def foo():
    read, write = os.pipe()
    process_1 = await asyncio.create_subprocess_exec('ls', stdout=write)
    process_2 = await asyncio.create_subprocess_exec('wc', stdin=read, stdout=asyncio.subprocess.PIPE)
    
    async for l in process_2.stdout:
        # streaming process data
        print(l)
    
    os.close(write)
    os.close(read)

await foo() # jupyter call
# async.run(foo()) # python call

do you get error message? always put full error message (starting at word "Traceback") in question (not comment) as text (not screenshot, not link to external portal). There are other useful information. — furas, Feb 16 '21 at 12:55

furas · Accepted Answer · 2021-02-16T13:08:45.833

2

Code works for me if I move close() in places like in your link.

But probably it is not what you expect.

import asyncio
import os

async def foo():
    read, write = os.pipe()
    
    process_1 = await asyncio.create_subprocess_exec('ls', stdout=write)
    os.close(write)

    process_2 = await asyncio.create_subprocess_exec('wc', stdin=read, stdout=asyncio.subprocess.PIPE)
    os.close(read)
    
    async for line in process_2.stdout:
        # streaming process data
        print(line.decode())

#await foo() # jupyter call
asyncio.run(foo()) # python call

Eventually I can close read later but I have to close write before for-loop.

import asyncio
import os

async def foo():
    read, write = os.pipe()
    
    process_1 = await asyncio.create_subprocess_exec('ls', stdout=write)
    process_2 = await asyncio.create_subprocess_exec('wc', stdin=read, stdout=asyncio.subprocess.PIPE)
    
    os.close(write)

    async for line in process_2.stdout:
        # streaming process data
        print(line.decode())

    os.close(read)

#await foo() # jupyter call
asyncio.run(foo()) # python call

edited Feb 16 '21 at 13:08

answered Feb 16 '21 at 12:59

furas

134,197
12
106
148

Does it mean, that process_1 writes all data into pipe at first and all data is stored in RAM? In case of process_1 is `cat` of big file and process_2 will be slow to process data from stdin? – Jan Feb 16 '21 at 15:32
I'd like to read from process_2 while the process_1 is running. – Jan Feb 16 '21 at 15:55
I'm sorry, that was my misunderstanding of how the pipe works... I didn't understand why I should close the pipe before reading the process and forgot that the subprocess has another copy of the pipe. Yes, you are right, it works, thank you! – Jan Feb 16 '21 at 18:58
Could be fine to explain why the program is blocked if I don't close the write pipe before reading `process_2.stdout` (it sould be fine for another users). – Jan Feb 18 '21 at 11:37

Asyncio stream process data with pipe between two subprocesses

1 Answers1