stroblmeandClaude Opus 5
a9136c7811
The limit was applied in `apply_outputs`, which the executor reaches after the item is off the queue — so a subscriber told to publish every 15s still cost a queue entry, a `cascade_started`, a run record and a walk of everything reachable from it per inbound message. Seven relay nodes behind one inverter ran 192 times a minute to publish six. Two halves, matching the two shapes it takes: `trigger()` now keeps a value whose every port is inside its window and journals nothing at all. The window split came out of `_throttled` as a read-only `_window_split`, so the question is asked the same way in both places and the exact split is still made once, at claim time. A cascade carries the names it actually published, and the wave runs only the nodes something in that set feeds. A node whose triggering inputs were all held back is completed without running, which frees its own consumers to be judged the same way — the case where a node re-published 619 messages a minute off inputs that changed six times. Redeliveries and emissions carry no such set and still walk everything, since one has a half-finished wave to finish and the other is the value already being in state. Skipping a node can make one ready that the scheduling pass has already walked past, so `submit_ready` runs to a fixpoint. That also closes the same latent hole on the replay path, where a done-marker skip could strand a join with no future outstanding to come back for it. Measured with the new `scripts/bench_engine.py`, 500 messages through the house's shape: a limited source went from 500 cascades / 3500 node runs / 5009 events to 1 / 7 / 19, publishing the same 8 values; an unlimited source into limited relays took the node reading them from 500 runs to 1. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01BpfSinyCBfjuieikyfMPbf
267 lines
8.4 KiB
Python
267 lines
8.4 KiB
Python
"""Per-port intervals: deliver at most every x seconds."""
|
|
|
|
import time
|
|
|
|
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.queue import MemoryWorkQueue
|
|
|
|
# Short enough to wait out in a test, long enough not to race the engine.
|
|
WINDOW = 0.05
|
|
|
|
|
|
def spec(name: str, interval: float = 0) -> MessageSpec:
|
|
return MessageSpec(name=name, dtype=DType.FLOAT, interval=interval)
|
|
|
|
|
|
def make_node(node_id: str, f, requires=(), provides=()) -> Node:
|
|
node = Node(f=f, requires=list(requires), provides=list(provides), name=node_id)
|
|
node.assign_flow("demo", node_id)
|
|
return node
|
|
|
|
|
|
def test_a_limited_output_publishes_once_inside_its_window():
|
|
readings = iter([1.0, 2.0, 3.0])
|
|
source = make_node(
|
|
"source",
|
|
lambda params: {"temp": next(readings)},
|
|
provides=[spec("temp", interval=60)],
|
|
)
|
|
pipeline = Pipeline(nodes=[source])
|
|
|
|
pipeline.run({})
|
|
assert pipeline.state["demo.temp"] == 1.0
|
|
|
|
# Same window: the reading is taken but not published.
|
|
pipeline.run({})
|
|
assert pipeline.state["demo.temp"] == 1.0
|
|
|
|
|
|
def test_an_unlimited_output_publishes_every_time():
|
|
readings = iter([1.0, 2.0])
|
|
source = make_node(
|
|
"source",
|
|
lambda params: {"temp": next(readings)},
|
|
provides=[spec("temp")],
|
|
)
|
|
pipeline = Pipeline(nodes=[source])
|
|
|
|
pipeline.run({})
|
|
pipeline.run({})
|
|
assert pipeline.state["demo.temp"] == 2.0
|
|
|
|
|
|
def test_a_limited_input_wakes_its_node_once_inside_the_window():
|
|
seen: list[float] = []
|
|
source = make_node("source", lambda params: {"temp": 20.0}, provides=[spec("temp")])
|
|
consumer = make_node(
|
|
"consumer",
|
|
lambda temp, params: seen.append(temp),
|
|
requires=[spec("temp", interval=60)],
|
|
)
|
|
# Binding the nodes is what the pipeline is for here.
|
|
Pipeline(nodes=[source, consumer])
|
|
|
|
source.inject({"temp": 20.0})
|
|
source.inject({"temp": 21.0})
|
|
|
|
assert seen == [20.0]
|
|
|
|
|
|
def test_an_unthrottled_input_still_wakes_a_node_beside_a_throttled_one():
|
|
seen: list[tuple[float, float]] = []
|
|
fast = make_node("fast", lambda params: None, provides=[spec("quick")])
|
|
slow = make_node("slow", lambda params: None, provides=[spec("rare")])
|
|
consumer = make_node(
|
|
"consumer",
|
|
lambda quick, rare, params: seen.append((quick, rare)),
|
|
requires=[spec("quick"), spec("rare", interval=60)],
|
|
)
|
|
pipeline = Pipeline(nodes=[fast, slow, consumer])
|
|
pipeline.state["demo.rare"] = 1.0
|
|
|
|
fast.inject({"quick": 1.0})
|
|
fast.inject({"quick": 2.0})
|
|
|
|
# The throttled port holds back only itself.
|
|
assert [quick for quick, _ in seen] == [1.0, 2.0]
|
|
|
|
|
|
def _running(nodes: list[Node]) -> tuple[Pipeline, MemoryWorkQueue, ExecutionService]:
|
|
"""A pipeline with the timer the engine uses for held-back values."""
|
|
queue = MemoryWorkQueue()
|
|
pipeline = Pipeline(nodes=nodes, work_queue=queue)
|
|
service = ExecutionService(queue)
|
|
service.bind(pipeline)
|
|
return pipeline, queue, service
|
|
|
|
|
|
def _run_due(queue: MemoryWorkQueue, service: ExecutionService) -> int:
|
|
"""What the engine's timer thread does once the window has passed."""
|
|
moved = queue.move_due(time.time())
|
|
for item in queue.claim(10, 10):
|
|
service._run_item(item)
|
|
return moved
|
|
|
|
|
|
def test_a_limited_output_publishes_its_last_value_when_the_window_ends():
|
|
"""A producer going quiet must not strand the reading it held back."""
|
|
readings = iter([1.0, 2.0])
|
|
source = make_node(
|
|
"source",
|
|
lambda params: {"temp": next(readings)},
|
|
provides=[spec("temp", interval=WINDOW)],
|
|
)
|
|
pipeline, queue, service = _running([source])
|
|
|
|
pipeline.run({})
|
|
pipeline.run({})
|
|
# Inside the window, so the second reading is held rather than published.
|
|
assert pipeline.state["demo.temp"] == 1.0
|
|
|
|
time.sleep(WINDOW * 2)
|
|
assert _run_due(queue, service) == 1
|
|
|
|
assert pipeline.state["demo.temp"] == 2.0
|
|
|
|
|
|
def test_a_limited_input_wakes_its_node_when_the_window_ends():
|
|
seen: list[float] = []
|
|
source = make_node("source", lambda params: None, provides=[spec("temp")])
|
|
consumer = make_node(
|
|
"consumer",
|
|
lambda temp, params: seen.append(temp),
|
|
requires=[spec("temp", interval=WINDOW)],
|
|
)
|
|
_pipeline, queue, service = _running([source, consumer])
|
|
|
|
source.inject({"temp": 20.0}, durable=False)
|
|
source.inject({"temp": 21.0}, durable=False)
|
|
assert seen == [20.0]
|
|
|
|
time.sleep(WINDOW * 2)
|
|
assert _run_due(queue, service) == 1
|
|
|
|
# The value that arrived inside the window is delivered at the end of it.
|
|
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")])
|
|
consumer = make_node(
|
|
"consumer",
|
|
lambda temp, params: seen.append(temp),
|
|
requires=[spec("temp", interval=3600)],
|
|
)
|
|
pipeline = Pipeline(nodes=[source, consumer])
|
|
|
|
# Pressing Run is an explicit ask; the interval governs the flow's own
|
|
# traffic, not what the person in front of it asked for.
|
|
pipeline.run({})
|
|
pipeline.run({})
|
|
|
|
assert len(seen) == 2
|