fix(log_lib): run_with_log hangs forever when an orphan holds the child's pipe

sky/skylet/log_lib.py

@@ -9,6 +9,7 @@
 import os
 import queue as queue_lib
 import shlex
+import signal
 import subprocess
 import sys
 import tempfile
@@ -155,6 +156,59 @@ def process_subprocess_stream(proc, stdout_stream_handler,
     return stdout, stderr
 
 
+#: How long to wait, after the child has exited, for its output pipes to reach
+#: EOF before concluding that an orphaned grandchild is holding them open.
+_ORPHANED_PIPE_GRACE_SECONDS = 30
+
+
+def _force_eof_when_orphans_hold_the_pipe(proc: subprocess.Popen, pgid: int,
+                                          drained: threading.Event,
+                                          grace: float) -> None:
+    """Unblock the log readers when a grandchild outlives the child holding its pipe.
+
+    ``process_subprocess_stream`` reads until EOF, and EOF requires EVERY write
+    end of the pipe to be closed -- including the ones a grandchild inherited.
+    So a child that dies leaving orphans behind leaves the reader blocked
+    forever, which means ``proc.wait()`` below is never reached and this
+    function never returns a returncode. The caller is then hung with no
+    timeout: a dead two-node training job whose ranks aborted was still being
+    reported as RUNNING by the jobs controller two hours later, because the Ray
+    task that ran it never finished.
+
+    ``kill_children_processes`` cannot help here -- it walks the process TREE,
+    and orphans are reparented to init, so they are no longer in it. The process
+    GROUP is the handle that still reaches them: ``start_new_session=True``
+    makes the child a group leader (so ``pgid == proc.pid``), and neither
+    reparenting nor reaping the leader changes the group of the orphans left in
+    it.
+
+    Fires only once the child has exited AND the readers are still blocked
+    ``grace`` seconds later. That is a state the caller never recovers from on
+    its own, so forcing EOF cannot lose anything that was still working -- while
+    an unconditional kill would break commands that deliberately leave daemons
+    in the group, such as ``ray start``.
+    """
+    try:
+        proc.wait()
+    except Exception:  # pylint: disable=broad-except
+        return
+    if drained.wait(timeout=grace):
+        return
+    try:
+        # Never signal our own group: that would take the caller down too.
+        if pgid == os.getpgid(0):
+            return
+    except OSError:
+        return
+    logger.warning(
+        f'Command exited but its output pipe is still open after {grace}s; '
+        f'killing leftover process group {pgid} so the log reader can finish.')
+    try:
+        os.killpg(pgid, signal.SIGKILL)
+    except OSError:
+        pass
+
+
 def run_with_log(
     cmd: Union[List[str], str],
     log_path: str,
@@ -299,6 +353,18 @@ def _timeout_handler():
                 timer = threading.Timer(timeout, _timeout_handler)
                 timer.start()
 
+            # start_new_session=True above makes the child a session and group
+            # leader, so its process group id IS its pid -- no lookup, which
+            # would be a race: a short-lived child is already a zombie by now
+            # and os.getpgid() can fail on it. See
+            # _force_eof_when_orphans_hold_the_pipe.
+            drained = threading.Event()
+            if ctx is not None or process_stream:
+                threading.Thread(target=_force_eof_when_orphans_hold_the_pipe,
+                                 args=(proc, proc.pid, drained,
+                                       _ORPHANED_PIPE_GRACE_SECONDS),
+                                 daemon=True).start()
+
             try:
                 if ctx is not None:
                     # When runs in a coroutine, always process the subprocess
@@ -317,6 +383,7 @@ def _timeout_handler():
                     stdout, stderr = process_subprocess_stream(
                         proc, stdout_stream_handler, stderr_stream_handler)
             finally:
+                drained.set()
                 if timer is not None:
                     timer.cancel()
 

tests/unit_tests/test_sky/skylet/test_log_lib.py

@@ -3,7 +3,9 @@
 from io import StringIO
 import subprocess
 import tempfile
+import time
 import unittest
+from unittest import mock
 
 from sky.skylet import log_lib
 
@@ -168,3 +170,43 @@ def test_no_stream_timeout_not_exceeded(self):
 
 if __name__ == '__main__':
     unittest.main()
+
+
+class TestRunWithLogOrphanedPipe(unittest.TestCase):
+    """run_with_log must not hang when a grandchild outlives the child holding its pipe."""
+
+    def test_returns_when_an_orphan_holds_the_output_pipe(self):
+        """A child that exits leaving a background grandchild must not hang the caller.
+
+        EOF needs EVERY write end of the pipe closed, so a grandchild that inherited
+        stdout keeps the log reader blocked; run_with_log then never reaches proc.wait()
+        and never returns a returncode. That is how a dead two-node training job was
+        still reported as RUNNING by the jobs controller two hours later.
+
+        kill_children_processes cannot reach such a grandchild -- it is reparented to
+        init, outside the process tree that helper walks -- so the fix signals the
+        process GROUP, which reparenting does not change.
+        """
+        with tempfile.NamedTemporaryFile(suffix='.log', delete=False) as f:
+            log_path = f.name
+
+        # The child exits at once with a distinctive code; the grandchild holds
+        # stdout open far longer than this test would ever wait.
+        cmd = ['bash', '-c', 'sleep 300 & exit 7']
+        # create=True so this test still RUNS against a tree without the fix,
+        # where it blocks on the orphan's pipe instead of erroring on a missing
+        # attribute -- the failure has to be the hang itself, or the test proves
+        # only that a constant exists.
+        with mock.patch.object(log_lib,
+                               '_ORPHANED_PIPE_GRACE_SECONDS',
+                               2,
+                               create=True):
+            start = time.time()
+            returncode = log_lib.run_with_log(cmd,
+                                              log_path,
+                                              process_stream=True)
+            elapsed = time.time() - start
+
+        self.assertEqual(returncode, 7)
+        self.assertLess(elapsed, 60,
+                        'run_with_log stayed blocked on the orphan\'s pipe')