Melvin Strobl
10d7024f85
The panel is going to sparkline how a value moved, and nothing kept more
than the latest one. Emissions now append to a capped list beside the
value — 120 points, enough to fill a sparkline without making Redis a
time-series store — and `/flows/{flow}/history/{message}` hands it back
oldest first.
Only numbers are recorded, bools included as neither, so a string
message costs nothing at all. The response carries `numeric` so an empty
series reads as "not plottable" rather than "nothing yet". The append is
one pipelined round-trip per emission batch and sits outside the lock,
since the list is append-only.
MemoryState keeps the same window in a deque, so the endpoint answers
without Redis too.
488 lines
18 KiB
Python
488 lines
18 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",
|
|
"unauthenticated_hook",
|
|
]
|
|
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[str, Any]) -> 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[str, Any] | 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},
|
|
)
|
|
# Append-only, so it needs no lock of its own.
|
|
state.append_history(result, ts)
|
|
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,
|
|
# A node that returns nothing ran but published nothing,
|
|
# which is a different thing to show than one that emitted.
|
|
"outputs": len(result or {}),
|
|
"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[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 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[str, Any] | 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[str, Any] | 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})
|
|
state.append_history(outputs, ts)
|
|
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,
|
|
}
|
|
)
|
|
# An injecting node — an MQTT subscriber, a webhook — publishes
|
|
# without going through the executor, but it did emit.
|
|
self._publish(
|
|
{
|
|
"type": "node_executed",
|
|
"flow": node.flow,
|
|
"node": node.id,
|
|
"outputs": len(outputs),
|
|
"duration_ms": 0,
|
|
"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()
|