stroblme
40864d5ca5
The rate the media dtypes could carry was one frame every second or two: each
was a file on the data volume, an event on the socket, and a request back for
the bytes. This closes both halves of that, and they are one feature.
`save_artifact(..., volatile=True)` writes to a `VolatileStore` — the same
content-addressed store, in `/dev/shm`, bounded by size with the oldest falling
out (`ARTIFACT_VOLATILE_BYTES`, 48 MB under the container's raised `shm_size`).
Nothing sweeps it: a frame nobody kept is not worth walking the store to find.
`ArtifactStore.path` falls through to it, which is what lets a volatile frame be
an ordinary reference everywhere else — the dtype check, a panel's digest scope,
`load_artifact` in a node, and the widget's own fetch all work on one unchanged.
`adopt` copies one into the store when a run records it, so "returned media is
kept, emitted media is not" stays true.
The bytes then go down the flows websocket as a length-prefixed binary frame,
sent just ahead of the `message_value` naming them, so a tile has the frame when
it hears the value moved. Nothing is pushed unasked: a client names the messages
it is drawing (`{"type":"media","names":[…]}`), a panel's list is intersected
with the scope it already had, and only the newest frame per name in a batch is
sent — a client that fell behind is not handed frames it would draw over. The
tunnel relays text only, so a screen reached through a portal falls back to
fetching, which is why the rate table now has two rows.
Around the edges: the remote worker's fetch cache is bounded at last
(`FLUKSIO_ARTIFACT_CACHE_BYTES`), since content addressing means nothing in it
ever expires and a media stream fills it with chunks nothing asks for twice; a
port carrying an image draws the frame in the node panel rather than only
saying `image/png · frame.png · 1.79kB`; and an edge chip says that much instead
of a line of hash. The media screenshot stops waiting for `networkidle` — a
camera is a socket that never goes quiet, which is the point of it.
76 lines
2.7 KiB
Python
76 lines
2.7 KiB
Python
"""The Fluksio engine, and the decorators that declare flows in your own code.
|
|
|
|
Two audiences, one import name. In a research repository, ``from fluksio
|
|
import Port, node, Flow`` is the authoring API: it says which of your existing
|
|
functions are nodes and which nodes make up a flow, and ``fluksio sync``
|
|
uploads what it finds. Those names come from :mod:`fluksio.sdk`, which imports
|
|
nothing but the standard library.
|
|
|
|
Inside a node, ``import fluksio`` is not this package at all: the worker
|
|
installs a reporter of its own under that name before any node code runs, so
|
|
what a node gets is :mod:`fluksio_worker.worker_main`'s ``emit`` and
|
|
``save_artifact`` — and inert copies of the decorators, since a module the
|
|
node imports may well declare them at its top. The stubs below stand in the
|
|
same place everywhere else, and say so rather than failing as a missing
|
|
attribute.
|
|
|
|
Nothing is imported from the rest of the package here. Every ``from fluksio.x
|
|
import y`` in the engine passes through this module, and an import cycle or a
|
|
second of start-up cost would both begin here.
|
|
"""
|
|
|
|
import logging
|
|
from importlib.metadata import PackageNotFoundError
|
|
from importlib.metadata import version as _version
|
|
from typing import Any
|
|
|
|
from fluksio.sdk import Flow, Port, node, use
|
|
|
|
try:
|
|
__version__ = _version("fluksio")
|
|
except PackageNotFoundError: # pragma: no cover - a checkout that was never installed
|
|
__version__ = "0.0.0+unknown"
|
|
|
|
__all__ = [
|
|
"Flow",
|
|
"Port",
|
|
"emit",
|
|
"load_artifact",
|
|
"node",
|
|
"save_artifact",
|
|
"use",
|
|
]
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_OUTSIDE = (
|
|
"fluksio.{name}() only works inside a node: the worker running it installs "
|
|
"the real one. There is nothing to {verb} out here."
|
|
)
|
|
|
|
|
|
def emit(**ports: Any) -> None:
|
|
"""Publish values on a node's declared output ports, mid-run.
|
|
|
|
Outside a node there is nowhere to publish to, and this does nothing on
|
|
purpose: a node function called directly — in a test, in a notebook, under
|
|
a debugger — is ordinary Python, and having it die on a progress report
|
|
would defeat the point of the code staying yours.
|
|
"""
|
|
logger.debug("fluksio.emit(%s) outside a node", ", ".join(ports))
|
|
|
|
|
|
def save_artifact(
|
|
source: Any,
|
|
name: str = "",
|
|
media_type: str = "application/octet-stream",
|
|
volatile: bool = False,
|
|
) -> dict[str, Any]:
|
|
"""Put bytes in the artifact store and return a reference to them."""
|
|
raise RuntimeError(_OUTSIDE.format(name="save_artifact", verb="save to"))
|
|
|
|
|
|
def load_artifact(ref: dict[str, Any]) -> str:
|
|
"""Fetch what an artifact reference points at, and return a path to it."""
|
|
raise RuntimeError(_OUTSIDE.format(name="load_artifact", verb="load from"))
|