From da528340a90ddcd8485c16bb195c9b1b1b4060a8 Mon Sep 17 00:00:00 2001 From: stroblme Date: Sat, 29 Aug 2026 19:58:39 +0200 Subject: [PATCH] Cut the round trips a message costs the engine MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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) Claude-Session: https://claude.ai/code/session_01M6hPWS6YEbT1P8LxhhFb2T --- backend/fluksio/flow/controller.py | 24 ++- backend/fluksio/flow/executor.py | 62 ++++++-- backend/fluksio/flow/nodes/delay.py | 10 +- backend/fluksio/flow/nodes/http.py | 23 ++- backend/fluksio/flow/nodes/mqtt.py | 5 +- backend/fluksio/flow/pipeline.py | 197 ++++++++++++++++--------- backend/fluksio/flow/queue.py | 68 ++++++--- backend/fluksio/flow/runs.py | 21 ++- backend/fluksio/flow/state.py | 111 ++++++++++++++ backend/fluksio/flow/store.py | 32 ++-- backend/fluksio/main.py | 52 +++++-- backend/tests/flow/test_queue.py | 42 ++++++ backend/tests/flow/test_round_trips.py | 126 ++++++++++++++++ 13 files changed, 639 insertions(+), 134 deletions(-) create mode 100644 backend/tests/flow/test_round_trips.py diff --git a/backend/fluksio/flow/controller.py b/backend/fluksio/flow/controller.py index 1ffce99..6f43120 100644 --- a/backend/fluksio/flow/controller.py +++ b/backend/fluksio/flow/controller.py @@ -825,13 +825,19 @@ class FlowController: return merged async def _teardown(self, flow: str | None = None) -> None: - """Stop everything the previous pipeline started, or one flow's share.""" - for entry in self.loaded.values(): + """Stop everything the previous pipeline started, or one flow's share. + + Together rather than one after another: each node is given + `NODE_STOP_TIMEOUT`, so in sequence a flow whose broker is unreachable + took that many seconds *per node* — long enough for a rebuild to + outlast `REBUILD_WAIT` and answer 503 on every deploy. Concurrently it + is five seconds flat however many nodes there are. + """ + + async def stop(entry: LoadedNode) -> None: node = entry.node if node is None: - continue - if flow is not None and entry.flow != flow: - continue + return try: await asyncio.wait_for(node.stop(self.app), NODE_STOP_TIMEOUT) except TimeoutError: @@ -844,6 +850,14 @@ class FlowController: ) except Exception: logger.exception("Error stopping node '%s'", entry.id) + + await asyncio.gather( + *( + stop(entry) + for entry in list(self.loaded.values()) + if flow is None or entry.flow == flow + ) + ) # After the nodes, so a loop still winding down is not restarted. if flow is None: await self.supervisor.cancel_all() diff --git a/backend/fluksio/flow/executor.py b/backend/fluksio/flow/executor.py index cbbef7a..ab9a2dd 100644 --- a/backend/fluksio/flow/executor.py +++ b/backend/fluksio/flow/executor.py @@ -41,6 +41,13 @@ TOUCH_INTERVAL_S = 20.0 # 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 — @@ -172,11 +179,17 @@ class ExecutionService: if not self._intake.is_set(): self._intake.wait(timeout=0.5) continue - free = self._await_capacity() + free, due_only = self._await_capacity() if not free: continue try: - items = self.queue.claim(free, CLAIM_BLOCK_MS) + 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 @@ -189,7 +202,7 @@ class ExecutionService: for item in items: self._dispatch(item) - def _sleep_until_due(self) -> None: + 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 @@ -209,6 +222,10 @@ class ExecutionService: 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.""" @@ -216,11 +233,12 @@ class ExecutionService: last_touch = 0.0 last_backlog = 0.0 while not self._stop.is_set(): - self._sleep_until_due() + promote = self._sleep_until_due() if self._stop.is_set(): break try: - self.queue.move_due(time.time()) + 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 @@ -255,6 +273,12 @@ class ExecutionService: 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) @@ -291,8 +315,8 @@ class ExecutionService: } ) - def _await_capacity(self) -> int: - """How many cascades may be claimed now. Zero means the service stops. + 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 @@ -300,11 +324,29 @@ class ExecutionService: 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 self._inflight >= self.max_cascades and not self._stop.is_set(): - self._inflight_lock.wait(0.5) - return 0 if self._stop.is_set() else self.max_cascades - self._inflight + 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: diff --git a/backend/fluksio/flow/nodes/delay.py b/backend/fluksio/flow/nodes/delay.py index 0cc1cf3..42871ec 100644 --- a/backend/fluksio/flow/nodes/delay.py +++ b/backend/fluksio/flow/nodes/delay.py @@ -132,7 +132,7 @@ class DelayNode(Node): def _f(self, params: dict[str, Any], **kwargs: Any) -> dict[str, Any] | None: """Forward messages with optional delay, rate-limiting, and alarm.""" - logger.info("[%s] Received %s", self.name, kwargs) + logger.debug("[%s] Received %s", self.name, kwargs) # Store last input for cron use if kwargs: @@ -142,7 +142,7 @@ class DelayNode(Node): if self.interval > 0: ts = time.time() if ts <= self.ts + self.interval: - logger.info("[%s] Stashing %s", self.name, kwargs) + logger.debug("[%s] Stashing %s", self.name, kwargs) return None self.ts = ts @@ -162,11 +162,13 @@ class DelayNode(Node): if self._pipeline is not None and self._pipeline.defer( self, self._to_messages(output) or {}, self.delay ): - logger.info("[%s] Sending %s in %ss", self.name, output, self.delay) + logger.debug( + "[%s] Sending %s in %ss", self.name, output, self.delay + ) return None time.sleep(self.delay) - logger.info("[%s] Sending %s", self.name, output) + logger.debug("[%s] Sending %s", self.name, output) return output # ----------------------------------------------------------------- diff --git a/backend/fluksio/flow/nodes/http.py b/backend/fluksio/flow/nodes/http.py index b5879b2..7a4e19c 100644 --- a/backend/fluksio/flow/nodes/http.py +++ b/backend/fluksio/flow/nodes/http.py @@ -24,16 +24,35 @@ logger = logging.getLogger(__name__) # One pooled client for every sender node: connections are the expensive part # of an HTTP request, and a node that fires every second should keep its own. +# The ceiling is process-wide rather than per node, which is why it is a +# constant here rather than a `Params` field on the node. +HTTP_POOL_LIMIT = 32 _client: httpx.Client | None = None _client_lock = threading.Lock() def shared_client() -> httpx.Client: - """The process-wide HTTP client, built on first use.""" + """The process-wide HTTP client, built on first use. + + Bounded on purpose: httpx's default pool is 100 connections with no + per-host cap, so one endpoint that stops answering could take the whole + pool and every other sender node with it. Read outside the lock once it + exists — this is on the per-request path, and rebinding the global is + what the lock is for. + """ global _client + if _client is not None: + return _client with _client_lock: if _client is None: - _client = httpx.Client() + _client = httpx.Client( + limits=httpx.Limits( + max_connections=HTTP_POOL_LIMIT, + max_keepalive_connections=HTTP_POOL_LIMIT // 2, + keepalive_expiry=30.0, + ), + transport=httpx.HTTPTransport(retries=1), + ) return _client diff --git a/backend/fluksio/flow/nodes/mqtt.py b/backend/fluksio/flow/nodes/mqtt.py index 80ffaef..b6763ea 100644 --- a/backend/fluksio/flow/nodes/mqtt.py +++ b/backend/fluksio/flow/nodes/mqtt.py @@ -613,7 +613,10 @@ class MqttNode(Node): payload = message.payload.decode("utf-8") incoming_topic = str(message.topic) - logger.info( + # Debug, not info: this is one formatted line and + # one write per message the broker sends, and the + # payload goes to the server log verbatim. + logger.debug( "[%s] Received on %s: %s", self.name, incoming_topic, diff --git a/backend/fluksio/flow/pipeline.py b/backend/fluksio/flow/pipeline.py index a2c82d8..b356816 100644 --- a/backend/fluksio/flow/pipeline.py +++ b/backend/fluksio/flow/pipeline.py @@ -23,6 +23,7 @@ from collections import deque from collections.abc import Callable, Iterator from concurrent.futures import Future, ThreadPoolExecutor, wait from contextlib import contextmanager +from functools import lru_cache from typing import Any, Literal, Protocol from pydantic import BaseModel, computed_field @@ -98,6 +99,17 @@ def node_source(node: Node) -> ValueSource: return ValueSource(kind="node", id=node.id, label=node.local_id) +@lru_cache(maxsize=2048) +def _node_source_dump(node_id: str, local_id: str) -> dict[str, Any]: + """The `source` block of a `message_value`, built once per node. + + It is a pure function of the two names, and it was a pydantic model + constructed and dumped per emission — the same cost the live path already + refuses to pay for `NodeOutcome`. + """ + return ValueSource(kind="node", id=node_id, label=local_id).model_dump() + + class NodeOutcome(BaseModel): """How one node execution went. @@ -184,19 +196,33 @@ def run_cache_key(fingerprint: str, inputs: dict[str, Any], flow: str = "") -> s def _derive( nodes: list[Node], -) -> tuple[dict[str, list[Node]], dict[Node, frozenset[Node]]]: +) -> tuple[ + dict[str, list[Node]], + dict[Node, frozenset[Node]], + dict[str, list[Node]], + dict[str, Node], +]: """Work out the wiring the node list implies: producers, then dependencies. Done over the whole list rather than one flow's share of it, because a flow is not a subgraph — its nodes can read and write messages another flow owns — so there is no deriving one flow's edges on their own. + + The two lookups come out of the same walk: who consumes a message, and + which node an id names. Both were linear scans over every node in the + installation, on the per-message path. """ # A message may have several producers; every one of them is upstream # of the nodes consuming it. produces: dict[str, list[Node]] = {} + consumes: dict[str, list[Node]] = {} + by_id: dict[str, Node] = {} for node in nodes: for msg in node.provides: produces.setdefault(msg, []).append(node) + for msg in node.requires: + consumes.setdefault(msg, []).append(node) + by_id[node.id] = node # A node never depends on itself: reading a message it also provides is # how state is carried between runs, not a cycle. An input marked @@ -211,7 +237,7 @@ def _derive( ) for node in nodes } - return produces, dependencies + return produces, dependencies, consumes, by_id class Pipeline: @@ -224,6 +250,8 @@ class Pipeline: "_max_workers", "produces", "dependencies", + "consumes", + "_by_id", "_edges", "_execution_order", "_downstream_cache", @@ -298,7 +326,12 @@ class Pipeline: # than the default puts its message in here. Swapped, never mutated. self.history_limits: dict[str, int] = {} - self.produces, self.dependencies = _derive(self._nodes) + ( + self.produces, + self.dependencies, + self.consumes, + self._by_id, + ) = _derive(self._nodes) self._edges: dict[Node, set[Node]] | None = None self._execution_order: list[Node] | None = None @@ -376,7 +409,7 @@ class Pipeline: if not placed: spliced.extend(nodes) - produces, dependencies = _derive(spliced) + produces, dependencies, consumes, by_id = _derive(spliced) # Assigned only once everything above has succeeded, and rebound # rather than mutated: a cascade already walking the graph holds @@ -385,6 +418,8 @@ class Pipeline: self._nodes = spliced self.produces = produces self.dependencies = dependencies + self.consumes = consumes + self._by_id = by_id self._edges = None self._execution_order = None self._downstream_cache = {} @@ -423,7 +458,7 @@ class Pipeline: self._state.increment_multi([self._version_key(name) for name in seeded]) def get_node_by_id(self, nid: str) -> Node | None: - return next((n for n in self._nodes if n.id == nid), None) + return self._by_id.get(nid) def flow_nodes(self, flow: str) -> set[Node]: return {n for n in self._nodes if n.flow == flow} @@ -634,16 +669,23 @@ class Pipeline: due_at = min(due_at or window_ends, window_ends) return passed, held, due_at, stamps.get(flush_key) - def _throttled(self, node: Node, result: dict[str, Any]) -> dict[str, Any]: + def _throttled( + self, node: Node, result: dict[str, Any] + ) -> tuple[dict[str, Any], list[str]]: """Hold back the outputs whose port is not due to publish yet. The value is not lost: it is kept and published when the window ends, so a producer that goes quiet still delivers its last reading rather than leaving the consumer on the one before it. Nothing declaring an interval means nothing to look up. + + Returns what passed and the held keys the publish makes stale — the + caller drops them in the same round trip it writes the values with, + rather than one DEL per port here. """ now = time.time() passed, held, due_at, pending = self._window_split(node, result, now) + stale: list[str] = [] if self._queue is not None: # Without a queue there is no timer to let the value out later, so @@ -652,13 +694,13 @@ class Pipeline: spec = node.provides.get(name) if spec is not None and spec.interval > 0: # A fresh publish makes anything held for that port stale. - self._state.delete(self._held_key(name)) + stale.append(self._held_key(name)) if held: self._state.update( {self._held_key(name): value for name, value in held.items()} ) self._schedule_flush(node, due_at, now, pending) - return passed + return passed, stale def _schedule_flush( self, node: Node, at: float, now: float, pending: float | None @@ -791,25 +833,34 @@ class Pipeline: self._state.update({self._delivered_key(node.id, name): now for name in due}) return True - def _is_node_ready(self, node: Node, state: StateBackend) -> bool: - # A non-triggering input is read if it happens to be there; waiting for - # it would make an accumulator's first run impossible, since it is what - # the node is about to write. - waited_on = [msg for msg, spec in node.requires.items() if spec.trigger] + def _is_node_ready( + self, node: Node, state: StateBackend + ) -> tuple[bool, dict[str, Any]]: + """Whether the node may run, and the inputs the answer was read from. + + The read covers every input rather than only the triggering ones, so + the node itself does not have to read the same keys again a moment + later — one MGET per node per message instead of two. A non-triggering + input is read if it happens to be there; waiting for it would make an + accumulator's first run impossible, since it is what the node is about + to write. + """ # One MGET rather than an EXISTS per input under the global state lock. - if len(state.get_present(waited_on)) != len(waited_on): - return False + values = state.get_present(list(node.requires)) + waited_on = [msg for msg, spec in node.requires.items() if spec.trigger] + if any(msg not in values for msg in waited_on): + return False, values if not self._input_is_due(node): - return False + return False, values if not node.synchronous: - return True + return True, values is_ready, current_versions = self._check_synchronous_ready(node) if not is_ready: - return False - return self._try_acquire_synchronous_execution(node, current_versions) + return False, values + return self._try_acquire_synchronous_execution(node, current_versions), values # ------------------------------------------------------------------------- # Execution @@ -952,15 +1003,25 @@ class Pipeline: state: StateBackend, entry_id: str = "", overrides: dict[str, Any] | None = None, + inputs: dict[str, Any] | None = None, ) -> dict[str, Any] | None: - """Run one node and record its outputs. Never raises.""" + """Run one node and record its outputs. Never raises. + + ``inputs`` is what the readiness check already read — the same keys, a + moment earlier. Reading them again here was the second of two MGETs + per node per message; running on the values that made the node ready + is also what ``overrides`` already assumes. + """ started = time.perf_counter() collected = logs.Collector() try: # One MGET. The lock this used to be read under bought nothing a # single bulk read does not, and it was the engine's one global # mutex — every node of every cascade queued behind it. - inputs = state.get_present(list(node.requires)) + if inputs is None: + inputs = state.get_present(list(node.requires)) + else: + inputs = dict(inputs) if overrides: # The value this wave is delivering wins over whatever state # holds by now. Not written back: the newest value is still the @@ -999,11 +1060,12 @@ class Pipeline: except Exception as exc: logger.warning("Could not mark '%s' done: %s", node.id, exc) + stale: list[str] = [] if result: - result = self._throttled(node, result) + result, stale = self._throttled(node, result) if result: - self._record_outputs(node, result, state) + self._record_outputs(node, result, state, stale) duration_ms = round((time.perf_counter() - started) * 1000, 2) self._publish( @@ -1058,24 +1120,34 @@ class Pipeline: return None def _record_outputs( - self, node: Node, outputs: dict[str, Any], state: StateBackend - ) -> None: + self, + node: Node, + outputs: dict[str, Any], + state: StateBackend, + drop: list[str] | None = None, + ) -> float: """Put a node's outputs where everything downstream of them looks. State, the timestamp beside it, the series, the version counter and the event the canvas draws from. Shared by a node returning and a node emitting mid-execution, because those are the same act: a value the node produced, leaving through a port it declared. + + The four writes go in one round trip: a pipeline is a transaction, so + they land together exactly as they did when they were four calls. + Returns the timestamp they were recorded under, which the caller + stamps its own events with so they agree. """ ts = time.time() - # Value and timestamp in one write, which is what the lock around two - # of them was for — a pipeline is a transaction, so they still land - # together and nobody waits on a mutex to do it. - state.update({**outputs, **{self._timestamp_key(name): ts for name in outputs}}) - # Append-only, so it needs no lock of its own. - state.append_history(outputs, ts, self.history_limits) - self._increment_message_versions(outputs) - origin = node_source(node).model_dump() + state.record( + outputs, + {self._timestamp_key(name): ts for name in outputs}, + [self._version_key(name) for name in outputs], + ts, + self.history_limits, + drop, + ) + origin = _node_source_dump(node.id, node.local_id) for name, value in outputs.items(): self._publish( { @@ -1087,6 +1159,7 @@ class Pipeline: "source": origin, } ) + return ts def publish_emission(self, node: Node, outputs: dict[str, Any]) -> None: """Publish what a node produced while it is still running. @@ -1103,10 +1176,10 @@ class Pipeline: with no meaning. """ self._observe_emission(node, outputs) - passed = self._throttled(node, outputs) + passed, stale = self._throttled(node, outputs) if not passed: return - self._record_outputs(node, passed, self._state) + self._record_outputs(node, passed, self._state, stale) if self._queue is not None: # Journalled carrying the emitted values, as an ``emission`` item: # the executor hands them to the nodes reading them instead of @@ -1277,11 +1350,15 @@ class Pipeline: complete(n) progressed = True continue - if check_ready and not self._is_node_ready(n, state): - if n.synchronous: - # Not ready now; a later trigger may make it ready. - skipped.add(n) - continue + ready_inputs: dict[str, Any] | None = None + if check_ready: + ready, ready_inputs = self._is_node_ready(n, state) + if not ready: + if n.synchronous: + # Not ready now; a later trigger may make it + # ready. + skipped.add(n) + continue if replay and entry_id and self._already_done(entry_id, n): # Its side effect happened on an earlier delivery; its # outputs are still in state, so downstream carries on. @@ -1290,7 +1367,7 @@ class Pipeline: continue submitted.add(n) node_futures[n] = executor.submit( - self._execute_node, n, state, entry_id, overrides + self._execute_node, n, state, entry_id, overrides, ready_inputs ) def drain(executor: ThreadPoolExecutor) -> None: @@ -1345,31 +1422,16 @@ class Pipeline: Returns the message names that actually reached state — post rate limiting — which is what the cascade behind it has to walk from. """ + stale: list[str] = [] if outputs: # This is where a chatty subscriber gets thinned out, so a port set # to publish every 60s does so whatever the broker sends. - outputs = self._throttled(node, outputs) + outputs, stale = self._throttled(node, outputs) if not outputs: return set() - state = self._state - ts = time.time() - state.update({**outputs, **{self._timestamp_key(name): ts for name in outputs}}) - state.append_history(outputs, ts, self.history_limits) - self._increment_message_versions(outputs) - origin = node_source(node).model_dump() - for name, value in outputs.items(): - self._publish( - { - "type": "message_value", - "flow": flow_of(name), - "name": name, - "value": value, - "ts": ts, - "source": origin, - } - ) + ts = self._record_outputs(node, outputs, self._state, stale) # An injecting node — an MQTT subscriber, a webhook — publishes # without going through the executor, but it did emit. self._publish( @@ -1559,11 +1621,13 @@ class Pipeline: origin = source or ValueSource(kind="api", label="API") ts = time.time() - self._state.update( - {**values, **{self._timestamp_key(name): ts for name in values}} + self._state.record( + values, + {self._timestamp_key(name): ts for name in values}, + [self._version_key(name) for name in values], + ts, + self.history_limits, ) - self._state.append_history(values, ts, self.history_limits) - self._increment_message_versions(values) source_dump = origin.model_dump() for name, value in values.items(): self._publish( @@ -1579,10 +1643,9 @@ class Pipeline: targets: set[Node] = set() for name in values: - for consumer in self._nodes: - if name in consumer.requires: - targets.add(consumer) - targets.update(self._get_downstream(consumer)) + for consumer in self.consumes.get(name, ()): + targets.add(consumer) + targets.update(self._get_downstream(consumer)) if targets: self._execute_parallel(targets, self._state, check_ready=True) diff --git a/backend/fluksio/flow/queue.py b/backend/fluksio/flow/queue.py index f40581d..417d9e5 100644 --- a/backend/fluksio/flow/queue.py +++ b/backend/fluksio/flow/queue.py @@ -17,7 +17,7 @@ import logging import threading import time from abc import ABC, abstractmethod -from collections import deque +from collections import OrderedDict, deque from collections.abc import Callable from dataclasses import dataclass, field from typing import Any, cast @@ -41,6 +41,10 @@ DUE_PREFIX = "due:" # A consumer this far past its last read belongs to an engine that is gone. A # live one interacts every claim, so nothing in service comes close. STALE_CONSUMER_IDLE_MS = 3_600_000 +# How many "this side effect already happened" markers the memory queue keeps. +# The Redis one expires each after an hour; this is the same idea sized by +# count, since redelivery happens seconds after the claim and never later. +DONE_MARKERS = 10_000 @dataclass @@ -154,8 +158,15 @@ class WorkQueue(ABC): """ @abstractmethod - def claim(self, count: int, block_ms: int) -> list[WorkItem]: - """Take up to ``count`` items, waiting up to ``block_ms`` for one.""" + def claim( + self, count: int, block_ms: int, due_only: bool = False + ) -> list[WorkItem]: + """Take up to ``count`` items, waiting up to ``block_ms`` for one. + + ``due_only`` takes nothing but promoted timers. A saturated engine + uses it to let a motor's stop past the cascade limit while every slot + is held by a long node. + """ @abstractmethod def ack(self, item: WorkItem) -> None: @@ -240,9 +251,17 @@ class MemoryWorkQueue(WorkQueue): def __init__(self) -> None: self._items: deque[WorkItem] = deque() + # Promoted timers, kept apart from the backlog the way Redis keeps + # two streams — so they can be claimed on their own. + self._due: deque[WorkItem] = deque() self._delayed: list[tuple[float, int, WorkItem]] = [] self._parked: dict[str, list[WorkItem]] = {} - self._done: set[tuple[str, str]] = set() + # Insertion-ordered and capped, because this grows one entry per + # non-idempotent node per item and nothing ever removed one — the + # Redis side expires its markers after an hour, this one leaked for + # the life of the process. Redelivery is what the marker guards, and + # that happens within seconds of the claim. + self._done: OrderedDict[tuple[str, str], None] = OrderedDict() # Claimed and not yet acknowledged, which is what Redis's `pending` is. self._in_flight = 0 self._counter = 0 @@ -277,17 +296,25 @@ class MemoryWorkQueue(WorkQueue): with self._lock: return self._delayed[0][0] if self._delayed else None - def claim(self, count: int, block_ms: int) -> list[WorkItem]: + def claim( + self, count: int, block_ms: int, due_only: bool = False + ) -> list[WorkItem]: deadline = time.monotonic() + block_ms / 1000.0 with self._wake: - while not self._items: + while not self._due and not (self._items and not due_only): remaining = deadline - time.monotonic() if remaining <= 0: return [] self._wake.wait(remaining) - claimed = [ - self._items.popleft() for _ in range(min(count, len(self._items))) - ] + # Due first: an item that has waited out a deadline is late by + # however long it queues here, while work merely enqueued is not + # waiting on a clock. + claimed = [self._due.popleft() for _ in range(min(count, len(self._due)))] + if not due_only: + claimed += [ + self._items.popleft() + for _ in range(min(count - len(claimed), len(self._items))) + ] self._in_flight += len(claimed) return claimed @@ -306,10 +333,7 @@ class MemoryWorkQueue(WorkQueue): while self._delayed and self._delayed[0][0] <= now: due.append(heapq.heappop(self._delayed)[2]) if due: - # In front of the backlog, in due order: an item that has - # waited out a deadline is late by however long it queues - # here, while work merely enqueued is not waiting on a clock. - self._items.extendleft(reversed(due)) + self._due.extend(due) self._wake.notify() return len(due) @@ -335,12 +359,12 @@ class MemoryWorkQueue(WorkQueue): def backlog(self) -> int: with self._lock: - return len(self._items) + return len(self._items) + len(self._due) def stats(self) -> dict[str, Any]: with self._lock: return { - "backlog": len(self._items), + "backlog": len(self._items) + len(self._due), "pending": self._in_flight, "delayed": len(self._delayed), "parked": sum(len(v) for v in self._parked.values()), @@ -354,7 +378,9 @@ class MemoryWorkQueue(WorkQueue): def mark_done(self, entry_id: str, node: str) -> None: with self._lock: - self._done.add((entry_id, node)) + self._done[(entry_id, node)] = None + while len(self._done) > DONE_MARKERS: + self._done.popitem(last=False) def was_done(self, entry_id: str, node: str) -> bool: with self._lock: @@ -467,7 +493,9 @@ class RedisWorkQueue(WorkQueue): return self._due_stream, entry_id[len(DUE_PREFIX) :] return self._stream, entry_id - def claim(self, count: int, block_ms: int) -> list[WorkItem]: + def claim( + self, count: int, block_ms: int, due_only: bool = False + ) -> list[WorkItem]: """Take up to ``count`` items, due timers before anything queued. One read over both streams rather than a read each: the block has to @@ -483,7 +511,11 @@ class RedisWorkQueue(WorkQueue): self._redis.xreadgroup( GROUP, self._consumer, - {self._due_stream: ">", self._stream: ">"}, + ( + {self._due_stream: ">"} + if due_only + else {self._due_stream: ">", self._stream: ">"} + ), count=count, block=block_ms, ), diff --git a/backend/fluksio/flow/runs.py b/backend/fluksio/flow/runs.py index e8a18f1..f6ea1f6 100644 --- a/backend/fluksio/flow/runs.py +++ b/backend/fluksio/flow/runs.py @@ -968,8 +968,12 @@ class RunService: while not self._stop.is_set(): try: # Runs put back to wait for a worker come due here. The claim - # below blocks for a second, so this is about once a second. - self.queue.move_due(time.time()) + # below blocks for a second, so this is about once a second — + # and asks the queue for nothing at all while nothing is + # waiting, which is the common case. + due = self.queue.next_due() + if due is not None and due <= time.time(): + self.queue.move_due(time.time()) items = self.queue.claim( max(CLAIM_COUNT, self.parallel), CLAIM_BLOCK_MS ) @@ -1199,11 +1203,18 @@ class RunService: self._release_cards(run) # Its values were only ever this run's; nothing reads them once it # has a result. On Redis the namespace would expire anyway. - if state is not None and status != "error": + if state is not None: + if status != "error": + try: + state.clear() + except Exception: + logger.warning("Could not clear state of run %s", run_id) + # Even when the values are kept: on Redis this backend is a + # client and a connection pool of its own, built per run. try: - state.clear() + state.close() except Exception: - logger.warning("Could not clear state of run %s", run_id) + logger.warning("Could not close state of run %s", run_id) def _release_cards(self, run: Run) -> None: """Hand a GPU run's device memory back when the run is over. diff --git a/backend/fluksio/flow/state.py b/backend/fluksio/flow/state.py index a3b646d..9875536 100644 --- a/backend/fluksio/flow/state.py +++ b/backend/fluksio/flow/state.py @@ -98,6 +98,15 @@ class StateBackend(ABC): """Clear all keys in the state.""" ... + def close(self) -> None: # noqa: B027 + """Release whatever the backend holds outside this process. + + A run builds a state backend of its own and drops it when it + finishes; on Redis that is a client and its connection pool, which + nothing was giving back. Concrete rather than abstract: a backend that + holds nothing has nothing to answer here. + """ + @abstractmethod def keys(self) -> list[str]: """ @@ -221,6 +230,36 @@ class StateBackend(ABC): """ ... + def record( + self, + values: dict[str, Any], + stamps: dict[str, float], + counters: list[str], + ts: float, + limits: dict[str, int] | None = None, + drop: list[str] | None = None, + ) -> None: + """Everything a published value owes state, in one round trip. + + The value itself, the timestamp beside it, its series and its version + counter used to be four calls and four round trips; a backend that can + batch them does so here. This default is those four calls, so a + backend that cannot batch gains nothing and breaks nothing. + + :param values: Message names mapped to the value just published. + :param stamps: The timestamp key of each of those messages. + :param counters: The version keys to bump. + :param ts: When they were published. + :param limits: How many points to keep per message. + :param drop: Keys to forget in the same trip — a released rate-limit + hold, which is written and cleared on this same path. + """ + for key in drop or (): + self.delete(key) + self.update({**values, **stamps}) + self.append_history(values, ts, limits) + self.increment_multi(counters) + @abstractmethod def history(self, key: str) -> list[tuple[float, float]]: """ @@ -369,6 +408,35 @@ class MemoryState(StateBackend): self._history[key] = series series.append((ts, number)) + def record( + self, + values: dict[str, Any], + stamps: dict[str, float], + counters: list[str], + ts: float, + limits: dict[str, int] | None = None, + drop: list[str] | None = None, + ) -> None: + """The four writes under one acquisition of the one lock.""" + with self._lock: + for key in drop or (): + self._data.pop(key, None) + self._history.pop(key, None) + self._data.update(values) + self._data.update(stamps) + for key, value in values.items(): + number = as_number(value) + if number is None: + continue + cap = (limits or {}).get(key, HISTORY_LIMIT) + series = self._history.get(key) + if series is None or series.maxlen != cap: + series = deque(series or (), maxlen=cap) + self._history[key] = series + series.append((ts, number)) + for key in counters: + self._data[key] = self._data.get(key, 0) + 1 + def history(self, key: str) -> list[tuple[float, float]]: """The recorded points of one message, oldest first.""" with self._lock: @@ -648,6 +716,10 @@ class RedisState(StateBackend): # Another client modified one of the watched keys return False + def close(self) -> None: + """Give the connection pool back.""" + self._client.close() + def _history_key(self, key: str) -> str: return self._key(f"__history__:{key}") @@ -671,6 +743,45 @@ class RedisState(StateBackend): if queued: pipe.execute() + def record( + self, + values: dict[str, Any], + stamps: dict[str, float], + counters: list[str], + ts: float, + limits: dict[str, int] | None = None, + drop: list[str] | None = None, + ) -> None: + """Value, timestamp, series and version counter in one round trip. + + These were four pipelines — four round trips — for one published + value, and a value crossing an edge pays them twice. Redis executes a + pipeline as one transaction, so batching them changes nothing about + what lands together and removes three of the four trips. + """ + pipe = self._client.pipeline() + for key in drop or (): + pipe.delete(self._key(key), self._history_key(key)) + for key, value in {**values, **stamps}.items(): + data = self._serialize(value) + if self._ttl: + pipe.setex(self._key(key), self._ttl, data) + else: + pipe.set(self._key(key), data) + for key, value in values.items(): + number = as_number(value) + if number is None: + continue + history_key = self._history_key(key) + cap = (limits or {}).get(key, HISTORY_LIMIT) + pipe.lpush(history_key, orjson.dumps([ts, number])) + pipe.ltrim(history_key, 0, cap - 1) + if self._ttl: + pipe.expire(history_key, self._ttl) + for key in counters: + pipe.incr(self._key(key)) + pipe.execute() + def history(self, key: str) -> list[tuple[float, float]]: """The recorded points of one message, oldest first.""" entries = cast(list[bytes], self._client.lrange(self._history_key(key), 0, -1)) diff --git a/backend/fluksio/flow/store.py b/backend/fluksio/flow/store.py index 4f857d7..3d3e56d 100644 --- a/backend/fluksio/flow/store.py +++ b/backend/fluksio/flow/store.py @@ -116,6 +116,10 @@ class FlowStore: #: it therefore counts from zero per process and misses an edit made on #: disk behind the API, which no writer here does. self.revision = 0 + #: `read_all` memoised against that revision — see the method. + self._read_all: list[FlowDef] = [] + self._read_all_at = -1 + self._read_all_lock = threading.Lock() if not (self.root / ".git").exists(): self._git("init", "-q") self._commit("Initialise flow store", allow_empty=True) @@ -394,15 +398,25 @@ class FlowStore: return FlowDef.model_validate_json(path.read_text()) def read_all(self) -> list[FlowDef]: - """Every published flow — what the engine runs.""" - flows = [] - for path in sorted(self.root.glob("*/flow.json")): - name = path.parent.name - try: - flows.append(self.read_flow(name)) - except Exception: - logger.exception("Skipping unreadable flow '%s'", name) - return flows + """Every published flow — what the engine runs. + + Cached against `revision`, because this sits on the publish path: a + dashboard slider moving asked every flow's file to be read and + validated again, per value. The list is rebuilt on the next commit, + and returned as a copy so a caller sorting it cannot disturb the next. + """ + with self._read_all_lock: + if self._read_all_at != self.revision: + flows = [] + for path in sorted(self.root.glob("*/flow.json")): + name = path.parent.name + try: + flows.append(self.read_flow(name)) + except Exception: + logger.exception("Skipping unreadable flow '%s'", name) + self._read_all = flows + self._read_all_at = self.revision + return list(self._read_all) def write_flow(self, flow: FlowDef) -> bool: """Publish a flow directly. Returns False when nothing actually changed.""" diff --git a/backend/fluksio/main.py b/backend/fluksio/main.py index 8b271c7..dbd4819 100644 --- a/backend/fluksio/main.py +++ b/backend/fluksio/main.py @@ -1,8 +1,9 @@ import asyncio import contextlib import logging -from collections.abc import AsyncIterator +from collections.abc import AsyncIterator, Coroutine from contextlib import AbstractAsyncContextManager, asynccontextmanager +from typing import Any from fastapi import FastAPI, Request from fastapi.concurrency import run_in_threadpool @@ -223,14 +224,36 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: ) watchdog = LoopWatchdog(event_bus) app.state.watchdog = watchdog - watchdog_task = asyncio.create_task(watchdog.run(), name="loop-watchdog") - alerts_task = asyncio.create_task(alerts.run(), name="alert-manager") - metrics_task = asyncio.create_task( - MetricsCollector(event_bus).run(), name="metrics-collector" - ) - gc_task = asyncio.create_task( - _sweep_artifacts(artifacts, controller), name="artifact-gc" - ) + + def _background(coro: Coroutine[Any, Any, None], name: str) -> asyncio.Task[None]: + """Start a long-lived task that says something if it ever stops. + + Each of these loops catches its own exceptions *inside* the loop, so + one raised anywhere else simply ended the task — an engine that went + on serving with no metrics, no alerts or no artifact sweep and nothing + anywhere saying so. + """ + task = asyncio.create_task(coro, name=name) + + def _finished(done: asyncio.Task[None]) -> None: + if done.cancelled(): + return + exc = done.exception() + if exc is None: + logger.warning("Background task '%s' stopped on its own", name) + return + logger.error("Background task '%s' died: %s", name, exc, exc_info=exc) + event_bus.publish( + {"type": "engine_degraded", "detail": f"{name} stopped: {exc}"} + ) + + task.add_done_callback(_finished) + return task + + watchdog_task = _background(watchdog.run(), "loop-watchdog") + alerts_task = _background(alerts.run(), "alert-manager") + metrics_task = _background(MetricsCollector(event_bus).run(), "metrics-collector") + gc_task = _background(_sweep_artifacts(artifacts, controller), "artifact-gc") await controller.start() run_service.start() # Optional, and off unless someone enrolled this installation: the @@ -246,8 +269,8 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: # Watched whether or not one exists now: enrolling from the CLI writes the # config from another process entirely, and an engine already serving # should pick it up rather than need restarting. - enrol_task = asyncio.create_task( - cloud_connector.watch_enrolment(app), name="cloud-enrolment-watch" + enrol_task = _background( + cloud_connector.watch_enrolment(app), "cloud-enrolment-watch" ) try: # A mounted sub-app gets no lifespan of its own, so the MCP session @@ -266,12 +289,15 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: running_cloud.cancel() await run_in_threadpool(run_service.stop) await controller.stop() - pool.stop() + # On a thread, like the two above: stopping a worker waits up to five + # seconds on each child, and with a pool per declared environment that + # is a shutdown the event loop should not be holding. + await run_in_threadpool(pool.stop) # A machine asked for and not yet arrived would hold an allocation # nobody is going to use. for provisioner in placer.provisioners: await run_in_threadpool(provisioner.shutdown) - close_shared_client() + await run_in_threadpool(close_shared_client) if settings.MCP_ENABLED: from fluksio.mcp.http import aclose diff --git a/backend/tests/flow/test_queue.py b/backend/tests/flow/test_queue.py index bde61bf..a3a8bc2 100644 --- a/backend/tests/flow/test_queue.py +++ b/backend/tests/flow/test_queue.py @@ -8,6 +8,7 @@ from fluksio.flow.executor import ExecutionService from fluksio.flow.messages import DType, MessageSpec from fluksio.flow.nodes import Node from fluksio.flow.pipeline import Pipeline +from fluksio.flow import queue as queue_module from fluksio.flow.queue import MemoryWorkQueue, WorkItem from fluksio.flow.state import MemoryState @@ -584,3 +585,44 @@ def test_a_throttled_emission_wakes_nothing(): # The first is let through; the second is held by the interval, and a value # nothing published is nothing to wake on. assert seen == [1.0] + + +def test_a_due_timer_is_claimable_while_the_backlog_is_not(): + """What a saturated engine uses to get a motor's stop out. + + Every cascade slot held by a long node used to mean the due lane was not + merely behind the backlog but unread, because the capacity gate sits in + front of the claim. + """ + queue = MemoryWorkQueue() + queue.add(WorkItem(kind="cascade", node="f.backlog", flow="f")) + queue.add_delayed( + WorkItem(kind="cascade", node="f.timer", flow="f"), time.time() - 1 + ) + assert queue.move_due(time.time()) == 1 + + claimed = queue.claim(10, 10, due_only=True) + assert [item.node for item in claimed] == ["f.timer"] + + # And the backlog is still there for whoever comes next. + assert [item.node for item in queue.claim(10, 10)] == ["f.backlog"] + + +def test_a_due_only_claim_of_nothing_does_not_take_the_backlog(): + queue = MemoryWorkQueue() + queue.add(WorkItem(kind="cascade", node="f.backlog", flow="f")) + + assert queue.claim(10, 10, due_only=True) == [] + assert len(queue.claim(10, 10)) == 1 + + +def test_done_markers_do_not_grow_without_bound(): + """The Redis side expires them after an hour; this one is capped by count.""" + queue = MemoryWorkQueue() + for i in range(queue_module.DONE_MARKERS + 500): + queue.mark_done(f"entry-{i}", "f.node") + + assert len(queue._done) == queue_module.DONE_MARKERS + # The newest are what redelivery would ask about. + assert queue.was_done(f"entry-{queue_module.DONE_MARKERS + 499}", "f.node") + assert not queue.was_done("entry-0", "f.node") diff --git a/backend/tests/flow/test_round_trips.py b/backend/tests/flow/test_round_trips.py new file mode 100644 index 0000000..b36ba8a --- /dev/null +++ b/backend/tests/flow/test_round_trips.py @@ -0,0 +1,126 @@ +"""What one message costs the state backend, in operations. + +The engine's time goes on round trips, not on Python: a profile against a +real Redis put its own code at 8.5% of self-time and Redis I/O at 61%. So the +thing worth a regression test is the *count* — a change that quietly turns one +pipelined write back into four would not fail any behavioural test and would +cost a fifth of the throughput. + +Counted against the memory backend, because the number is a property of the +call sites rather than of the transport. +""" + +from collections import Counter +from typing import Any + +from fluksio.flow.messages import DType, MessageSpec +from fluksio.flow.nodes import Node +from fluksio.flow.pipeline import Pipeline +from fluksio.flow.state import MemoryState + +COUNTED = ( + "get", + "set", + "get_multi", + "get_present", + "update", + "record", + "append_history", + "increment_multi", + "delete", +) + + +class CountingState(MemoryState): + """A state backend that remembers how often it was asked for something.""" + + # MemoryState uses __slots__; this one needs an attribute of its own. + __slots__ = ("calls",) + + def __init__(self) -> None: + super().__init__() + self.calls: Counter[str] = Counter() + + def __getattribute__(self, name: str) -> Any: + if name in COUNTED: + object.__getattribute__(self, "calls")[name] += 1 + return object.__getattribute__(self, name) + + +def spec(name: str) -> MessageSpec: + return MessageSpec(name=name, dtype=DType.FLOAT) + + +def make_node(node_id: str, flow: str, f, requires=(), provides=()) -> Node: + node = Node(f=f, requires=list(requires), provides=list(provides), name=node_id) + node.assign_flow(flow, node_id) + return node + + +def build() -> tuple[Pipeline, Node, CountingState]: + source = make_node( + "source", "chain", lambda params: {"a": 1.0}, provides=[spec("a")] + ) + relay = make_node( + "relay", + "chain", + lambda a, params: {"b": a + 1}, + requires=[spec("a")], + provides=[spec("b")], + ) + sink = make_node( + "sink", "chain", lambda b, params: None, requires=[spec("b")] + ) + state = CountingState() + pipeline = Pipeline([source, relay, sink], state=state) + return pipeline, source, state + + +def test_recording_a_value_is_one_write() -> None: + """Value, timestamp, history and version counter go in one round trip. + + They were four calls — `update`, `append_history`, `increment_multi` and a + `delete` per rate-limited port — and a value crossing an edge pays them + twice. + """ + pipeline, source, state = build() + state.calls.clear() + + pipeline.apply_outputs(source, {"chain.a": 1.0}) + + assert state.calls["record"] == 1 + for method in ("update", "append_history", "increment_multi"): + assert state.calls[method] == 0, f"{method} should be folded into record()" + + +def test_a_node_reads_its_inputs_once() -> None: + """Readiness and execution share one read of the same keys. + + The readiness check used to read the triggering inputs and throw the + values away, and the node then read the same keys again to run on them. + """ + pipeline, source, state = build() + state.calls.clear() + + changed = pipeline.apply_outputs(source, {"chain.a": 1.0}) + pipeline.run_downstream(source, changed=changed) + + # Two nodes ran downstream (relay, sink); one bulk read each. + assert state.calls["get_present"] == 2 + + +def test_a_hop_costs_a_bounded_number_of_state_operations() -> None: + """The whole of one value crossing one edge, counted. + + A ceiling rather than an exact number, so an unrelated change does not + fail it — but low enough that reintroducing a per-value read or a + split-up write does. + """ + pipeline, source, state = build() + state.calls.clear() + + changed = pipeline.apply_outputs(source, {"chain.a": 1.0}) + pipeline.run_downstream(source, changed=changed) + + total = sum(state.calls[name] for name in COUNTED) + assert total <= 8, f"state operations per hop grew: {dict(state.calls)}"