Fix Python BrokenPipeError with head and Verify Pipeline Exit Codes

Piping a Python export into head -n 3 can print the requested rows and then a BrokenPipeError: [Errno 32] Broken pipe traceback. A background job using pipefail may also report failure.

For beginner-to-intermediate Linux users, this tutorial reproduces the issue with the standard library, adds explicit handling to an output CLI, and checks complete output, early termination, and real failure. The demo generates text only; it performs no database updates or batch file changes.

1. Prepare and reproduce

Use Linux, Python 3.9 or newer, Bash, and head. No pip packages are required. The examples were tested on Linux with Python 3.11.4.

1. Create a demo directory and run subsequent commands there. Use a different directory name when repeating the experiment:

python3 --version
command -v bash
head --help
mkdir broken-pipe-demo
cd broken-pipe-demo

2. Run an output command without pipe handling:

python3 -u -c 'for n in range(1000000): print(f"row={n:06d}")' | head -n 3

You will generally see three rows followed by a traceback. -u makes writes happen promptly. A small buffered output may fit into the pipe before the consumer closes, so three output rows alone cannot reliably test a broken pipe.

After reading enough, head closes its input while the producer continues writing to a pipe without readers. Python’s SIGPIPE note explains the exception and entry-point handling.

If the symptom is delayed logs, start with Python output buffering and pipefail. -u changes buffering; a closed pipe remains unwritable.

2. Method 1: handle closed stdout at the CLI entry point

1. Save this complete program as emit_rows.py:

"""Emit demo rows with explicit pipe failure handling. Python 3.9+."""
import argparse
import os
import sys


def positive_int(value):
    number = int(value)
    if number < 1:
        raise argparse.ArgumentTypeError("must be at least 1")
    return number


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--rows", type=positive_int, default=1_000_000)
    parser.add_argument("--fail", action="store_true", help="simulate task failure")
    args = parser.parse_args()
    if args.fail:
        print("ERROR: simulated task failure", file=sys.stderr)
        return 7
    for number in range(1, args.rows + 1):
        print(f"row={number:06d}")
    return 0


def cli():
    try:
        status = main()
        sys.stdout.flush()
        return status
    except BrokenPipeError:
        sink = os.open(os.devnull, os.O_WRONLY)
        try:
            os.dup2(sink, sys.stdout.fileno())
        finally:
            os.close(sink)
        return 1


if __name__ == "__main__":
    raise SystemExit(cli())

2. Inspect help and check ordinary output:

python3 emit_rows.py --help
python3 emit_rows.py --rows 3

The last command should print:

row=000001
row=000002
row=000003

cli() explicitly flushes stdout inside try, including output left buffered after the loop. On a broken pipe, it redirects stdout’s underlying descriptor to the null device before exit. The demo returns 1 for incomplete output, following the handling approach in the SIGPIPE note above.

os.dup2() copies the null-device descriptor onto stdout; finally closes the extra descriptor. This entry point targets Linux CLIs with ordinary terminal streams, rather than Notebooks or embedded hosts that replace sys.stdout.

Only stdout writes need this policy here. In networked programs, narrow the catch to stdout writes and flushing, preserving socket error handling. Avoid globally setting SIGPIPE to SIG_DFL; the Python note explains its effect on interrupted socket writes.

3. Check pipeline status and output completeness

1. Use Bash explicitly and save PIPESTATUS immediately. This demo leaves set -e disabled so it can inspect the expected nonzero result:

bash <<'BASH'
set -o pipefail
python3 emit_rows.py | head -n 3
codes=("${PIPESTATUS[@]}")
printf 'python=%s head=%s\n' "${codes[0]}" "${codes[1]}"
test "${codes[0]}" -eq 1 && test "${codes[1]}" -eq 0
BASH

Expect the first three rows followed by python=1 head=0, with no stderr traceback. Python did not finish all output, while head completed its request. The final test is a demonstration assertion: its success means those statuses matched expectations.

Later commands update PIPESTATUS, so copy it immediately after the pipeline. Run Bash’s help set to inspect the installed pipefail rule: with it enabled, the pipeline returns the rightmost nonzero command status, here 1. Without it, the final head usually determines the status and can hide producer errors.

2. If you only need three rows, run python3 emit_rows.py --rows 3 directly. If the full task must succeed before previewing, save all output first:

bash <<'BASH'
set -euo pipefail
python3 emit_rows.py --rows 10 > complete.txt
head -n 3 complete.txt
wc -l < complete.txt
BASH

The last line should be the count 10. This demo overwrites complete.txt in its demo directory. For production exports, choose a new filename and validate it before passing it downstream. Avoid blanket || true handling or discarding error output to hide the cause.

When Python launches the command, use subprocess timeouts, errors, and logs to keep checking its actual exit status.

4. Test buffering, closed pipes, and real errors

Run this in the directory containing emit_rows.py. It closes the pipe’s read end before starting the child, removing dependence on head timing. Each child has a ten-second timeout:

python3 - <<'PY_CHECK'
import os
import subprocess
import sys

command = [sys.executable, "emit_rows.py"]
normal = subprocess.run(
    [*command, "--rows", "3"], capture_output=True, text=True, timeout=10,
)
assert normal.returncode == 0, normal
assert normal.stdout == "row=000001\nrow=000002\nrow=000003\n", normal
assert normal.stderr == "", normal

for options in ([], ["-u"]):
    for count in (1, 1_000_000):
        reader, writer = os.pipe()
        os.close(reader)  # No reader exists before the child starts.
        try:
            closed = subprocess.run(
                [sys.executable, *options, "emit_rows.py", "--rows", str(count)],
                stdout=writer, stderr=subprocess.PIPE, text=True, timeout=10,
            )
        finally:
            os.close(writer)
        assert closed.returncode == 1, (options, count, closed)
        assert closed.stderr == "", (options, count, closed)

failure = subprocess.run(
    [*command, "--fail"], capture_output=True, text=True, timeout=10,
)
assert failure.returncode == 7, failure
assert failure.stdout == "" and "simulated task failure" in failure.stderr, failure
invalid = subprocess.run(
    [*command, "--rows", "0"], capture_output=True, text=True, timeout=10,
)
assert invalid.returncode == 2 and invalid.stderr, invalid
print("OK: full output, closed pipe (buffered/-u), task failure, bad arguments")
PY_CHECK

Expected output:

OK: full output, closed pipe (buffered/-u), task failure, bad arguments

One buffered row exercises the explicit flush() path; large output exercises writes in the loop; -u covers unbuffered startup. --fail still returns 7 and invalid arguments return 2, so handling a broken pipe does not swallow every failure.

5. Troubleshoot directly

For background jobs with path or interpreter failures, use Linux Python scheduled-job path troubleshooting to fix execution conditions before inspecting the output chain.

6. Summary

When a consumer closes early, the producer cannot continue guaranteeing complete output. Explicit flushing and pipe handling remove repeated tracebacks while preserving a clear nonzero status. Test both buffering modes with a pre-closed reader, then decide whether the caller needs a complete export or a limited preview.