Skip to content

Piping Data Between Subprocesses in asyncio

A shell writes pg_dump db | gzip | aws s3 cp - s3://bucket/x and the kernel does everything: three processes, two pipes, automatic back-pressure, no data through the shell itself. Reproducing that in asyncio takes more thought, because asyncio.subprocess.PIPE routes every byte through your process, and the default way of waiting for a process that has a pipe attached deadlocks. This guide covers both routes — pumping through Python when you need to see the data, and handing kernel pipes to the children when you do not — with the failure modes reproduced so they are recognisable when they appear in a service.

Prerequisites

  • Python 3.11+ on a Unix-like system. The measurements below are from Linux; os.pipe() behaves the same on macOS, and the kernel-pipe route does not apply on Windows.
  • Subprocess basics from Subprocesses & File I/O, particularly create_subprocess_exec and communicate().
  • Stream back-pressure from Streams, Transports & Protocols, since drain() is the whole back-pressure story here.
Two routes for the same 200 MB 2 columns contrasting through your process, os.pipe() between children. Two routes for the same 200 MB through your process 0.25 s wall, 0.22 s parent CPU every byte is copied twice you can inspect and transform back-pressure via drain() the loop does the work os.pipe() between children 0.09 s wall, 0.00 s parent CPU the kernel moves the bytes the data is invisible to you back-pressure is automatic the loop stays idle Measured on this machine; the gap widens with volume and with loop load.

1. Pump the data through your process

When you need to inspect, transform, count or checksum the stream, the data has to pass through Python. Read from one child, write to the other, and let drain() apply back-pressure:

import asyncio


async def pump(source: asyncio.StreamReader, sink: asyncio.StreamWriter) -> None:
    while chunk := await source.read(65536):
        sink.write(chunk)
        await sink.drain()                             # blocks here if the consumer is slow
    sink.close()
    await sink.wait_closed()                           # the consumer needs the EOF


async def run() -> bytes:
    src = await asyncio.create_subprocess_exec(
        "pg_dump", "mydb", stdout=asyncio.subprocess.PIPE)
    dst = await asyncio.create_subprocess_exec(
        "gzip", "-c", stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.PIPE)
    _, compressed = await asyncio.gather(pump(src.stdout, dst.stdin), dst.stdout.read())
    await src.wait()
    await dst.wait()
    return compressed

Two details are load-bearing. await sink.drain() is what makes the pipeline self-limiting: without it, write() buffers in your process and a producer faster than the consumer will grow that buffer until memory runs out. And the gather is required — reading the consumer's output concurrently with pumping, because waiting to pump everything before reading anything reproduces the deadlock in step 3.

Moving 200 MB this way took 0.25 s of wall time and 0.22 s of parent CPU. The CPU figure is the point: your event loop did that work, so every other task in the process waited behind it.

Verify: the consumer's byte count matches the producer's, and the parent's memory stays flat during the transfer.

Pumping one stream into another 5 ordered steps. Pumping one stream into another read a chunk await src.stdout.read(65536) write it dst.stdin.write(chunk) await drain() this is the back-pressure close stdin at EOF the consumer needs it await both exit codes a failure is silent otherwise Without the drain, a fast producer buffers the whole stream in your process.

2. Hand the kernel a pipe instead

When Python does not need to see the bytes — compression, encryption, upload, a filter chain — give both children the two ends of one kernel pipe. stdin and stdout accept raw file descriptors:

import os


async def run_fast() -> bytes:
    read_fd, write_fd = os.pipe()
    src = await asyncio.create_subprocess_exec("pg_dump", "mydb", stdout=write_fd)
    os.close(write_fd)                                 # the parent must not hold it open
    dst = await asyncio.create_subprocess_exec(
        "gzip", "-c", stdin=read_fd, stdout=asyncio.subprocess.PIPE)
    os.close(read_fd)
    compressed = await dst.stdout.read()
    await src.wait()
    await dst.wait()
    return compressed

The same 200 MB took 0.09 s of wall time and no measurable parent CPU — the kernel copied the data between the two processes and the event loop stayed free.

os.close() on both descriptors after spawning is not optional. A pipe reports end-of-file only when every write end is closed, so a parent that keeps write_fd open leaves the consumer waiting forever after the producer exits. This is the second-most-common subprocess-pipeline hang, and the reason it is so hard to spot is that the code that causes it looks like resource tidiness deferred to later.

Chaining more stages is the same pattern. A three-stage printf | sort | uniq -c built from two os.pipe() pairs returned 2 a 2 b 1 c with exit codes [0, 0, 0].

Verify: the consumer terminates on its own after the producer exits — if it hangs, a descriptor is still open in the parent.

Four ways a subprocess pipeline hangs or lies A grid of 4 rows by 2 columns. Four ways a subprocess pipeline hangs or lies mistake symptom fix await proc.wait() with a PIPE hangs once 64 KB is buffered read the pipe, or communicate() communicate() while pumping ConnectionResetError on drain read stdout directly instead consumer exits early BrokenPipeError while writing catch it and stop pumping parent keeps the write end the consumer never sees EOF os.close(w) after spawning The first is the classic: a pipe holds 64 KB by default, then the writer blocks.

3. Never wait() on a process whose pipe you are not reading

A kernel pipe holds 64 KB by default (/proc/sys/fs/pipe-max-size caps it at 1 MB on this machine). Once it is full, the writing process blocks. If that process is your child and you are waiting for it to exit, both sides wait forever:

proc = await asyncio.create_subprocess_exec(*cmd, stdout=asyncio.subprocess.PIPE)
await proc.wait()                                      # deadlock as soon as 64 KB is buffered

Measured with a child writing 10 MB: wait() was still blocked after 2 seconds, and would have stayed blocked indefinitely. The documented alternatives are await proc.communicate(), which reads both pipes while waiting, or reading proc.stdout yourself until EOF and then calling wait().

The mirror-image mistake is calling communicate() on a process you are also pumping into:

ConnectionResetError: Connection lost

communicate() closes stdin as part of its contract, so the next drain() in your pump finds the stream gone. Use communicate() or pump manually — never both on the same process.

Verify: the deadlocking version times out under asyncio.wait_for while the communicate() version completes.

4. Handle a consumer that exits first

head -c 10, a downstream process that fails, a grep -q that stops at the first match — any of these closes the read end while the producer is still writing. What happens next depends on which route you took.

Pumping through Python, the parent gets the error:

try:
    while chunk := await src.stdout.read(65536):
        dst.stdin.write(chunk)
        await dst.stdin.drain()
except (BrokenPipeError, ConnectionResetError):
    pass                                               # the consumer is gone; stop pumping

That is an expected outcome, not a bug — head exiting early is the correct behaviour of head. Catch it, stop pumping, and decide whether the producer should be killed (src.kill()) or allowed to finish.

With a kernel pipe, the parent never sees it: the producer receives SIGPIPE, or an error if it ignores the signal. In the measured run the consumer exited with 0 and the producer exited with 1, reporting BrokenPipeError: [Errno 32] Broken pipe on stderr. Any code that treats a non-zero exit as failure will report a spurious error for what was a normal early exit — so check the consumer's status first, and treat the producer's broken pipe as expected when the consumer succeeded.

Verify: running the pipeline with head -c 10 as consumer yields ten bytes and no unhandled exception.

5. Collect every exit code and stderr

A pipeline that produces the right bytes can still have failed. gzip writing a truncated archive after a disk error exits non-zero while your output looks fine, and the shell's own $PIPESTATUS exists precisely because this is easy to miss.

codes = await asyncio.gather(src.wait(), dst.wait())
if any(codes):
    raise RuntimeError(f"pipeline failed with exit codes {codes}")

Give every stage stderr=asyncio.subprocess.PIPE and read it concurrently — a child that fills its stderr pipe blocks exactly as one filling stdout does. Under a TaskGroup, a stderr reader per stage plus the pump is a clean structure, and cancellation of the group tears the whole pipeline down.

Kill the processes on the way out. A cancelled task leaves running children behind unless you terminate() them in a finally, as covered in graceful shutdown and signals.

Verify: a deliberately failing stage makes the pipeline raise rather than return partial output.

Where the data sits 3 bars comparing kernel pipe, default with the others. Where the data sits kernel pipe, default 64 KB kernel pipe, max on this host 1 MB StreamWriter without drain() unbounded From /proc/sys/fs/pipe-max-size; the Python buffer grows until memory runs out. The fixed kernel buffer is what makes a pipeline self-limiting — keep it in the path.

Verification

A subprocess pipeline is correct when:

  • Every pipe is drained: no wait() on a process with an unread PIPE.
  • EOF propagates: the parent closes descriptors it handed to children, and closes stdin when the stream ends.
  • Back-pressure exists: every write() into a child is followed by await drain().
  • Early exits are handled: BrokenPipeError is caught and interpreted against the consumer's exit code.
  • All exit codes are checked: every stage's return code is collected, with stderr captured.
  • Children never outlive the task: cancellation terminates every process.

Pitfalls & edge cases

  • shell=True to get a pipeline. create_subprocess_shell("a | b") works and gives the shell's pipe handling, but only the last stage's exit code, plus quoting hazards with any untrusted input.
  • Reading line by line from binary streams. readline() on a binary stream with no newlines buffers until LimitOverrunError; use read(n) for arbitrary data.
  • Assuming the pipe buffer is large. 64 KB is the default; do not design around holding a message in the pipe.
  • Inheriting descriptors accidentally. os.pipe() descriptors are non-inheritable by default in Python 3.4+, which is why passing them explicitly as stdin/stdout works while relying on inheritance does not.
  • Windows. Passing raw descriptors as stdin/stdout does not work the same way, and the ProactorEventLoop has its own subprocess constraints.
  • Very large single messages. If a stage needs the whole input in memory anyway, a temporary file is simpler and lets you retry a stage without rerunning the producer.

Frequently Asked Questions

How do I pipe one subprocess into another with asyncio?

Either pump the data yourself — read from the first process's stdout, write to the second's stdin, and await drain() between writes — or create a kernel pipe with os.pipe() and pass the descriptors directly as stdout and stdin, closing both in the parent afterwards. The second route keeps the bytes out of your process entirely.

Why does await proc.wait() hang in asyncio?

Because the process was started with stdout=PIPE or stderr=PIPE and nobody is reading it. The pipe holds 64 KB, then the child blocks writing while you block waiting. Use await proc.communicate(), which drains both pipes, or read the streams to EOF before calling wait().

Why do I get ConnectionResetError when writing to a subprocess stdin?

Usually because communicate() was called on the same process — it closes stdin as part of its contract, so the next drain() fails. The other cause is a consumer that has exited, which surfaces as BrokenPipeError or ConnectionResetError; catch both and stop pumping.

How do I know which stage of a subprocess pipeline failed?

Collect every stage's exit code with asyncio.gather(*(p.wait() for p in stages)) and give each stage its own stderr=PIPE reader. The shell exposes the same information as PIPESTATUS; asyncio gives you no aggregate, so a pipeline that checks only the last stage will miss a failed producer.

Is os.pipe() faster than asyncio.subprocess.PIPE?

Measurably, when the data is large and you do not need to see it. Moving 200 MB between two children took 0.25 s of wall time and 0.22 s of event-loop CPU through Python, versus 0.09 s and no measurable parent CPU with a kernel pipe. Use PIPE when you need to transform or inspect the stream.