Keep the pipeline descriptor blocking for shell producers

Making the pipe non-blocking regressed shell commands piped to a terminal
consumer: do_shell() hands the same open file description to the child, which
writes to it directly and failed with EAGAIN once the pipe was full. Leave the
descriptor blocking. PipelineWriter still returns to Python regularly, as the
job-control relay requires, by polling for room and writing at most PIPE_BUF
bytes at a time, which cannot block once the pipe reports it is writable.

The job-control test's timeout diagnostics now cope with a sandbox that cannot
run ps.

Author: tleonhardt

cmd2/utils.py

@@ -836,7 +836,6 @@ class PipelineWriter(io.FileIO):
     def __init__(self, fd: int, reader: ProcReader) -> None:
         """Take ownership of a pipe descriptor managed by reader."""
         super().__init__(fd, "w")
-        os.set_blocking(fd, False)
         self._reader = reader
 
     def write(self, b: Any) -> int:
@@ -846,25 +845,27 @@ def write(self, b: Any) -> int:
         writes, even for an instant, would stop a consumer that had just resumed a
         terminal read with SIGTTIN.
 
-        A full pipe is awaited in short polls rather than a blocking write. Only the
-        main thread runs Python signal handlers, and the job-control stop ProcReader
-        relays may wake another thread, so the main thread has to return to Python
-        code on its own for the handler to run.
+        A full pipe is awaited in short polls rather than in one blocking write. Only
+        the main thread runs Python signal handlers, and the job-control stop ProcReader
+        relays may wake another thread, so the main thread has to return to Python code
+        on its own for the handler to run. The descriptor itself stays blocking: a shell
+        command inherits it, and a producer that found it non-blocking would fail with
+        EAGAIN once the pipe filled.
         """
         import select
         import signal
 
         view = memoryview(b).cast("B")
+        fd = self.fileno()
         poller = select.poll()
-        poller.register(self.fileno(), select.POLLOUT)
+        poller.register(fd, select.POLLOUT)
         try:
             with self._reader.lend_terminal():
                 written = 0
                 while written < len(view):
-                    try:
-                        written += os.write(self.fileno(), view[written:])
-                    except BlockingIOError:
-                        poller.poll(100)
+                    # Once there is room, a write of at most PIPE_BUF bytes does not block.
+                    if poller.poll(100):
+                        written += os.write(fd, view[written : written + select.PIPE_BUF])
                 return written
         except BrokenPipeError:
             # Ctrl-C during a blocking write must cancel the command, even if it

tests/test_pipeline_job_control.py

@@ -29,9 +29,12 @@ def describe_processes(root: int, master: int) -> str:
         foreground: object = os.tcgetpgrp(master)
     except OSError as error:
         foreground = error
-    listing = subprocess.run(
-        ["ps", "-e", "-o", "pid,ppid,pgid,stat,wchan,command"], capture_output=True, text=True, check=False
-    )
+    try:
+        listing = subprocess.run(
+            ["ps", "-e", "-o", "pid,ppid,pgid,stat,wchan,command"], capture_output=True, text=True, check=False
+        )
+    except OSError as error:
+        return f"foreground process group: {foreground}\nno process listing: {error}"
     rows = listing.stdout.splitlines()
     parents = {}
     for row in rows[1:]:

tests/test_utils.py

@@ -1,5 +1,6 @@
 """Unit testing for cmd2/utils.py module."""
 
+import contextlib
 import errno
 import math
 import os
@@ -280,6 +281,48 @@ def test_proc_reader_terminal_group() -> None:
     assert reader.terminal_group is None
 
 
+@pytest.mark.skipif(sys.platform == "win32", reason="POSIX pipes")
+@pytest.mark.parametrize("producer", ["writer", "child"])
+def test_pipeline_writer_delivers_more_than_the_pipe_holds(producer) -> None:
+    """Both cmd2's writes and a child inheriting the descriptor must wait for a slow consumer.
+
+    A shell producer gets the descriptor itself, so it must stay blocking: a child that
+    inherits O_NONBLOCK fails with EAGAIN once the pipe is full.
+    """
+    import subprocess
+    import threading
+
+    payload = b"x" * 4 * 1024 * 1024
+    read_fd, write_fd = os.pipe()
+    received = bytearray()
+
+    def drain() -> None:
+        while chunk := os.read(read_fd, 65536):
+            received.extend(chunk)
+            time.sleep(0.001)
+
+    reader = mock.Mock(lend_terminal=contextlib.nullcontext)
+    writer = cu.PipelineWriter(write_fd, reader)
+    consumer = threading.Thread(target=drain)
+    consumer.start()
+    try:
+        if producer == "writer":
+            assert writer.write(payload) == len(payload)
+        else:
+            child = subprocess.run(
+                [sys.executable, "-c", f"import sys; sys.stdout.buffer.write(b'x' * {len(payload)})"],
+                stdout=writer.fileno(),
+                stderr=subprocess.PIPE,
+                check=False,
+            )
+            assert child.returncode == 0, child.stderr.decode()
+    finally:
+        writer.close()
+        consumer.join()
+        os.close(read_fd)
+    assert bytes(received) == payload
+
+
 def test_proc_reader_terminate(pr_none) -> None:
     assert pr_none._proc.poll() is None
     pr_none.terminate()