From d54838e4a693c281c00b3b31d6fe487b245999c6 Mon Sep 17 00:00:00 2001 From: Tigist Diriba Date: Fri, 11 Sep 2026 12:01:44 +0300 Subject: [PATCH] fix(log_lib): run_with_log hangs forever when an orphan holds the child's pipe `process_subprocess_stream` reads the child's output until EOF, and only then does `run_with_log` call `proc.wait()` -- with the comment "Stream processing already waited for process completion". That assumption breaks whenever a grandchild outlives the child: EOF needs EVERY write end of the pipe closed, including the ones the grandchild inherited, so the reader stays blocked, `proc.wait()` is never reached, and `run_with_log` never returns a returncode. The caller hangs with no timeout. Seen in production on a two-node training job: the ranks aborted on an RDMA transport error, torchrun exited, but five orphaned multiprocessing-spawn children kept the stdout pipe open. The Ray task running the command never finished, so the managed-jobs controller reported the job as RUNNING for two hours while 16 GPUs sat idle. `kill_children_processes` cannot help: it walks the process tree, and the orphans are reparented to init, so they are no longer in it. The process group still reaches them -- `start_new_session=True` makes the child a group leader, so the group id is its pid, and neither reparenting nor reaping the leader moves the orphans out of that group. A watchdog thread now waits for the child, then waits a grace period for the readers to drain, and only if they are still blocked kills the leftover process group to force EOF. That trigger matters: an unconditional kill after exit would break commands that deliberately leave daemons in the group, such as `ray start`. When it does fire, the caller was already hung forever, so forcing EOF cannot lose anything that was still working. Test: a child that exits while a background grandchild holds its stdout. Without the fix it hangs (killed at 90 s); with it, `run_with_log` returns the child's exit code after the grace period. Co-Authored-By: Claude Opus 5 --- sky/skylet/log_lib.py | 67 +++++++++++++++++++ .../test_sky/skylet/test_log_lib.py | 42 ++++++++++++ 2 files changed, 109 insertions(+) diff --git a/sky/skylet/log_lib.py b/sky/skylet/log_lib.py index 54dcc9612..5b0530b69 100644 --- a/sky/skylet/log_lib.py +++ b/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() diff --git a/tests/unit_tests/test_sky/skylet/test_log_lib.py b/tests/unit_tests/test_sky/skylet/test_log_lib.py index 9a7f9e05e..06d5c6618 100644 --- a/tests/unit_tests/test_sky/skylet/test_log_lib.py +++ b/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')