Melvin Strobl
1d6f39a5de
The browser half of M3. Flows open on a full-bleed canvas with their chrome floating over it: flow tabs top, dock bottom, node settings in a panel on the right that leaves the graph visible and running behind it. - Connections are derived, not stored. A node declares the messages it reads and publishes; every matching pair draws an edge, so two producers of one message converge on their consumer. Dragging output to input is shorthand for pointing that input at the producer's message, and asks before it replaces an existing one. - Values land on the edges as they flow, over a websocket that feeds a store outside React, so a value arriving re-renders its own chip and nothing else. Clicking an edge shows the last payload and when it arrived. - Node source is edited in Monaco, loaded only when a panel opens and themed from the design tokens. - Edits autosave; identical documents are skipped server-side, so a quiet canvas writes nothing. - Validation from the API shows on the node it belongs to and is summarised in the dock, where each entry pans to its node. - Works on a phone: touch-connect, 44px dock targets, and the node panel becomes a full-screen sheet. Two new tokens (--status-success, --font-mono) are mirrored in the website repo and recorded in DESIGN-GUIDELINES.md.
469 lines
17 KiB
Python
469 lines
17 KiB
Python
"""Pipeline: the executable graph.
|
|
|
|
One pipeline holds the nodes of every loaded flow. Edges are not declared —
|
|
they follow from message names, so a node consuming ``heating.setpoint`` is
|
|
downstream of every node providing it. Several producers of one message are
|
|
allowed: each publication triggers the consumers, and the latest value wins.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import time
|
|
from collections import deque
|
|
from concurrent.futures import Future, ThreadPoolExecutor, wait
|
|
from typing import Any, Literal
|
|
|
|
from pydantic import BaseModel
|
|
|
|
from app.flow.events import EventBus
|
|
from app.flow.messages import flow_of
|
|
from app.flow.nodes import Node
|
|
from app.flow.state import MemoryState, StateBackend
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class ValidationIssue(BaseModel):
|
|
"""A problem that keeps a flow from running correctly."""
|
|
|
|
code: Literal[
|
|
"cycle",
|
|
"unconnected_input",
|
|
"missing_initial_value",
|
|
"node_error",
|
|
]
|
|
message: str
|
|
flow: str = ""
|
|
nodes: list[str] = []
|
|
node: str | None = None
|
|
port: str | None = None
|
|
message_name: str | None = None
|
|
|
|
|
|
class Pipeline:
|
|
"""Directed graph of nodes with automatic dependency resolution."""
|
|
|
|
__slots__ = (
|
|
"_nodes",
|
|
"_state",
|
|
"_events",
|
|
"_max_workers",
|
|
"produces",
|
|
"dependencies",
|
|
"_edges",
|
|
"_execution_order",
|
|
"_downstream_cache",
|
|
)
|
|
|
|
def __init__(
|
|
self,
|
|
nodes: list[Node] | None = None,
|
|
state: StateBackend | None = None,
|
|
events: EventBus | None = None,
|
|
max_workers: int | None = None,
|
|
initial_values: dict[str, Any] | None = None,
|
|
) -> None:
|
|
self._nodes = nodes or []
|
|
self._state: StateBackend = state or MemoryState()
|
|
self._events = events
|
|
self._max_workers = max_workers
|
|
|
|
# A message may have several producers; every one of them is upstream
|
|
# of the nodes consuming it.
|
|
self.produces: dict[str, list[Node]] = {}
|
|
for node in self._nodes:
|
|
for msg in node.provides:
|
|
self.produces.setdefault(msg, []).append(node)
|
|
|
|
self.dependencies: dict[Node, frozenset[Node]] = {
|
|
node: frozenset(
|
|
producer
|
|
for msg in node.requires
|
|
for producer in self.produces.get(msg, ())
|
|
if producer is not node
|
|
)
|
|
for node in self._nodes
|
|
}
|
|
|
|
self._edges: dict[Node, set[Node]] | None = None
|
|
self._execution_order: list[Node] | None = None
|
|
self._downstream_cache: dict[Node, list[Node]] = {}
|
|
|
|
if initial_values:
|
|
with self._state.lock():
|
|
for name, value in initial_values.items():
|
|
if name not in self._state:
|
|
self._state[name] = value
|
|
|
|
for node in self._nodes:
|
|
node.bind(self)
|
|
|
|
# -------------------------------------------------------------------------
|
|
# Graph
|
|
# -------------------------------------------------------------------------
|
|
|
|
@property
|
|
def nodes(self) -> list[Node]:
|
|
return self._nodes
|
|
|
|
@property
|
|
def state(self) -> StateBackend:
|
|
return self._state
|
|
|
|
@property
|
|
def edges(self) -> dict[Node, set[Node]]:
|
|
"""Producers mapped to their consumers, built on first access."""
|
|
if self._edges is None:
|
|
self._edges = {node: set() for node in self._nodes}
|
|
for consumer, producers in self.dependencies.items():
|
|
for producer in producers:
|
|
self._edges[producer].add(consumer)
|
|
return self._edges
|
|
|
|
def get_node_by_id(self, nid: str) -> Node | None:
|
|
return next((n for n in self._nodes if n.id == nid), None)
|
|
|
|
def flow_nodes(self, flow: str) -> set[Node]:
|
|
return {n for n in self._nodes if n.flow == flow}
|
|
|
|
def _topological_sort(self) -> list[Node]:
|
|
"""Kahn's algorithm; nodes left over are part of a cycle."""
|
|
if self._execution_order is not None:
|
|
return self._execution_order
|
|
|
|
in_degree = {node: len(deps) for node, deps in self.dependencies.items()}
|
|
queue = deque(n for n, deg in in_degree.items() if deg == 0)
|
|
result: list[Node] = []
|
|
|
|
while queue:
|
|
node = queue.popleft()
|
|
result.append(node)
|
|
for consumer in self.edges[node]:
|
|
in_degree[consumer] -= 1
|
|
if in_degree[consumer] == 0:
|
|
queue.append(consumer)
|
|
|
|
self._execution_order = result
|
|
return result
|
|
|
|
def _get_downstream(self, start: Node) -> list[Node]:
|
|
if start not in self._downstream_cache:
|
|
reachable: set[Node] = set()
|
|
queue = deque([start])
|
|
while queue:
|
|
for consumer in self.edges[queue.popleft()]:
|
|
if consumer not in reachable:
|
|
reachable.add(consumer)
|
|
queue.append(consumer)
|
|
order = self._topological_sort()
|
|
ordered = [n for n in order if n in reachable]
|
|
# Nodes inside a cycle never make it into the topological order.
|
|
ordered += [n for n in reachable if n not in order]
|
|
self._downstream_cache[start] = ordered
|
|
return self._downstream_cache[start]
|
|
|
|
# -------------------------------------------------------------------------
|
|
# Validation
|
|
# -------------------------------------------------------------------------
|
|
|
|
def validate(
|
|
self, flow_inputs: dict[str, bool] | None = None
|
|
) -> list[ValidationIssue]:
|
|
"""Report everything that would keep this graph from running.
|
|
|
|
:param flow_inputs: Messages declared as inputs of a flow rather than
|
|
computed by it, mapped to whether they carry an initial value.
|
|
"""
|
|
declared = flow_inputs or {}
|
|
issues: list[ValidationIssue] = []
|
|
|
|
ordered = set(self._topological_sort())
|
|
if len(ordered) != len(self._nodes):
|
|
cyclic = sorted(n.id for n in self._nodes if n not in ordered)
|
|
issues.append(
|
|
ValidationIssue(
|
|
code="cycle",
|
|
message=(
|
|
"These nodes depend on each other in a loop, so none of "
|
|
"them can run: " + ", ".join(cyclic)
|
|
),
|
|
flow=flow_of(cyclic[0]) if cyclic else "",
|
|
nodes=cyclic,
|
|
)
|
|
)
|
|
|
|
for node in self._nodes:
|
|
for msg_name, spec in node.requires.items():
|
|
if msg_name in self.produces:
|
|
continue
|
|
|
|
if msg_name not in declared:
|
|
issues.append(
|
|
ValidationIssue(
|
|
code="unconnected_input",
|
|
message=(
|
|
f"'{node.local_id}' waits for '{msg_name}', "
|
|
"which nothing provides."
|
|
),
|
|
flow=node.flow,
|
|
node=node.id,
|
|
port=spec.port,
|
|
message_name=msg_name,
|
|
)
|
|
)
|
|
elif not declared[msg_name]:
|
|
# Declared as a flow input, but nothing ever sets it, so
|
|
# the node waits forever.
|
|
issues.append(
|
|
ValidationIssue(
|
|
code="missing_initial_value",
|
|
message=(
|
|
f"'{msg_name}' has no starting value, so "
|
|
f"'{node.local_id}' never runs."
|
|
),
|
|
flow=node.flow,
|
|
node=node.id,
|
|
port=spec.port,
|
|
message_name=msg_name,
|
|
)
|
|
)
|
|
return issues
|
|
|
|
# -------------------------------------------------------------------------
|
|
# Readiness (synchronous nodes)
|
|
# -------------------------------------------------------------------------
|
|
|
|
def _version_key(self, msg_name: str) -> str:
|
|
return f"__version__:{msg_name}"
|
|
|
|
def _last_seen_key(self, node_name: str, msg_name: str) -> str:
|
|
return f"__last_seen__:{node_name}:{msg_name}"
|
|
|
|
def _timestamp_key(self, msg_name: str) -> str:
|
|
return f"__ts__:{msg_name}"
|
|
|
|
def _increment_message_versions(self, outputs: dict) -> None:
|
|
for msg_name in outputs:
|
|
self._state.increment(self._version_key(msg_name))
|
|
|
|
def _check_synchronous_ready(self, node: Node) -> tuple[bool, dict[str, int]]:
|
|
"""A synchronous node runs once every input is newer than last time."""
|
|
if not node.requires:
|
|
return True, {}
|
|
|
|
version_keys = [self._version_key(msg) for msg in node.requires]
|
|
last_seen_keys = [self._last_seen_key(node.id, msg) for msg in node.requires]
|
|
values = self._state.get_multi(version_keys + last_seen_keys)
|
|
|
|
current_versions = {}
|
|
all_newer = True
|
|
|
|
for msg_name in node.requires:
|
|
current = values.get(self._version_key(msg_name)) or 0
|
|
last_seen = values.get(self._last_seen_key(node.id, msg_name)) or 0
|
|
current_versions[msg_name] = current
|
|
if current == 0 or current <= last_seen:
|
|
all_newer = False
|
|
|
|
return all_newer, current_versions
|
|
|
|
def _try_acquire_synchronous_execution(
|
|
self, node: Node, current_versions: dict[str, int]
|
|
) -> bool:
|
|
"""Claim the right to execute, so concurrent triggers run a node once."""
|
|
if not current_versions:
|
|
return True
|
|
|
|
expected = {}
|
|
updates = {}
|
|
for msg_name, version in current_versions.items():
|
|
expected[self._version_key(msg_name)] = version
|
|
updates[self._last_seen_key(node.id, msg_name)] = version
|
|
|
|
return self._state.compare_and_swap_multi(expected, updates)
|
|
|
|
def _is_node_ready(self, node: Node, state: StateBackend) -> bool:
|
|
with state.lock():
|
|
for msg_name in node.requires:
|
|
if msg_name not in state:
|
|
return False
|
|
|
|
if not node.synchronous:
|
|
return True
|
|
|
|
is_ready, current_versions = self._check_synchronous_ready(node)
|
|
if not is_ready:
|
|
return False
|
|
return self._try_acquire_synchronous_execution(node, current_versions)
|
|
|
|
# -------------------------------------------------------------------------
|
|
# Execution
|
|
# -------------------------------------------------------------------------
|
|
|
|
def _publish(self, event: dict[str, Any]) -> None:
|
|
if self._events is not None:
|
|
self._events.publish(event)
|
|
|
|
def _execute_node(self, node: Node, state: StateBackend) -> dict | None:
|
|
"""Run one node and record its outputs. Never raises."""
|
|
started = time.perf_counter()
|
|
try:
|
|
with state.lock():
|
|
inputs = {k: state[k] for k in node.requires if k in state}
|
|
|
|
result = node.execute(inputs)
|
|
|
|
if result:
|
|
ts = time.time()
|
|
with state.lock():
|
|
state.update(result)
|
|
state.update(
|
|
{self._timestamp_key(name): ts for name in result},
|
|
)
|
|
self._increment_message_versions(result)
|
|
for name, value in result.items():
|
|
self._publish(
|
|
{
|
|
"type": "message_value",
|
|
"flow": flow_of(name),
|
|
"name": name,
|
|
"value": value,
|
|
"ts": ts,
|
|
}
|
|
)
|
|
|
|
self._publish(
|
|
{
|
|
"type": "node_executed",
|
|
"flow": node.flow,
|
|
"node": node.id,
|
|
"duration_ms": round((time.perf_counter() - started) * 1000, 2),
|
|
"ts": time.time(),
|
|
}
|
|
)
|
|
return result
|
|
except Exception as exc:
|
|
# One failing node must not take the rest of the graph down.
|
|
logger.exception("Node '%s' failed", node.id)
|
|
self._publish(
|
|
{
|
|
"type": "node_error",
|
|
"flow": node.flow,
|
|
"node": node.id,
|
|
"error": f"{type(exc).__name__}: {exc}",
|
|
"ts": time.time(),
|
|
}
|
|
)
|
|
return None
|
|
|
|
def _execute_parallel(
|
|
self,
|
|
nodes_subset: set[Node] | None,
|
|
state: StateBackend,
|
|
check_ready: bool = False,
|
|
) -> StateBackend:
|
|
"""Execute nodes concurrently, scheduling each as its inputs arrive."""
|
|
target_nodes = nodes_subset if nodes_subset is not None else set(self._nodes)
|
|
if not target_nodes:
|
|
return state
|
|
|
|
in_degree = {
|
|
n: sum(1 for dep in self.dependencies[n] if dep in target_nodes)
|
|
for n in target_nodes
|
|
}
|
|
|
|
submitted: set[Node] = set()
|
|
skipped: set[Node] = set()
|
|
node_futures: dict[Node, Future] = {}
|
|
|
|
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 submit_ready(executor: ThreadPoolExecutor) -> None:
|
|
for n in target_nodes:
|
|
if n not in submitted and n not in skipped and is_ready(n):
|
|
submitted.add(n)
|
|
node_futures[n] = executor.submit(self._execute_node, n, state)
|
|
elif n not in submitted and n.synchronous and in_degree[n] == 0:
|
|
# Not ready now; a later trigger may make it ready.
|
|
skipped.add(n)
|
|
|
|
with ThreadPoolExecutor(max_workers=self._max_workers) as executor:
|
|
submit_ready(executor)
|
|
|
|
while node_futures:
|
|
done, _ = wait(node_futures.values(), return_when="FIRST_COMPLETED")
|
|
completed = [n for n, f in node_futures.items() if f in done]
|
|
|
|
for n in completed:
|
|
result = node_futures.pop(n).result()
|
|
# A node returning nothing (rate limiting, an error) stops
|
|
# propagation along its branch.
|
|
if result is not None:
|
|
for consumer in self.edges[n]:
|
|
if consumer in target_nodes:
|
|
in_degree[consumer] -= 1
|
|
submit_ready(executor)
|
|
|
|
return state
|
|
|
|
def run(
|
|
self, inputs: dict | None = None, nodes: set[Node] | None = None
|
|
) -> StateBackend:
|
|
"""Execute the graph (or one flow's nodes) against the shared state."""
|
|
if inputs:
|
|
with self._state.lock():
|
|
self._state.update(inputs)
|
|
self._increment_message_versions(inputs)
|
|
|
|
return self._execute_parallel(nodes, self._state, check_ready=False)
|
|
|
|
def trigger(self, node: Node, outputs: dict | None) -> StateBackend:
|
|
"""Publish a node's outputs and run everything downstream of it."""
|
|
state = self._state
|
|
|
|
if outputs:
|
|
ts = time.time()
|
|
with state.lock():
|
|
state.update(outputs)
|
|
state.update({self._timestamp_key(name): ts for name in outputs})
|
|
self._increment_message_versions(outputs)
|
|
for name, value in outputs.items():
|
|
self._publish(
|
|
{
|
|
"type": "message_value",
|
|
"flow": flow_of(name),
|
|
"name": name,
|
|
"value": value,
|
|
"ts": ts,
|
|
}
|
|
)
|
|
|
|
downstream = set(self._get_downstream(node))
|
|
if not downstream:
|
|
return state
|
|
|
|
return self._execute_parallel(downstream, state, check_ready=True)
|
|
|
|
def values(self, flow: str | None = None) -> dict[str, dict[str, Any]]:
|
|
"""Last value and timestamp of every message, optionally one flow's."""
|
|
out: dict[str, dict[str, Any]] = {}
|
|
with self._state.lock():
|
|
keys = [k for k in self._state.keys() if not k.startswith("__")]
|
|
for key in keys:
|
|
if flow and flow_of(key) != flow:
|
|
continue
|
|
out[key] = {
|
|
"value": self._state.get(key),
|
|
"ts": self._state.get(self._timestamp_key(key)),
|
|
}
|
|
return out
|
|
|
|
def reset(self) -> None:
|
|
self._state.clear()
|