Continous output with asyncio subprocess

Question

I'm using asyncio subprocess to execute a subcommand. I want to see the long-running process and save the content at the same time to a buffer for later use. Furthermore, I found this related question (Getting live output from asyncio subprocess), but it mainly centers around the use case for ssh.

The asyncio subprocess docs have an example for reading the output line-by-line, which goes into the direction of what I want to achieve. (https://docs.python.org/3/library/asyncio-subprocess.html#examples)

import asyncio
import sys

async def get_date():
    code = 'import datetime; print(datetime.datetime.now())'

    # Create the subprocess; redirect the standard output
    # into a pipe.
    proc = await asyncio.create_subprocess_exec(
        sys.executable, '-c', code,
        stdout=asyncio.subprocess.PIPE)

    # Read one line of output.
    data = await proc.stdout.readline()
    line = data.decode('ascii').rstrip()

    # Wait for the subprocess exit.
    await proc.wait()
    return line

date = asyncio.run(get_date())
print(f"Current date: {date}")

I adapted this example to the following:

async def subprocess_async(cmd, **kwargs):
    cmd_list = shlex.split(cmd)
    proc = await asyncio.create_subprocess_exec(
            *cmd_list,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.STDOUT, **kwargs)

    full_log = ""
    while True:
        buf = await proc.stdout.readline()
        if not buf:
            break
        full_log += buf.decode()
        print(f' {buf.decode().rstrip()}')
    await proc.wait()
    res = subprocess.CompletedProcess(cmd, proc.returncode,  stdout=full_log.encode(), stderr=b'')
    return res

The issue here is, that the proc.returncode value sometimes becomes None. I guess, I have a misunderstanding, how proc.wait() works and when it is safe to stop reading the output. How do I achieve continuous output using asyncio subprocess?

Copyholder

Can you post an example where proc.returncode becomes None? – Macknair

You can use semaphores and a global variable to write results and control locking resource but this will slow down the process, wait for pool to complete and get results is the best and efficient approach. Safe implementation is in here docs.python.org/3/library/… – Chatterbox

Answer 1

Your code is working fine as-is for me. What command are you trying to run that is causing your issue?

Two things I can think of to help, are

1. Instead of calling .wait() afterwards, set the returncode as the loop condition to keep running.
2. Don't wait for full line returns in case the program is like ffmpeg where it will do some tricks to paste over itself in console and not actually send newline characters.

Example code:

import asyncio, shlex, subprocess, sys


async def subprocess_async(cmd, **kwargs):
    cmd_list = shlex.split(cmd)
    proc = await asyncio.create_subprocess_exec(
        *cmd_list,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.STDOUT,
        **kwargs)

    full_log = b""
    while proc.returncode is None:
        buf = await proc.stdout.read(20)
        if not buf:
            break
        full_log += buf
        sys.stdout.write(buf.decode())
    res = subprocess.CompletedProcess(cmd, proc.returncode,  stdout=full_log, stderr=b'')
    return res


if __name__ == '__main__':
    asyncio.run(subprocess_async("ffprobe -i video.mp4"))

Anteroom