stroblme
99534a82ee
A wheel whose top-level module is `app` collides with anything else in a user's venv, so the package that is about to be published takes the name it is published under. Only the Python package moves; the repo, the Docker WORKDIR and the compose project keep theirs.
85 lines
3.2 KiB
Python
85 lines
3.2 KiB
Python
"""Event bus bridging the engine's worker threads to async subscribers.
|
|
|
|
Nodes execute in a thread pool; websocket clients live on the event loop.
|
|
Publishers are therefore thread-safe and never block: a subscriber that cannot
|
|
keep up loses its oldest queued events rather than stalling the engine.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
from collections import deque
|
|
from collections.abc import AsyncIterator
|
|
from contextlib import asynccontextmanager
|
|
from typing import Any
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
QUEUE_SIZE = 256
|
|
LOG_HISTORY = 400
|
|
|
|
|
|
class EventBus:
|
|
"""Fan-out of engine events to any number of async subscribers."""
|
|
|
|
def __init__(self) -> None:
|
|
self._loop: asyncio.AbstractEventLoop | None = None
|
|
self._subscribers: set[asyncio.Queue[dict[str, Any]]] = set()
|
|
# Kept whether or not anyone is listening, so opening the log panel
|
|
# shows what just happened rather than an empty box.
|
|
self.recent_logs: deque[dict[str, Any]] = deque(maxlen=LOG_HISTORY)
|
|
# How often each qualified node has emitted, for the same reason: a
|
|
# client that reconnects between two pages would otherwise start from
|
|
# zero and draw a busy graph as idle. A session count, not a metric —
|
|
# the rollups in `metrics.py` are what survives a restart.
|
|
self.emits: dict[str, int] = {}
|
|
|
|
def bind(self, loop: asyncio.AbstractEventLoop) -> None:
|
|
"""Attach the bus to the running event loop (called once at startup)."""
|
|
self._loop = loop
|
|
|
|
def publish(self, event: dict[str, Any]) -> None:
|
|
"""Publish an event from any thread."""
|
|
kind = event.get("type")
|
|
if kind == "node_log":
|
|
# deque.append is atomic, so worker threads need no lock here.
|
|
self.recent_logs.append(event)
|
|
elif kind == "node_executed" and event.get("outputs"):
|
|
# The condition the canvas animates on, so both ends agree on what
|
|
# counts as an emission. Two worker threads racing here can lose an
|
|
# increment, which is a count nobody is going to miss.
|
|
node = str(event.get("node") or "")
|
|
self.emits[node] = self.emits.get(node, 0) + 1
|
|
loop = self._loop
|
|
if loop is None or not self._subscribers:
|
|
return
|
|
try:
|
|
loop.call_soon_threadsafe(self._dispatch, event)
|
|
except RuntimeError:
|
|
# Loop already closed — shutting down.
|
|
pass
|
|
|
|
def _dispatch(self, event: dict[str, Any]) -> None:
|
|
for queue in self._subscribers:
|
|
if queue.full():
|
|
# Drop the oldest so a slow client never blocks the engine.
|
|
try:
|
|
queue.get_nowait()
|
|
except asyncio.QueueEmpty:
|
|
pass
|
|
queue.put_nowait(event)
|
|
|
|
@asynccontextmanager
|
|
async def subscribe(self) -> AsyncIterator[asyncio.Queue[dict[str, Any]]]:
|
|
"""Yield a queue receiving every event published while subscribed."""
|
|
queue: asyncio.Queue[dict[str, Any]] = asyncio.Queue(maxsize=QUEUE_SIZE)
|
|
self._subscribers.add(queue)
|
|
try:
|
|
yield queue
|
|
finally:
|
|
self._subscribers.discard(queue)
|
|
|
|
|
|
event_bus = EventBus()
|