diff --git a/agent/periodic_scheduler.py b/agent/periodic_scheduler.py index a1ba07d666661..ae5d73ed0a35c 100644 --- a/agent/periodic_scheduler.py +++ b/agent/periodic_scheduler.py @@ -3,15 +3,24 @@ Replaces the per-child ``while not stop.wait(interval): body()`` daemon threads (delegate heartbeat, durable turn-lease refresher, turn-liveness watchdog). With ~130 in-process subagents those added 2-3 sleeping OS -threads per child; this module runs every periodic body on ONE daemon -thread ordered by a heap of due times. +threads per child. This module keeps ONE daemon thread that only orders due +times; every due body runs on its own short-lived daemon worker. A blocked +callback therefore cannot delay unrelated lease/liveness timers, while +steady-state thread use stays near zero (workers exist only while a body is +actually running, never one per scheduled handle). Semantics match the loop they replace: the first call happens ``interval`` seconds after :func:`schedule`, and each following call ``interval`` seconds after the previous body *returned* (drift-free wrt. body duration was never a property of the old loops either). A body that returns ``False`` stops itself; a body that raises is logged at debug and rescheduled — one bad -callback must never kill the shared thread. +callback must never kill the shared thread. A handle never overlaps itself: +it is re-queued only once its in-flight run has returned. A worker-start +failure never retires the handle: it is re-queued and logged at warning. + +Prior art: per-handle-worker isolation follows #102458 (jfreshpicks); +bounded-pool alternatives #102615/#102657/#104488 were rejected for +preserving a shared starvation domain; originates from #102030. """ from __future__ import annotations @@ -26,18 +35,20 @@ logger = logging.getLogger(__name__) _THREAD_NAME = "hermes-periodic-scheduler" +_CALLBACK_THREAD_PREFIX = "hermes-periodic-callback" class ScheduledHandle: """Cancel token for one scheduled periodic callback.""" - __slots__ = ("_fn", "_interval", "_cancelled", "_scheduler") + __slots__ = ("_fn", "_interval", "_cancelled", "_scheduler", "_runner") def __init__(self, scheduler: "PeriodicScheduler", fn: Callable[[], object], interval: float): self._scheduler = scheduler self._fn = fn self._interval = interval self._cancelled = False + self._runner: Optional[threading.Thread] = None @property def cancelled(self) -> bool: @@ -56,7 +67,6 @@ def __init__(self) -> None: self._heap: list = [] # (due, seq, handle) self._seq = itertools.count() self._thread: Optional[threading.Thread] = None - self._running: Optional[ScheduledHandle] = None def schedule(self, fn: Callable[[], object], interval: float) -> ScheduledHandle: handle = ScheduledHandle(self, fn, float(interval)) @@ -72,8 +82,54 @@ def _cancel(self, handle: ScheduledHandle, wait: Optional[float]) -> None: with self._cond: handle._cancelled = True self._cond.notify() - if wait and self._running is handle and threading.current_thread() is not self._thread: - self._cond.wait_for(lambda: self._running is not handle, timeout=wait) + runner = handle._runner + if wait and runner is not None and threading.current_thread() is not runner: + runner.join(wait) + + def _dispatch(self, handle: ScheduledHandle) -> None: + """Start ``handle``'s body on its own worker. Called with ``_cond`` held + so ``cancel`` can never observe a half-set runner.""" + runner = threading.Thread( + target=self._run_callback, + args=(handle,), + name=f"{_CALLBACK_THREAD_PREFIX}-{id(handle):x}", + daemon=True, + ) + handle._runner = runner + try: + runner.start() + except Exception: + handle._runner = None + logger.warning( + "failed to start periodic callback worker %r; retrying in %s s", + handle._fn, + handle._interval, + exc_info=True, + ) + if not handle._cancelled: + heapq.heappush( + self._heap, + (time.monotonic() + handle._interval, next(self._seq), handle), + ) + self._cond.notify_all() + + def _run_callback(self, handle: ScheduledHandle) -> None: + stop = False + try: + stop = handle._fn() is False + except Exception: + logger.debug("periodic callback %r raised", handle._fn, exc_info=True) + finally: + with self._cond: + handle._runner = None + if stop: + handle._cancelled = True + elif not handle._cancelled: + heapq.heappush( + self._heap, + (time.monotonic() + handle._interval, next(self._seq), handle), + ) + self._cond.notify_all() def _run(self) -> None: while True: @@ -91,28 +147,13 @@ def _run(self) -> None: self._cond.wait(delay) continue heapq.heappop(self._heap) - self._running = handle + self._dispatch(handle) break - stop = False - try: - stop = handle._fn() is False - except Exception: - logger.debug("periodic callback %r raised", handle._fn, exc_info=True) - with self._cond: - self._running = None - if stop: - handle._cancelled = True - elif not handle._cancelled: - heapq.heappush( - self._heap, - (time.monotonic() + handle._interval, next(self._seq), handle), - ) - self._cond.notify_all() _DEFAULT = PeriodicScheduler() def schedule(fn: Callable[[], object], interval: float) -> ScheduledHandle: - """Run ``fn()`` every ``interval`` seconds on the shared scheduler thread.""" + """Run ``fn()`` every ``interval`` seconds via the shared scheduler.""" return _DEFAULT.schedule(fn, interval) diff --git a/agent/turn_facade_lease.py b/agent/turn_facade_lease.py index da33122f6f892..9ae9c93a02e84 100644 --- a/agent/turn_facade_lease.py +++ b/agent/turn_facade_lease.py @@ -4,7 +4,8 @@ resume, gateway, background delivery). ``admit_durable_turn_lease`` acquires the row lease (or returns the early result the façade must hand back); ``DurableTurnLease`` owns the periodic refresher, the turn-liveness watchdog wiring, and the lease-loss / stall interrupt plumbing. Both -timers run on the shared scheduler thread (``agent/periodic_scheduler.py``), not per-turn threads. +timers run via the shared scheduler (``agent/periodic_scheduler.py``; timer thread orders, +bodies run on per-handle workers), not per-turn threads. """ import logging import os @@ -172,7 +173,7 @@ def clear_interrupt(self) -> None: _set_interrupt(False, agent._execution_thread_id) def refresh_tick(self): - """One periodic renewal (every ``refresh_interval`` on the shared scheduler); a miss or + """One periodic renewal (every ``refresh_interval`` via the shared scheduler); a miss or error interrupts the turn. Returning False stops the timer. The holder-qualified UPDATE fences a late refresher from a successor lease. The façade's diff --git a/agent/turn_liveness.py b/agent/turn_liveness.py index 538ca2a583035..7b0562815dd05 100644 --- a/agent/turn_liveness.py +++ b/agent/turn_liveness.py @@ -86,8 +86,8 @@ def resolve_turn_liveness_settings( class TurnLivenessWatchdog: - """Sampled-idle watchdog bound to one conversation turn (polls on the - shared periodic scheduler thread). + """Sampled-idle watchdog bound to one conversation turn (via the shared + periodic scheduler; timer thread orders, body runs on its own worker). ``activity_lock`` must be the SAME lock ``AIAgent._touch_activity`` stamps the activity clock with; run_agent owns the lease state and callbacks. @@ -110,7 +110,7 @@ def __init__( self._deactivate_turn = deactivate_turn def schedule(self): - """Start polling on the shared periodic scheduler thread; returns the cancel handle. + """Start polling via the shared periodic scheduler; returns the cancel handle. Scheduled at turn entry, after the turn-active flag and activity clock are stamped.""" from agent.periodic_scheduler import schedule diff --git a/tests/agent/test_periodic_scheduler.py b/tests/agent/test_periodic_scheduler.py index e1a2b3765a550..883311838bd6a 100644 --- a/tests/agent/test_periodic_scheduler.py +++ b/tests/agent/test_periodic_scheduler.py @@ -1,4 +1,4 @@ -"""agent/periodic_scheduler: one shared thread runs every periodic timer.""" +"""agent/periodic_scheduler: one timer thread dispatches isolated callbacks.""" import threading import time @@ -24,7 +24,7 @@ def test_two_intervals_fire_proportionally_and_cancel_stops_one(): assert _wait_until(lambda: len(slow) >= 3) assert len(fast) > len(slow) # 5x interval ratio -> clearly more fast ticks - # Both ran on this scheduler's single thread, not on new threads. + # Scheduling another timer creates no persistent per-handle thread. before = threading.active_count() sched.schedule(lambda: None, 0.01).cancel() assert threading.active_count() == before @@ -90,3 +90,90 @@ def test_module_level_schedule_uses_shared_default(): assert threading.active_count() == before for handle in handles: handle.cancel() + + +def test_blocked_callback_does_not_stall_due_sibling(monkeypatch): + scheduler = PeriodicScheduler() + monkeypatch.setattr(periodic_scheduler, "_DEFAULT", scheduler) + blocker_entered = threading.Event() + release_blocker = threading.Event() + sibling_ran = threading.Event() + + def blocker(): + blocker_entered.set() + release_blocker.wait(2.0) + return False + + def sibling(): + sibling_ran.set() + return False + + blocker_handle = schedule(blocker, 0.01) + assert blocker_entered.wait(1.0) + sibling_handle = schedule(sibling, 0.01) + try: + assert sibling_ran.wait(0.30), ( + "a blocked periodic callback stalled an unrelated due callback" + ) + finally: + release_blocker.set() + blocker_handle.cancel(wait=1.0) + sibling_handle.cancel(wait=1.0) + + +def test_slow_callback_never_overlaps_itself(): + sched = PeriodicScheduler() + lock = threading.Lock() + two_runs = threading.Event() + running = 0 + peak_running = 0 + completed = 0 + + def slow(): + nonlocal running, peak_running, completed + with lock: + running += 1 + peak_running = max(peak_running, running) + time.sleep(0.05) + with lock: + running -= 1 + completed += 1 + if completed >= 2: + two_runs.set() + + handle = sched.schedule(slow, 0.01) + try: + assert two_runs.wait(2.0) + assert peak_running == 1, "a handle ran concurrently with itself" + finally: + handle.cancel(wait=2.0) + + +def test_worker_start_failure_keeps_timer(monkeypatch): + sched = PeriodicScheduler() + fired: list = [] + real_thread = threading.Thread + attempts = {"n": 0} + + def flaky(*args, **kwargs): + name = kwargs.get("name", "") + if name.startswith(periodic_scheduler._CALLBACK_THREAD_PREFIX) and attempts["n"] == 0: + attempts["n"] += 1 + + class Boom: + def start(self): + raise RuntimeError("no threads") + + return Boom() + attempts["n"] += 1 + return real_thread(*args, **kwargs) + + monkeypatch.setattr(periodic_scheduler.threading, "Thread", flaky) + handle = sched.schedule(lambda: fired.append(1), 0.01) + try: + assert _wait_until(lambda: bool(fired), timeout=3.0), ( + "worker-start failure silently retired the timer" + ) + assert not handle.cancelled + finally: + handle.cancel() diff --git a/tools/delegate_tool_child_run.py b/tools/delegate_tool_child_run.py index 6fc38b059f8a3..7d49038a8df4b 100644 --- a/tools/delegate_tool_child_run.py +++ b/tools/delegate_tool_child_run.py @@ -203,7 +203,7 @@ def _attr_line(attr): # ── Per-run helpers ────────────────────────────────────────────────────────── class _Heartbeat: - """One child's parent-activity heartbeat on the shared periodic scheduler thread + """One child's parent-activity heartbeat via the shared periodic scheduler (``agent.periodic_scheduler``) — not one daemon thread per child. NOT started at construction: the caller calls ``start()`` inside its ``try`` so a failed schedule (OS thread exhaustion on first use) leaves ``handle`` None and ``stop()`` is a no-op."""