fix(agent): real start/stop state machine — kill the stuck "stopping" pill
The Status pill hung on "stopping" forever (operator-flagged 2026-07-01). Root cause: the backend had no lifecycle state — status() only returned running/stopped — so the UI FABRICATED "stopping" in JS as `!running && active>0`. That pill only cleared when the backend's `active` counter hit 0, but stop() (a) blocked the HTTP handler on lease-release calls to curator and (b) left `active>0` whenever a consumer wedged mid-submit/release to an overloaded curator → "stopping" that never resolved. Give the backend a real, truthful state it drives itself: stopped → starting → running → stopping → stopped - start(): → starting; a downloader flips it to running on its FIRST successful lease (so "running" means curator is actually answering, not just "Start was clicked"). If curator's down it honestly stays "starting". - stop(): → stopping; returns immediately (no handler block). A background monitor waits for the worker threads to actually exit, releases leases, then → stopped — bounded by STOPPING_TIMEOUT (20s) so a wedged submit can NEVER hold the UI in "stopping" again. In-flight work is handed back safely. - Buttons follow the real state (Start only from stopped; both disabled through the transition), so you can't fight a transition. - Log every Start/Stop button press (routes) and every transition (worker), so the Logs panel shows exactly what each button did. Frontend now trusts s.state (drops the active>0 hack); VERSION → .8. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
+83
-13
@@ -133,6 +133,23 @@ STATS_INTERVAL = 30.0
|
||||
QUEUE_POLL_INTERVAL = 5.0
|
||||
UI_IDLE_GRACE = 20.0
|
||||
|
||||
# Stop backstop: pressing Stop signals every worker thread to wind down and hand
|
||||
# its lease back; the UI shows "stopping" until those threads have actually
|
||||
# exited, then "stopped" — so the state the operator sees is always truthful.
|
||||
# If a thread wedges (a stuck submit/release to an overloaded curator), NEVER
|
||||
# hang in "stopping": declare "stopped" after this many seconds and let the lone
|
||||
# straggler finish detached. This is the guarantee the old active>0 UI hack lacked.
|
||||
STOPPING_TIMEOUT = 20.0
|
||||
|
||||
# The lifecycle the operator sees. `_state` is the source of truth the UI reads;
|
||||
# `_running` is the low-level flag the worker THREADS gate on (True while
|
||||
# starting/running; False from the Stop press onward, so they wind down at once).
|
||||
# stopped → starting (spun up, not yet reached curator)
|
||||
# → running (leased at least once — curator is answering)
|
||||
# → stopping (Stop pressed; threads winding down + handing work back)
|
||||
# → stopped (threads exited, or the timeout backstop fired)
|
||||
STOPPED, STARTING, RUNNING, STOPPING = "stopped", "starting", "running", "stopping"
|
||||
|
||||
log = logging.getLogger("fc_agent.worker")
|
||||
|
||||
|
||||
@@ -147,6 +164,12 @@ class Worker:
|
||||
)
|
||||
self._lock = threading.Lock()
|
||||
self._running = False
|
||||
# The lifecycle state the UI shows (see the STOPPED/STARTING/... consts).
|
||||
# It stays STOPPING — a truthful "winding down" — from the Stop press until
|
||||
# the worker threads have actually exited (or the timeout backstop fires),
|
||||
# then STOPPED. `_running` flips False the instant Stop is pressed.
|
||||
self._state = STOPPED
|
||||
self._shutdown_thread: threading.Thread | None = None
|
||||
self._auto = bool(cfg.auto_scale) # autoscale the downloader count
|
||||
self._dl_target = max(1, min(DL_MAX, cfg.concurrency))
|
||||
self._consumer_target = 1 # GPU is fast — start with one
|
||||
@@ -311,42 +334,88 @@ class Worker:
|
||||
|
||||
# --- control -----------------------------------------------------------
|
||||
def start(self):
|
||||
# Only a fully-stopped worker can start — ignore a click while already
|
||||
# up or still winding down (the UI also disables the button then). This
|
||||
# keeps a Start during "stopping" from racing the shutdown monitor's
|
||||
# lease-release against a fresh pool's new leases.
|
||||
with self._lock:
|
||||
if self._state != STOPPED:
|
||||
return
|
||||
self._running = True
|
||||
self._state = STARTING # → RUNNING on the first successful lease
|
||||
self._dl_target = max(1, self._dl_target)
|
||||
self._consumer_target = max(1, self._consumer_target)
|
||||
self._reconcile_locked()
|
||||
log.info("▶ start → starting (spinning up, awaiting curator)")
|
||||
# (Re)start the autoscaler control loop.
|
||||
if self._ctrl_thread is None or not self._ctrl_thread.is_alive():
|
||||
self._ctrl_stop.clear()
|
||||
self._ctrl_thread = threading.Thread(target=self._control_loop, daemon=True)
|
||||
self._ctrl_thread.start()
|
||||
|
||||
def stop(self):
|
||||
# Flip the flag FIRST (atomic bool), before any lock, so /status and the
|
||||
# loops observe "stopped" immediately even if _lock is momentarily held —
|
||||
# the state can never lag behind the click.
|
||||
self._running = False
|
||||
self._ctrl_stop.set()
|
||||
def _mark_running(self) -> None:
|
||||
"""STARTING → RUNNING once a downloader has actually leased from curator —
|
||||
so 'running' means 'curator is answering', not just 'Start was clicked'.
|
||||
First caller wins; the rest are a cheap no-op."""
|
||||
if self._state != STARTING:
|
||||
return
|
||||
with self._lock:
|
||||
if self._state != STARTING:
|
||||
return
|
||||
self._state = RUNNING
|
||||
log.info("state → running (curator reachable — leasing)")
|
||||
|
||||
def stop(self):
|
||||
# Enter STOPPING and signal every worker thread to wind down. The handler
|
||||
# returns at once — the actual wait-for-exit + lease-release runs in a
|
||||
# background monitor, so a slow curator can NEVER block the Stop button or
|
||||
# wedge the state (the bug: the old UI faked "stopping" from active>0 and
|
||||
# sat there forever when a consumer hung mid-release).
|
||||
with self._lock:
|
||||
if self._state not in (STARTING, RUNNING):
|
||||
return # already stopping/stopped
|
||||
self._state = STOPPING
|
||||
self._running = False # every thread's _stopped() now trips → wind down
|
||||
dls, self._dls = self._dls, []
|
||||
cons, self._consumers = self._consumers, []
|
||||
self._active = 0 # no consumers left → the meter reads 0 at once;
|
||||
# any lagging decrement is clamped (see _bump)
|
||||
self._active = 0
|
||||
self._ctrl_stop.set()
|
||||
for _, ev in dls:
|
||||
ev.set()
|
||||
for _, ev in cons:
|
||||
ev.set()
|
||||
# Wake any consumer blocked on an empty buffer.
|
||||
# Wake any consumer blocked on an empty buffer so it sees the stop at once.
|
||||
for _ in range(CONSUMER_MAX):
|
||||
try:
|
||||
self._buffer.put_nowait(None)
|
||||
except queue.Full:
|
||||
break
|
||||
# Drain the buffer + release every still-held lease in one shot so orphaned
|
||||
# work is re-leased at once. A downloader/consumer mid-flight may also
|
||||
# release its own job — a duplicate release is a harmless no-op.
|
||||
log.info("■ stop → stopping (winding down, handing work back)")
|
||||
threads = [t for t, _ in dls] + [t for t, _ in cons]
|
||||
self._shutdown_thread = threading.Thread(
|
||||
target=self._finish_stop, args=(threads,), daemon=True)
|
||||
self._shutdown_thread.start()
|
||||
|
||||
def _finish_stop(self, threads: list[threading.Thread]) -> None:
|
||||
"""Wait for the signalled worker threads to actually exit, then land on
|
||||
STOPPED — so 'stopped' is truthful. Bounded by STOPPING_TIMEOUT so a
|
||||
wedged submit/release can never hold the UI in 'stopping' forever."""
|
||||
deadline = time.monotonic() + STOPPING_TIMEOUT
|
||||
for t in threads:
|
||||
t.join(timeout=max(0.0, deadline - time.monotonic()))
|
||||
# Hand back every lease still held in one shot (background — no handler
|
||||
# block); a straggler releasing its own job later is a harmless dup.
|
||||
self._drain_and_release()
|
||||
with self._lock:
|
||||
if self._state == STOPPING: # a Start may have re-entered; if so, leave it
|
||||
self._state = STOPPED
|
||||
self._active = 0 # any straggler's -1 is clamped (see _bump)
|
||||
stragglers = sum(1 for t in threads if t.is_alive())
|
||||
if stragglers:
|
||||
log.info("state → stopped (%d thread(s) still winding down past %ds — detached)",
|
||||
stragglers, int(STOPPING_TIMEOUT))
|
||||
else:
|
||||
log.info("state → stopped (all work handed back)")
|
||||
|
||||
def _drain_and_release(self) -> None:
|
||||
while True:
|
||||
@@ -419,7 +488,7 @@ class Worker:
|
||||
# under the GIL) and this backs the UI poll — it must NEVER be able to
|
||||
# block behind a thread holding _lock, or the whole status view freezes.
|
||||
return {
|
||||
"state": "running" if self._running else "stopped",
|
||||
"state": self._state, # stopped|starting|running|stopping (truthful)
|
||||
"concurrency": self._dl_target, # the UI dial = downloader count
|
||||
"max_concurrency": DL_MAX,
|
||||
"auto": self._auto,
|
||||
@@ -456,6 +525,7 @@ class Worker:
|
||||
with self._timed("lease"):
|
||||
jobs = self.client.lease(self.cfg.batch_size)
|
||||
backoff = self.cfg.poll_idle_seconds # server answered → reset
|
||||
self._mark_running() # curator answered → leave "starting"
|
||||
except Exception:
|
||||
# curator unreachable (redeploy, network drop): wait it out with
|
||||
# exponential backoff, capped — resume on our own when it returns.
|
||||
|
||||
Reference in New Issue
Block a user