Skip to content

macOS pipe writable notification can leave subprocess.communicate() blocked past timeout #162016

Open

Description

@hakula139

Bug report

On macOS 27.0.1 (26A434), ARM64, CPython 3.13.17's subprocess.run() did not return before its two-second timeout. An independent watchdog ended the worker after 5.13 seconds. The expected result is 32,768 output bytes or TimeoutExpired.

subprocess.run(
    ["/bin/cat"],
    input=b"x" * 32768,
    stdout=subprocess.PIPE,
    timeout=2,
)

A traced run ended after 5.08 seconds under the same watchdog. A faulthandler snapshot at one second showed:

Timeout (0:00:01)!
subprocess.py:2166 in _communicate
subprocess.py:1223 in communicate
subprocess.py:557 in run

In this CPython build, line 2166 is the os.write() call to the child's stdin. This snapshot records the stack at one second. The watchdog later expired before the inner timeout returned.

A separate C probe using one pipe and no child process observed poll() report POLLOUT while nonblocking writes returned EAGAIN. It accepted 512 bytes before EAGAIN. The probe's fill loop is bounded at 1 MiB minus 512 bytes, reserving 512 bytes for subsequent control writes.

Full native probe output
pipe_buf=512 pipe_buf_errno=0
pollout=4 pollerr=8 pollhup=16 pollnval=32 eagain=35
accepted_bytes=512 cap_bytes=1048576 fill_cap_bytes=1048064
full_poll=1 revents=4 errno=0
full_write_512=-1 errno=35
full_write_1=-1 errno=35
drain_read=512 errno=0
drained_poll=1 revents=4 errno=0
drained_write_512=512 errno=0

After draining 512 bytes, another 512-byte write succeeded. This confirms the readiness/write discrepancy on this host independently of Python. The evidence does not identify whether the defect belongs to CPython, macOS, or their interaction. No global pipe pressure was induced, and the trigger is unknown.

CPython 3.13.17 and 3.13.18 use the same relevant POSIX communicate() logic. I have not runtime-tested 3.13.18 or CPython main. Could subprocess.communicate() safely handle a writable notification followed by a blocking write, or is more OS-level evidence needed to determine the right owner?

Bounded Python reproducer

The script runs one 32 KiB /bin/cat round trip, with a two-second inner timeout and an independent five-second worker watchdog. It does not induce pipe pressure.

#!/usr/bin/env python3
# ======================================================================================
# Bounded subprocess pipe reproducer with traceback
# Capture one worker traceback before the independent watchdog deadline.
# ======================================================================================

import argparse
import contextlib
import faulthandler
import os
import signal
import subprocess
import sys


def run_worker():
    payload = b'x' * 32768

    faulthandler.dump_traceback_later(1, file=sys.stderr)
    try:
        result = subprocess.run(
            ['/bin/cat'],
            input=payload,
            stdout=subprocess.PIPE,
            timeout=2,
            check=True,
        )
    except subprocess.TimeoutExpired:
        print('Inner subprocess timeout fired.', flush=True)
        return 2
    except Exception as error:
        print(f'Worker failed: {type(error).__name__}.', flush=True)
        return 3
    finally:
        faulthandler.cancel_dump_traceback_later()

    if result.stdout != payload:
        print('Round trip returned different bytes.', flush=True)
        return 4

    print('Round trip completed: 32768 bytes matched.', flush=True)
    return 0


def run_watchdog():
    worker = None
    interrupted = False

    def defer_interrupt(signum, frame):
        nonlocal interrupted
        interrupted = True

    original_handler = signal.getsignal(signal.SIGINT)

    try:
        signal.signal(signal.SIGINT, defer_interrupt)
        try:
            worker = subprocess.Popen(
                [sys.executable, os.path.abspath(__file__), '--worker'],
                stdin=subprocess.DEVNULL,
                start_new_session=True,
            )
        finally:
            signal.signal(signal.SIGINT, original_handler)

        if interrupted:
            raise KeyboardInterrupt

        try:
            return worker.wait(timeout=5)
        except subprocess.TimeoutExpired:
            print('Outer watchdog expired before worker completion.', flush=True)
            return 124
    except KeyboardInterrupt:
        print('Interrupted. Stopping the owned worker group.', flush=True)
        return 130
    except OSError as error:
        print(f'Worker launch failed: {type(error).__name__}.', flush=True)
        return 125
    finally:
        signal.signal(signal.SIGINT, defer_interrupt)
        try:
            if worker is not None and worker.returncode is None:
                with contextlib.suppress(ProcessLookupError):
                    os.killpg(worker.pid, signal.SIGKILL)

                worker.wait()
        finally:
            signal.signal(signal.SIGINT, original_handler)


def main():
    parser = argparse.ArgumentParser(
        description=(
            'Run one traced 32768-byte cat round trip with a two-second subprocess '
            'timeout and an independent five-second worker watchdog.'
        )
    )
    parser.add_argument('--worker', action='store_true', help=argparse.SUPPRESS)
    args = parser.parse_args()

    if args.worker:
        return run_worker()

    if os.name != 'posix':
        print('This reproducer requires POSIX process groups.', flush=True)
        return 125

    return run_watchdog()


if __name__ == '__main__':
    sys.exit(main())
Native pipe probe source
// =================================================================================================
// Bounded Native Pipe Probe
// Measure one pipe's nonblocking writes and readiness without a child process.
// =================================================================================================

#include <errno.h>
#include <fcntl.h>
#include <poll.h>
#include <stdio.h>
#include <string.h>
#include <unistd.h>

int main(void) {
    int descriptors[2];
    if (pipe(descriptors) == -1) {
        printf("pipe_errno=%d\n", errno);
        return 1;
    }

    int result = 1;
    int flags = fcntl(descriptors[1], F_GETFL);
    if (flags == -1 || fcntl(descriptors[1], F_SETFL, flags | O_NONBLOCK) == -1) {
        printf("fcntl_errno=%d\n", errno);
        goto done;
    }

    errno = 0;
    long pipe_buf = fpathconf(descriptors[1], _PC_PIPE_BUF);
    int pipe_buf_errno = errno;
    printf("pipe_buf=%ld pipe_buf_errno=%d\n", pipe_buf, pipe_buf_errno);
    printf("pollout=%d pollerr=%d pollhup=%d pollnval=%d eagain=%d\n",
           POLLOUT, POLLERR, POLLHUP, POLLNVAL, EAGAIN);

    char data[512];
    memset(data, 'x', sizeof(data));
    size_t accepted = 0;
    const size_t cap = 1024 * 1024;
    const size_t fill_cap = cap - sizeof(data);

    while (accepted < fill_cap) {
        size_t count = fill_cap - accepted;
        if (count > sizeof(data)) {
            count = sizeof(data);
        }

        ssize_t written = write(descriptors[1], data, count);
        if (written > 0) {
            accepted += (size_t)written;
        } else if (written == -1 && errno == EAGAIN) {
            break;
        } else {
            printf("fill_write=%zd fill_errno=%d accepted_bytes=%zu\n",
                   written, errno, accepted);
            goto done;
        }
    }

    printf("accepted_bytes=%zu cap_bytes=%zu fill_cap_bytes=%zu\n", accepted, cap, fill_cap);
    if (accepted == fill_cap) {
        printf("fill_cap_reached=1\n");
        result = 0;
        goto done;
    }

    struct pollfd descriptor = {.fd = descriptors[1], .events = POLLOUT};
    errno = 0;
    int ready = poll(&descriptor, 1, 0);
    printf("full_poll=%d revents=%d errno=%d\n", ready, descriptor.revents, errno);
    if (ready == -1) {
        goto done;
    }

    errno = 0;
    ssize_t written = write(descriptors[1], data, sizeof(data));
    int write_errno = errno;
    printf("full_write_512=%zd errno=%d\n", written, write_errno);
    if (written > 0) {
        accepted += (size_t)written;
    } else if (written == -1 && write_errno == EAGAIN) {
        errno = 0;
        ssize_t byte_written = write(descriptors[1], data, 1);
        int byte_errno = errno;
        printf("full_write_1=%zd errno=%d\n", byte_written, byte_errno);
        if (byte_written > 0) {
            accepted += (size_t)byte_written;
        } else if (byte_written != -1 || byte_errno != EAGAIN) {
            goto done;
        }
    } else {
        goto done;
    }

    if (accepted < sizeof(data)) {
        printf("drain_skipped=1\n");
        result = 0;
        goto done;
    }

    errno = 0;
    ssize_t consumed = read(descriptors[0], data, sizeof(data));
    int read_errno = errno;
    printf("drain_read=%zd errno=%d\n", consumed, read_errno);
    if (consumed <= 0) {
        goto done;
    }

    descriptor.revents = 0;
    errno = 0;
    ready = poll(&descriptor, 1, 0);
    printf("drained_poll=%d revents=%d errno=%d\n", ready, descriptor.revents, errno);
    if (ready == -1) {
        goto done;
    }

    errno = 0;
    written = write(descriptors[1], data, sizeof(data));
    write_errno = errno;
    printf("drained_write_512=%zd errno=%d\n", written, write_errno);
    result = written == (ssize_t)sizeof(data) ? 0 : 1;

done:
    close(descriptors[0]);
    close(descriptors[1]);
    return result;
}
Source references

CPython versions tested on:

3.13

Operating systems tested on:

macOS

Activity

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Metadata

Metadata

Assignees

No one assigned

    Labels

    No labels
    No labels

    Projects

    No projects

      Milestone

      No milestone

      Relationships

      None yet

      Development

      No branches or pull requests

      Issue actions