From 664494c8b05184bba46e4192445984dc71314695 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 5 Aug 2026 20:38:42 +0000 Subject: [PATCH 1/2] fix(log): scope fd-capture stop() to the owning process MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Console capture died partway through a torchtitan run: the last uploaded line was mid-way through a torch.compile warning ~10s after init(), while the terminal kept every line to the end of training. FdCapture.stop() writes a sentinel into the pipe so the reader knows where "everything before stop()" ends. But that pipe is shared with every process that inherited fd 1/2 — forked DataLoader and inductor compile workers, the sync subprocess — and forked children inherit pluto's atexit handlers, so a child exiting through the normal interpreter path runs Op.finish() -> flush_console_buffers() -> FdCapture.stop(). The child's sentinel lands in the shared pipe and the parent's reader treats it as its own: it sets _enqueue_enabled = False and drops into tee-only drain mode. Terminal output is unaffected, the reader thread stays alive, nothing is logged — the run's console section just stops. Scope both halves to the process that called start(): - stop() is a no-op off the owner pid. Restoring fds and signalling the reader belong to the owner. - The sentinel carries the owner's pid, and the reader honours only its own. A foreign sentinel is dropped from the tee and the capture rather than leaking control bytes into the terminal or the uploaded lines. Split-across-reads handling grows with the variable-length pid field: _partial_sentinel_suffix now also holds back a complete prefix whose pid (and closing suffix) is still arriving. Both new fork tests fail on the previous code and pass here. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01HGGPRf5pRf6sz9tgoucTm4 --- CLAUDE.md | 27 +++++++++++ pluto/_fd_capture.py | 88 ++++++++++++++++++++++++++++------ tests/test_fd_capture.py | 100 +++++++++++++++++++++++++++++++++++++++ 3 files changed, 200 insertions(+), 15 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index c3dc931..46a9382 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -349,6 +349,33 @@ an auth failure: is definitive, so it no longer says "token may still be valid" (5xx/network failures keep that softer wording). +### fd-level console capture and forked children + +`FdCapture` (`pluto/_fd_capture.py`) `dup2`s a pipe over fds 1/2 so logging +handlers bound before `init()` still get uploaded. That pipe is **shared with +every process that inherits the fd** — forked DataLoader/`torch.compile` +workers, the sync subprocess, anything spawned during training — and forked +children inherit pluto's `atexit` handlers too. So a child exiting through the +normal interpreter path runs `Op.finish()` → `flush_console_buffers()` → +`FdCapture.stop()`. + +Anything in `stop()` that touches the shared pipe must therefore be scoped to +the process that called `start()`: + +- `stop()` is a **no-op off the owner pid**. Restoring fds and writing the + flush sentinel are the owner's business. +- The flush sentinel **carries the owner's pid** (`_stop_sentinel()`), and the + reader only honours its own; a foreign one is dropped from both the tee and + the capture. + +Without that scoping the failure is silent and total: the child's sentinel +lands in the shared pipe, the parent's reader sets `_enqueue_enabled = False` +and drops into tee-only drain mode, so the terminal keeps every line while the +run's console section stops dead — mid-batch, with no error anywhere. Seen on a +torchtitan job where capture died ~10 s in, exactly when inductor spun up its +compile workers. Regression tests: +`tests/test_fd_capture.py::TestForkedChildCannotDisableParentCapture`. + ### Network Filesystems (NFS/Lustre/SMB) and SQLite WAL The sync DB uses SQLite WAL mode (`pluto/sync/store.py`), which relies on POSIX diff --git a/pluto/_fd_capture.py b/pluto/_fd_capture.py index 106e224..38aa121 100644 --- a/pluto/_fd_capture.py +++ b/pluto/_fd_capture.py @@ -21,6 +21,15 @@ restored. When the reader sees it, everything written before ``stop()`` has been enqueued — ``stop()`` waits on that (bounded), giving a deterministic flush without racing the reader. +- The pipe is SHARED with every process that inherited the fd (forked + DataLoader/compile workers, the sync subprocess, anything spawned during + training). So both the sentinel and ``stop()`` itself are scoped to the + process that called ``start()``: the sentinel carries that pid and the + reader only honours its own, and ``stop()`` is a no-op anywhere else. + Without that scoping, a child that runs pluto's teardown on its way out + (atexit handlers are inherited across fork) writes the sentinel into the + shared pipe, and the parent's reader — still very much alive, still + teeing to the terminal — mutes its own uploads for the rest of the run. - The reader thread is NOT joined/killed. Child processes forked while capture was active (e.g. DataLoader workers) inherit the pipe write end; if nobody drained it, their writes would block once the pipe buffer @@ -32,6 +41,7 @@ import logging import os +import re import threading import time from typing import Any, List, Optional, Tuple @@ -42,7 +52,21 @@ # pre-stop output". Control bytes make a collision with real output # effectively impossible; writes <= PIPE_BUF are atomic so it can't # interleave with a concurrent writer's bytes. -_STOP_SENTINEL = b'\x00\x1dpluto:fdcap:flush\x1d\x00' +# +# The trailing pid names the capture that emitted it. The pipe is shared +# with every process that inherited the fd, so a reader must ignore any +# sentinel it did not write itself (see module docstring). +_SENTINEL_PREFIX = b'\x00\x1dpluto:fdcap:flush:' +_SENTINEL_SUFFIX = b'\x1d\x00' +_SENTINEL_RE = re.compile( + re.escape(_SENTINEL_PREFIX) + rb'(\d+)' + re.escape(_SENTINEL_SUFFIX) +) +# Upper bound on the pid field, for detecting a sentinel split across reads. +_MAX_PID_DIGITS = 20 + + +def _stop_sentinel(pid: int) -> bytes: + return _SENTINEL_PREFIX + str(pid).encode('ascii') + _SENTINEL_SUFFIX class FdCapture: @@ -70,6 +94,11 @@ def __init__( self._thread: Optional[threading.Thread] = None self._started = False self._stopped = False + # Owner of the redirected fd. Set in start(); everything that + # touches the shared pipe checks it (see module docstring). + self._owner_pid: Optional[int] = None + self._own_sentinel = b'' + self._own_sentinel_pid = b'' # Guards line/batch state shared between the reader thread and a # stop() caller doing a last-resort flush. @@ -92,6 +121,9 @@ def start(self) -> None: """Redirect self.fd into a pipe drained by a daemon reader thread.""" if self._started or self._stopped: return + self._owner_pid = os.getpid() + self._own_sentinel = _stop_sentinel(self._owner_pid) + self._own_sentinel_pid = str(self._owner_pid).encode('ascii') self._orig_fd = os.dup(self.fd) read_fd, write_fd = os.pipe() os.dup2(write_fd, self.fd) @@ -113,6 +145,13 @@ def stop(self, timeout: float = 2.0) -> None: """ if not self._started or self._stopped: return + if os.getpid() != self._owner_pid: + # A forked child inherited this object (pluto's atexit handlers + # come along with it, so a child exiting through the normal + # interpreter path lands here). It shares the parent's pipe: + # writing the sentinel would mute the parent's uploads for the + # rest of the run, and restoring fds is the parent's business. + return self._stopped = True # 1. Sentinel into the pipe while self.fd still points at it — it @@ -122,7 +161,7 @@ def stop(self, timeout: float = 2.0) -> None: # paths that must never block (DDP). def _write_sentinel() -> None: try: - os.write(self.fd, _STOP_SENTINEL) + os.write(self.fd, self._own_sentinel) except OSError: pass @@ -161,18 +200,23 @@ def _reader_loop(self) -> None: data = held + chunk held = b'' - idx = data.find(_STOP_SENTINEL) - if idx != -1: - before = data[:idx] - after = data[idx + len(_STOP_SENTINEL) :] + # Sentinels are never teed or ingested — they're control bytes. + # Only our own marks the flush point; one written by a process + # that inherited this pipe is dropped and capture continues. + while True: + match = _SENTINEL_RE.search(data) + if match is None: + break + before = data[: match.start()] self._tee(before) self._ingest(before) - with self._state_lock: - self._flush_locked(drain_partial=True) - self._enqueue_enabled = False - self._flushed.set() - self._tee(after) - continue # drain mode: _ingest below is a no-op now + data = data[match.end() :] + if match.group(1) == self._own_sentinel_pid: + with self._state_lock: + self._flush_locked(drain_partial=True) + self._enqueue_enabled = False + self._flushed.set() + # drain mode from here: _ingest below is a no-op now # A sentinel prefix at the very end of the chunk may be the # sentinel split across reads — hold those bytes back until the @@ -197,10 +241,24 @@ def _reader_loop(self) -> None: @staticmethod def _partial_sentinel_suffix(data: bytes) -> int: - """Length of the longest proper sentinel prefix that ends ``data``.""" - max_k = min(len(_STOP_SENTINEL) - 1, len(data)) + """Length of the trailing bytes that could still become a sentinel. + + Two shapes to hold back: a partial ``_SENTINEL_PREFIX``, or a + complete prefix whose pid field (and closing suffix) is still + arriving on the next read. + """ + idx = data.rfind(_SENTINEL_PREFIX) + if idx != -1: + tail = data[idx + len(_SENTINEL_PREFIX) :] + pid_part = tail[:-1] if tail.endswith(_SENTINEL_SUFFIX[:1]) else tail + if len(tail) <= _MAX_PID_DIGITS + 1 and ( + pid_part == b'' or pid_part.isdigit() + ): + return len(data) - idx + + max_k = min(len(_SENTINEL_PREFIX) - 1, len(data)) for k in range(max_k, 0, -1): - if data.endswith(_STOP_SENTINEL[:k]): + if data.endswith(_SENTINEL_PREFIX[:k]): return k return 0 diff --git a/tests/test_fd_capture.py b/tests/test_fd_capture.py index 3496b4e..bc1d6ac 100644 --- a/tests/test_fd_capture.py +++ b/tests/test_fd_capture.py @@ -131,6 +131,106 @@ def test_start_after_stop_is_a_noop(self): assert not any('zombie' in line for line in sm.lines) +class TestForkedChildCannotDisableParentCapture: + """A forked child shares the parent's pipe — and pluto's atexit handlers. + + Any child exiting through the normal interpreter path runs Op.finish() + -> flush_console_buffers() -> FdCapture.stop(). Before the sentinel was + scoped to the process that called start(), that stop() wrote the flush + mark into the SHARED pipe, and the parent's reader — alive and still + teeing to the terminal — muted its own uploads for the rest of the run. + Observed on a torchtitan job: console logs stopped ~10s in, at the point + torch.compile spun up its workers, while the terminal stayed complete. + """ + + def test_child_stop_does_not_mute_parent(self): + with capture_fd(2, logging.ERROR) as (cap, sm): + os.write(2, b'before-fork\n') + time.sleep(0.3) + + pid = os.fork() + if pid == 0: + try: + cap.stop() # inherited teardown, in the child + finally: + os._exit(0) + os.waitpid(pid, 0) + time.sleep(0.3) + + os.write(2, b'after-fork\n') + time.sleep(0.3) + cap.stop() + + assert any('before-fork' in line for line in sm.lines) + assert any('after-fork' in line for line in sm.lines), ( + 'parent capture was silenced by a forked child' + ) + + def test_child_stop_leaks_no_control_bytes(self): + """A foreign sentinel is swallowed, not teed or logged as output.""" + with capture_fd(2, logging.ERROR) as (cap, sm): + pid = os.fork() + if pid == 0: + try: + cap.stop() + finally: + os._exit(0) + os.waitpid(pid, 0) + time.sleep(0.3) + + os.write(2, b'still here\n') + time.sleep(0.3) + cap.stop() + + assert any('still here' in line for line in sm.lines) + assert not any('fdcap:flush' in line for line in sm.lines) + assert not any('\x00' in line for line in sm.lines) + + def test_owner_stop_still_flushes_and_mutes(self): + """The scoping must not break the flush stop() exists to provide.""" + with capture_fd(2, logging.ERROR) as (cap, sm): + os.write(2, b'pre-stop') # no trailing newline + cap.stop() + os.write(2, b'post-stop\n') + time.sleep(0.1) + assert any('pre-stop' in line for line in sm.lines) + assert not any('post-stop' in line for line in sm.lines) + + +class TestSentinelSplitAcrossReads: + def test_partial_sentinel_suffix_holds_incomplete_marks(self): + from pluto import _fd_capture as fc + + hold = FdCapture._partial_sentinel_suffix + assert hold(b'plain output') == 0 + # partial prefix at the edge + assert hold(b'out' + fc._SENTINEL_PREFIX[:5]) == 5 + # complete prefix, pid still arriving + assert hold(b'out' + fc._SENTINEL_PREFIX) == len(fc._SENTINEL_PREFIX) + assert hold(b'out' + fc._SENTINEL_PREFIX + b'123') == ( + len(fc._SENTINEL_PREFIX) + 3 + ) + # pid complete, closing suffix half arrived + assert hold(b'out' + fc._SENTINEL_PREFIX + b'123\x1d') == ( + len(fc._SENTINEL_PREFIX) + 4 + ) + + def test_sentinel_split_across_two_writes_still_flushes(self): + """stop()'s mark can land across a read boundary; it must still hit.""" + with capture_fd(2, logging.ERROR) as (cap, sm): + os.write(2, b'line-a\n') + sentinel = cap._own_sentinel + os.write(2, sentinel[:8]) + time.sleep(0.15) + os.write(2, sentinel[8:]) + time.sleep(0.15) + os.write(2, b'line-b\n') + time.sleep(0.15) + assert any('line-a' in line for line in sm.lines) + assert not any('line-b' in line for line in sm.lines) + assert not any('fdcap:flush' in line for line in sm.lines) + + class TestLineHandling: def test_partial_writes_coalesce_into_one_line(self): with capture_fd(2, logging.ERROR) as (cap, sm): From 8861d30ed9a8787d5d002cda954489435dc6612a Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 5 Aug 2026 20:44:43 +0000 Subject: [PATCH 2/2] style: match the pinned ruff formatter on the new assert CI pins ruff 0.4.10 via poetry.lock, which formats an assert message inline rather than parenthesized. No behaviour change. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01HGGPRf5pRf6sz9tgoucTm4 --- tests/test_fd_capture.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/tests/test_fd_capture.py b/tests/test_fd_capture.py index bc1d6ac..1dbc692 100644 --- a/tests/test_fd_capture.py +++ b/tests/test_fd_capture.py @@ -162,9 +162,9 @@ def test_child_stop_does_not_mute_parent(self): cap.stop() assert any('before-fork' in line for line in sm.lines) - assert any('after-fork' in line for line in sm.lines), ( - 'parent capture was silenced by a forked child' - ) + assert any( + 'after-fork' in line for line in sm.lines + ), 'parent capture was silenced by a forked child' def test_child_stop_leaks_no_control_bytes(self): """A foreign sentinel is swallowed, not teed or logged as output."""