Fix pager hangs and a startup race in terminal pipelines

Three bugs in the terminal-pipeline job control, found in review and
reproduced on a pseudo-terminal with real less:

1. Output written straight to the pipe's descriptor hung the pager.

   cmd2 lent the pager the terminal only during its own writes through
   PipelineWriter. A subprocess given self.stdout (for example
   subprocess.run(..., stdout=self.stdout) in a custom command, or a `!`
   command run by `run_script s.txt | less`) writes to the descriptor
   itself, so no lend happened. Once the pipe filled, less stopped with
   SIGTTIN reading the keyboard, the watcher waited for a lend that never
   came, and the producer blocked on the full pipe forever. On main these
   cases worked, because the pager ran in its own session.

   PipelineWriter.fileno() now hands out the write end of a relay pipe.
   A thread passes that output on to the consumer and lends it the
   terminal while a producer is blocked on the relay's full pipe, which
   is the only time the consumer needs the terminal to make progress.
   Once no producer is waiting, command code gets the terminal back, as
   between cmd2's own writes, so a later input() still works. cmd2's own
   writes first wait for pending relay output, which keeps output in
   order. When the consumer exits, the relay closes its pipe, so
   producers still get EPIPE or SIGPIPE.

2. A pipe nested in a piped command took the terminal from the outer
   pager.

   A pipeline counted as the terminal's job when either its stdout or its
   stderr was the foreground terminal. In `run_script s.txt | less` with
   `big | cat` in the script, the inner pipeline's stdout is the outer
   pipe, but its stderr is the terminal. It became a terminal job and its
   lends went to cat's group, so less was never lent the terminal, and
   the command hung as in (1).

   Only a pipeline whose stdout is the foreground terminal is now the
   terminal's job. Others, including nested pipelines and pipelines whose
   stdout cmd2 captures, run in their own session, as on main.

3. The pager could start before it owned the terminal.

   The consumer could run between Popen() and cmd2's first terminal
   lend. A pager such as less sets its terminal modes as it starts, and a
   background tcsetattr() stops it with SIGTTOU. On macOS the call then
   fails with EINTR once the process continues, and less carries on with
   a cooked terminal, so q needs Enter and keys are echoed. This was
   reproduced by delaying cmd2 after it starts the pipeline.

   A terminal pipeline now starts as
   `/bin/sh -c 'read -r _ || exit 1; exec "$SHELL" -c <command>'`. cmd2
   sends a single newline down the consumer's stdin pipe only after
   making the pipeline's group the foreground one. From a pipe, the read
   builtin takes no more than that line, so no extra descriptor is needed.
   That matters because dash, /bin/sh on Debian and Ubuntu, cannot
   redirect descriptors above 9. Checked with sh, bash, dash and zsh.
   If cmd2 fails before sending the newline, it closes the pipe, and the
   held pipeline reads EOF and exits rather than waiting forever.

Tests:
- A PTY test covering the three hang cases in (1) and (2): a subprocess
  writer, a script's shell command, and a nested pipe. It fails without
  this change.
- The pager-startup PTY test gains a variant that delays cmd2 after it
  starts the pipeline, which fails without this change.
- Unit tests for the relay: lending only while a producer waits, giving
  the terminal back whether or not output is pending, ordering with
  cmd2's own writes, and passing the consumer's exit on to producers.

Author: tleonhardt

CHANGELOG.md

@@ -6,8 +6,8 @@
       does. It had run in a separate session that never received the terminal, so Ctrl-Z and `fg`
       did not suspend and resume it together with cmd2. The program now owns the terminal as it
       starts, so a pager can set its terminal modes, and Ctrl-C and Ctrl-Z reach the whole pipeline.
-      Pipes started from a worker thread, or whose output cmd2 captures, still run in their own
-      session
+      Pipes started from a worker thread, or whose output does not go to the terminal, such as one
+      nested in a command whose own output is piped, still run in their own session
     - A `shell` command piped to an interactive program, such as `shell git log | less`, now joins
       the pipeline's job, so both processes receive Ctrl-C and Ctrl-Z
 

cmd2/cmd2.py

@@ -3346,22 +3346,40 @@ def _redirect_output(self, statement: Statement) -> utils.RedirectionSavedState:
             pipe_stderr = None if isinstance(sys.stderr, utils.StdSim) else sys.stderr
 
             terminal_fd = None
+            popen_command = statement.redirect_to
             if sys.platform != "win32":
-                # Job control installs signal handlers, which only the main thread may do.
-                # Elsewhere, keep the pipeline in its own session as before.
-                if threading.current_thread() is threading.main_thread():
-                    for stream in (pipe_stdout, pipe_stderr):
-                        if stream is not None and stream.isatty():
-                            with contextlib.suppress(OSError, ValueError):
-                                if os.tcgetpgrp(stream.fileno()) == os.getpgrp():
-                                    terminal_fd = stream.fileno()
-                                    break
+                # Only a pipeline whose output goes to the terminal is the terminal's job. One
+                # nested in a command whose output is piped, for instance, feeds that outer
+                # pipeline, whose consumer needs the terminal instead. Job control installs
+                # signal handlers, which only the main thread may do. Otherwise, keep the
+                # pipeline in its own session as before.
+                if threading.current_thread() is threading.main_thread() and pipe_stdout is not None and pipe_stdout.isatty():
+                    with contextlib.suppress(OSError, ValueError):
+                        if os.tcgetpgrp(pipe_stdout.fileno()) == os.getpgrp():
+                            terminal_fd = pipe_stdout.fileno()
                 if terminal_fd is None:
                     kwargs["start_new_session"] = True
                 else:
                     kwargs["process_group"] = 0
 
-            with contextlib.ExitStack() as terminal_stack:
+                    # A pager such as less sets its terminal modes as it starts, before it reads
+                    # the pipe. It must own the terminal by then: a background tcsetattr() stops
+                    # it with SIGTTOU, and on macOS that call fails with EINTR when the process
+                    # is continued instead of being restarted. less ignores the failure and runs
+                    # on a cooked terminal. So hold the pipeline in a POSIX sh until cmd2 has
+                    # made its group the foreground one and sent a newline down the pipe. The
+                    # read builtin takes no more than that line from a pipe. Then exec the
+                    # user's shell as before.
+                    import shlex
+
+                    user_shell = shlex.quote(kwargs.get("executable", "/bin/sh"))
+                    popen_command = f"read -r _ || exit 1; exec {user_shell} -c {shlex.quote(statement.redirect_to)}"
+                    kwargs["executable"] = "/bin/sh"
+
+            with contextlib.ExitStack() as terminal_stack, contextlib.ExitStack() as gate_stack:
+                if terminal_fd is not None:
+                    # Should cmd2 fail before opening the gate, the held pipeline reads EOF and exits.
+                    gate_stack.callback(new_stdout.close)
                 with contextlib.ExitStack() as spawn_stack:
                     if terminal_fd is not None and os.getpgrp() == os.getsid(0):
                         import signal
@@ -3372,7 +3390,7 @@ def _redirect_output(self, statement: Statement) -> utils.RedirectionSavedState:
                         previous_tstp = signal.signal(signal.SIGTSTP, signal.SIG_IGN)
                         spawn_stack.callback(signal.signal, signal.SIGTSTP, previous_tstp)
                     proc = subprocess.Popen(  # noqa: S602
-                        statement.redirect_to,
+                        popen_command,
                         stdin=subproc_stdin,
                         stdout=subprocess.PIPE if pipe_stdout is None else pipe_stdout,
                         stderr=subprocess.PIPE if pipe_stderr is None else pipe_stderr,
@@ -3394,12 +3412,11 @@ def _redirect_output(self, statement: Statement) -> utils.RedirectionSavedState:
                     if cmd_pipe_proc_reader is None:
                         proc.wait(0.2)
                     else:
-                        # A pager such as less sets its terminal modes as it starts, before it
-                        # reads the pipe. It must own the terminal by then: a background
-                        # tcsetattr() stops it with SIGTTOU, and on macOS that call fails with
-                        # EINTR when the process is continued instead of being restarted. less
-                        # ignores the failure and runs on a cooked terminal.
+                        # Open the start gate only once the pipeline owns the terminal.
                         with cmd_pipe_proc_reader.lend_terminal():
+                            with contextlib.suppress(OSError):
+                                os.write(new_stdout.fileno(), b"\n")
+                            gate_stack.pop_all()
                             cmd_pipe_proc_reader.wait_for_exit(0.2)
 
                 # Check if the pipe process already exited

cmd2/utils.py

@@ -921,13 +921,158 @@ def _write_bytes(stream: StdSim | TextIO, to_write: bytes | str) -> None:
             stream.buffer.write(to_write)
 
 
+class _DescriptorRelay:
+    """Carry output that subprocesses write to a terminal pipeline's descriptor.
+
+    A subprocess given :meth:`PipelineWriter.fileno` writes on its own, so cmd2 cannot lend
+    the terminal write by write. It writes into this relay's pipe instead, and a thread passes
+    that output on to the consumer. The consumer is lent the terminal while a producer is
+    blocked on the relay's full pipe, since only then does it need the terminal to make
+    progress. Otherwise command code keeps it, as it does between cmd2's own writes.
+    """
+
+    def __init__(self, out_fd: int, reader: ProcReader) -> None:
+        """Start relaying to out_fd, a descriptor the relay takes ownership of.
+
+        :param out_fd: the consumer's pipe
+        :param reader: the terminal pipeline that reads out_fd
+        """
+        self._out_fd = out_fd
+        self._reader = reader
+        # write_fd is the descriptor handed to producers. PipelineWriter closes it; once every
+        # producer has closed its copy too, the relay reads EOF and closes the consumer's pipe.
+        self._in_fd, self.write_fd = os.pipe()
+        self._write_fd_open = True
+        self._lock = threading.Condition()
+        # Bytes read from the relay's pipe, and bytes passed on to the consumer's pipe
+        self._received = 0
+        self._sent = 0
+        self._done = False
+        threading.Thread(name="pipe_relay", target=self._relay, daemon=True).start()
+
+    def close_write_fd(self) -> None:
+        """Close cmd2's copy of the producers' descriptor. The relay finishes once theirs close."""
+        with self._lock:
+            if self._write_fd_open:
+                os.close(self.write_fd)
+                self._write_fd_open = False
+
+    def _unread(self) -> int:
+        """Bytes producers have written that the relay has not read yet. Requires _lock."""
+        import fcntl
+        import struct
+        import termios
+
+        if self._done:
+            return 0
+        return int(struct.unpack("i", fcntl.ioctl(self._in_fd, termios.FIONREAD, b"\0" * 4))[0])
+
+    def _producer_blocked(self) -> bool:
+        """Whether the relay's pipe is full, which means a producer is waiting on the consumer."""
+        import select
+
+        with self._lock:
+            if not self._write_fd_open:
+                # The command is done. ProcReader.wait() lends the terminal from here on.
+                return False
+            poller = select.poll()
+            poller.register(self.write_fd, select.POLLOUT)
+            return not poller.poll(0)
+
+    def flush(self) -> None:
+        """Wait until output already written to the relay has reached the consumer's pipe.
+
+        cmd2's own writes go straight to the consumer's pipe, so they wait for this first to
+        keep their order with the output of a producer that has finished. The wait is in short
+        polls so that the main thread still runs Python signal handlers.
+        """
+        with self._lock:
+            target = self._received + self._unread()
+            while not self._done and self._sent < target:
+                self._lock.wait(0.1)
+
+    def _relay(self) -> None:
+        """Pass producer output on to the consumer until producers close or the consumer exits."""
+        import select
+
+        poller = select.poll()
+        poller.register(self._out_fd, select.POLLOUT)
+        lend = contextlib.ExitStack()
+        lending = False
+        try:
+            while True:
+                with self._lock:
+                    idle = not self._unread()
+                if idle and lending:
+                    # No producer is waiting. Let command code have the terminal back.
+                    lend.close()
+                    lending = False
+                data = os.read(self._in_fd, 65536)
+                if not data:
+                    return
+                with self._lock:
+                    self._received += len(data)
+                view = memoryview(data)
+                written = 0
+                while written < len(view):
+                    # Once there is room, a write of at most PIPE_BUF bytes does not block.
+                    if poller.poll(100):
+                        count = os.write(self._out_fd, view[written : written + select.PIPE_BUF])
+                        written += count
+                        with self._lock:
+                            self._sent += count
+                            self._lock.notify_all()
+                    elif self._producer_blocked():
+                        if not lending:
+                            lend.enter_context(self._reader.lend_terminal())
+                            lending = True
+                    elif lending:
+                        # A producer that stopped writing does not need the consumer to go on.
+                        lend.close()
+                        lending = False
+        except OSError:
+            # The consumer exited. Closing the relay's pipe below passes that on to producers,
+            # which get EPIPE or SIGPIPE just as they would writing to the consumer directly.
+            return
+        finally:
+            with contextlib.suppress(OSError):
+                lend.close()
+            with self._lock:
+                self._done = True
+                os.close(self._in_fd)
+                os.close(self._out_fd)
+                self._lock.notify_all()
+
+
 class PipelineWriter(io.FileIO):
     """A pipe whose blocking writes temporarily give the consumer terminal access."""
 
     def __init__(self, fd: int, reader: ProcReader) -> None:
         """Take ownership of a pipe descriptor managed by reader."""
         super().__init__(fd, "w")
         self._reader = reader
+        self._relay: _DescriptorRelay | None = None
+
+    def fileno(self) -> int:
+        """Return a descriptor for subprocesses, such as a shell command's stdout.
+
+        A subprocess writes to it directly, bypassing :meth:`write`. It is the write end of a
+        relay (see :class:`_DescriptorRelay`), which lends the consumer the terminal whenever
+        such a producer is waiting for the consumer to drain the pipe.
+        """
+        if self.closed:
+            raise ValueError("I/O operation on closed file")
+        if self._relay is None:
+            self._relay = _DescriptorRelay(os.dup(super().fileno()), self._reader)
+        return self._relay.write_fd
+
+    def close(self) -> None:
+        """Close the pipe. The consumer sees EOF once every producer has closed its descriptor too."""
+        try:
+            super().close()
+        finally:
+            if self._relay is not None:
+                self._relay.close_write_fd()
 
     def write(self, b: Any) -> int:
         """Write all of b while the consumer can interact with the terminal.
@@ -947,11 +1092,13 @@ def write(self, b: Any) -> int:
         import signal
 
         view = memoryview(b).cast("B")
-        fd = self.fileno()
+        fd = super().fileno()
         poller = select.poll()
         poller.register(fd, select.POLLOUT)
         try:
             with self._reader.lend_terminal():
+                if self._relay is not None:
+                    self._relay.flush()
                 written = 0
                 while written < len(view):
                     # Once there is room, a write of at most PIPE_BUF bytes does not block.

tests/test_pipeline_job_control.py

@@ -461,12 +461,16 @@ def wait_until(predicate):
         process.wait(timeout=5)
 
 
-def test_pipeline_pager_can_set_terminal_modes_at_startup(tmp_path) -> None:
+@pytest.mark.parametrize("parent_delay", [False, True])
+def test_pipeline_pager_can_set_terminal_modes_at_startup(tmp_path, parent_delay) -> None:
     """A pager such as less puts the terminal in raw mode as it starts, before reading its pipe.
 
     It has to own the terminal by then. A background tcsetattr() stops it with SIGTTOU, and
     on macOS the call then fails with EINTR once it is continued rather than being restarted.
     less ignores that failure, leaving a cooked terminal: q needs Enter and keys are echoed.
+
+    However late cmd2 gets to hand the terminal over after starting the pipeline, the pager
+    must not start before it has.
     """
     import pty
     import termios
@@ -511,6 +515,13 @@ def test_pipeline_pager_can_set_terminal_modes_at_startup(tmp_path) -> None:
         "        time.sleep(0.01)\n"
         "    return startup_wait(reader, timeout)\n"
         "utils.ProcReader.wait_for_exit = held_startup_wait\n"
+        f"if {parent_delay!r}:\n"
+        # Stall between starting the pipeline and handing it the terminal.
+        "    start_job_control = utils.ProcReader.manage_terminal\n"
+        "    def late_job_control(reader):\n"
+        "        time.sleep(0.5)\n"
+        "        return start_job_control(reader)\n"
+        "    utils.ProcReader.manage_terminal = late_job_control\n"
         "app = Cmd()\n"
         "app.prompt = 'TEST> '\n"
         "app.cmdloop()\n",
@@ -731,3 +742,98 @@ def wait_until(predicate):
         os.close(master)
         process.kill()
         process.wait(timeout=5)
+
+
+@pytest.mark.parametrize("producer", ["subprocess", "script_shell", "nested_pipe"])
+def test_pager_gets_the_terminal_for_output_written_to_the_descriptor(tmp_path, producer) -> None:
+    """A producer that writes to the pipe's descriptor itself, bypassing cmd2's writes, cannot trigger a lend per write.
+
+    Such as a subprocess given self.stdout, a shell command run by a script, or a pipeline nested
+    in a piped command. Once the pipe is full, the pager has to be lent the terminal to read the
+    keys that let it go on. Otherwise it stops with SIGTTIN while the producer waits on it forever.
+    """
+    import pty
+
+    shell = shutil.which("bash")
+    if shell is None:
+        pytest.skip("requires an interactive bash shell")
+    big = tmp_path / "big.py"
+    big.write_text("import sys\nsys.stdout.write('x' * 1048576)\n", encoding="utf-8")
+    pager = tmp_path / "pager.py"
+    pager.write_text(
+        "import os, signal, sys\n"
+        # Interactive bash leaves TTIN ignored in what it execs, which turns a background read into EIO.
+        "signal.signal(signal.SIGTTIN, signal.SIG_DFL)\n"
+        "sys.stdin.buffer.read(4096)\n"
+        "os.write(2, b'PAGER_READY\\n')\n"
+        # Like less, read the keyboard from an inherited terminal descriptor. Quit without
+        # draining the pipe, which the producer must then learn of.
+        "os.write(2, b'PAGER_GOT ' + os.read(2, 1) + b'\\n')\n",
+        encoding="utf-8",
+    )
+    python = shlex.quote(sys.executable)
+    script = tmp_path / "script.txt"
+    script.write_text(
+        f"!{python} {shlex.quote(str(big))}\n" if producer == "script_shell" else "big | cat\n", encoding="utf-8"
+    )
+    application = tmp_path / "application.py"
+    application.write_text(
+        "import subprocess, sys\n"
+        "from cmd2 import Cmd\n"
+        "class App(Cmd):\n"
+        "    def do_sub(self, _):\n"
+        f"        subprocess.run([sys.executable, {str(big)!r}], stdout=self.stdout, check=False)\n"
+        "    def do_big(self, _):\n"
+        "        self.poutput('y' * 1048576)\n"
+        "app = App()\n"
+        "app.prompt = 'TEST> '\n"
+        "app.cmdloop()\n",
+        encoding="utf-8",
+    )
+    master, slave = pty.openpty()
+    bootstrap = (
+        "import os, fcntl, termios; os.setsid(); "
+        "fcntl.ioctl(0, termios.TIOCSCTTY, 0); "
+        "os.execv(os.environ['TEST_SHELL'], ['bash', '--noprofile', '--norc', '-i'])"
+    )
+    env = dict(os.environ, TERM="xterm-256color", PS1="OUTER> ", TEST_SHELL=shell, SHELL=shell)
+    env["PYTHONPATH"] = str(Path(__file__).resolve().parents[1])
+    process = subprocess.Popen([sys.executable, "-c", bootstrap], stdin=slave, stdout=slave, stderr=slave, env=env)
+    os.close(slave)
+    decoder = codecs.getincrementaldecoder("utf-8")("replace")
+    transcript = ""
+
+    def wait_until(predicate):
+        nonlocal transcript
+        deadline = time.monotonic() + 10
+        while time.monotonic() < deadline:
+            if select.select([master], [], [], 0.05)[0]:
+                data = decoder.decode(os.read(master, 65536))
+                transcript += data
+                if "\x1b[6n" in data:
+                    # Answer prompt-toolkit's cursor-position request as a terminal would.
+                    os.write(master, b"\x1b[1;1R")
+            if predicate():
+                return
+        pytest.fail(f"terminal condition timed out:\n{transcript}\n{describe_processes(process.pid, master)}")
+
+    command = "sub" if producer == "subprocess" else f"run_script {shlex.quote(str(script))}"
+    try:
+        wait_until(lambda: "OUTER> " in transcript)
+        os.write(master, f"{python} {shlex.quote(str(application))}\n".encode())
+        wait_until(lambda: "TEST>" in transcript)
+        os.write(master, f"{command} | {python} {shlex.quote(str(pager))}\n".encode())
+        wait_until(lambda: "PAGER_READY\r\n" in transcript)
+        # The pipe fills once the pager has read its first chunk.
+        time.sleep(0.5)
+        start = len(transcript)
+        # The terminal is still canonical: this pager, unlike less, sets no modes.
+        os.write(master, b"q\n")
+        wait_until(lambda: "PAGER_GOT q" in transcript[start:])
+        wait_until(lambda: "TEST>" in transcript[start:])
+        os.write(master, b"quit\n")
+        wait_until(lambda: os.tcgetpgrp(master) == process.pid)
+    finally:
+        os.close(master)
+        process.kill()
+        process.wait(timeout=5)