Coordinate Ctrl-Z across a pipeline's job and fix review findings

Fixes from further review of the terminal pipeline job control. Each was
reproduced before being fixed, except where noted.

- A `shell` producer could hang on Ctrl-Z when cmd2 is a session leader
  (started with exec, so there is no outer shell). The pipeline's
  consumer inherited an ignored SIGTSTP, but the producer that joins its
  job did not. Ctrl-Z during `shell sleep 3 | cat` stopped the producer
  for good while cat ran on, and cmd2 waited on the producer forever.
  Both spawns now share one helper, _session_leader_job_stops(), that
  applies the session leader's Ctrl-Z behavior.

- A producer's stop was dropped when the consumer ignores Ctrl-Z. The
  producer's wait relayed its stop only once the consumer's watcher had
  exited, assuming the consumer would stop too. With a consumer that
  ignores SIGTSTP, the producer stayed stopped and cmd2 hung.

- A direct Ctrl-Z to cmd2 counted the SIGSTOP it sent the pipeline, so
  that the watcher could skip the resulting stop report. A consumer that
  was already stopped, for instance waiting for the terminal, produces no
  report, and the count never came back down. A later genuine stop of the
  consumer would then have been ignored.

  The last two are fixed together. Suspensions of the job now take one
  lock, and count as they finish. Whichever stop is reported first, the
  consumer's, a producer's, or cmd2's own, suspends the whole job. A
  suspension ends by continuing the whole pipeline, so a stop reported
  before one finished needs nothing more and is skipped. This replaces
  the count of cmd2's own stops.

- send_sigint() and terminate() used the pipeline's process ID even after
  its watcher had reaped it, when the system may already have given that
  ID to an unrelated process. Both now check the return code first.
  (Confirmed from the code; pid reuse cannot be forced in a test.)

- The descriptor relay could report output as passed on while it was
  still on its way: bytes read from the relay's pipe were neither in the
  pipe nor counted until the relay thread took its lock again. cmd2's
  next write could then overtake the end of a subprocess's output. The
  relay now waits for output outside the lock, then reads and counts it
  under the lock.

The next release is 4.3.0 rather than 4.2.5: the changelog heading is
updated, as these changes have grown beyond a patch release.

Tests:
- PTY test: Ctrl-Z during `shell <producer> | <consumer>`, run from a
  shell or as a session leader, with a consumer that stops or one that
  ignores Ctrl-Z. Three of the four cases hung before this change.
- Unit tests: a producer's stop suspends the job while the watcher runs,
  unless a finished suspension already resolved it; the watcher skips such
  stops and follows newer ones while it waits for the terminal; a
  suspension waits for one in progress; a direct Ctrl-Z while another
  thread relays a stop leaves the pipeline to that thread; a reaped
  pipeline is never signaled; the relay counts output it is reading.

Author: tleonhardt

CHANGELOG.md

@@ -1,4 +1,4 @@
-## 4.2.5 (TBD)
+## 4.3.0 (TBD)
 
 - Bug Fixes
     - On POSIX, piping a command's output to an interactive program such as `less`

cmd2/cmd2.py

@@ -3380,15 +3380,7 @@ def _redirect_output(self, statement: Statement) -> utils.RedirectionSavedState:
                 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
-
-                        # A session leader's job has no outer shell to resume it.
-                        # Its pipeline must inherit the same Ctrl-Z behavior: the
-                        # new group would otherwise make SIGTSTP actionable again.
-                        previous_tstp = signal.signal(signal.SIGTSTP, signal.SIG_IGN)
-                        spawn_stack.callback(signal.signal, signal.SIGTSTP, previous_tstp)
+                with utils._session_leader_job_stops() if terminal_fd is not None else contextlib.nullcontext():
                     proc = subprocess.Popen(  # noqa: S602
                         popen_command,
                         stdin=subproc_stdin,
@@ -5025,8 +5017,13 @@ def do_shell(self, args: argparse.Namespace) -> None:
             while True:
                 try:
                     # For any stream that is a StdSim, we will use a pipe so we can capture its output.
-                    # A command joining the pipeline is spawned inside the lend, which blocks SIGTTOU.
-                    with utils._unblocked_sigttou() if "process_group" in kwargs else contextlib.nullcontext():
+                    # A command joining the pipeline is spawned inside the lend, which blocks SIGTTOU,
+                    # and with the job's Ctrl-Z behavior.
+                    joining = "process_group" in kwargs
+                    with (
+                        utils._unblocked_sigttou() if joining else contextlib.nullcontext(),
+                        utils._session_leader_job_stops() if joining else contextlib.nullcontext(),
+                    ):
                         proc = subprocess.Popen(  # noqa: S602
                             expanded_command,
                             stdout=subprocess.PIPE if isinstance(self.stdout, utils.StdSim) else self.stdout,  # type: ignore[unreachable]

cmd2/utils.py

@@ -552,6 +552,26 @@ def _unblocked_sigttou() -> Iterator[None]:
         signal.pthread_sigmask(signal.SIG_SETMASK, previous_mask)
 
 
+@contextlib.contextmanager
+def _session_leader_job_stops() -> Iterator[None]:
+    """Spawn a process for a terminal pipeline's job with the same Ctrl-Z behavior as cmd2.
+
+    A session leader's job has no outer shell to resume it, so Ctrl-Z must not stop any part
+    of it. A new process group would otherwise make SIGTSTP actionable again. An ignored
+    signal stays ignored across exec, so ignore SIGTSTP while spawning.
+    """
+    import signal
+
+    if os.getpgrp() != os.getsid(0):
+        yield
+        return
+    previous = signal.signal(signal.SIGTSTP, signal.SIG_IGN)
+    try:
+        yield
+    finally:
+        signal.signal(signal.SIGTSTP, previous)
+
+
 class ProcReader:
     """Used to capture stdout and stderr from a Popen process if any of those were set to subprocess.PIPE.
 
@@ -591,10 +611,13 @@ def __init__(
         self._terminal_lock = threading.RLock()
         self._job_resumed = threading.Event()
         # Set by a thread that relays a pipeline's stop to cmd2's own job. Otherwise Ctrl-Z
-        # reached cmd2 directly, and cmd2 stops the pipeline itself. The watcher skips the
-        # SIGSTOPs cmd2 sends that way, counted under _terminal_lock.
+        # reached cmd2 directly, and cmd2 stops the pipeline itself.
         self._relaying_stop = False
-        self._own_stops = 0
+        # One suspension of the whole job at a time, and how many have finished. A suspension
+        # continues the whole pipeline as it ends, so a stop reported before it finished is
+        # already dealt with: see _suspend_with_cmd2().
+        self._suspension_lock = threading.Lock()
+        self._suspensions = 0
         if terminal_fd is not None:
             self._original_group = os.tcgetpgrp(terminal_fd)
 
@@ -618,21 +641,22 @@ def send_sigint(self) -> None:
             self._proc.send_signal(signal.CTRL_BREAK_EVENT)
         else:
             # Since cmd2 uses shell=True in its Popen calls, we need to send the SIGINT to
-            # the whole process group to make sure it propagates further than the shell
-            try:
-                group_id = os.getpgid(self._proc.pid)
-            except ProcessLookupError:
-                # Pipelines lead their own group. A shell command that joined it, such as
-                # `shell sleep 100 | head -1`, can outlive the reaped consumer. Find the group
-                # through that command, which cmd2 has not reaped yet. Never signal the
-                # consumer's own ID: once its group is gone, the system may reuse it.
-                for producer in self._joined:
-                    if producer.returncode is None:
-                        with contextlib.suppress(ProcessLookupError):
-                            group_id = os.getpgid(producer.pid)
-                            break
-                else:
-                    return
+            # the whole process group to make sure it propagates further than the shell.
+            # Once reaped, the process's ID may already belong to another process.
+            group_id = None
+            if self._proc.returncode is None:
+                with contextlib.suppress(ProcessLookupError):
+                    group_id = os.getpgid(self._proc.pid)
+            # Pipelines lead their own group. A shell command that joined it, such as
+            # `shell sleep 100 | head -1`, can outlive the reaped consumer. Find the group
+            # through that command, which cmd2 has not reaped yet. Never signal the
+            # consumer's own ID: once its group is gone, the system may reuse it.
+            for producer in self._joined:
+                if group_id is None and producer.returncode is None:
+                    with contextlib.suppress(ProcessLookupError):
+                        group_id = os.getpgid(producer.pid)
+            if group_id is None:
+                return
             # Never re-signal our own group: other ProcReader callers may share it
             # and already received Ctrl-C.
             if group_id != os.getpgrp():
@@ -647,8 +671,10 @@ def terminate(self) -> None:
             import signal
 
             # Popen.terminate() polls first, which would compete with our waitpid thread.
-            with contextlib.suppress(ProcessLookupError):
-                os.kill(self._proc.pid, signal.SIGTERM)
+            # Once the watcher has reaped the process, its ID may belong to another one.
+            if self._proc.returncode is None:
+                with contextlib.suppress(ProcessLookupError):
+                    os.kill(self._proc.pid, signal.SIGTERM)
 
     @property
     def _terminal_group(self) -> int | None:
@@ -694,23 +720,20 @@ def _manage_terminal(self) -> Iterator[None]:
         def suspend_job(signum: int, frame: Any) -> None:
             relayed = self._relaying_stop
             self._relaying_stop = False
-            stopped_pipeline = False
+            own_suspension = False
             try:
                 if previous_handler != signal.SIG_DFL:
                     if callable(previous_handler):
                         previous_handler(signum, frame)
                     return
                 if os.tcgetpgrp(terminal_fd) == self._proc.pid:
                     self._set_foreground_group(terminal_fd, self._original_group)
-                if not relayed and self._proc.returncode is None:
-                    # Ctrl-Z reached only cmd2's group, which owns the terminal between pipe
-                    # writes. Stop the pipeline too, as a shell stops its whole job.
-                    with self._terminal_lock:
-                        self._own_stops += 1
-                    stopped_pipeline = self._signal_pipeline(signal.SIGSTOP)
-                    if not stopped_pipeline:
-                        with self._terminal_lock:
-                            self._own_stops -= 1
+                # Ctrl-Z reached only cmd2's group, which owns the terminal between pipe writes.
+                # Stop the pipeline too, as a shell stops its whole job. Should another thread
+                # be relaying a stop already, it has stopped the pipeline.
+                if not relayed and self._suspension_lock.acquire(blocking=False):
+                    own_suspension = True
+                    self._signal_pipeline(signal.SIGSTOP)
                 # Ignore our group-directed copy, then stop this thread synchronously.
                 # Wrappers in our job must stop too. Unlike SIGSTOP, SIGTSTP is
                 # discarded for orphaned groups, which have no shell to resume them.
@@ -726,8 +749,10 @@ def suspend_job(signum: int, frame: Any) -> None:
                     if self._terminal_available.is_set() and os.tcgetpgrp(terminal_fd) == self._original_group:
                         self._set_foreground_group(terminal_fd, self._proc.pid)
                 finally:
-                    if stopped_pipeline:
+                    if own_suspension:
                         self._signal_pipeline(signal.SIGCONT)
+                        self._suspensions += 1
+                        self._suspension_lock.release()
                     self._job_resumed.set()
 
         signal.signal(signal.SIGTSTP, suspend_job)
@@ -820,16 +845,11 @@ def _watch_job(self, terminal_fd: int) -> None:
         import signal
 
         while True:
+            seen = self._suspensions
             _, status = os.waitpid(self._proc.pid, os.WUNTRACED)
             if not os.WIFSTOPPED(status):
                 self._proc.returncode = os.waitstatus_to_exitcode(status)
                 return
-            if os.WSTOPSIG(status) == signal.SIGSTOP:
-                with self._terminal_lock:
-                    if self._own_stops:
-                        # cmd2 stopped the pipeline along with itself, and continues it too.
-                        self._own_stops -= 1
-                        continue
             if os.WSTOPSIG(status) in (signal.SIGTTIN, signal.SIGTTOU):
                 # Command code owns the terminal between pipe writes. Defer
                 # consumer terminal access until the next write or final wait.
@@ -838,10 +858,14 @@ def _watch_job(self, terminal_fd: int) -> None:
                     with self._terminal_lock:
                         # A stopped consumer can be killed before another write.
                         # Keep reaping even while command code owns the terminal.
+                        recent = self._suspensions
                         pid, pending_status = os.waitpid(self._proc.pid, os.WNOHANG | os.WUNTRACED)
-                        if pid and not os.WIFSTOPPED(pending_status):
-                            self._proc.returncode = os.waitstatus_to_exitcode(pending_status)
-                            return
+                        if pid:
+                            if not os.WIFSTOPPED(pending_status):
+                                self._proc.returncode = os.waitstatus_to_exitcode(pending_status)
+                                return
+                            # A newer stop: the one a suspension would deal with.
+                            seen = recent
                         # A short write may already have returned the terminal.
                         # Do not turn that ordinary handoff into a job suspension.
                         if not self._terminal_available.is_set():
@@ -854,37 +878,55 @@ def _watch_job(self, terminal_fd: int) -> None:
                 if foreground == self._proc.pid:
                     continue
 
-            self._suspend_with_cmd2(terminal_fd)
+            self._suspend_with_cmd2(terminal_fd, seen)
+
+    def _suspend_with_cmd2(self, terminal_fd: int, seen: int) -> None:
+        """Relay a stop in the pipeline's job to cmd2's own, and continue the pipeline once cmd2 resumes.
+
+        Ctrl-Z stops every process of the job that does not ignore it, and each stop may be
+        reported: the consumer's to its watcher, a producer's to the thread that waits for it.
+        Only the first to arrive suspends the job. A suspension ends by continuing the whole
+        pipeline, so a stop reported before it finished needs nothing more.
 
-    def _suspend_with_cmd2(self, terminal_fd: int) -> None:
-        """Relay a pipeline's stop to cmd2's own job, and continue the pipeline once cmd2 resumes."""
+        :param terminal_fd: the controlling terminal
+        :param seen: the number of finished suspensions when the stop was reported
+        """
         import signal
 
-        with self._terminal_lock:
-            if os.tcgetpgrp(terminal_fd) == self._proc.pid:
-                self._set_foreground_group(terminal_fd, self._original_group)
-        # Stop every terminal reader before returning control to the outer shell.
-        self._signal_pipeline(signal.SIGSTOP)
-        self._job_resumed.clear()
-        self._relaying_stop = True
-        # Signal the main thread itself. Only it runs Python signal handlers, and a
-        # process-directed signal may be taken by another thread while the main
-        # thread sleeps in a system call, which then never returns to run the handler.
-        signal.pthread_kill(threading.main_thread().ident or 0, signal.SIGTSTP)
-        self._job_resumed.wait()
-        self._signal_pipeline(signal.SIGCONT)
-
-    def _relay_producer_stop(self) -> None:
-        """Suspend the shell's whole job for a stopped producer that outlived the consumer.
-
-        Ctrl-Z reaches only the foreground group, and the producer may be all that is
-        left of it. The watcher ended with the consumer, so nothing else relays the stop.
+        # Wait in short polls: on the main thread, the suspension being waited for may need
+        # this thread to run the SIGTSTP handler.
+        while not self._suspension_lock.acquire(timeout=0.1):
+            pass
+        try:
+            if self._suspensions != seen:
+                return
+            with self._terminal_lock:
+                if os.tcgetpgrp(terminal_fd) == self._proc.pid:
+                    self._set_foreground_group(terminal_fd, self._original_group)
+            # Stop every terminal reader before returning control to the outer shell.
+            self._signal_pipeline(signal.SIGSTOP)
+            self._job_resumed.clear()
+            self._relaying_stop = True
+            # Signal the main thread itself. Only it runs Python signal handlers, and a
+            # process-directed signal may be taken by another thread while the main
+            # thread sleeps in a system call, which then never returns to run the handler.
+            signal.pthread_kill(threading.main_thread().ident or 0, signal.SIGTSTP)
+            self._job_resumed.wait()
+            self._signal_pipeline(signal.SIGCONT)
+            self._suspensions += 1
+        finally:
+            self._suspension_lock.release()
+
+    def _relay_producer_stop(self, seen: int) -> None:
+        """Suspend the shell's whole job for a stopped producer that joined this pipeline.
+
+        The consumer need not stop with it: it may ignore Ctrl-Z, or be gone already, and
+        then nothing else would relay the stop.
+
+        :param seen: the number of finished suspensions when the stop was reported
         """
-        terminal_fd = self._terminal_fd
-        if terminal_fd is None or not self._process_done.is_set():
-            # A live watcher relays the consumer's stop and continues the whole group.
-            return
-        self._suspend_with_cmd2(terminal_fd)
+        if self._terminal_fd is not None:
+            self._suspend_with_cmd2(self._terminal_fd, seen)
 
     def _wait_for_producer(self, pipeline: "ProcReader", timeout: float | None) -> None:
         """Wait for a producer in a terminal pipeline's job, relaying its job-control stops.
@@ -896,6 +938,7 @@ def _wait_for_producer(self, pipeline: "ProcReader", timeout: float | None) -> N
 
         deadline = None if timeout is None else time.monotonic() + timeout
         while self._proc.returncode is None:
+            seen = pipeline._suspensions
             try:
                 pid, status = os.waitpid(self._proc.pid, os.WNOHANG | os.WUNTRACED)
             except ChildProcessError:
@@ -907,7 +950,7 @@ def _wait_for_producer(self, pipeline: "ProcReader", timeout: float | None) -> N
                     raise subprocess.TimeoutExpired(self._proc.args, timeout or 0)
                 time.sleep(0.05)
             elif os.WIFSTOPPED(status):
-                pipeline._relay_producer_stop()
+                pipeline._relay_producer_stop(seen)
             else:
                 self._proc.returncode = os.waitstatus_to_exitcode(status)
 
@@ -1069,6 +1112,8 @@ def _relay(self) -> None:
 
         poller = select.poll()
         poller.register(self._out_fd, select.POLLOUT)
+        incoming = select.poll()
+        incoming.register(self._in_fd, select.POLLIN)
         lend = contextlib.ExitStack()
         lending = False
         try:
@@ -1079,11 +1124,15 @@ def _relay(self) -> None:
                     # 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
+                # Wait for output outside the lock, then take and count it under the lock. Output
+                # taken out of the pipe but not yet counted would look passed on to idle() and
+                # flush(), and cmd2's next write could overtake it.
+                incoming.poll()
                 with self._lock:
+                    data = os.read(self._in_fd, 65536)
                     self._received += len(data)
+                if not data:
+                    return
                 view = memoryview(data)
                 written = 0
                 while written < len(view):