fix isolated_runner: TimeoutExpired has no .pid (e-261003004801) — Popen start_new_session + killpg(proc.pid), group+children reaped; scheduler None-guard in r2 double-close guard (mocked es in tests); regression test test_timeout_kill
This commit is contained in:
parent
98511453d7
commit
03d3dffeb8
3 changed files with 135 additions and 46 deletions
|
|
@ -50,61 +50,70 @@ def run_llm_tick(level_cfg, prompt: str, session_id=None, trace_prefix=None, tim
|
|||
# просто поглощает такты (TickAbsorbed), это норма архитектуры
|
||||
timeout = getattr(level_cfg, "tick_timeout", 300)
|
||||
timeout = max(timeout, int(level_cfg.every * 0.8))
|
||||
# start_new_session: воркер становится лидером СВОЕЙ процесс-группы —
|
||||
# добивание на таймауте достаёт всю группу LLM-потомков по proc.pid,
|
||||
# без чтения /proc/<pid>/status.
|
||||
proc = subprocess.Popen(
|
||||
[_VENV_PY, _WORKER],
|
||||
stdin=subprocess.PIPE,
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE,
|
||||
cwd=_HARNESS_DIR,
|
||||
env=env,
|
||||
start_new_session=True,
|
||||
)
|
||||
try:
|
||||
proc = subprocess.run(
|
||||
[_VENV_PY, _WORKER],
|
||||
input=json.dumps(payload).encode(),
|
||||
capture_output=True,
|
||||
timeout=timeout,
|
||||
cwd=_HARNESS_DIR,
|
||||
env=env,
|
||||
)
|
||||
except subprocess.TimeoutExpired as e:
|
||||
# воркер убит по таймауту. subprocess.run убивает только основной
|
||||
# процесс (Popen.kill через os.kill без группы) — сабпроцесс _llm_worker
|
||||
# успевал сам заспавнить LLM-подпроцессы, которые оставались ЖИВЫМИ и
|
||||
# продолжали писать в стор после SchedulerTimeout (BudgetSpent/ToolCalled
|
||||
# с завершённым trace — race, burn 9router, t-261002090242-cf7cf9).
|
||||
# Добиваем всю процесс-группу и ЖДЁМ фактической смерти.
|
||||
pgid = None
|
||||
stdout, stderr = proc.communicate(
|
||||
input=json.dumps(payload).encode(), timeout=timeout)
|
||||
except subprocess.TimeoutExpired:
|
||||
# воркер не уложился в таймаут. subprocess.communicate() бросает
|
||||
# TimeoutExpired БЕЗ .pid — subprocess.run не отдавал Popen, и
|
||||
# аварийная зачистка падала с AttributeError ('TimeoutExpired'
|
||||
# object has no attribute 'pid', e-261003004801) ДО добивания
|
||||
# группы: LLM-потомки выживали и писали в стор после смерти тика
|
||||
# (race, burn роутера, t-261002090242). С Popen pid есть всегда.
|
||||
pgid = proc.pid
|
||||
try:
|
||||
with open(f"/proc/{e.pid}/status") as f:
|
||||
for line in f:
|
||||
if line.startswith("Tgid:"):
|
||||
pgid = line.split()[1]
|
||||
break
|
||||
except OSError:
|
||||
pass
|
||||
try:
|
||||
if e.pid > 1:
|
||||
os.killpg(pgid or e.pid, signal.SIGKILL)
|
||||
if pgid > 1:
|
||||
os.killpg(pgid, signal.SIGKILL)
|
||||
except (ProcessLookupError, PermissionError, OSError):
|
||||
pass
|
||||
for _ in range(10): # до 5с на смерть группы, иначе — SIGKILL ещё раз
|
||||
try:
|
||||
os.killpg(pgid or e.pid, 0)
|
||||
os.killpg(pgid, 0)
|
||||
time.sleep(0.5)
|
||||
except (ProcessLookupError, PermissionError) as ex:
|
||||
break
|
||||
else:
|
||||
try:
|
||||
os.killpg(pgid or e.pid, signal.SIGKILL)
|
||||
os.killpg(pgid, signal.SIGKILL)
|
||||
except OSError:
|
||||
pass
|
||||
for stream in (proc.stdout, proc.stderr, proc.stdin):
|
||||
try:
|
||||
if stream:
|
||||
stream.close()
|
||||
except OSError:
|
||||
pass
|
||||
raise
|
||||
out = None
|
||||
if os.path.exists(result_path):
|
||||
try:
|
||||
with open(result_path) as f:
|
||||
out = json.load(f)
|
||||
except ValueError:
|
||||
out = None
|
||||
try:
|
||||
os.unlink(result_path)
|
||||
except OSError:
|
||||
pass
|
||||
if out is None:
|
||||
return None, {"error": f"worker result file missing: stdout={(proc.stdout or b'')[:200]!r} stderr={(proc.stderr or b'').decode(errors='replace')[-300:]}"}, None
|
||||
if not out.get("ok"):
|
||||
return None, {"error": (out.get("error") or "unknown")[:500]}, None
|
||||
return out.get("text", ""), out.get("usage", {}), out.get("trace_id")
|
||||
proc.wait(timeout=5) # reap: без этого — зомби до смерти родителя
|
||||
except subprocess.TimeoutExpired:
|
||||
pass
|
||||
raise
|
||||
else:
|
||||
out = None
|
||||
if os.path.exists(result_path):
|
||||
try:
|
||||
with open(result_path) as f:
|
||||
out = json.load(f)
|
||||
except ValueError:
|
||||
out = None
|
||||
try:
|
||||
os.unlink(result_path)
|
||||
except OSError:
|
||||
pass
|
||||
if out is None:
|
||||
return None, {"error": f"worker result file missing: stdout={(stdout or b'')[:200]!r} stderr={(stderr or b'').decode(errors='replace')[-300:]}"}, None
|
||||
if not out.get("ok"):
|
||||
return None, {"error": (out.get("error") or "unknown")[:500]}, None
|
||||
return out.get("text", ""), out.get("usage", {}), out.get("trace_id")
|
||||
|
|
@ -540,9 +540,10 @@ def run_level_tick(level_cfg, seq: int, model: dict):
|
|||
# содержит целевой ответ; повторный append поверх перезаписал бы answer
|
||||
# дайджестом тика и испортил бы проверенный результат (перезапись
|
||||
# «последний-победил» в проекции). Закрываем только ещё-running.
|
||||
_model = es.run_projections()
|
||||
_model = es.run_projections() or {} # стаб/None в тестах — не валим
|
||||
_tasks = _model.get("tasks") or {}
|
||||
_done = {tid for tid in (claimed or [])
|
||||
if ((_model.get("tasks") or {}).get(tid) or {}).get("status") == "done"}
|
||||
if (_tasks.get(tid) or {}).get("status") == "done"}
|
||||
_first = True
|
||||
for tid in (claimed or []):
|
||||
if tid in _done:
|
||||
|
|
|
|||
79
tests/test_timeout_kill.py
Normal file
79
tests/test_timeout_kill.py
Normal file
|
|
@ -0,0 +1,79 @@
|
|||
"""Регресс e-261003004801: таймаут LLM-тика не должен валить зачистку.
|
||||
|
||||
Было: subprocess.run + TimeoutExpired без .pid → AttributeError в
|
||||
изолированном runner'е ДО добивания процесс-группы. LLM-потомки воркера
|
||||
выживали и писали в стор после смерти тика (race, burn роутера).
|
||||
Теперь: Popen(start_new_session=True) → killpg(proc.pid); исключение —
|
||||
честный TimeoutExpired, группа (с потомками) мертва до re-raise.
|
||||
"""
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import time
|
||||
import unittest
|
||||
|
||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
from harness import isolated_runner
|
||||
|
||||
|
||||
class TimeoutKillTest(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.tmp = tempfile.mkdtemp(prefix="test_timeout_kill_")
|
||||
self.worker = os.path.join(self.tmp, "fake_worker.py")
|
||||
# фейковый воркер: спавнит потомка (имитация LLM-сабпроцесса) и спит
|
||||
with open(self.worker, "w") as f:
|
||||
f.write(
|
||||
"import subprocess, time\n"
|
||||
"child = subprocess.Popen(['sleep', '287'])\n"
|
||||
"print(child.pid, flush=True)\n"
|
||||
"time.sleep(300)\n"
|
||||
)
|
||||
self._orig = (isolated_runner._VENV_PY, isolated_runner._WORKER)
|
||||
isolated_runner._VENV_PY = sys.executable
|
||||
isolated_runner._WORKER = self.worker
|
||||
|
||||
def tearDown(self):
|
||||
isolated_runner._VENV_PY, isolated_runner._WORKER = self._orig
|
||||
|
||||
def _cfg(self):
|
||||
class Cfg:
|
||||
name = "test"
|
||||
model = "x"
|
||||
effort = "low"
|
||||
mandate = ""
|
||||
can_shell = False
|
||||
ctx_tokens = 1000
|
||||
every = 300
|
||||
return Cfg()
|
||||
|
||||
def test_timeout_kills_group_without_pid_crash(self):
|
||||
with self.assertRaises(subprocess.TimeoutExpired):
|
||||
isolated_runner.run_llm_tick(self._cfg(), "ping", timeout=3)
|
||||
# группа воркера убита: потомок-«LLM» не выжил
|
||||
time.sleep(0.5)
|
||||
r = subprocess.run(["pgrep", "-c", "-f", "sleep 287"],
|
||||
capture_output=True)
|
||||
# pgrep: exit 1 = совпадений нет → потомок-«LLM» мёртв
|
||||
self.assertEqual(r.returncode, 1,
|
||||
f"LLM-потомок пережил таймаут добивания: {r.stdout!r}")
|
||||
|
||||
def test_normal_run_unaffected(self):
|
||||
# короткий воркер: пишет результат в result_path как настоящий
|
||||
worker = os.path.join(self.tmp, "ok_worker.py")
|
||||
with open(worker, "w") as f:
|
||||
f.write(
|
||||
"import json, os, sys\n"
|
||||
"payload = json.loads(sys.stdin.read())\n"
|
||||
"json.dump({'ok': True, 'text': 'done',\n"
|
||||
" 'usage': {}, 'trace_id': 't'}, open(payload['result_path'], 'w'))\n"
|
||||
)
|
||||
isolated_runner._WORKER = worker
|
||||
text, usage, trace = isolated_runner.run_llm_tick(self._cfg(),
|
||||
"ping", timeout=15)
|
||||
self.assertEqual(text, "done")
|
||||
self.assertEqual(trace, "t")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main(verbosity=2)
|
||||
Loading…
Reference in a new issue