mirror of
https://github.com/Niko1221/Strata.git
synced 2026-11-07 06:15:58 -10:00
serve: the lifeline identity is the child's start time, the sampler survives a bad source (#3590, #3407)
The portable lifeline watcher built a child's identity from `ps -o lstart=,stat=` and kept the whole line, so the token changed with the process state (R while it starts, S asleep). The watcher compared its own reading with the token taken at start, decided it was not the child it was told about and exited without ending it (test_lifeline "the child outlived a killed server (watcher)", #3407). The identity is now the start time alone: field 22 of /proc/<pid>/stat on Linux (a zombie has none), `ps -o lstart=` elsewhere; strata_control/procs.start_token drops the state column the same way. A launch that is not a real Popen is not watched. Telemetry._loop no longer ends on the first failing sample (a machine whose GPU or psutil source is unreadable): it prints one line and keeps sampling, and NoLeakedSampler checks the thread itself instead of counting threads. New test: the token is the same while the child runs (R) and sleeps (S); on the old code it fails on Linux ('...R' != '...S'). Ported from PR #3590.
This commit is contained in:
1 parent
fd8d37e93d
commit
b3db5a1c83
5 files changed
+107
-15
No files matched your search
+42
-4
@@ -25,6 +25,8 @@ import sys
|
||||
import threading
|
||||
import time
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
from serve import winjob
|
||||
from strata_control import procs
|
||||
|
||||
@@ -99,11 +101,25 @@ pid, token = int(sys.argv[1]), sys.argv[2]
|
||||
for s in (signal.SIGINT, signal.SIGHUP, signal.SIGTERM):
|
||||
signal.signal(s, signal.SIG_IGN) # it ends with its pipe, not with the terminal
|
||||
def start_of(p):
|
||||
# Start time ONLY. `ps`'s stat column changes while a process runs (R while it starts, S while it
|
||||
# sleeps), so including it made this token disagree with the one taken at start and the watcher
|
||||
# spared the child it was meant to end.
|
||||
try:
|
||||
out = subprocess.run(["ps", "-o", "lstart=,stat=", "-p", str(p)], capture_output=True, text=True, timeout=5).stdout.strip()
|
||||
with open("/proc/%d/stat" % p, "rb") as f: # closed explicitly: start_of runs up to 42 times per child
|
||||
raw = f.read().decode("utf-8", "replace")
|
||||
fields = raw[raw.rindex(")") + 2:].split()
|
||||
if fields[0] == "Z": # exited; only its parent has not collected it
|
||||
return None
|
||||
return "l:" + fields[19]
|
||||
except Exception:
|
||||
pass
|
||||
if os.path.exists("/proc/self/stat"): # Linux, and that PID has no /proc entry any more
|
||||
return None
|
||||
try:
|
||||
out = subprocess.run(["ps", "-o", "lstart=", "-p", str(p)], capture_output=True, text=True, timeout=5).stdout.strip()
|
||||
except Exception:
|
||||
return None
|
||||
return None if not out or out.endswith("Z") else "s:" + out
|
||||
return None if not out else "p:" + out
|
||||
os.set_blocking(0, True)
|
||||
data = b""
|
||||
while True:
|
||||
@@ -134,15 +150,37 @@ _watchers_lock = threading.Lock()
|
||||
|
||||
|
||||
def _start_token(pid: int) -> str:
|
||||
"""The same string for the life of one process, different for a later process with that PID.
|
||||
|
||||
Identity is the start time alone. The previous version ran `ps -o lstart=,stat=` and kept the whole
|
||||
line, so the token changed whenever the process changed state - measured here: one child reported
|
||||
'... R' once and '... S' eleven times inside 150 ms. The watcher then compared its own reading with
|
||||
the token from start time, saw "different process", and exited without ending the child, which is
|
||||
exactly the flake `serve/test_lifeline.py` reported as "the child outlived a killed server".
|
||||
"""
|
||||
if sys.platform.startswith("linux"):
|
||||
try:
|
||||
fields = Path(f"/proc/{pid}/stat").read_text(encoding="utf-8", errors="replace")
|
||||
rest = fields[fields.rindex(")") + 2:].split()
|
||||
return "" if rest[0] == "Z" else "l:" + rest[19] # starttime, clock ticks since boot
|
||||
except (OSError, ValueError, IndexError):
|
||||
# procfs IS there and the entry is not readable: the process is gone (or is not ours), and there is
|
||||
# nothing to fall back to. Without procfs at all, fall through to `ps` - otherwise this side would
|
||||
# return "" while the watcher's start_of() returns "p:...", the two can never match, and the child
|
||||
# outlives the server: exactly the failure this change removes, restored for that environment.
|
||||
if os.path.exists("/proc/self/stat"):
|
||||
return ""
|
||||
try:
|
||||
out = subprocess.run(["ps", "-o", "lstart=,stat=", "-p", str(pid)], capture_output=True, text=True,
|
||||
out = subprocess.run(["ps", "-o", "lstart=", "-p", str(pid)], capture_output=True, text=True,
|
||||
timeout=5).stdout.strip()
|
||||
except (OSError, subprocess.SubprocessError):
|
||||
return ""
|
||||
return "" if not out or out.endswith("Z") else "s:" + out
|
||||
return "" if not out else "p:" + out
|
||||
|
||||
|
||||
def _watch(child: subprocess.Popen) -> None:
|
||||
if not isinstance(child, _REAL_POPEN) or getattr(child, "pid", None) is None:
|
||||
return
|
||||
try:
|
||||
watcher = subprocess.Popen([sys.executable, "-S", "-c", _WATCHER, str(child.pid), _start_token(child.pid)],
|
||||
stdin=subprocess.PIPE, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
|
||||
|
||||
+18
-2
@@ -581,7 +581,8 @@ class Telemetry:
|
||||
}
|
||||
self._disk_prev = None
|
||||
self._stop = threading.Event()
|
||||
threading.Thread(target=self._loop, daemon=True).start()
|
||||
self._thread = threading.Thread(target=self._loop, daemon=True)
|
||||
self._thread.start()
|
||||
|
||||
def _disk(self):
|
||||
if not self.ps:
|
||||
@@ -641,8 +642,23 @@ class Telemetry:
|
||||
self._stop.set()
|
||||
|
||||
def _loop(self):
|
||||
"""One sample a second, for as long as the server lives.
|
||||
|
||||
A failing sample must not end the thread: on a machine where the GPU/psutil source is not readable
|
||||
(a CI runner, a container without NVML) the first exception used to kill the sampler outright, so
|
||||
`GET /metrics` silently stopped updating and `test_responses.NoLeakedSampler` saw "the server did
|
||||
not start its hardware sampler" - the thread had started and died between two readings.
|
||||
"""
|
||||
complained = False
|
||||
while not self._stop.is_set():
|
||||
s = self.sample()
|
||||
try:
|
||||
s = self.sample()
|
||||
except Exception as e: # noqa: BLE001 - a monitor must outlive its source
|
||||
if not complained: # once, not once a second
|
||||
print(f"[strata] metrics: a sample failed ({e}); the sampler keeps trying", flush=True)
|
||||
complained = True
|
||||
self._stop.wait(1.0)
|
||||
continue
|
||||
with self.lock:
|
||||
self.now = s
|
||||
for k in ("gpu_util", "gpu_mem_used", "gpu_temp", "gpu_power", "gpu_pcie_rx_mb", "cpu", "ram_used",
|
||||
|
||||
@@ -105,6 +105,45 @@ class Lifeline(unittest.TestCase):
|
||||
def test_the_portable_watcher_ends_the_child_when_the_server_is_killed(self):
|
||||
self.assertEqual(self._killed_parent_takes_the_child(STRATA_LIFELINE="watcher"), "watcher")
|
||||
|
||||
@unittest.skipUnless(POSIX, "the portable watcher is the POSIX path")
|
||||
def test_the_watcher_token_does_not_change_with_the_child_state(self):
|
||||
"""The watcher's token has to be the child's START TIME, and nothing that moves while it runs.
|
||||
|
||||
The token used to come from `ps -o lstart=,stat=`, so it carried the state character as well. A child
|
||||
is R (running) for the first milliseconds of its life and then settles to S (sleeping); the token is
|
||||
taken right after `popen`, so the watcher then compared 'S' against its 'R' token, decided the child
|
||||
was not the one it had been told about, and exited WITHOUT signalling it. That is the "child outlived
|
||||
a killed server" failure, and it appeared or vanished depending on which state the spawn was sampled
|
||||
in - which is why it read as a flake. Sampled while the child spins (R) and again while it sleeps
|
||||
(S), the token must be the same either way.
|
||||
"""
|
||||
from serve import lifeline
|
||||
from unittest import mock
|
||||
# in-process popen() means procs.write_record() would use the REAL per-user state dir; setUp's env only
|
||||
# reaches subprocesses. Keep this test's launch record in the temp dir like the other cases.
|
||||
with mock.patch.dict(os.environ, {"XDG_STATE_HOME": self.temp.name, "LOCALAPPDATA": self.temp.name}):
|
||||
child = lifeline.popen([sys.executable, "-c",
|
||||
"import time\nend = time.time() + 1.5\nwhile time.time() < end:\n pass\n"
|
||||
"time.sleep(30)"], "engine")
|
||||
self.spawned.append(child.pid)
|
||||
self.addCleanup(lambda: child.poll() is None and child.kill())
|
||||
|
||||
def state():
|
||||
"""The child's state character, read independently of the code under test ('R' running, 'S' asleep)."""
|
||||
out = subprocess.run(["ps", "-o", "stat=", "-p", str(child.pid)], capture_output=True, text=True).stdout
|
||||
return out.strip()[:1]
|
||||
|
||||
while_spinning = lifeline._start_token(child.pid)
|
||||
spinning_state = state()
|
||||
time.sleep(2.0) # asleep by now
|
||||
while_sleeping = lifeline._start_token(child.pid)
|
||||
sleeping_state = state()
|
||||
self.assertEqual((spinning_state, sleeping_state), ("R", "S"),
|
||||
"the child did not go from running to sleeping, so this proves nothing about the token")
|
||||
self.assertTrue(while_spinning and while_sleeping, (while_spinning, while_sleeping))
|
||||
self.assertEqual(while_spinning, while_sleeping,
|
||||
"the token moved with the child's state, so the watcher would skip the kill")
|
||||
|
||||
@unittest.skipUnless(sys.platform.startswith("linux"), "PR_SET_PDEATHSIG is Linux")
|
||||
def test_a_child_started_on_a_request_thread_outlives_that_thread_but_not_the_server(self):
|
||||
proc, info = self.launch("thread")
|
||||
|
||||
@@ -276,19 +276,18 @@ class NoLeakedSampler(unittest.TestCase):
|
||||
running (21 after this file), polling NVML and psutil every second, which skewed serve/test_parallel's timings."""
|
||||
|
||||
def test_server_close_ends_the_sampler(self):
|
||||
import threading
|
||||
import time
|
||||
tok = ByteTokenizer()
|
||||
before = sum(1 for t in threading.enumerate() if t.name.endswith("(_loop)"))
|
||||
httpd = serve(Service(MockEngine(tok, ANSWER, max_context=16384), tok, TEMPLATE), port=0)
|
||||
self.assertEqual(sum(1 for t in threading.enumerate() if t.name.endswith("(_loop)")), before + 1)
|
||||
sampler = getattr(getattr(httpd, "svc", None), "telemetry", None)
|
||||
self.assertIsNotNone(sampler, "the server did not initialize telemetry")
|
||||
thread = getattr(sampler, "_thread", None)
|
||||
self.assertIsNotNone(thread, "the server did not start its hardware sampler thread")
|
||||
self.assertTrue(thread.is_alive(), "the hardware sampler thread died immediately")
|
||||
httpd.shutdown()
|
||||
httpd.server_close()
|
||||
for _ in range(50):
|
||||
if sum(1 for t in threading.enumerate() if t.name.endswith("(_loop)")) == before:
|
||||
break
|
||||
time.sleep(0.1)
|
||||
self.assertEqual(sum(1 for t in threading.enumerate() if t.name.endswith("(_loop)")), before)
|
||||
thread.join(timeout=5.0)
|
||||
self.assertFalse(thread.is_alive(), "a stopped server left its hardware sampler running")
|
||||
|
||||
|
||||
# ------------------------------------------------------------------------------------------------ over HTTP
|
||||
|
||||
@@ -75,7 +75,7 @@ def start_token(pid: int) -> str | None:
|
||||
timeout=5).stdout.strip()
|
||||
except (OSError, subprocess.SubprocessError):
|
||||
return None
|
||||
return None if not out or out.endswith("Z") else "s:" + out
|
||||
return None if not out or out.endswith("Z") else "s:" + out.rsplit(None, 1)[0]
|
||||
|
||||
|
||||
def started_epoch(pid: int) -> float | None:
|
||||
|
||||
Reference in new issue
Block a user