Skip to content

Streaming Subprocess Output Without Deadlocks

A deployment tool runs database migrations as a child process and shows progress in a web UI. It works in staging. In production, where the migration logs a warning for each of forty thousand rows, the job hangs at 3% forever with the child sleeping in write(). Nothing crashed and nothing timed out — the child filled the kernel's pipe buffer while the parent was busy awaiting proc.wait(), and each is now waiting for the other. communicate() would avoid the deadlock, but it buffers the entire output until the child exits, which rules out live progress and holds hundreds of megabytes of logs in memory. Long-running children need their output streamed: both pipes drained for the whole life of the process, lines handed over as they arrive, memory bounded, and the ability to stop the child as soon as the output says it is time. This guide builds that, including the less common directions — feeding large input through stdin and reading binary or unterminated output.

Prerequisites

Waiting versus draining 2 columns contrasting await proc.wait(), drain, then wait. Waiting versus draining await proc.wait() nobody reads the pipe child writes 64 KiB pipe full: child blocks parent waits for exit neither side moves drain, then wait reader per pipe child writes freely readers empty the pipe EOF on both streams wait() returns Start the readers before anything waits for the child to exit.

1. Reproduce the pipe deadlock

Seeing the deadlock once makes the rule stick. The child below writes one megabyte to stdout and then exits. The parent requests a pipe and waits for the exit without reading. Linux pipe buffers hold 64 KiB by default, so the child blocks after writing the first 64 KiB and never exits.

import asyncio
import sys

BIG_WRITER = "import sys; sys.stdout.write('x' * 1_000_000); sys.stdout.flush()"


async def main() -> None:
    proc = await asyncio.create_subprocess_exec(
        sys.executable, "-c", BIG_WRITER, stdout=asyncio.subprocess.PIPE)
    try:
        async with asyncio.timeout(2):
            await proc.wait()                    # child is blocked writing; this never returns
        print("no deadlock")
    except TimeoutError:
        print("deadlocked: the pipe is full and nobody is reading")
        out, _ = await proc.communicate()        # reading unblocks the child
        print("drained", len(out), "bytes; exit code", proc.returncode)


asyncio.run(main())

The asyncio documentation warns about exactly this for wait() with PIPE. The same deadlock happens with two pipes read sequentially: reading stdout to the end while the child is blocked writing a burst of stderr stalls both sides just as effectively.

Verify: the script prints deadlocked: ... after two seconds, then drained 1000000 bytes; exit code 0 — the child finishes the moment someone reads.

2. Drain stdout and stderr concurrently

The fix is structural: one reader task per pipe, running for the entire life of the process, started before anything waits for exit. A TaskGroup expresses that directly, and a bounded deque keeps a tail of each stream for error reports without holding everything.

import asyncio
import sys
from collections import deque
from typing import Callable

CHATTY = """
import sys, time
for i in range(5):
    print(f'progress {i * 25}%', flush=True)
    sys.stderr.write('warn ' + 'z' * 20000 + '\\n'); sys.stderr.flush()
    time.sleep(0.05)
print('progress 100%', flush=True)
"""


async def drain_lines(stream: asyncio.StreamReader, sink: Callable[[str], None],
                      keep: deque[str]) -> None:
    while True:
        try:
            raw = await stream.readline()
        except ValueError:                        # a line longer than the reader limit
            keep.append("<line over limit dropped>")
            continue
        if not raw:                               # EOF: the child closed this pipe
            return
        line = raw.decode(errors="replace").rstrip("\n")
        keep.append(line)
        sink(line)


async def run_streaming(argv: list[str], on_stdout: Callable[[str], None],
                        on_stderr: Callable[[str], None], tail: int = 50):
    proc = await asyncio.create_subprocess_exec(
        *argv, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE,
        stdin=asyncio.subprocess.DEVNULL, limit=256 * 1024)
    out_tail: deque[str] = deque(maxlen=tail)
    err_tail: deque[str] = deque(maxlen=tail)
    try:
        async with asyncio.TaskGroup() as tg:      # both pipes, for the whole life of the child
            tg.create_task(drain_lines(proc.stdout, on_stdout, out_tail))
            tg.create_task(drain_lines(proc.stderr, on_stderr, err_tail))
        return await proc.wait(), list(out_tail), list(err_tail)
    finally:
        if proc.returncode is None:                # cancelled, or a sink raised
            proc.kill()
            await proc.wait()


async def main() -> None:
    progress: list[str] = []
    code, out, err = await run_streaming([sys.executable, "-c", CHATTY],
                                         progress.append, lambda line: None)
    print(code, progress[-1], len(err))           # 0 progress 100% 5


asyncio.run(main())

Waiting for exit only after both readers have reached end-of-file guarantees the child can never block on a full pipe. If a sink raises — a progress callback that fails — the task group cancels the other reader, and the finally block makes sure the child does not outlive the failure. When the two streams do not need to be distinguished, stderr=asyncio.subprocess.STDOUT merges them into one pipe and one reader, which also preserves their relative order.

Verify: progress lines arrive while the child is still running rather than all at the end — log a timestamp in the sink to see them spaced 50 ms apart — and the stderr tail contains the five 20,000-character warnings.

3. Handle long lines and binary output

readline() refuses to buffer an unbounded line. The limit defaults to 64 KiB, set per process with limit=; a longer line raises ValueError: Separator is not found, and chunk exceed the limit. After that error the reader has already discarded the buffered data, so the rest of the oversized line arrives later as a fragment that looks like a normal line. Either raise the limit to cover the longest legitimate line, or stop treating the stream as lines.

import asyncio
import codecs
from typing import AsyncIterator


async def text_chunks(stream: asyncio.StreamReader, size: int = 65536) -> AsyncIterator[str]:
    """Bounded chunks of text, decoded incrementally so multi-byte characters never split."""
    decoder = codecs.getincrementaldecoder("utf-8")(errors="replace")
    while chunk := await stream.read(size):
        yield decoder.decode(chunk)
    if tail := decoder.decode(b"", final=True):
        yield tail


async def copy_binary(stream: asyncio.StreamReader, path: str, size: int = 1 << 20) -> int:
    """Stream binary output (an archive, an image) to disk without holding it in memory."""
    total = 0
    f = await asyncio.to_thread(open, path, "wb")
    try:
        while chunk := await stream.read(size):
            await asyncio.to_thread(f.write, chunk)          # one hop per megabyte, not per byte
            total += len(chunk)
    finally:
        await asyncio.to_thread(f.close)
    return total

The incremental decoder matters for chunked text: a UTF-8 character split across two read() calls would otherwise be replaced with garbage on both sides of the boundary. For binary output, chunk size sets the trade-off between thread hops and memory; one megabyte per hop keeps the file-I/O overhead small, as measured in async file I/O with aiofiles vs asyncio.to_thread.

Verify: a child printing a 300,000-character line with the default limit produces ValueError followed by a fragment; with limit=1024 * 1024 it arrives as one line. text_chunks fed "héllo" split between the two bytes of é yields "héllo" intact.

Choosing how to read child output A grid of 4 rows by 3 columns. Choosing how to read child output strategy memory live output long lines communicate() all output no fine readline(), 64 KiB limit one line yes ValueError readline(), raised limit one line yes up to limit read(n) chunks one chunk yes fine, no lines Line reading needs a limit that fits the data; chunks need no limit at all.

4. Feed stdin while reading stdout

Filters — compressors, formatters, jq, database import tools — read input and write output at the same time. Writing all input first and reading afterwards deadlocks as soon as the child's output pipe fills while the parent is still writing. Run the writer and the reader concurrently, and respect backpressure on the writer with drain().

import asyncio
import sys

UPPERCASE = "import sys\nfor line in sys.stdin: sys.stdout.write(line.upper())"


async def filter_rows(rows: int = 200_000) -> int:
    proc = await asyncio.create_subprocess_exec(
        sys.executable, "-c", UPPERCASE,
        stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.PIPE)

    async def feed() -> None:
        for i in range(rows):
            proc.stdin.write(f"row {i}\n".encode())
            if i % 1000 == 0:
                await proc.stdin.drain()          # pause when the child's stdin buffer is full
        await proc.stdin.drain()
        proc.stdin.close()                        # EOF: the child's loop over stdin ends
        await proc.stdin.wait_closed()

    received = 0

    async def collect() -> None:
        nonlocal received
        async for _ in proc.stdout:
            received += 1

    async with asyncio.TaskGroup() as tg:
        tg.create_task(feed())
        tg.create_task(collect())
    await proc.wait()
    return received


print(asyncio.run(filter_rows()))                 # 200000

write() never blocks; it appends to the transport's buffer. Without periodic drain() calls, a fast producer grows that buffer without limit in the parent's memory. Closing stdin is not optional either: most filters only finish, and flush their last output, when they see end-of-file.

Verify: the function returns 200000, and the parent's memory stays flat when rows is raised to ten million, because drain() holds the writer back to the child's pace.

5. Stop the child as soon as the output says so

Streaming makes it possible to act on output: fail fast on a fatal log line, return as soon as a server prints that it is listening, or stop a search once the first match appears. Leaving the read loop early must also end the child, or it keeps running — and eventually blocks on the pipe nobody reads any more.

import asyncio
import re
import sys

SERVER = ("import time\nprint('booting', flush=True)\ntime.sleep(0.2)\n"
          "print('listening on :8080', flush=True)\ntime.sleep(60)")


async def wait_for_output(argv: list[str], pattern: str, timeout: float) -> str | None:
    proc = await asyncio.create_subprocess_exec(
        *argv, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.STDOUT)
    rx = re.compile(pattern)
    try:
        async with asyncio.timeout(timeout):
            async for raw in proc.stdout:
                if match := rx.search(raw.decode(errors="replace")):
                    return match.group(0)
        return None                                # exited without printing the pattern
    finally:
        if proc.returncode is None:                # matched early, timed out, or cancelled
            proc.terminate()
            await proc.wait()


print(asyncio.run(wait_for_output([sys.executable, "-c", SERVER], r"listening on :\d+", 5)))

Here the child is terminated once the pattern is seen, which suits one-shot checks. When the child must keep running after the match — a test fixture that starts a server — keep a reader task draining its output for the rest of its life instead of terminating it, or redirect its output to a file, because a server that logs into an unread pipe freezes as soon as the buffer fills.

Verify: the call prints listening on :8080 about 200 ms after starting, not after 60 seconds, and no Python child remains in ps afterwards.

What should happen after the match? A decision on Does the child keep running with 3 outcomes. What should happen after the match? Does the child keep running? no, one-shot check terminate and reap finally block yes, parent stays up keep draining output reader task for life yes, unattended log to a file no pipe to fill A pipe that stops being read will eventually freeze the child writing to it.

Verification

Subprocess streaming is correct when:

  • No undrained pipes exist: every PIPE requested has a reader for the whole life of the child, or stderr is merged or sent to DEVNULL.
  • Output is live: consumers receive lines while the child runs, with timestamps spread across the run rather than bunched at exit.
  • Memory is bounded: only fixed-size tails are retained, and parent memory does not grow with child output volume.
  • Long lines are deliberate: the reader limit covers the longest legitimate line, and binary output is read in chunks.
  • Early exit ends the child: returning from a read loop, a timeout or a cancellation terminates and reaps the process.

Pitfalls & edge cases

  • Output buffering in the child. Many programs switch to block buffering when stdout is a pipe, so "live" output arrives in 4–8 KiB bursts. Use the tool's line-buffering flag, python -u for Python children, or stdbuf -oL for C programs that use stdio.
  • Sequential reads of two pipes. await proc.stdout.read() followed by await proc.stderr.read() deadlocks when the child fills stderr first. Read both concurrently or merge them.
  • Progress parsing on carriage returns. Tools that redraw a progress bar with \r never emit \n until the end. Read chunks and split on both \r and \n instead of using readline().
  • Forgetting to close stdin. A filter waiting for more input never exits, and proc.wait() hangs. Close stdin after the last write, or pass stdin=DEVNULL when there is no input.
  • Grandchildren holding pipes. End-of-file on stdout arrives only when every process holding the pipe has closed it. Terminate the whole process group, as in running subprocesses with asyncio.create_subprocess_exec.

Frequently Asked Questions

Why does asyncio proc.wait() hang when stdout=PIPE?

The child filled the operating system pipe buffer, typically 64 KiB on Linux, and is blocked writing because nothing is reading. The parent is waiting for the child to exit, so neither can make progress. Read the pipe while the process runs, use communicate(), or avoid requesting a pipe you do not read.

How do I read subprocess stdout and stderr at the same time in asyncio?

Start one reader task for each stream, for example inside an asyncio.TaskGroup, and only await proc.wait() after both readers have reached end of file. If you do not need to distinguish the streams, pass stderr=asyncio.subprocess.STDOUT to merge them into a single pipe with one reader.

What does Separator is not found, and chunk exceed the limit mean?

readline() met a line longer than the stream reader's limit, which defaults to 64 KiB. The buffered data is discarded and the remainder of that line arrives later as a fragment. Pass a larger limit to create_subprocess_exec, or read the stream in fixed-size chunks with read() when lines can be arbitrarily long.

How do I write to a subprocess stdin without running out of memory?

Call await proc.stdin.drain() regularly while writing, so the writer pauses whenever the child is not consuming input fast enough, and read the child's output concurrently so its output pipe cannot fill. Close stdin after the last write so the child sees end of file and finishes.