diff --git a/backend/fluksio/flow/executor.py b/backend/fluksio/flow/executor.py index e3a4b47..510766f 100644 --- a/backend/fluksio/flow/executor.py +++ b/backend/fluksio/flow/executor.py @@ -387,10 +387,17 @@ class ExecutionService: } ) + replay = item.deliveries > 1 try: - pipeline.apply_outputs(node, item.outputs or None) + published = pipeline.apply_outputs(node, item.outputs or None) pipeline.run_downstream( - node, entry_id=item.entry_id, replay=item.deliveries > 1 + 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, ) finally: # Paired, or a cascade that raised — state backend gone, say — is a diff --git a/backend/fluksio/flow/pipeline.py b/backend/fluksio/flow/pipeline.py index 69cb639..39fa0e4 100644 --- a/backend/fluksio/flow/pipeline.py +++ b/backend/fluksio/flow/pipeline.py @@ -569,13 +569,14 @@ class Pipeline: """When a rate-limit window of this node is due to be let through.""" return f"__flush__:{node_name}" - def _throttled(self, node: Node, result: dict[str, Any]) -> dict[str, Any]: - """Hold back the outputs whose port is not due to publish yet. + def _window_split( + self, node: Node, result: dict[str, Any], now: float + ) -> tuple[dict[str, Any], dict[str, Any], float, float | None]: + """What may publish now, what its window holds back, when it ends, and + whether a flush of this node is already booked. - 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. + Reads state and writes nothing, so the same question can be asked when + a value arrives and again when the cascade carrying it is claimed. """ limited = { name: spec.interval @@ -583,9 +584,8 @@ class Pipeline: if spec.interval > 0 and name in result } if not limited: - return result + return result, {}, 0.0, None - now = time.time() flush_key = self._flush_key(node.id) stamps = self._state.get_multi( [self._timestamp_key(name) for name in limited] + [flush_key] @@ -604,19 +604,32 @@ class Pipeline: else: held[name] = value 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]: + """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. + """ + now = time.time() + passed, held, due_at, pending = self._window_split(node, result, now) if self._queue is not None: # Without a queue there is no timer to let the value out later, so # holding it would only mean losing it more slowly. - for name in limited: - if name in passed: + for name in passed: + 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)) if held: self._state.update( {self._held_key(name): value for name, value in held.items()} ) - self._schedule_flush(node, due_at, now, stamps.get(flush_key)) + self._schedule_flush(node, due_at, now, pending) return passed def _schedule_flush( @@ -652,10 +665,14 @@ class Pipeline: } if held: # Publishing runs the limit again, which is what clears the hold — - # or puts the timer back, if this fired a hair early. - self.apply_outputs(node, held) - self.run_downstream(node) + # or puts the timer back, if this fired a hair early. In that case + # nothing came out and there is nothing downstream to wake. + published = self.apply_outputs(node, held) + if published: + self.run_downstream(node, changed=published) if any(spec.interval > 0 for spec in node.requires.values()): + # A wake-up that was held back, not a value that just changed — + # so this one deliberately walks everything reachable. self._execute_parallel( {node, *self._get_downstream(node)}, self._state, check_ready=True ) @@ -1134,8 +1151,15 @@ class Pipeline: check_ready: bool = False, entry_id: str = "", replay: bool = False, + changed: set[str] | None = None, ) -> StateBackend: - """Execute nodes concurrently, scheduling each as its inputs arrive.""" + """Execute nodes concurrently, scheduling each as its inputs arrive. + + ``changed`` restricts the wave to nodes something published to: one + whose triggering inputs are all untouched is completed without being + run, so what is downstream of *it* is judged on the same footing. None + runs everything the subset holds, which is what a manual run means. + """ # One view of the graph for the whole wave: a flow replaced halfway # through must not have this wave asking the new dependencies about a # node the old edges knew. @@ -1154,42 +1178,64 @@ class Pipeline: n: sum(1 for dep in deps[n] if dep in target_nodes) for n in target_nodes } + # Copied: the caller's set is not this wave's to grow. + fresh = set(changed) if changed is not None else None + submitted: set[Node] = set() skipped: set[Node] = set() node_futures: dict[Node, Future[dict[str, Any] | None]] = {} - def is_ready(n: Node) -> bool: - if in_degree[n] != 0: - return False - if check_ready: - return self._is_node_ready(n, state) - return True + def untouched(n: Node) -> bool: + """Did this wave publish nothing this node waits on?""" + assert fresh is not None + return not any( + spec.trigger and name in fresh for name, spec in n.requires.items() + ) + + def complete(n: Node) -> None: + """Count a node as done without running it, freeing its consumers.""" + skipped.add(n) + for consumer in edges[n]: + if consumer in target_nodes: + in_degree[consumer] -= 1 def submit_ready(executor: ThreadPoolExecutor) -> None: - for n in target_nodes: - if n in submitted or n in skipped: - continue - # Gate before the readiness check, so a held-back node does not - # spend the synchronous claim it would need once it may run. - if self._gate_blocks(n): - continue - if is_ready(n): - submitted.add(n) + # To a fixpoint: completing a node without running it frees its + # consumers, and one this pass has already walked past would + # otherwise be stranded — there is no future outstanding to bring + # the drain loop back round for it. + progressed = True + while progressed: + progressed = False + for n in target_nodes: + if n in submitted or n in skipped: + continue + # Gate before the readiness check, so a held-back node does + # not spend the synchronous claim it needs once it may run. + if self._gate_blocks(n) or in_degree[n] != 0: + continue + if fresh is not None and untouched(n): + # Ahead of the readiness check, so an unchanged node + # spends neither a delivery stamp nor a synchronous + # claim on a run it is not going to make. + 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 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. - skipped.add(n) - submitted.discard(n) - for consumer in edges[n]: - if consumer in target_nodes: - in_degree[consumer] -= 1 + complete(n) + progressed = True continue + submitted.add(n) node_futures[n] = executor.submit( self._execute_node, n, state, entry_id ) - elif n.synchronous and in_degree[n] == 0: - # Not ready now; a later trigger may make it ready. - skipped.add(n) def drain(executor: ThreadPoolExecutor) -> None: submit_ready(executor) @@ -1203,6 +1249,10 @@ class Pipeline: # A node returning nothing (rate limiting, an error) stops # propagation along its branch. if result is not None: + # Before the decrement: a consumer freed by this node + # is judged on what it just published. + if fresh is not None: + fresh.update(result) for consumer in edges[n]: if consumer in target_nodes: in_degree[consumer] -= 1 @@ -1229,11 +1279,14 @@ class Pipeline: return self._execute_parallel(nodes, self._state, check_ready=False) - def apply_outputs(self, node: Node, outputs: dict[str, Any] | None) -> None: + def apply_outputs(self, node: Node, outputs: dict[str, Any] | None) -> set[str]: """Record what a node emitted: state, history, versions and events. Shared by the direct path and by the execution service replaying a journaled item, so a value looks the same on the canvas either way. + + Returns the message names that actually reached state — post rate + limiting — which is what the cascade behind it has to walk from. """ if outputs: # This is where a chatty subscriber gets thinned out, so a port set @@ -1241,7 +1294,7 @@ class Pipeline: outputs = self._throttled(node, outputs) if not outputs: - return + return set() state = self._state ts = time.time() @@ -1274,9 +1327,14 @@ class Pipeline: "ts": ts, } ) + return set(outputs) def run_downstream( - self, node: Node, entry_id: str = "", replay: bool = False + self, + node: Node, + entry_id: str = "", + replay: bool = False, + changed: set[str] | None = None, ) -> StateBackend: """Run everything downstream of a node that has just published. @@ -1284,7 +1342,17 @@ class Pipeline: node with outside side effects can record that it ran. On a ``replay`` — the same item handed back after a crash — that record is checked first: at-least-once delivery must not mean two of the same request. + + ``changed`` names the messages this wave actually published. A node + none of whose triggering inputs are in it would re-run on values it has + already read, which is how one node came to publish 619 messages a + minute off inputs that changed six times. ``None`` means walk + everything reachable, which is what a manual run wants. """ + if changed is not None and not changed: + # Everything the cascade carried was held back by a rate limit, so + # there is nothing new for anything downstream to read. + return self._state downstream = set(self._get_downstream(node)) if not downstream: return self._state @@ -1294,6 +1362,7 @@ class Pipeline: check_ready=True, entry_id=entry_id, replay=replay, + changed=changed, ) def trigger( @@ -1309,9 +1378,13 @@ class Pipeline: A stopped flow drops the event: its subscriptions and schedules are torn down anyway, and anything still arriving from another thread would be - work the flow was explicitly told not to do. A *paused* flow still - publishes, so the incoming value is visible on the canvas, and holds - the nodes downstream of it. + work the flow was explicitly told not to do. A *paused* flow parks the + item, so nothing of it publishes until the flow is resumed or stepped. + + A value whose every port is inside its rate-limit window is kept here + and never journaled at all: the limit is about how often the value + goes out, and enqueueing a cascade to discover that costs a run record + and a walk of everything downstream per message. """ state = self._state @@ -1321,11 +1394,37 @@ class Pipeline: if durable is None: durable = self._queue is not None if durable and self._queue is not None: - self._enqueue_cascade(node, outputs) + if not self._held_at_source(node, outputs): + self._enqueue_cascade(node, outputs) return state return self._run_here(node, outputs) + def _held_at_source(self, node: Node, outputs: dict[str, Any] | None) -> bool: + """Keep a value back before it costs a cascade, or say it has to run. + + Only when *every* port is inside its window: anything that would + publish leaves the exact split to ``apply_outputs`` at claim time, + which is the one place it has always been made. + """ + if not outputs: + # A cascade carrying nothing exists to walk, not to publish. + return False + try: + now = time.time() + passed, held, due_at, pending = self._window_split(node, outputs, now) + if passed or not held: + return False + self._state.update( + {self._held_key(name): value for name, value in held.items()} + ) + self._schedule_flush(node, due_at, now, pending) + return True + except Exception: + # Journalling it is what this was avoiding, not what it depends on. + logger.exception("Could not hold '%s' back at the source", node.id) + return False + def _run_here( self, node: Node, outputs: dict[str, Any] | None, cause: str = "manual" ) -> StateBackend: @@ -1348,8 +1447,14 @@ class Pipeline: } ) try: - self.apply_outputs(node, outputs) - state = self.run_downstream(node, entry_id=run_id) + published = self.apply_outputs(node, outputs) + state = self.run_downstream( + node, + entry_id=run_id, + # No payload means the value is already in state and this is a + # wake-up, which has nothing to name as changed. + changed=published if outputs else None, + ) finally: # Paired, or a cascade that raised leaves the run open until the # abandoned sweep ten minutes later. diff --git a/backend/scripts/bench_engine.py b/backend/scripts/bench_engine.py index 768cfdf..6ed1164 100644 --- a/backend/scripts/bench_engine.py +++ b/backend/scripts/bench_engine.py @@ -14,6 +14,11 @@ Three scenarios, each answering a different question: Reports cascades enqueued, node executions and bus events per injection — the numbers, not the wall clock, are the point. +``relay`` + The same, with the *source* unlimited: the cascade really is owed, and + what the relays hold back is what the node reading them should not be + woken for. + ``throughput`` How many messages a second, and how late is the last node? A relay chain, driven until the queue drains, reporting end-to-end latency percentiles. @@ -170,7 +175,27 @@ class Engine: def stop(self) -> None: self.service.stop() - def drain(self, timeout: float = 30.0) -> bool: + def deliver(self, source: Node, outputs: dict[str, Any]) -> None: + """Inject one value and run every cascade it owes, in this thread. + + The counting scenarios drive the queue by hand rather than starting + the consumer: injecting flat out means five hundred messages land + before the first cascade writes the timestamp its window is measured + from, so every one of them looks due and the thing being counted never + happens. A house sends three readings a second into an engine that + finishes a cascade in four milliseconds, which is this. + """ + source.inject(outputs) + while True: + self.queue.move_due(0.0) + items = self.queue.claim(16, 1) + if not items: + return + for item in items: + self.service._run_item(item) + self.queue.ack(item) + + def drain(self, timeout: float = 60.0) -> bool: """Wait for the queue to go quiet. False means it never did. Delayed items are not waited for: a rate-limit flush is due a whole @@ -224,13 +249,11 @@ def scenario_throttle(args: argparse.Namespace) -> dict[str, Any]: nodes.append(relay(f"relay{i}", "raw", f"out{i}", args.window, counters)) engine = Engine(nodes, args.redis, counters) - engine.start() try: started = time.perf_counter() for i in range(args.messages): - source.inject({"raw": float(i)}) - injected = time.perf_counter() - started - drained = engine.drain() + engine.deliver(source, {"raw": float(i)}) + elapsed = time.perf_counter() - started finally: engine.stop() @@ -239,8 +262,7 @@ def scenario_throttle(args: argparse.Namespace) -> dict[str, Any]: "messages": args.messages, "relays": args.relays, "window_s": args.window, - "drained": drained, - "inject_s": round(injected, 3), + "elapsed_s": round(elapsed, 3), # The three numbers the bug is about. "cascades_started": counters.by_event.get("cascade_started", 0), "node_executions": counters.executions, @@ -256,6 +278,62 @@ def scenario_throttle(args: argparse.Namespace) -> dict[str, Any]: } +def scenario_relay(args: argparse.Namespace) -> dict[str, Any]: + """An *unlimited* source into limited relays, and one node reading them all. + + The other half of the house's shape, and the one holding it at the source + cannot help with: the readings really do arrive, so a cascade is owed — + but the relays hold their outputs back, and what reads them has nothing + new to read. That node published 619 messages a minute off inputs that + changed six times. + """ + counters = Counters() + source = node("source", lambda params: None, provides=[spec("raw")]) + nodes = [source] + for i in range(args.relays): + nodes.append(relay(f"relay{i}", "raw", f"out{i}", args.window, counters)) + + ports = [f"out{i}" for i in range(args.relays)] + + def watch(**kwargs: Any) -> dict[str, Any]: + counters.executions += 1 + return {"total": sum(kwargs[p] for p in ports)} + + nodes.append( + node( + "watch", + watch, + requires=[spec(p) for p in ports], + provides=[spec("total")], + ) + ) + + engine = Engine(nodes, args.redis, counters) + try: + started = time.perf_counter() + for i in range(args.messages): + engine.deliver(source, {"raw": float(i)}) + elapsed = time.perf_counter() - started + finally: + engine.stop() + + return { + "scenario": "relay", + "messages": args.messages, + "relays": args.relays, + "window_s": args.window, + "elapsed_s": round(elapsed, 3), + "cascades_started": counters.by_event.get("cascade_started", 0), + "node_executions": counters.executions, + "bus_events": counters.events, + "published": counters.by_event.get("message_value", 0), + "per_message": { + "executions": round(counters.executions / args.messages, 3), + "events": round(counters.events / args.messages, 3), + }, + } + + def scenario_throughput(args: argparse.Namespace) -> dict[str, Any]: """A relay chain, driven flat out: messages a second, and how late the tail is.""" counters = Counters() @@ -289,7 +367,7 @@ def scenario_throughput(args: argparse.Namespace) -> dict[str, Any]: with lock: starts[value] = time.perf_counter() source.inject({"v0": value}) - drained = engine.drain() + drained = engine.drain(args.drain_timeout) elapsed = time.perf_counter() - started finally: engine.stop() @@ -330,7 +408,7 @@ def scenario_fanout(args: argparse.Namespace) -> dict[str, Any]: started = time.perf_counter() for i in range(args.messages): source.inject({"raw": float(i)}) - drained = engine.drain() + drained = engine.drain(args.drain_timeout) elapsed = time.perf_counter() - started finally: engine.stop() @@ -349,6 +427,7 @@ def scenario_fanout(args: argparse.Namespace) -> dict[str, Any]: SCENARIOS: dict[str, Callable[[argparse.Namespace], dict[str, Any]]] = { "throttle": scenario_throttle, + "relay": scenario_relay, "throughput": scenario_throughput, "fanout": scenario_fanout, } @@ -391,6 +470,7 @@ def main() -> int: metavar="HOST", help="Run against a real Redis (namespaces bench/benchq) rather than memory.", ) + parser.add_argument("--drain-timeout", type=float, default=60.0) parser.add_argument("--profile", action="store_true", help="cProfile, top 25.") parser.add_argument("--json", action="store_true", help="Machine-readable output.") args = parser.parse_args() diff --git a/backend/tests/flow/test_discretization.py b/backend/tests/flow/test_discretization.py index 8670179..cae205c 100644 --- a/backend/tests/flow/test_discretization.py +++ b/backend/tests/flow/test_discretization.py @@ -148,6 +148,106 @@ def test_a_limited_input_wakes_its_node_when_the_window_ends(): assert seen == [20.0, 21.0] +def test_an_inject_inside_the_window_costs_no_cascade(): + """The limit used to thin the messages and not the work. + + It was applied after the item came off the queue, so a subscriber told to + publish every 15s still cost a queue entry, a run record and a walk of + everything downstream for every message the broker sent. + """ + source = make_node( + "source", + lambda params: None, + provides=[spec("temp", interval=WINDOW)], + ) + pipeline, queue, service = _running([source]) + + source.inject({"temp": 20.0}) + for item in queue.claim(10, 10): + service._run_item(item) + assert pipeline.state["demo.temp"] == 20.0 + + # Inside the window: held where it is, with nothing journaled for it. + source.inject({"temp": 21.0}) + assert queue.claim(10, 10) == [] + + # And the timer still lets it out at the end of the window. + time.sleep(WINDOW * 2) + assert _run_due(queue, service) == 1 + assert pipeline.state["demo.temp"] == 21.0 + + +def test_a_node_whose_inputs_did_not_change_is_not_run(): + """A cascade used to walk everything reachable, changed or not.""" + seen: list[float] = [] + source = make_node("source", lambda params: None, provides=[spec("raw")]) + relay = make_node( + "relay", + lambda raw, params: {"level": raw}, + requires=[spec("raw")], + provides=[spec("level", interval=WINDOW)], + ) + watcher = make_node( + "watcher", + lambda level, params: seen.append(level), + requires=[spec("level")], + ) + _pipeline, queue, service = _running([source, relay, watcher]) + + def deliver(value: float) -> None: + source.inject({"raw": value}) + for item in queue.claim(10, 10): + service._run_item(item) + + deliver(1.0) + assert seen == [1.0] + + # The relay runs — its own input did change — but publishes nothing, so + # the watcher is left on the value it already has. + deliver(2.0) + assert seen == [1.0] + + # The held value reaches it when the window ends. + time.sleep(WINDOW * 2) + _run_due(queue, service) + assert seen == [1.0, 2.0] + + +def test_a_join_behind_a_skipped_branch_still_runs(): + """Skipping a node frees its consumers, which may already have been passed.""" + seen: list[tuple[float, float]] = [] + source = make_node( + "source", + lambda params: None, + provides=[spec("fast"), spec("slow", interval=WINDOW)], + ) + middle = make_node( + "middle", + lambda slow, params: {"derived": slow}, + requires=[spec("slow")], + provides=[spec("derived")], + ) + join = make_node( + "join", + lambda fast, derived, params: seen.append((fast, derived)), + requires=[spec("fast"), spec("derived")], + ) + _pipeline, queue, service = _running([source, middle, join]) + + def deliver(value: float) -> None: + source.inject({"fast": value, "slow": value}) + for item in queue.claim(10, 10): + service._run_item(item) + + deliver(1.0) + assert seen == [(1.0, 1.0)] + + # `slow` is held this time, so `middle` is skipped — and `join` still has + # to run, because `fast` did change. + deliver(2.0) + assert seen == [(1.0, 1.0), (2.0, 1.0)] + + def test_a_manual_run_is_never_throttled_on_its_inputs(): seen: list[float] = [] source = make_node("source", lambda params: {"temp": 20.0}, provides=[spec("temp")])