1
\$\begingroup\$

My task is to run a subprocess using asyncio.create_subprocess_shell and yield the lines of output. I am using asyncio to avoid creating threads just to pump the streams.

  • It is necessary to separate stdout lines from stderr lines.
  • Both outputs need to be read simultaneously, it is not possible to read all of stdout or stderr first, because then the other output buffer in the subprocess may fill and it would stop running.
  • This function is incorporated into a gRPC service, which is the reason for wrapping the yielded lines in instances of cli_pb2.ExecuteReply(...)
  • .readline() returns "" on EOF, so when I see it, that is when to stop reading that stream.

I want to ask if my approach to reading stdout and stderr simultaneously can be improved. I was looking for something akin to UNIX select() and found asyncio.wait({...}, return_when=asyncio.FIRST_COMPLETED). Then I had to figure out which stream (stdout, stderr) the result corresponds to. I hacked at it by looking at _coro.

import asyncio

import cli_pb2
import cli_pb2_grpc

async def run(cmd):
    proc = await asyncio.create_subprocess_shell(
        cmd,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
    )

    stderr_readline = proc.stderr.readline()
    stdout_readline = proc.stdout.readline()
    pending = {stderr_readline, stdout_readline}

    while pending:
        done, pending = await asyncio.wait(pending, return_when=asyncio.FIRST_COMPLETED)
        for f in done:
            line = (await f).decode()
            if f._coro is stderr_readline:  # bad, but it works and can't think of better approach
                if line:
                    yield cli_pb2.ExecuteReply(stderr=line)
                    stderr_readline = proc.stderr.readline()
                    pending.add(stderr_readline)  # put it back into the `pending` set
            if f._coro is stdout_readline:
                if line:
                    print(line)
                    yield cli_pb2.ExecuteReply(stdout=line)
                    stdout_readline = proc.stdout.readline()
                    pending.add(stdout_readline)
    await proc.wait()
    status = proc.returncode
    yield cli_pb2.ExecuteReply(status=status)

Usage example

async def printlines():
    async for line in run("echo a; sleep 5; echo b; echo c"):
        print(line)

outputs (with a 5 second pause after first line

stdout: "a\n"
stdout: "b\n"
stdout: "c\n"

The protobuf definition is

syntax = "proto3";

option java_multiple_files = true;
option java_package = "djtests.cli";
option java_outer_classname = "CliProto";

package cli;

service Executor {
    rpc Execute (ExecuteRequest) returns (stream ExecuteReply) {
    };
}

message ExecuteRequest {
    string cmd = 1;
}

message ExecuteReply {
    oneof reply_fields {
        string stdout = 1;
        string stderr = 2;
        int32 status = 3;  // sent in the last message
    }
}
```
\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.