stroblmeandClaude Opus 5
da528340a9
Measured with `make bench-engine` against a real Redis: 103.6 -> 164.4 messages a second on a five-node chain (p50 latency 2125 -> 1171 ms) and 34.8 -> 63.2 on a fan-out of twenty. Against the memory backend, which is what a pip install runs on, 262 -> 626. The two that bought most of it: - `StateBackend.record` puts a published value, its timestamp, its series and its version counter in one round trip. They were four calls building four pipelines, and a value crossing an edge pays them twice. A released rate-limit hold rides along instead of a DEL per port. - the readiness check reads a node's inputs and hands them to the node, rather than reading the triggering ones to count them and having the node read the same keys again a moment later. `apply_outputs` was a second copy of `_record_outputs` and is now the same code plus the event that distinguishes it. The rest, each small: - `_derive` builds a node-by-id map and a `consumes` index, so dispatching an item and publishing a value stop scanning every node in the installation. - `read_all` is memoised against the store revision — it sits on the publish path, so a dashboard slider was reading and validating every flow file per value. Same mechanism `_wiring` already uses. - the `message_value` source block is built once per node instead of per emission. - both timer threads ask the queue to promote only when something is actually due, which takes an idle engine from ~4 Redis round trips a second to one. - the shared httpx client is bounded (32 connections, one retry); its default pool is 100 with no per-host cap, so one slow endpoint could take it and every other sender node with it. - the MQTT and delay nodes no longer log a line per message at INFO. Robustness, in the same pass: - `MemoryWorkQueue._done` was a set nothing ever removed from — one entry per non-idempotent node per item, for the life of the process, in the default configuration. Capped, the way the Redis side expires its markers. - a saturated engine can claim from the due lane past the cascade limit. The capacity gate sits in front of the claim, so the due lane's priority — decided inside it — did not apply while every slot was held: a motor's stop was not behind the long nodes, it was unread. Only after a slot has genuinely failed to free for half a second, and briefly, so the backlog is not starved in turn. - `reclaim_stale` dispatches through that same gate. It could return sixty entries and push in-flight far past the limit the gate exists to hold. - a flow's nodes are stopped together rather than one after another. Each gets `NODE_STOP_TIMEOUT`, so a flow whose broker was unreachable took five seconds per node — long enough to outlast `REBUILD_WAIT` and 503 the deploy. - the worker pool and the HTTP client are closed on a thread, not on the event loop, and a run closes the state backend it built (on Redis, a client and a connection pool per run). - the five background tasks say something when they die. Each catches exceptions inside its loop, so one raised anywhere else left the engine serving with no metrics, no alerts or no artifact sweep, silently. `tests/flow/test_round_trips.py` counts the state operations one message costs — four, where it was about eleven — because none of the above would fail a behavioural test if it were undone. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01M6hPWS6YEbT1P8LxhhFb2T
545 lines
22 KiB
Python
545 lines
22 KiB
Python
"""The execution service: what turns journaled work into node runs.
|
|
|
|
A consumer thread claims items from the work queue and hands each one to a
|
|
dispatch pool, which drives the wave it starts. It claims only what that pool
|
|
can start, so a backlog waits in the queue rather than inside the process.
|
|
Node bodies run on a second, separate pool: if cascade drivers and node bodies
|
|
shared one, a wave waiting for its own nodes could occupy every thread and
|
|
deadlock.
|
|
|
|
A reaper takes back items claimed by an engine that died before acknowledging
|
|
them, which is the mechanism that makes a crash mid-cascade recoverable rather
|
|
than lossy.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import threading
|
|
import time
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from typing import TYPE_CHECKING, Any
|
|
|
|
from fluksio.flow.queue import MAX_DELIVERIES, WorkItem, WorkQueue
|
|
|
|
if TYPE_CHECKING:
|
|
from fluksio.flow.events import EventBus
|
|
from fluksio.flow.pipeline import Pipeline
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
CLAIM_BLOCK_MS = 1000
|
|
# Long enough that a busy cascade is not mistaken for a dead one.
|
|
RECLAIM_IDLE_MS = 60_000
|
|
RECLAIM_INTERVAL_S = 30.0
|
|
# How often to tell the queue that what we hold is still being worked on. A
|
|
# node may run for as long as it likes, so what marks an item abandoned is this
|
|
# stopping — which is what an engine that died does.
|
|
TOUCH_INTERVAL_S = 20.0
|
|
# The longest the timer thread sleeps with nothing due. It is a housekeeping
|
|
# cadence and a backstop for a deadline written by another process, not the
|
|
# resolution of a delay: a delayed item is waited for exactly, so what a timer
|
|
# fires late by is a wake-up and a promotion rather than up to a whole second.
|
|
DELAYED_INTERVAL_S = 1.0
|
|
# Cascade slots a promoted timer may use past `max_cascades`. A due item was
|
|
# already waited for, so making it queue behind whatever long node happens to
|
|
# hold the pool is the one lateness the sleeping timer thread cannot remove.
|
|
DUE_RESERVE = 2
|
|
# How long the saturated engine waits on the due lane before going back to
|
|
# check whether a cascade slot has freed.
|
|
DUE_CLAIM_BLOCK_MS = 200
|
|
#: How many cascades may be in flight, unless the service is given a number.
|
|
#: Sustained throughput is this over the mean cascade time, so an installation
|
|
#: whose nodes wait on a network rather than a CPU may want more of them —
|
|
#: `FLOW_MAX_CASCADES` is where that is said.
|
|
MAX_CASCADES = 4
|
|
#: Node threads, unless the service is given a number. Both this and the one
|
|
#: above are taken as written: only ``None`` means "nobody said", so a number
|
|
#: that reached here is one somebody chose, and an unusable one is the pool's
|
|
#: ``ValueError`` rather than a silent 4.
|
|
MAX_WORKERS = 4
|
|
# How long a reload waits for claimed work to finish before rebuilding anyway.
|
|
DRAIN_TIMEOUT_S = 10.0
|
|
# Work waiting in the stream, undelivered. A burst is normal — the pool claims
|
|
# only what it can start — so what marks an engine as falling behind is the
|
|
# backlog staying up across several checks rather than any one reading.
|
|
BACKLOG_INTERVAL_S = 5.0
|
|
BACKLOG_DEGRADED = 50
|
|
BACKLOG_STRIKES = 3
|
|
|
|
|
|
class ExecutionService:
|
|
"""Owns the engine's worker threads and the queue they read from."""
|
|
|
|
def __init__(
|
|
self,
|
|
queue: WorkQueue,
|
|
max_workers: int | None = None,
|
|
events: EventBus | None = None,
|
|
max_cascades: int | None = None,
|
|
) -> None:
|
|
self.queue = queue
|
|
self._events = events
|
|
self.max_cascades = MAX_CASCADES if max_cascades is None else max_cascades
|
|
self._pipeline: Pipeline | None = None
|
|
self._stop = threading.Event()
|
|
# Set when a deadline moves closer, so the timer thread stops waiting
|
|
# on the one it read and goes back for the new one.
|
|
self._timer_wake = threading.Event()
|
|
queue.on_delayed = self._timer_wake.set
|
|
self._intake = threading.Event()
|
|
self._intake.set()
|
|
self._inflight = 0
|
|
self._inflight_lock = threading.Condition()
|
|
# Entry ids claimed and still running, under _inflight_lock.
|
|
self._active: set[str] = set()
|
|
self.node_pool = ThreadPoolExecutor(
|
|
max_workers=MAX_WORKERS if max_workers is None else max_workers,
|
|
thread_name_prefix="node",
|
|
)
|
|
self._cascade_pool = ThreadPoolExecutor(
|
|
max_workers=self.max_cascades, thread_name_prefix="cascade"
|
|
)
|
|
self._consumer: threading.Thread | None = None
|
|
self._timers: threading.Thread | None = None
|
|
# Consecutive backlog readings over the threshold, and whether the last
|
|
# of them said so out loud.
|
|
self._backlog_strikes = 0
|
|
self.behind = False
|
|
|
|
# -------------------------------------------------------------------------
|
|
# Lifecycle
|
|
# -------------------------------------------------------------------------
|
|
|
|
def start(self) -> None:
|
|
if self._consumer is not None:
|
|
return
|
|
self._consumer = threading.Thread(
|
|
target=self._consume, name="queue-consumer", daemon=True
|
|
)
|
|
self._consumer.start()
|
|
self._timers = threading.Thread(
|
|
target=self._tick, name="queue-timers", daemon=True
|
|
)
|
|
self._timers.start()
|
|
|
|
def stop(self) -> None:
|
|
self._stop.set()
|
|
# The timer thread sleeps on this, not on _stop.
|
|
self._timer_wake.set()
|
|
for thread in (self._consumer, self._timers):
|
|
if thread is not None:
|
|
thread.join(timeout=5)
|
|
self._consumer = None
|
|
self._timers = None
|
|
self._cascade_pool.shutdown(wait=False)
|
|
self.node_pool.shutdown(wait=False)
|
|
self.queue.close()
|
|
|
|
def bind(self, pipeline: Pipeline) -> None:
|
|
"""Point the service at the pipeline it should execute against."""
|
|
self._pipeline = pipeline
|
|
|
|
def pause_intake(self) -> None:
|
|
"""Stop claiming, and wait for what is already claimed to finish.
|
|
|
|
Called around a rebuild: items claimed against the old pipeline should
|
|
finish there rather than half-run against the new one.
|
|
"""
|
|
self._intake.clear()
|
|
deadline = time.monotonic() + DRAIN_TIMEOUT_S
|
|
with self._inflight_lock:
|
|
while self._inflight:
|
|
remaining = deadline - time.monotonic()
|
|
if remaining <= 0:
|
|
logger.warning(
|
|
"Rebuild did not wait out %d cascades", self._inflight
|
|
)
|
|
return
|
|
self._inflight_lock.wait(remaining)
|
|
|
|
def resume_intake(self) -> None:
|
|
self._intake.set()
|
|
|
|
def alive(self) -> bool:
|
|
return self._consumer is not None and self._consumer.is_alive()
|
|
|
|
@property
|
|
def inflight(self) -> int:
|
|
"""Cascades claimed and still running."""
|
|
return self._inflight
|
|
|
|
# -------------------------------------------------------------------------
|
|
# Threads
|
|
# -------------------------------------------------------------------------
|
|
|
|
def _consume(self) -> None:
|
|
failures = 0
|
|
while not self._stop.is_set():
|
|
if not self._intake.is_set():
|
|
self._intake.wait(timeout=0.5)
|
|
continue
|
|
free, due_only = self._await_capacity()
|
|
if not free:
|
|
continue
|
|
try:
|
|
items = self.queue.claim(
|
|
free,
|
|
# Briefly, in the due-only case: this is the saturated
|
|
# engine, and a slot freeing has to be noticed promptly.
|
|
DUE_CLAIM_BLOCK_MS if due_only else CLAIM_BLOCK_MS,
|
|
due_only,
|
|
)
|
|
failures = 0
|
|
except Exception as exc:
|
|
failures += 1
|
|
logger.error("Could not claim work: %s", exc)
|
|
self._publish_unavailable(exc)
|
|
# Backing off hard: a queue that is down stays down for a while.
|
|
self._stop.wait(min(30.0, 2.0**failures))
|
|
continue
|
|
|
|
for item in items:
|
|
self._dispatch(item)
|
|
|
|
def _sleep_until_due(self) -> bool:
|
|
"""Wait for the soonest deadline, the housekeeping cap, or a new one.
|
|
|
|
A fixed poll here made every delayed item late by 0-1000ms whatever the
|
|
load — on a rollershutter driven for a measured 26 seconds, 2-4% of its
|
|
travel every time, accumulating in the position its node believes it is
|
|
at. Sleeping to the deadline instead leaves a wake-up and a promotion,
|
|
which is milliseconds.
|
|
"""
|
|
self._timer_wake.clear()
|
|
try:
|
|
# Read after the clear: a deadline arriving in between sets the
|
|
# event again, so the wait below returns immediately rather than
|
|
# sleeping through work that landed in the gap.
|
|
due = self.queue.next_due()
|
|
except Exception as exc:
|
|
logger.error("Could not read the next deadline: %s", exc)
|
|
due = None
|
|
wait = DELAYED_INTERVAL_S if due is None else due - time.time()
|
|
self._timer_wake.wait(min(max(wait, 0.0), DELAYED_INTERVAL_S))
|
|
# What the caller promotes for: the deadline this woke for has passed,
|
|
# or the read failed and it should look anyway. An idle engine reads
|
|
# `next_due` once a second and asks for nothing.
|
|
return due is None or due <= time.time()
|
|
|
|
def _tick(self) -> None:
|
|
"""Promote delayed items, and take back what a dead engine dropped."""
|
|
last_reclaim = 0.0
|
|
last_touch = 0.0
|
|
last_backlog = 0.0
|
|
while not self._stop.is_set():
|
|
promote = self._sleep_until_due()
|
|
if self._stop.is_set():
|
|
break
|
|
try:
|
|
if promote:
|
|
self.queue.move_due(time.time())
|
|
except Exception as exc:
|
|
logger.error("Could not promote delayed work: %s", exc)
|
|
# The item is still due, so the wait above would be zero and
|
|
# this would spin on a queue that is down. Back off to what a
|
|
# fixed poll used to cost.
|
|
self._stop.wait(DELAYED_INTERVAL_S)
|
|
|
|
now = time.monotonic()
|
|
if now - last_backlog >= BACKLOG_INTERVAL_S:
|
|
last_backlog = now
|
|
try:
|
|
self._check_backlog()
|
|
except Exception as exc:
|
|
logger.error("Could not read the queue backlog: %s", exc)
|
|
|
|
if now - last_touch >= TOUCH_INTERVAL_S:
|
|
last_touch = now
|
|
with self._inflight_lock:
|
|
running = list(self._active)
|
|
try:
|
|
self.queue.touch(running)
|
|
except Exception as exc:
|
|
logger.error("Could not touch claimed work: %s", exc)
|
|
|
|
if now - last_reclaim < RECLAIM_INTERVAL_S:
|
|
continue
|
|
last_reclaim = now
|
|
try:
|
|
for item in self.queue.reclaim_stale(RECLAIM_IDLE_MS):
|
|
logger.info(
|
|
"Reclaimed work for '%s' (delivery %d)",
|
|
item.node,
|
|
item.deliveries,
|
|
)
|
|
# Through the same gate the main loop uses: a reclaim can
|
|
# return sixty-odd entries at once, and dispatching them
|
|
# all would push `_inflight` far past `max_cascades` —
|
|
# exactly the overcommit the gate exists to prevent.
|
|
if not self._await_capacity()[0]:
|
|
break
|
|
self._dispatch(item)
|
|
except Exception as exc:
|
|
logger.error("Could not reclaim stale work: %s", exc)
|
|
|
|
def _check_backlog(self) -> None:
|
|
"""Say so when work has been waiting in the stream for a while.
|
|
|
|
A flow enqueuing faster than the pool drains produces no event of its
|
|
own: the backlog simply grows, every timer and connector poll drifts
|
|
behind it, and nothing on the health screen moves. This is that event.
|
|
The flow named is the one most of the waiting work belongs to, which is
|
|
the half somebody can act on.
|
|
"""
|
|
backlog = self.queue.backlog()
|
|
if backlog < BACKLOG_DEGRADED:
|
|
self._backlog_strikes = 0
|
|
self.behind = False
|
|
return
|
|
|
|
self._backlog_strikes += 1
|
|
if self._backlog_strikes < BACKLOG_STRIKES or self.behind:
|
|
return
|
|
|
|
self.behind = True
|
|
flows = self.queue.backlog_flows()
|
|
worst = max(flows, key=lambda f: flows[f], default="")
|
|
logger.warning("engine behind: %d items waiting (%s)", backlog, worst or "?")
|
|
self._publish(
|
|
{
|
|
"type": "engine_degraded",
|
|
"reason": f"{backlog} items waiting in the queue",
|
|
"flow": worst,
|
|
"ts": time.time(),
|
|
}
|
|
)
|
|
|
|
def _await_capacity(self) -> tuple[int, bool]:
|
|
"""How many cascades may be claimed now, and whether only due ones.
|
|
|
|
Claiming past what the pool can run makes nothing faster: the extra
|
|
items queue up inside the pool, count as in flight and hold their
|
|
journal entries open the whole time, which is how four cascade threads
|
|
came to report hundreds busy on a healthy engine. Work left in the
|
|
stream is work that is still anyone's to take; work that is claimed is
|
|
work that is actually being run.
|
|
|
|
The gate sat in front of the claim, though, and the due lane's
|
|
priority is decided *inside* it — so with every slot held by a long
|
|
node, a motor's stop was not merely behind them, it was unread. Past
|
|
the limit this therefore keeps claiming, from the due lane alone:
|
|
a promoted timer is work that was already waited for, and there are
|
|
only ever as many of them as there are deadlines.
|
|
"""
|
|
with self._inflight_lock:
|
|
while not self._stop.is_set():
|
|
free = self.max_cascades - self._inflight
|
|
if free > 0:
|
|
return free, False
|
|
# Only once a slot has genuinely failed to free: the due lane
|
|
# is usually empty, and going to look at it ahead of waiting
|
|
# would leave the backlog unclaimed for the length of that
|
|
# read every time the pool filled up.
|
|
if self._inflight_lock.wait(0.5):
|
|
continue
|
|
reserve = self.max_cascades + DUE_RESERVE - self._inflight
|
|
if reserve > 0:
|
|
return reserve, True
|
|
return 0, False
|
|
|
|
def _dispatch(self, item: WorkItem) -> None:
|
|
with self._inflight_lock:
|
|
self._inflight += 1
|
|
if item.entry_id:
|
|
self._active.add(item.entry_id)
|
|
try:
|
|
self._cascade_pool.submit(self._handle, item)
|
|
except RuntimeError:
|
|
# Pool already shutting down.
|
|
self._done(item)
|
|
|
|
def _done(self, item: WorkItem) -> None:
|
|
with self._inflight_lock:
|
|
self._inflight -= 1
|
|
self._active.discard(item.entry_id)
|
|
self._inflight_lock.notify_all()
|
|
|
|
# -------------------------------------------------------------------------
|
|
# Handling one item
|
|
# -------------------------------------------------------------------------
|
|
|
|
def step(self, flow: str) -> str | None:
|
|
"""Run one item a pause is holding, and hold everything else still.
|
|
|
|
Blocking, so the caller sees the wave finish. Returns the node the item
|
|
came from, or None when nothing is parked for this flow.
|
|
"""
|
|
pipeline = self._pipeline
|
|
if pipeline is None:
|
|
return None
|
|
item = self.queue.unpark_one(flow)
|
|
if item is None:
|
|
return None
|
|
with pipeline.stepping(flow):
|
|
try:
|
|
self._run_item(item)
|
|
except Exception:
|
|
logger.exception("Step of '%s' failed", item.node)
|
|
return item.node
|
|
|
|
def _handle(self, item: WorkItem) -> None:
|
|
handled = True
|
|
try:
|
|
handled = self._run_item(item)
|
|
except Exception:
|
|
logger.exception("Work item for '%s' failed", item.node)
|
|
finally:
|
|
# Leaving it unacknowledged is how it comes back: the reaper hands
|
|
# it to whoever can actually run it.
|
|
if handled:
|
|
try:
|
|
self.queue.ack(item)
|
|
except Exception as exc:
|
|
logger.error(
|
|
"Could not acknowledge work for '%s': %s", item.node, exc
|
|
)
|
|
self._done(item)
|
|
|
|
def _run_item(self, item: WorkItem) -> bool:
|
|
"""Run one item. False means it was not handled and must come back."""
|
|
pipeline = self._pipeline
|
|
if pipeline is None:
|
|
logger.warning("No pipeline bound; leaving work for '%s'", item.node)
|
|
return False
|
|
|
|
if item.deliveries > MAX_DELIVERIES:
|
|
self.queue.dead_letter(item, f"{item.deliveries} deliveries")
|
|
self._publish(
|
|
{
|
|
"type": "cascade_dropped",
|
|
"flow": item.flow,
|
|
"node": item.node,
|
|
"deliveries": item.deliveries,
|
|
"ts": time.time(),
|
|
}
|
|
)
|
|
return True
|
|
|
|
node = pipeline.get_node_by_id(item.node)
|
|
if node is None:
|
|
# The flow was edited while this was queued; its values are already
|
|
# in state, so there is nothing to salvage.
|
|
logger.debug("Work item for unknown node '%s', dropped", item.node)
|
|
return True
|
|
|
|
if pipeline.is_disabled(node.flow):
|
|
return True
|
|
|
|
if pipeline.is_paused(node.flow) and not pipeline.is_stepping(node.flow):
|
|
self.queue.park(node.flow, item)
|
|
return True
|
|
|
|
if item.kind == "flush":
|
|
# A rate-limit window ended; nothing to replay, only to let out.
|
|
# ponytail: no run record for a flush — it is the tail of the run
|
|
# that scheduled it, not a run of its own.
|
|
pipeline.flush(node)
|
|
return True
|
|
|
|
if item.guard_key and str(node.recall(item.guard_key, "")) != item.guard_value:
|
|
# The node moved on while this waited — a restarted timer, say.
|
|
logger.debug("Guard no longer holds for '%s', dropped", item.node)
|
|
return True
|
|
|
|
# Only here is the item certain to run, which is what a run record is.
|
|
now = time.time()
|
|
self._publish(
|
|
{
|
|
"type": "cascade_started",
|
|
"run": item.entry_id,
|
|
"flow": item.flow,
|
|
"node": item.node,
|
|
"cause": item.cause,
|
|
"deliveries": item.deliveries,
|
|
"ts": now,
|
|
}
|
|
)
|
|
if item.deliveries == 1:
|
|
# A redelivery waited for the reaper, not for the engine.
|
|
self._publish(
|
|
{
|
|
"type": "work_latency",
|
|
"flow": item.flow,
|
|
"node": item.node,
|
|
"lag_ms": max(
|
|
0.0,
|
|
(now - max(item.enqueued_at, item.not_before)) * 1000,
|
|
),
|
|
"ts": now,
|
|
}
|
|
)
|
|
|
|
replay = item.deliveries > 1
|
|
emission = item.kind == "emission"
|
|
try:
|
|
# An emission's values went into state when the node produced them;
|
|
# this item carries them so its readers get the chunk that caused
|
|
# the wave rather than whichever is newest by the time they run.
|
|
# Applying them again would let a mid-node emission overwrite what
|
|
# the node returned at the end.
|
|
published = (
|
|
set(item.outputs)
|
|
if emission
|
|
else pipeline.apply_outputs(node, item.outputs or None)
|
|
)
|
|
pipeline.run_downstream(
|
|
node,
|
|
entry_id=item.entry_id,
|
|
replay=replay,
|
|
# A redelivery has to finish a walk that may be half done, and
|
|
# an item with no payload is the value already being in state.
|
|
# Neither can say what changed, so neither filters on it.
|
|
changed=None if replay or not item.outputs else published,
|
|
overrides=item.outputs if emission else None,
|
|
)
|
|
finally:
|
|
# Paired, or a cascade that raised — state backend gone, say — is a
|
|
# run left open until the abandoned sweep ten minutes later.
|
|
self._publish(
|
|
{
|
|
"type": "cascade_finished",
|
|
"run": item.entry_id,
|
|
"flow": item.flow,
|
|
"ts": time.time(),
|
|
}
|
|
)
|
|
return True
|
|
|
|
# -------------------------------------------------------------------------
|
|
# Reporting
|
|
# -------------------------------------------------------------------------
|
|
|
|
def stats(self) -> dict[str, Any]:
|
|
try:
|
|
stats = self.queue.stats()
|
|
except Exception as exc:
|
|
return {"error": str(exc), "consumer_alive": self.alive()}
|
|
stats["consumer_alive"] = self.alive()
|
|
stats["cascades_busy"] = self._inflight
|
|
stats["behind"] = self.behind
|
|
return stats
|
|
|
|
def _publish_unavailable(self, exc: Exception) -> None:
|
|
self._publish(
|
|
{
|
|
"type": "queue_unavailable",
|
|
"error": f"{type(exc).__name__}: {exc}",
|
|
"ts": time.time(),
|
|
}
|
|
)
|
|
|
|
def _publish(self, event: dict[str, Any]) -> None:
|
|
if self._events is not None:
|
|
self._events.publish(event)
|