From 7132d26f3672aeceb221861dcea72c7f3daa54d9 Mon Sep 17 00:00:00 2001 From: stroblme Date: Wed, 11 Nov 2026 12:03:43 +0100 Subject: [PATCH] Stopping a live flow ends the calls it has running, on workers and in the local pool, and deleting one now ends its local calls too; what a call returns after its flow was stopped or deleted is dropped by a per-flow count the node compares, so a flow started again never takes an old call's answer. A live remote call's id carries a suffix, so two calls of one node no longer share it, and a call is registered before it queues for a slot, so ending its flow or run reaches it and it is never sent; a run's cancel goes through the call's own thread as well --- backend/fluksio/flow/controller.py | 27 ++++- backend/fluksio/flow/nodes/base.py | 9 +- backend/fluksio/flow/pipeline.py | 31 +++++- backend/fluksio/flow/remote.py | 50 +++++---- backend/tests/flow/test_remote.py | 131 ++++++++++++++++++------ backend/tests/flow/test_service_node.py | 5 +- docs/code/workers.md | 8 +- docs/concepts/flows.md | 9 +- 8 files changed, 195 insertions(+), 75 deletions(-) diff --git a/backend/fluksio/flow/controller.py b/backend/fluksio/flow/controller.py index e4fd9ee..30c706c 100644 --- a/backend/fluksio/flow/controller.py +++ b/backend/fluksio/flow/controller.py @@ -699,7 +699,9 @@ class FlowController: else: self.disabled = self.disabled | {flow} self.pipeline.set_disabled(self.disabled) + self.pipeline.end_running(flow) await self._teardown(flow) + await self._end_calls(flow) finally: self._lock.release() @@ -862,10 +864,7 @@ class FlowController: if built is None: if name: pipeline.remove_flow(name) - # A call still out on a worker would keep its slot and its - # process there, for a result that is now dropped. - if self.remote is not None: - self.remote.cancel_flow(name) + await self._end_calls(name) self.loaded = { node_id: entry for node_id, entry in self.loaded.items() @@ -1142,6 +1141,26 @@ class FlowController: else: await self.supervisor.cancel_flow(flow) + async def _end_calls(self, flow: str) -> None: + """End the calls a stopped or deleted flow has running, here and on + workers. A call left out there would keep its slot and its process for + a result that is dropped (`Pipeline.end_running`).""" + if self.remote is not None: + self.remote.cancel_flow(flow) + pool = self.workers + if pool is None: + return + nodes = [ + node_id for node_id, entry in self.loaded.items() if entry.flow == flow + ] + + def cancel() -> None: + for node_id in nodes: + pool.cancel(node_id) + + # Off the loop: a killed worker is waited for. + await run_in_threadpool(cancel) + async def _activate(self, flow: str | None = None) -> None: """Start subscriptions, schedules and webhooks of the new pipeline. diff --git a/backend/fluksio/flow/nodes/base.py b/backend/fluksio/flow/nodes/base.py index ad406c1..42a0972 100644 --- a/backend/fluksio/flow/nodes/base.py +++ b/backend/fluksio/flow/nodes/base.py @@ -464,11 +464,12 @@ class Node: # A source node asked to inject nothing produces its own data. if not outputs and not self.requires: + ended = self._pipeline.ended(self.flow) collected = logs.Collector() with logs.capture(collected): outputs = self.f(params=self.params) or {} - if self._pipeline is None: - # Its flow was deleted while it ran (`Pipeline.remove_flow`). + if self._pipeline is None or self._pipeline.ended(self.flow) != ended: + # Its flow was stopped or deleted while it ran. return None self._pipeline.publish_log(self, collected, "") @@ -490,7 +491,11 @@ class Node: :returns: Node outputs if successful, or pipeline execution results if bound. :rtype: dict | None """ + ended = self._pipeline.ended(self.flow) if self._pipeline is not None else 0 outputs = self.execute(inputs) if self._pipeline is None: return outputs + if self._pipeline.ended(self.flow) != ended: + # Its flow was stopped while it ran. + return None return self._pipeline.trigger(self, outputs, durable=durable) diff --git a/backend/fluksio/flow/pipeline.py b/backend/fluksio/flow/pipeline.py index f7da81b..15978f6 100644 --- a/backend/fluksio/flow/pipeline.py +++ b/backend/fluksio/flow/pipeline.py @@ -308,6 +308,7 @@ class Pipeline: "emission_observer", "run_cache", "run_id", + "_ended", ) def __init__( @@ -373,6 +374,9 @@ class Pipeline: # How deep to keep each message's series; a chart asking for more # than the default puts its message in here. Swapped, never mutated. self.history_limits: dict[str, int] = {} + # How often each flow had what it was running ended (`end_running`). + # Swapped, never mutated. + self._ended: dict[str, int] = {} ( self.produces, @@ -493,6 +497,22 @@ class Pipeline: self.replace_flow(flow, []) for node in gone: node.bind(None) + self.end_running(flow) + + def end_running(self, flow: str) -> None: + """Drop what this flow's nodes return from what they are running now. + + For a flow that was stopped or deleted. Started again, its nodes run + as before, and a call from before still answering is not taken for + theirs. + """ + with self._gate_lock: + self._ended = {**self._ended, flow: self._ended.get(flow, 0) + 1} + + def ended(self, flow: str) -> int: + """How often *flow* had what it was running ended: a node compares it + before and after it runs.""" + return self._ended.get(flow, 0) def _seed(self, initial_values: dict[str, Any] | None) -> None: """Give messages a starting value, without overwriting one already there. @@ -1093,6 +1113,7 @@ class Pipeline: """ started = time.perf_counter() collected = logs.Collector() + ended = self.ended(node.flow) 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 @@ -1127,8 +1148,8 @@ class Pipeline: result = node.execute(inputs) finally: ARRIVED.reset(token) - if node._pipeline is not self: - # Its flow was deleted while it ran (`remove_flow`). + if self.ended(node.flow) != ended: + # Its flow was stopped or deleted while it ran (`end_running`). return None self.publish_log(node, collected, "") @@ -1191,9 +1212,9 @@ class Pipeline: ) return result or {} except Exception as exc: - if node._pipeline is not self: - # Deleted while it ran, and most likely failing because the - # call was ended for that: nobody is left to tell. + if self.ended(node.flow) != ended: + # Stopped or deleted while it ran, and most likely failing + # because the call was ended for that. return None # One failing node must not take the rest of the graph down. error = self.publish_error(node, exc, collected, entry_id) diff --git a/backend/fluksio/flow/remote.py b/backend/fluksio/flow/remote.py index 138053a..df6bdce 100644 --- a/backend/fluksio/flow/remote.py +++ b/backend/fluksio/flow/remote.py @@ -27,13 +27,19 @@ import queue import sys import threading import time +import uuid from collections.abc import Callable from dataclasses import dataclass from typing import Any from fluksio.flow.nodes.base import ARRIVED, NodeOutputError from fluksio.flow.resources import ResourceAccountant -from fluksio.flow.workers import NodeTimeout, RemoteError, _remote_class +from fluksio.flow.workers import ( + NodeCancelled, + NodeTimeout, + RemoteError, + _remote_class, +) logger = logging.getLogger(__name__) @@ -189,13 +195,19 @@ class RemoteWorker: call_id = str(payload["call_id"]) if inbox is None: inbox = queue.Queue() - # Blocking here is the backpressure, exactly as taking a slot is in the - # local pool. - if slot: - self._slots.acquire() + # Registered before it queues for a slot, so ending its flow or its run + # reaches a call still waiting for one. with self._lock: self._pending[call_id] = inbox try: + # Blocking here is the backpressure, exactly as taking a slot is in + # the local pool. + if slot: + self._slots.acquire() + if not inbox.empty(): + # Ended, or the worker gone, while it waited: nothing went out, + # so there is nothing on the worker to cancel. + raise NodeCancelled(f"ended before '{self.name}' was sent the call") future = asyncio.run_coroutine_threadsafe(self._send(payload), self._loop) future.result(timeout=SEND_TIMEOUT_S) @@ -271,12 +283,9 @@ class RemoteWorker: with self._lock: return len(self._pending) - def calls_of(self, run_id: str) -> list[str]: - with self._lock: - return [call for call in self._pending if call.startswith(f"{run_id}:")] - - def end_calls(self, prefix: str) -> None: - """End every call whose id starts with *prefix*, each once it is sent.""" + def end_calls(self, prefix: str) -> int: + """End every call whose id starts with *prefix*, each once it is sent, + and one still waiting for a slot before it is.""" with self._lock: inboxes = [ inbox @@ -285,6 +294,7 @@ class RemoteWorker: ] for inbox in inboxes: inbox.put(self.END) + return len(inboxes) def cancel(self, call_id: str) -> None: """Ask the worker to kill what it is running for this call.""" @@ -423,7 +433,11 @@ class RemoteWorkerHub: """ payload = { "op": "run", - "call_id": f"{run_id}:{node_id}" if run_id else node_id, + # A live node can run in two cascades at once, so its calls are + # told apart; a run runs each node once. + "call_id": ( + f"{run_id}:{node_id}" if run_id else f"{node_id}:{uuid.uuid4().hex[:8]}" + ), "flow": flow, "node": node, "source": source, @@ -529,18 +543,14 @@ class RemoteWorkerHub: def cancel_run(self, run_id: str) -> int: """Kill whatever this run has executing on any attached worker.""" - stopped = 0 - for worker in self.workers(): - for call in worker.calls_of(run_id): - worker.cancel(call) - stopped += 1 - return stopped + return sum(worker.end_calls(f"{run_id}:") for worker in self.workers()) def cancel_flow(self, flow: str) -> None: """End what a live flow has running on any attached worker. - A live call goes by its node's id, ``flow.node``. A run's carries the - run's id first and a compile its op, so neither is touched. + A live call goes by its node's id and a suffix, ``flow.node:…``. A + run's carries the run's id first and a compile its op, so neither is + touched. """ for worker in self.workers(): worker.end_calls(f"{flow}.") diff --git a/backend/tests/flow/test_remote.py b/backend/tests/flow/test_remote.py index aa11858..05d9a8e 100644 --- a/backend/tests/flow/test_remote.py +++ b/backend/tests/flow/test_remote.py @@ -244,49 +244,65 @@ def test_cancelling_a_run_reaches_only_that_run(loop): call_in_thread(lambda: call("run-b")) assert socket.arrived.wait(5) + socket.arrived.clear() assert hub.cancel_run("run-a") == 1 + # Sent by the call's own thread, once its frame is out. + assert socket.arrived.wait(5) cancels = [frame for frame in socket.sent if frame.get("op") == "cancel"] assert [frame["call_id"] for frame in cancels] == ["run-a:flow.node"] +class Loading(FakeSocket): + """A worker that loads whatever it is sent and leaves each call running.""" + + worker: RemoteWorker + + async def send_json(self, payload: dict) -> None: + await super().send_json(payload) + if payload["op"] == "compile": + self.worker.deliver({"call_id": payload["call_id"], "ok": True}) + + +def live_voice(loop, tmp_path, pool: PythonWorkerPool, local: str = ""): + """A live flow 'voice' whose node 'gpu' runs on a worker and, given + *local*, a node 'cpu' running that source in *pool*. Built and started.""" + store = FlowStore(tmp_path / "flows") + nodes = [ + NodeDef( + id="gpu", + worker="gpu", + provides=[MessageSpec(name="reply", dtype=DType.STR)], + ) + ] + if local: + nodes.append( + NodeDef(id="cpu", provides=[MessageSpec(name="local", dtype=DType.STR)]) + ) + store.write_node_source("voice", "cpu", local) + store.write_flow(FlowDef(name="voice", nodes=nodes)) + store.write_node_source("voice", "gpu", "def process():\n return None\n") + socket = Loading() + socket.worker = RemoteWorker( + name="gpu1", labels=["gpu"], send=socket.send_json, loop=loop, max_parallel=2 + ) + hub = RemoteWorkerHub() + hub.attach(socket.worker) + controller = FlowController(store, workers=pool, remote=hub) + asyncio.run(controller.reload()) + socket.arrived.clear() + return controller, store, socket + + def test_deleting_a_flow_ends_its_call_and_drops_a_late_answer(loop, tmp_path): """A node that never returns kept its worker slot past its flow's delete, and an answer that came anyway put the deleted flow's value back.""" - store = FlowStore(tmp_path / "flows") - store.write_flow( - FlowDef( - name="voice", - nodes=[ - NodeDef( - id="gpu", - worker="gpu", - provides=[MessageSpec(name="reply", dtype=DType.STR)], - ) - ], - ) - ) - store.write_node_source("voice", "gpu", "def process():\n return None\n") + pool = PythonWorkerPool(python=sys.executable, size=1) + controller, store, socket = live_voice(loop, tmp_path, pool) + worker = socket.worker - class Agent(FakeSocket): - """Loads whatever it is sent, and leaves a call running until told.""" - - async def send_json(self, payload: dict) -> None: - await super().send_json(payload) - if payload["op"] == "compile": - worker.deliver({"call_id": payload["call_id"], "ok": True}) - - socket = Agent() - worker = RemoteWorker(name="gpu1", labels=["gpu"], send=socket.send_json, loop=loop) - hub = RemoteWorkerHub() - hub.attach(worker) - controller = FlowController( - store, workers=PythonWorkerPool(python=sys.executable, size=1), remote=hub - ) - asyncio.run(controller.reload()) - - socket.arrived.clear() running = call_in_thread(lambda: controller.run_flow("voice")) assert socket.arrived.wait(5) + call = socket.sent[-1]["call_id"] assert socket.sent[-1]["op"] == "run" socket.arrived.clear() @@ -294,15 +310,62 @@ def test_deleting_a_flow_ends_its_call_and_drops_a_late_answer(loop, tmp_path): asyncio.run(controller.reload_flow("voice")) controller.forget_flow(key) assert socket.arrived.wait(5) - assert socket.sent[-1] == {"op": "cancel", "call_id": "voice.gpu"} + assert socket.sent[-1] == {"op": "cancel", "call_id": call} - worker.deliver({"call_id": "voice.gpu", "ok": True, "result": {"reply": "late"}}) + worker.deliver({"call_id": call, "ok": True, "result": {"reply": "late"}}) running.join(timeout=5) assert not running.is_alive() assert worker.in_flight == 0 assert "voice.reply" not in controller.state +def test_stopping_a_flow_ends_its_calls_here_and_on_workers(loop, tmp_path): + """A stopped flow's calls ran on, here and on the worker, and what they + returned was written to its state, or, once it was started again, taken + for a new call's: a live call's id was its node's.""" + pool = PythonWorkerPool(python=sys.executable, size=1) + pool.start() + local = "import time\n\n\ndef process():\n time.sleep(60)\n return None\n" + controller, _, socket = live_voice(loop, tmp_path, pool, local=local) + worker = socket.worker + try: + running = call_in_thread(lambda: controller.run_flow("voice")) + assert socket.arrived.wait(5) + old = socket.sent[-1]["call_id"] + deadline = time.monotonic() + 10 + while ("", "voice.cpu") not in pool._running: + assert time.monotonic() < deadline, "the local node never started" + time.sleep(0.01) + + socket.arrived.clear() + asyncio.run(controller.set_enabled("voice", "voice", False)) + assert socket.arrived.wait(5) + assert socket.sent[-1] == {"op": "cancel", "call_id": old} + deadline = time.monotonic() + 5 + while ("", "voice.cpu") in pool._running: + assert time.monotonic() < deadline, "the local node was left running" + time.sleep(0.01) + + # Started again before the worker has answered the old call. + asyncio.run(controller.set_enabled("voice", "voice", True)) + socket.arrived.clear() + gpu = controller.get_node("voice.gpu") + assert gpu is not None + again = call_in_thread(lambda: controller.pipeline.run({}, nodes={gpu})) + assert socket.arrived.wait(5) + new = socket.sent[-1]["call_id"] + + worker.deliver({"call_id": old, "ok": True, "result": {"reply": "old"}}) + running.join(timeout=5) + assert not running.is_alive() + assert "voice.reply" not in controller.state + worker.deliver({"call_id": new, "ok": True, "result": {"reply": "new"}}) + again.join(timeout=5) + assert controller.state.get("voice.reply") == "new" + finally: + pool.stop() + + def test_a_service_and_a_compile_leave_the_only_slot_to_calls(loop): """A service holding the slot for good would leave the worker blocked.""" hub = RemoteWorkerHub() diff --git a/backend/tests/flow/test_service_node.py b/backend/tests/flow/test_service_node.py index bcce29d..a8c89c3 100644 --- a/backend/tests/flow/test_service_node.py +++ b/backend/tests/flow/test_service_node.py @@ -180,10 +180,11 @@ def test_a_service_that_never_gets_healthy_reads_down(monkeypatch): asyncio.run(scenario()) -def test_a_stop_right_after_the_start_ends_the_serve_after_it(monkeypatch): +def test_a_stop_right_after_the_start_leaves_no_server_running(monkeypatch): """The stop's cancel went out on a thread of its own, so with the serve's thread slow to start it reached the worker first. A worker ignores a cancel for a call it does not know, and ran that server until its socket dropped. + Ended before its thread has sent it, the serve now never goes out. """ monkeypatch.setattr(remote, "SILENCE_S", 1.0) monkeypatch.setattr(service, "STOP_WAIT_S", 1.0) @@ -228,4 +229,4 @@ def test_a_stop_right_after_the_start_ends_the_serve_after_it(monkeypatch): await node.stop() asyncio.run(scenario()) - assert [frame["op"] for frame in agent.sent] == ["serve", "cancel"] + assert agent.sent == [] diff --git a/docs/code/workers.md b/docs/code/workers.md index ca53308..3149c89 100644 --- a/docs/code/workers.md +++ b/docs/code/workers.md @@ -96,10 +96,10 @@ Source code travels with each call and compiles on the worker, so remote-only dependencies do not need to be installed on the engine. Calls to `emit`, `save_artifact` and `load_artifact` work there too. -Cancelling a run stops its active node processes. Deleting a flow does the same -for the calls it has out, and drops an answer that still comes back. A -disconnected worker fails in-flight calls. While executing, the agent sends -heartbeats every ten seconds; 90 seconds without contact fails the call. +Cancelling a run stops its active node processes. Stopping or deleting a flow +does the same for the calls it has out, and drops an answer that still comes +back. A disconnected worker fails in-flight calls. While executing, the agent +sends heartbeats every ten seconds; 90 seconds without contact fails the call. Heartbeats do not reset the node's own silence timeout. Every call runs in a process group of its own. The agent kills the group when diff --git a/docs/concepts/flows.md b/docs/concepts/flows.md index cc4a524..f08f620 100644 --- a/docs/concepts/flows.md +++ b/docs/concepts/flows.md @@ -167,10 +167,11 @@ Build issues prevent the affected flow from running. `node_unhealthy` and `node_degraded` are runtime issues: other work continues, and they clear when the node recovers. A degraded node is a warning on Home rather than a fault. -You can **stop** a live flow to remove its subscriptions and schedules, or -**pause** it to hold pending work while values still arrive. **Step** releases -one held item. Repeated background crashes can quarantine a flow; publishing -a change gives it another chance. +You can **stop** a live flow to remove its subscriptions and schedules and end +the node calls it has running, or **pause** it to hold pending work while +values still arrive. What a stopped call still returns is dropped. **Step** +releases one held item. Repeated background crashes can quarantine a flow; +publishing a change gives it another chance. ## Where to next