From 400d7d9c5c342f0d13c403d18e3b81ed6b6b7c89 Mon Sep 17 00:00:00 2001 From: stroblme Date: Mon, 24 Aug 2026 20:31:31 +0200 Subject: [PATCH] Stage caching for batch runs, and an engine that lives in the command MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A code node in a batch run is now fingerprinted by its source, its raw settings and the values it reads — an artifact input counting as its digest, which is what the content addressing was always for. A run that finds the key restores what the earlier one returned and skips the node, recorded as `cached`. The run history is the cache: `run_node.outputs` beside the `cache_key` the schema already had, no second store. On for code nodes, never for the built-in and connector types that have side effects; off per node with `@node(cache=False)` and per run with `--no-cache`. Emissions are not replayed on a hit, so a cached training node returns its result without redrawing its curve. Recorded in NOTEPAD.md with the two other deliberate limits. `fluksio run --local` boots the real app in the command's own process and drives it through its ASGI interface behind the ordinary client, so a run no longer needs a `serve` terminal beside it — same data directory, same history, and the cache carries between the two. It always waits, because the engine it starts lives exactly as long as the command. Also: `fluksio sweep --param lr=0.1,0.01` for the product of the lists, `run --follow` for a run's numbers as they arrive, Ctrl-C cancelling a waited run rather than abandoning it, coloured statuses on a terminal, and `name` made optional on the metrics endpoint so a follower can ask for every series. Co-Authored-By: Claude Opus 5 (1M context) --- .../versions/c9a5d1e73b48_stage_cache.py | 42 +++ backend/fluksio/api/routes/runs.py | 24 +- backend/fluksio/flow/controller.py | 23 +- backend/fluksio/flow/nodes/base.py | 5 + backend/fluksio/flow/pipeline.py | 116 ++++++- backend/fluksio/flow/runs.py | 79 ++++- backend/fluksio/flow/schemas.py | 8 + backend/fluksio/main.py | 1 + backend/fluksio/models.py | 12 +- backend/fluksio/sdk/__init__.py | 10 + backend/fluksio/sdk/cli.py | 305 ++++++++++++++++-- backend/fluksio/sdk/client.py | 24 +- backend/tests/api/routes/test_runs.py | 77 +++++ backend/tests/flow/test_runs.py | 104 +++++- backend/tests/flow/test_stage_cache_build.py | 83 +++++ backend/tests/test_cli.py | 118 +++++++ docs/code/api.md | 4 +- docs/code/cli.md | 49 ++- docs/concepts/runs.md | 40 ++- docs/getting-started/data-science.md | 33 +- frontend/src/client/schemas.gen.ts | 32 ++ frontend/src/client/sdk.gen.ts | 2 +- frontend/src/client/types.gen.ts | 14 +- 23 files changed, 1147 insertions(+), 58 deletions(-) create mode 100644 backend/fluksio/alembic/versions/c9a5d1e73b48_stage_cache.py create mode 100644 backend/tests/api/routes/test_runs.py create mode 100644 backend/tests/flow/test_stage_cache_build.py diff --git a/backend/fluksio/alembic/versions/c9a5d1e73b48_stage_cache.py b/backend/fluksio/alembic/versions/c9a5d1e73b48_stage_cache.py new file mode 100644 index 0000000..1954ab6 --- /dev/null +++ b/backend/fluksio/alembic/versions/c9a5d1e73b48_stage_cache.py @@ -0,0 +1,42 @@ +"""run_node.outputs, run.no_cache + +What a node returned, so a later run with the same cache key can skip it and +restore its outputs instead of executing it again. `run.no_cache` is the +per-run way to ask for the whole graph to run regardless. + +Revision ID: c9a5d1e73b48 +Revises: b7c21f4d9e30 +Create Date: 2026-08-24 + +""" + +import sqlalchemy as sa +import sqlmodel.sql.sqltypes +from alembic import op + +# revision identifiers, used by Alembic. +revision = "c9a5d1e73b48" +down_revision = "b7c21f4d9e30" +branch_labels = None +depends_on = None + + +def upgrade(): + op.add_column( + "run_node", + sa.Column("outputs", sqlmodel.sql.sqltypes.AutoString(), nullable=True), + ) + op.add_column( + "run", + sa.Column( + "no_cache", + sa.Boolean(), + nullable=False, + server_default=sa.text("0"), + ), + ) + + +def downgrade(): + op.drop_column("run", "no_cache") + op.drop_column("run_node", "outputs") diff --git a/backend/fluksio/api/routes/runs.py b/backend/fluksio/api/routes/runs.py index c7332aa..d8bf099 100644 --- a/backend/fluksio/api/routes/runs.py +++ b/backend/fluksio/api/routes/runs.py @@ -30,6 +30,8 @@ class RunCreate(BaseModel): seed: int | None = None #: Run the unpublished draft instead of what is published. draft: bool = False + #: Execute every node, whatever an earlier run already worked out. + no_cache: bool = False class SweepEntry(BaseModel): @@ -40,6 +42,7 @@ class SweepEntry(BaseModel): class SweepCreate(BaseModel): runs: list[SweepEntry] = Field(default_factory=list) draft: bool = False + no_cache: bool = False class RunNodeRow(BaseModel): @@ -50,6 +53,9 @@ class RunNodeRow(BaseModel): worker: str error: str logs: str + #: What this node's result was looked up by. Empty when it may not be + #: reused; `status` is "cached" when it was. + cache_key: str = "" class ArtifactRow(BaseModel): @@ -96,6 +102,7 @@ class MetricPoint(BaseModel): step: int ts: float value: float + name: str = "" class MetricSeries(BaseModel): @@ -132,6 +139,7 @@ async def create_run( cause="api", actor=user.email, draft=body.draft, + no_cache=body.no_cache, ) except FlowNotFound as exc: raise HTTPException(status_code=404, detail=str(exc)) from exc @@ -168,6 +176,7 @@ async def create_sweep( cause="sweep", actor=user.email, draft=body.draft, + no_cache=body.no_cache, ) for entry in body.runs ] @@ -233,17 +242,18 @@ async def cancel_run(run_id: str, request: Request, session: SessionDep) -> Any: @router.get("/{run_id}/metrics", response_model=list[MetricPoint]) -def read_metrics(run_id: str, session: SessionDep, name: str, stride: int = 1) -> Any: - """One metric's series, in step order. +def read_metrics( + run_id: str, session: SessionDep, name: str = "", stride: int = 1 +) -> Any: + """One metric's series, in step order — or every one of them, unnamed. ``stride`` thins a long curve down: 3000 steps drawn on a 400-pixel chart is 3000 points nobody can see. """ - statement = ( - select(RunMetric) - .where(col(RunMetric.run_id) == run_id, col(RunMetric.name) == name) - .order_by(col(RunMetric.step)) - ) + statement = select(RunMetric).where(col(RunMetric.run_id) == run_id) + if name: + statement = statement.where(col(RunMetric.name) == name) + statement = statement.order_by(col(RunMetric.name), col(RunMetric.step)) rows = list(session.exec(statement)) if stride > 1: rows = rows[:: max(1, stride)] diff --git a/backend/fluksio/flow/controller.py b/backend/fluksio/flow/controller.py index 13b9e53..4786be1 100644 --- a/backend/fluksio/flow/controller.py +++ b/backend/fluksio/flow/controller.py @@ -15,6 +15,8 @@ nodes are built like any other's, so `set_enabled` only starts or stops them. from __future__ import annotations import asyncio +import hashlib +import json import logging import time import traceback @@ -51,7 +53,13 @@ from fluksio.flow.nodes import ( SwitchNode, TriggerNode, ) -from fluksio.flow.pipeline import NodeOutcome, Pipeline, ValidationIssue, ValueSource +from fluksio.flow.pipeline import ( + NodeOutcome, + Pipeline, + RunCacheLookup, + ValidationIssue, + ValueSource, +) from fluksio.flow.remote import RemoteWorkerHub from fluksio.flow.schemas import ( BrainEdge, @@ -904,6 +912,17 @@ class FlowController: params=params, name=node_def.id, ) + if run is not None and node_def.cache: + # The raw params, not the resolved ones: a secret's value + # must not end up in a key, and its name is what changes + # when the node is reconfigured anyway. + node.fingerprint = hashlib.sha256( + json.dumps( + {"source": code, "params": node_def.params}, + sort_keys=True, + separators=(",", ":"), + ).encode() + ).hexdigest() emissions.node = node else: node = node_type.cls( @@ -1339,6 +1358,7 @@ class FlowController: observer: Callable[[NodeOutcome], None] | None = None, emission_observer: Callable[[str, dict[str, Any]], None] | None = None, run: RunContext | None = None, + run_cache: RunCacheLookup | None = None, ) -> Pipeline: """Build one flow as a pipeline of its own, for a single run. @@ -1359,6 +1379,7 @@ class FlowController: initial_values=initial_values, observer=observer, emission_observer=emission_observer, + run_cache=run_cache, ) pipeline.history_limits = self.history_limits return pipeline diff --git a/backend/fluksio/flow/nodes/base.py b/backend/fluksio/flow/nodes/base.py index 6a8b024..cd097c4 100644 --- a/backend/fluksio/flow/nodes/base.py +++ b/backend/fluksio/flow/nodes/base.py @@ -91,6 +91,7 @@ class Node: "synchronous", "_on_health", "supervisor", + "fingerprint", ) def __init__( @@ -107,6 +108,10 @@ class Node: # Set by the controller before start(); a node built on its own (tests, # previews) runs its loops unsupervised. self.supervisor: Supervisor | None = None + # Everything about this node a cached result would depend on, hashed. + # Empty means it is not cacheable — which is every node type that is + # not code, and every node built for anything but a run. + self.fingerprint: str = "" self.params = dict(params) if params else {} self.synchronous = bool(self.params.get("synchronous", False)) diff --git a/backend/fluksio/flow/pipeline.py b/backend/fluksio/flow/pipeline.py index bc0fd81..ab4d9fc 100644 --- a/backend/fluksio/flow/pipeline.py +++ b/backend/fluksio/flow/pipeline.py @@ -13,6 +13,8 @@ construction. A deploy does that rather than building a second pipeline. from __future__ import annotations +import hashlib +import json import logging import threading import time @@ -21,7 +23,7 @@ from collections import deque from collections.abc import Callable, Iterator from concurrent.futures import Future, ThreadPoolExecutor, wait from contextlib import contextmanager -from typing import Any, Literal +from typing import Any, Literal, Protocol from pydantic import BaseModel @@ -99,6 +101,46 @@ class NodeOutcome(BaseModel): #: Artifact references this node emitted, keyed by the message carrying #: them — what a run records so a result can be opened later. artifacts: dict[str, dict[str, Any]] = {} + #: Restored from an earlier run rather than executed. + cached: bool = False + #: What an equal execution of this node would be looked up by. Empty when + #: the node is not cacheable at all. + cache_key: str = "" + #: What it returned, for whoever stores the cache. None when it published + #: nothing, which is a result a later run has to be able to restore too. + output_values: dict[str, Any] | None = None + + +class RunCacheLookup(Protocol): + """Where a pipeline asks whether a node has already been run. + + Kept to one method so the pipeline never learns there is a database: a run + hands it one of these, a test hands it a dict. + """ + + def lookup(self, key: str) -> tuple[bool, dict[str, Any] | None]: + """(hit, outputs). Outputs None on a hit means it published nothing.""" + + +def run_cache_key(fingerprint: str, inputs: dict[str, Any]) -> str: + """What this node, with these inputs, is known by. + + An artifact input counts as its digest: the reference carries a name and a + size beside it, and the same bytes under another name are the same input. + A value JSON cannot carry cannot be part of a key, and a node reading one + is simply not cacheable. + """ + reduced = { + name: value["digest"] if is_reference(value) else value + for name, value in inputs.items() + } + try: + canonical = json.dumps( + {"fp": fingerprint, "in": reduced}, sort_keys=True, separators=(",", ":") + ) + except (TypeError, ValueError): + return "" + return hashlib.sha256(canonical.encode()).hexdigest() def _derive( @@ -156,6 +198,7 @@ class Pipeline: "history_limits", "observer", "emission_observer", + "run_cache", ) def __init__( @@ -170,6 +213,7 @@ class Pipeline: node_pool: ThreadPoolExecutor | None = None, observer: Callable[[NodeOutcome], None] | None = None, emission_observer: Callable[[str, dict[str, Any]], None] | None = None, + run_cache: RunCacheLookup | None = None, ) -> None: self._nodes = nodes or [] # Stopped flows are stored and survive a restart; paused ones are a @@ -200,6 +244,10 @@ class Pipeline: # And every value a node produced on the way, which is what a # training curve is once it goes out a port rather than into a log. self.emission_observer = emission_observer + # Set by a run that may reuse earlier results. A live pipeline has + # none: a cascade is about what just happened, not about what a node + # once returned for the same inputs. + self.run_cache = run_cache # 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] = {} @@ -758,6 +806,60 @@ class Pipeline: # for a value that may never have been delivered at all. return False + def _from_cache( + self, node: Node, key: str, state: StateBackend, entry_id: str + ) -> tuple[bool, dict[str, Any] | None]: + """Restore an earlier run of this node: (hit, what it published). + + Both halves are needed, because a node that published nothing is a + result worth restoring and looks exactly like a miss otherwise. + + The outputs go into state as if the node had just returned them, which + is what everything downstream reads — a run's state namespace is its + own, so a skipped node leaves nothing behind for the next one to find. + What it emitted on the way is not restored: those values were the + story of an execution that is not happening this time. + """ + assert self.run_cache is not None + try: + hit, outputs = self.run_cache.lookup(key) + except Exception: + # A cache that cannot answer is a cache miss, never a failed node. + logger.exception("Cache lookup failed for '%s'", node.id) + return False, None + if not hit: + return False, None + + if outputs: + self._record_outputs(node, outputs, state) + self._publish( + { + "type": "node_executed", + "flow": node.flow, + "node": node.id, + "outputs": len(outputs or {}), + "duration_ms": 0.0, + "run": entry_id, + "ts": time.time(), + } + ) + self._observe( + NodeOutcome( + node=node.id, + ok=True, + cached=True, + cache_key=key, + outputs=len(outputs or {}), + output_values=outputs, + artifacts={ + name: value + for name, value in (outputs or {}).items() + if is_reference(value) + }, + ) + ) + return True, outputs + def _execute_node( self, node: Node, state: StateBackend, entry_id: str = "" ) -> dict[str, Any] | None: @@ -768,6 +870,14 @@ class Pipeline: with state.lock(): inputs = {k: state[k] for k in node.requires if k in state} + key = "" + if self.run_cache is not None and node.fingerprint: + key = run_cache_key(node.fingerprint, inputs) + if key: + hit, restored = self._from_cache(node, key, state, entry_id) + if hit: + return restored + with logs.capture(collected): result = node.execute(inputs) self.publish_log(node, collected, "") @@ -818,6 +928,10 @@ class Pipeline: for name, value in (result or {}).items() if is_reference(value) }, + cache_key=key, + # Post-throttle: what went into state is what a later run + # restoring this node has to find. + output_values=result, ) ) return result diff --git a/backend/fluksio/flow/runs.py b/backend/fluksio/flow/runs.py index 7fdeeac..a8f281a 100644 --- a/backend/fluksio/flow/runs.py +++ b/backend/fluksio/flow/runs.py @@ -43,6 +43,7 @@ from sqlalchemy.dialects.sqlite import insert as upsert from sqlmodel import Session, col, select from fluksio.core.db import engine as db_engine +from fluksio.flow.artifacts import ArtifactStore, is_reference from fluksio.flow.controller import FlowController, RunContext from fluksio.flow.messages import qualify from fluksio.flow.pipeline import NodeOutcome, Pipeline @@ -69,6 +70,10 @@ CLAIM_COUNT = 4 CLAIM_BLOCK_MS = 1000 ERROR_CAP = 2000 LOG_CAP = 8000 +#: How much of a node's returned outputs is kept so a later run can restore +#: them. Past it the node simply is not cacheable — a cache is not a store, +#: and an artifact is how a large result is meant to travel. +OUTPUT_CAP = 256_000 #: Reported numbers held before they are written. A training loop reporting #: every step must not be a round trip every step. METRIC_BATCH = 500 @@ -288,6 +293,68 @@ def origin_commit(flow: FlowDef) -> str: return f"{origin.commit}-dirty" if origin.dirty else origin.commit +def _cacheable(outcome: NodeOutcome) -> str | None: + """A node's outputs as stored, or None when it may not be reused.""" + if not outcome.ok or not outcome.cache_key: + return None + try: + text = json.dumps(outcome.output_values, sort_keys=True, separators=(",", ":")) + except (TypeError, ValueError): + return None + return text if len(text) <= OUTPUT_CAP else None + + +class RunCache: + """What earlier runs already worked out, looked up by cache key. + + The rows are the run history itself — a node that ran successfully and + whose outputs were small enough to keep is a cache entry, without a store + of its own. An entry whose artifacts have gone from the content store is + not a hit: the reference would name bytes nobody can open. + """ + + def __init__(self, artifacts: ArtifactStore | None = None) -> None: + self.artifacts = artifacts + + def lookup(self, key: str) -> tuple[bool, dict[str, Any] | None]: + if not key: + return False, None + try: + with Session(db_engine) as session: + rows = session.exec( + select(RunNode) + .where( + col(RunNode.cache_key) == key, + col(RunNode.status).in_(("ok", "cached")), + col(RunNode.outputs).is_not(None), + ) + # Run ids are time-ordered, so this is the most recent + # first. A handful is enough to get past entries whose + # artifacts have been collected. + .order_by(col(RunNode.run_id).desc()) + .limit(5) + ).all() + for row in rows: + outputs = json.loads(row.outputs or "null") + if outputs and self._unrestorable(outputs): + continue + return True, outputs + except Exception: + # A cache that cannot answer is a cache miss, never a failed run. + logger.exception("Cache lookup failed; running the node instead") + return False, None + + def _unrestorable(self, outputs: dict[str, Any]) -> bool: + return any( + is_reference(value) + and ( + self.artifacts is None + or self.artifacts.path(str(value["digest"])) is None + ) + for value in outputs.values() + ) + + class RunService: """Accepts runs, drives them, and writes down what they did.""" @@ -297,9 +364,11 @@ class RunService: queue: WorkQueue, state_factory: Callable[[str], StateBackend] | None = None, parallel: int = MAX_PARALLEL, + artifacts: ArtifactStore | None = None, ) -> None: self.controller = controller self.queue = queue + self._cache = RunCache(artifacts) # Without one, a run gets a private in-memory state — which is exactly # the isolation it wants, minus surviving the process. self._state_factory = state_factory or (lambda _ns: MemoryState()) @@ -356,6 +425,7 @@ class RunService: cause: str = "api", actor: str = "", draft: bool = False, + no_cache: bool = False, ) -> Run: """Journal a run and wake an engine up for it. Never blocks on it.""" flow = self.controller.store.read_flow(flow_name, draft=draft) @@ -378,6 +448,7 @@ class RunService: seed=seed, group_id=group_id, cause=cause, + no_cache=no_cache, status="queued", labels=required_labels(flow), created_at=datetime.now(UTC), @@ -605,6 +676,7 @@ class RunService: observer=observe, emission_observer=sink.handle, run=RunContext(run_id=run_id), + run_cache=None if run.no_cache else self._cache, ) with self._lock: self._active[run_id] = pipeline @@ -644,14 +716,19 @@ class RunService: logger.warning("Could not clear state of run %s", run_id) def _record_node(self, run_id: str, outcome: NodeOutcome) -> None: + outputs = _cacheable(outcome) row = RunNode( run_id=run_id, node=outcome.node[:255], - status="ok" if outcome.ok else "error", + status="cached" if outcome.cached else ("ok" if outcome.ok else "error"), started_at=datetime.now(UTC), duration_ms=outcome.duration_ms, error=outcome.error[:ERROR_CAP], logs=outcome.logs[:LOG_CAP], + # Together or not at all: a row carrying a key must be one a + # lookup can actually restore from. + cache_key=outcome.cache_key if outputs is not None else "", + outputs=outputs, ) try: with Session(db_engine) as session: diff --git a/backend/fluksio/flow/schemas.py b/backend/fluksio/flow/schemas.py index 5ffb677..9ac8f86 100644 --- a/backend/fluksio/flow/schemas.py +++ b/backend/fluksio/flow/schemas.py @@ -70,6 +70,14 @@ class NodeDef(BaseModel): "locally anyway." ), ) + cache: bool = Field( + default=True, + description=( + "Whether a batch run may reuse an earlier execution of this node " + "with the same source, settings and inputs. Turn it off for a " + "function whose answer can change on its own." + ), + ) @field_validator("id") @classmethod diff --git a/backend/fluksio/main.py b/backend/fluksio/main.py index f257102..c601c5b 100644 --- a/backend/fluksio/main.py +++ b/backend/fluksio/main.py @@ -157,6 +157,7 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: controller=controller, queue=_work_queue("run"), state_factory=_run_state, + artifacts=artifacts, ) app.state.run_service = run_service watchdog = LoopWatchdog(event_bus) diff --git a/backend/fluksio/models.py b/backend/fluksio/models.py index 1618092..557a685 100644 --- a/backend/fluksio/models.py +++ b/backend/fluksio/models.py @@ -342,6 +342,8 @@ class Run(SQLModel, table=True): parent_id: str | None = Field(default=None, max_length=64) #: api, hook, sweep or cli. cause: str = Field(default="api", max_length=32) + #: Re-execute every node, whatever the stage cache holds for it. + no_cache: bool = False #: queued, running, ok, error, cancelled or abandoned. status: str = Field(default="queued", index=True, max_length=16) #: Why it is where it is: what it waits for, or what went wrong. @@ -392,10 +394,14 @@ class RunNode(SQLModel, table=True): worker: str = Field(default="local", max_length=64) error: str = "" logs: str = "" - #: Everything this node's output depends on, hashed. Recorded from the - #: start so that skipping a stage whose inputs have not changed is later a - #: lookup rather than a migration. + #: Everything this node's output depends on, hashed: its source, its + #: settings and the values it read. What a later run looks itself up by. cache_key: str = Field(default="", index=True, max_length=64) + #: What it returned, as canonical JSON, so a node with this key can be + #: skipped and its outputs restored. None when it may not be reused — + #: opted out, too large, or a value JSON cannot carry. Written with + #: `cache_key` or not at all, so a keyed row is always restorable. + outputs: str | None = None class RunMetric(SQLModel, table=True): diff --git a/backend/fluksio/sdk/__init__.py b/backend/fluksio/sdk/__init__.py index 745929a..2215fdf 100644 --- a/backend/fluksio/sdk/__init__.py +++ b/backend/fluksio/sdk/__init__.py @@ -175,6 +175,7 @@ class NodeSpec: timeout: float | None, device: str | None, device_policy: str, + cache: bool = True, ) -> None: self.fn = fn self.id = id @@ -186,6 +187,7 @@ class NodeSpec: self.timeout = timeout self.device = device self.device_policy = device_policy + self.cache = cache def rebind( self, *, id: str = "", wire: dict[str, str] | None = None, **settings: Any @@ -216,6 +218,7 @@ class NodeSpec: timeout=self.timeout, device=self.device, device_policy=self.device_policy, + cache=self.cache, ) @@ -246,6 +249,7 @@ def node( timeout: float | None = None, device: str | None = None, device_policy: str = "require", + cache: bool = True, ) -> Callable[[F], F]: """Mark a function as a node, declaring its ports. @@ -257,6 +261,10 @@ def node( the canvas can tune them without editing code. ``device`` picks the worker the code runs on — ``"gpu"``, say — and ``device_policy="prefer"`` runs it locally when no such worker is attached rather than waiting for one. + + A batch run skips this node when an earlier one already ran the same source + with the same settings and the same input values, and restores what it + returned. ``cache=False`` says not to: the answer can change on its own. """ if device_policy not in ("require", "prefer"): raise SyncError("device_policy is 'require' or 'prefer'") @@ -275,6 +283,7 @@ def node( timeout=timeout, device=device, device_policy=device_policy, + cache=cache, ) _check_signature(spec) fn.__fluksio__ = spec # type: ignore[attr-defined] @@ -417,6 +426,7 @@ def _check(spec: NodeSpec, mode: str) -> dict[str, Any]: "timeout": spec.timeout, "device": spec.device, "device_policy": spec.device_policy, + "cache": spec.cache, } diff --git a/backend/fluksio/sdk/cli.py b/backend/fluksio/sdk/cli.py index 2ca5952..0b855c4 100644 --- a/backend/fluksio/sdk/cli.py +++ b/backend/fluksio/sdk/cli.py @@ -1,7 +1,8 @@ -"""The `fluksio sync`, `run`, `runs` and `login` commands. +"""The `fluksio sync`, `run`, `runs`, `sweep` and `login` commands. Kept beside the SDK rather than in `fluksio.cli`: these are the client half of -the tool, and none of them needs the engine to be importable. +the tool, and none of them needs the engine to be importable — `--local`, which +does, imports it inside the branch that asked for it. """ from __future__ import annotations @@ -9,9 +10,13 @@ from __future__ import annotations import argparse import getpass import importlib +import itertools import json import pkgutil import sys +import time +from collections.abc import Iterator +from contextlib import contextmanager from pathlib import Path from typing import Any @@ -20,6 +25,7 @@ from fluksio.sdk.client import ( GLOBAL_DATA_DIR, ApiError, Client, + RunHandle, data_dir, login, origin_of, @@ -29,6 +35,17 @@ from fluksio.sdk.client import ( __all__ = ["add_parsers", "discover"] +#: Statuses worth a colour, and the SGR code each gets. +_COLORS = { + "ok": "32", + "error": "31", + "cached": "36", + "cancelled": "33", + "abandoned": "31", + "running": "36", + "queued": "33", +} + def _say(message: str = "") -> None: print(message) @@ -39,6 +56,20 @@ def _fail(message: str) -> int: return 1 +def _status(text: str, width: int = 0) -> str: + """A status, coloured when a terminal is reading it. + + Padded before it is coloured: the escape sequences are characters as far + as `str.format` is concerned, and a column that lines up in a pipe would + not line up on screen. + """ + body = f"{text:<{width}}" if width else text + code = _COLORS.get(text) + if not code or not sys.stdout.isatty(): + return body + return f"\033[{code}m{body}\033[0m" + + # --------------------------------------------------------------------------- # Discovery # --------------------------------------------------------------------------- @@ -112,6 +143,72 @@ def discover(targets: list[str]) -> list[Flow]: return list(FLOWS.values()) +# --------------------------------------------------------------------------- +# Which engine +# --------------------------------------------------------------------------- + + +@contextmanager +def _engine_client() -> Iterator[Client]: + """The engine itself, in this process, behind the ordinary client. + + Everything `fluksio serve` does apart from listening on a socket: the same + data directory, the same database, the same admin, the same lifespan. The + app is driven through its ASGI interface, so a run costs what it costs on + a served engine and lands in the same history — which is what makes the + stage cache carry across the two. + + The engine only exists for the length of the command, so nothing here is + written back as a login: a stored token belongs to whichever engine + `fluksio login` was pointed at. + """ + # Imported here, not at module scope: everything else in this file is the + # client half and must keep working with no engine installed. + import logging + + from fluksio.cli import _data_dir, _prepare, _print_new_admin + + directory = _data_dir(None) + # Settings are read when the app is imported, so this comes first. + _prepare(directory) + # Nothing here goes over a network, so httpx logging each call as a + # request to "testserver" is noise that also happens to be untrue. + logging.getLogger("httpx").setLevel(logging.WARNING) + + from datetime import timedelta + + from fastapi.testclient import TestClient + from sqlmodel import Session + + from fluksio.core import security + from fluksio.core.bootstrap import ensure_superuser + from fluksio.core.db import engine + + with Session(engine) as session: + admin, generated = ensure_superuser(session) + admin_id = admin.id + if generated: + _print_new_admin(admin.email, generated) + token = security.create_access_token(admin_id, expires_delta=timedelta(hours=12)) + + from fluksio.main import app + + # Entering the client is what runs the lifespan: the worker pool, the + # controller and the run service all start here and stop on the way out. + with TestClient(app) as http: + yield Client(http=http, token=token) + + +@contextmanager +def _client_for(args: argparse.Namespace) -> Iterator[Client]: + """The engine this command talks to: one running somewhere, or this one.""" + if getattr(args, "local", False): + with _engine_client() as client: + yield client + else: + yield Client(url=args.url, token=args.token) + + # --------------------------------------------------------------------------- # Commands # --------------------------------------------------------------------------- @@ -196,12 +293,17 @@ def _coerce(value: str, dtype: str) -> Any: return json.loads(value) -def _params(definition: dict[str, Any], rest: list[str]) -> dict[str, Any]: - """Turn `--lr 0.05` into a typed parameter, using the flow's own inputs.""" - types = { +def _input_types(definition: dict[str, Any]) -> dict[str, str]: + """What each of a flow's inputs is declared to be.""" + return { str(entry["spec"]["name"]): str(entry["spec"].get("dtype", "float")) for entry in definition.get("inputs") or [] } + + +def _params(definition: dict[str, Any], rest: list[str]) -> dict[str, Any]: + """Turn `--lr 0.05` into a typed parameter, using the flow's own inputs.""" + types = _input_types(definition) params: dict[str, Any] = {} pending: str | None = None for token in rest: @@ -258,42 +360,156 @@ def _sync_first(client: Client) -> None: _say(f"synced {', '.join(r.flow for r in changed)}") -def cmd_run(args: argparse.Namespace, rest: list[str]) -> int: +def _follow(client: Client, handle: RunHandle, poll: float = 1.0) -> None: + """Print a run's numbers as they arrive, until it is over. + + Polled rather than pushed: the engine writes a metric down when it is + reported, so asking once a second draws the same curve a socket would + have, without either side holding a connection open. The status is read + before the numbers, so the last batch is never the one that gets missed. + """ + seen: set[tuple[str, int]] = set() + while True: + done = handle.refresh().done + for point in client.metrics(handle.id): + mark = (str(point.get("name", "")), int(point.get("step", -1))) + if mark in seen: + continue + seen.add(mark) + _say(f" {mark[0]}[{mark[1]}] = {point['value']:g}") + if done: + return + time.sleep(poll) + + +def _cancel(client: Client, handle: RunHandle) -> int: + """Ctrl-C means stop the run, not just stop watching it.""" try: - client = Client(url=args.url, token=args.token) - if not args.no_sync: - _sync_first(client) - stored = client.get_flow(args.flow) - if stored is None: - return _fail(f"no flow '{args.flow}' on that engine") - params = _params(stored.get("definition") or {}, rest) - handle = client.submit(args.flow, params, seed=args.seed) + client.cancel(handle.id) + except (SyncError, ApiError) as exc: + return _fail(f"could not cancel {handle.id}: {exc}") + _say(f"{handle.id} {_status('cancelled')}") + return 130 + + +def _cached_note(client: Client, handle: RunHandle) -> str: + """How much of the run earlier ones had already answered.""" + try: + nodes = client.run(handle.id).get("nodes") or [] + except (SyncError, ApiError): + return "" + cached = sum(1 for node in nodes if node.get("status") == "cached") + return f" ({cached}/{len(nodes)} {_status('cached')})" if cached else "" + + +def cmd_run(args: argparse.Namespace, rest: list[str]) -> int: + # An in-process engine lives exactly as long as this command, so a run + # nobody waits for would be thrown away with the queue holding it. + wait = args.wait or args.follow or args.local + try: + with _client_for(args) as client: + if not args.no_sync: + _sync_first(client) + stored = client.get_flow(args.flow) + if stored is None: + return _fail(f"no flow '{args.flow}' on that engine") + params = _params(stored.get("definition") or {}, rest) + handle = client.submit( + args.flow, params, seed=args.seed, no_cache=args.no_cache + ) + _say(f"{handle.id} queued {json.dumps(params)}") + if not wait: + return 0 + try: + if args.follow: + _follow(client, handle) + else: + handle.wait(timeout=args.timeout) + except KeyboardInterrupt: + return _cancel(client, handle) + _say( + f"{handle.id} {_status(handle.status)} " + f"{json.dumps(handle.result)}{_cached_note(client, handle)}" + ) + return 0 if handle.status == "ok" else 1 except (SyncError, ApiError) as exc: return _fail(str(exc)) - _say(f"{handle.id} queued {json.dumps(params)}") - if not args.wait: - return 0 - handle.wait(timeout=args.timeout) - _say(f"{handle.id} {handle.status} {json.dumps(handle.result)}") - return 0 if handle.status == "ok" else 1 def cmd_runs(args: argparse.Namespace) -> int: try: - rows = Client(url=args.url, token=args.token).runs( - flow=args.flow, limit=args.limit - ) + with _client_for(args) as client: + rows = client.runs(flow=args.flow, limit=args.limit) except (SyncError, ApiError) as exc: return _fail(str(exc)) for row in rows: commit = (row.get("origin_commit") or "")[:7] _say( - f"{row['id']} {row['status']:<9} {row['flow']:<16} " + f"{row['id']} {_status(row['status'], 9)} {row['flow']:<16} " f"{row['duration_ms'] / 1000:7.1f}s {commit:<8} {json.dumps(row['params'])}" ) return 0 +def _grid( + definition: dict[str, Any], values: list[str], seed: int | None +) -> list[dict[str, Any]]: + """`--param lr=0.1,0.01 --param epochs=10,50` — every combination of them.""" + types = _input_types(definition) + names: list[str] = [] + columns: list[list[Any]] = [] + for raw in values: + name, sep, listed = raw.partition("=") + name = name.replace("-", "_") + if not sep or not listed: + raise SyncError(f"--param takes name=value,value — got '{raw}'") + if name not in types: + raise SyncError( + f"'{name}' is not an input of this flow (it takes " + f"{', '.join(sorted(types)) or 'none'})" + ) + names.append(name) + columns.append([_coerce(item, types[name]) for item in listed.split(",")]) + return [ + {"params": dict(zip(names, combination, strict=True)), "seed": seed} + for combination in itertools.product(*columns) + ] + + +def cmd_sweep(args: argparse.Namespace) -> int: + wait = args.wait or args.local + try: + with _client_for(args) as client: + if not args.no_sync: + _sync_first(client) + stored = client.get_flow(args.flow) + if stored is None: + return _fail(f"no flow '{args.flow}' on that engine") + entries = _grid(stored.get("definition") or {}, args.param, args.seed) + handles = client.sweep(args.flow, entries, no_cache=args.no_cache) + for handle, entry in zip(handles, entries, strict=True): + _say(f"{handle.id} queued {json.dumps(entry['params'])}") + if not wait: + return 0 + failed = 0 + try: + for handle in handles: + handle.wait(timeout=args.timeout) + _say( + f"{handle.id} {_status(handle.status)} " + f"{json.dumps(handle.result)}" + ) + failed += handle.status != "ok" + except KeyboardInterrupt: + for handle in handles: + if not handle.refresh().done: + _cancel(client, handle) + return 130 + return 1 if failed else 0 + except (SyncError, ApiError) as exc: + return _fail(str(exc)) + + # --------------------------------------------------------------------------- # Wiring # --------------------------------------------------------------------------- @@ -302,11 +518,17 @@ def cmd_runs(args: argparse.Namespace) -> int: def add_parsers(subparsers: Any) -> None: """Register the client commands on `fluksio`'s parser.""" - def with_engine(sub: argparse.ArgumentParser) -> None: + def with_engine(sub: argparse.ArgumentParser, local: bool = False) -> None: sub.add_argument( "--url", default="", help="the engine (default: the last login)" ) sub.add_argument("--token", default="", help="override the stored token") + if local: + sub.add_argument( + "--local", + action="store_true", + help="boot the engine in this process instead of talking to one", + ) parser = subparsers.add_parser( "login", help="store a token for an engine elsewhere" @@ -348,17 +570,46 @@ def add_parsers(subparsers: Any) -> None: parser.add_argument("flow") parser.add_argument("--seed", type=int, default=None) parser.add_argument("--wait", action="store_true", help="block until it finishes") + parser.add_argument( + "--follow", + action="store_true", + help="wait, printing the numbers it reports as they arrive", + ) parser.add_argument("--timeout", type=float, default=0.0) parser.add_argument( "--no-sync", action="store_true", help="run what is already on the engine, without uploading first", ) - with_engine(parser) + parser.add_argument( + "--no-cache", + action="store_true", + help="execute every node, even one an earlier run already answered", + ) + with_engine(parser, local=True) parser.set_defaults(func=cmd_run) parser = subparsers.add_parser("runs", help="the runs an engine has recorded") parser.add_argument("--flow", default="") parser.add_argument("--limit", type=int, default=20) - with_engine(parser) + with_engine(parser, local=True) parser.set_defaults(func=cmd_runs) + + parser = subparsers.add_parser( + "sweep", help="one flow, once per combination of the parameters given" + ) + parser.add_argument("flow") + parser.add_argument( + "--param", + action="append", + default=[], + metavar="NAME=V1,V2", + help="an input and the values to try; repeat for a grid", + ) + parser.add_argument("--seed", type=int, default=None) + parser.add_argument("--wait", action="store_true", help="block until all finish") + parser.add_argument("--timeout", type=float, default=0.0) + parser.add_argument("--no-sync", action="store_true") + parser.add_argument("--no-cache", action="store_true") + with_engine(parser, local=True) + parser.set_defaults(func=cmd_sweep) diff --git a/backend/fluksio/sdk/client.py b/backend/fluksio/sdk/client.py index 5456793..539cc5b 100644 --- a/backend/fluksio/sdk/client.py +++ b/backend/fluksio/sdk/client.py @@ -197,13 +197,33 @@ class Client: # -- runs -------------------------------------------------------------- def submit( - self, flow: str, params: dict[str, Any] | None = None, seed: int | None = None + self, + flow: str, + params: dict[str, Any] | None = None, + seed: int | None = None, + no_cache: bool = False, ) -> RunHandle: row = self._call( - "POST", f"/runs/flows/{flow}", json={"params": params or {}, "seed": seed} + "POST", + f"/runs/flows/{flow}", + json={"params": params or {}, "seed": seed, "no_cache": no_cache}, ) return RunHandle(self, row["id"], row) + def sweep( + self, + flow: str, + entries: list[dict[str, Any]], + no_cache: bool = False, + ) -> list[RunHandle]: + """Many runs of one flow at once. The caller decides what varies.""" + rows = self._call( + "POST", + f"/runs/flows/{flow}/sweep", + json={"runs": entries, "no_cache": no_cache}, + ) + return [RunHandle(self, row["id"], row) for row in rows] + def run(self, run_id: str) -> dict[str, Any]: result: dict[str, Any] = self._call("GET", f"/runs/{run_id}") return result diff --git a/backend/tests/api/routes/test_runs.py b/backend/tests/api/routes/test_runs.py new file mode 100644 index 0000000..f1b0a19 --- /dev/null +++ b/backend/tests/api/routes/test_runs.py @@ -0,0 +1,77 @@ +"""The stage cache, from the side that needs a database. + +The pipeline half — what a hit restores and what a key is made of — is in +`tests/flow/test_runs.py`, which runs without one. +""" + +import json + +from sqlmodel import Session + +from fluksio.core.db import engine as db_engine +from fluksio.flow.artifacts import ArtifactStore +from fluksio.flow.pipeline import NodeOutcome +from fluksio.flow.runs import OUTPUT_CAP, RunCache, _cacheable +from fluksio.models import RunNode + + +def test_a_run_cache_finds_what_an_earlier_run_recorded(tmp_path): + """The run history is the cache; there is no second store to keep.""" + store = ArtifactStore(tmp_path / "artifacts") + reference = store.put([b"payload"], name="data.bin") + plain, with_artifact, collected = "k-plain", "k-artifact", "k-collected" + with Session(db_engine) as session: + session.add( + RunNode( + run_id="cache-1", + node="study.a", + status="ok", + cache_key=plain, + outputs=json.dumps({"study.loss": 1.5}), + ) + ) + session.add( + RunNode( + run_id="cache-2", + node="study.b", + status="ok", + cache_key=with_artifact, + outputs=json.dumps({"study.data": reference}), + ) + ) + session.add( + RunNode( + run_id="cache-3", + node="study.c", + status="ok", + cache_key=collected, + outputs=json.dumps( + {"study.data": {**reference, "digest": "sha256:" + "1" * 64}} + ), + ) + ) + session.commit() + + cache = RunCache(store) + assert cache.lookup(plain) == (True, {"study.loss": 1.5}) + assert cache.lookup(with_artifact) == (True, {"study.data": reference}) + # Its bytes have gone from the store, so the reference names nothing a + # restored run could open. That is a miss, not a broken run. + assert cache.lookup(collected) == (False, None) + assert cache.lookup("never-seen") == (False, None) + assert cache.lookup("") == (False, None) + + +def test_what_may_be_stored_as_a_cache_entry(): + """A row carries a key and its outputs together, or neither.""" + ok = NodeOutcome( + node="study.a", ok=True, cache_key="k", output_values={"study.loss": 1.0} + ) + assert _cacheable(ok) == '{"study.loss":1.0}' + # A node that published nothing is still an answer worth reusing. + assert _cacheable(ok.model_copy(update={"output_values": None})) == "null" + # Not cacheable: it failed, it has no key, or it returned too much. + assert _cacheable(ok.model_copy(update={"ok": False})) is None + assert _cacheable(ok.model_copy(update={"cache_key": ""})) is None + big = {"study.data": "x" * (OUTPUT_CAP + 1)} + assert _cacheable(ok.model_copy(update={"output_values": big})) is None diff --git a/backend/tests/flow/test_runs.py b/backend/tests/flow/test_runs.py index 5892cc6..7790713 100644 --- a/backend/tests/flow/test_runs.py +++ b/backend/tests/flow/test_runs.py @@ -10,7 +10,7 @@ import pytest from fluksio.flow.messages import DType, MessageSpec from fluksio.flow.nodes import Node -from fluksio.flow.pipeline import NodeOutcome, Pipeline +from fluksio.flow.pipeline import NodeOutcome, Pipeline, run_cache_key from fluksio.flow.runs import ( MetricSink, RunRejected, @@ -333,3 +333,105 @@ def test_a_runs_seed_fills_an_input_of_that_name(): def test_a_flow_without_a_seed_input_ignores_the_runs_seed(): flow = double_flow() assert seed_values(flow, {"lr": 1.0}, seed=7) == {"study.lr": 1.0} + + +# ----------------------------------------------------------------------------- +# The stage cache — a node whose inputs have not changed is not run again +# ----------------------------------------------------------------------------- + + +class FakeCache: + """A stage cache with no database behind it, and a record of what it was asked.""" + + def __init__(self, entries: dict[str, dict | None] | None = None) -> None: + self.entries = entries or {} + self.asked: list[str] = [] + + def lookup(self, key: str): + self.asked.append(key) + if key in self.entries: + return True, self.entries[key] + return False, None + + +def counting_node(flow: str = "study") -> tuple[Node, list[int]]: + """A node that says how many times it actually ran.""" + calls: list[int] = [] + + def train(lr, params): + calls.append(1) + return {"loss": lr * 2} + + node = make_node( + "train", flow, train, requires=[spec("lr")], provides=[spec("loss")] + ) + node.fingerprint = "fp-train" + return node, calls + + +def test_a_cache_hit_restores_the_outputs_without_running_the_node(): + flow = double_flow() + node, calls = counting_node() + state = MemoryState() + seen: list[NodeOutcome] = [] + key = run_cache_key("fp-train", {"study.lr": 0.5}) + cache = FakeCache({key: {"study.loss": 99.0}}) + + Pipeline(nodes=[node], state=state, observer=seen.append, run_cache=cache).run( + seed_values(flow, {"lr": 0.5}) + ) + + assert calls == [] + # Restored into this run's own state, which is where everything + # downstream of it looks — its namespace holds nothing otherwise. + assert collect_result(flow, state) == {"loss": 99.0} + assert seen[0].cached and seen[0].cache_key == key + + +def test_a_miss_runs_the_node_and_carries_what_would_be_stored(): + flow = double_flow() + node, calls = counting_node() + seen: list[NodeOutcome] = [] + cache = FakeCache() + + Pipeline( + nodes=[node], state=MemoryState(), observer=seen.append, run_cache=cache + ).run(seed_values(flow, {"lr": 0.5})) + + assert calls == [1] + assert cache.asked == [run_cache_key("fp-train", {"study.lr": 0.5})] + assert not seen[0].cached + assert seen[0].cache_key and seen[0].output_values == {"study.loss": 1.0} + + +def test_the_key_follows_the_inputs(): + first = run_cache_key("fp", {"lr": 0.5}) + assert first != run_cache_key("fp", {"lr": 0.6}) + assert first != run_cache_key("other", {"lr": 0.5}) + assert first == run_cache_key("fp", {"lr": 0.5}) + + +def test_an_artifact_input_counts_as_its_digest(): + digest = "sha256:" + "0" * 64 + # The same bytes under another name, of a size recorded differently, are + # the same input — the reference is a handle, the digest is the content. + assert run_cache_key( + "fp", {"data": {"digest": digest, "name": "a.csv", "size": 3}} + ) == run_cache_key("fp", {"data": {"digest": digest, "name": "b.csv", "size": 3}}) + + +def test_a_node_with_no_fingerprint_is_never_looked_up(): + """Built-in and connector nodes, and anything declared `cache=False`.""" + flow = double_flow() + node, calls = counting_node() + node.fingerprint = "" + seen: list[NodeOutcome] = [] + cache = FakeCache() + + Pipeline( + nodes=[node], state=MemoryState(), observer=seen.append, run_cache=cache + ).run(seed_values(flow, {"lr": 0.5})) + + assert calls == [1] + assert cache.asked == [] + assert seen[0].cache_key == "" diff --git a/backend/tests/flow/test_stage_cache_build.py b/backend/tests/flow/test_stage_cache_build.py new file mode 100644 index 0000000..2d036ce --- /dev/null +++ b/backend/tests/flow/test_stage_cache_build.py @@ -0,0 +1,83 @@ +"""What a node is fingerprinted with, decided where it is built. + +The pipeline half — what a hit restores — is in `test_runs.py`. This is the +link between the two: a node only carries a fingerprint when it is built for a +run, and only when it is allowed to be cached at all. +""" + +from pathlib import Path + +import pytest + +from fluksio.flow.controller import FlowController, RunContext +from fluksio.flow.messages import DType, MessageSpec +from fluksio.flow.schemas import FlowDef, NodeDef +from fluksio.flow.state import MemoryState +from fluksio.flow.store import FlowStore + +SOURCE = "def process(reading, factor):\n return {'scaled': reading * factor}\n" + + +def a_flow(**params: object) -> FlowDef: + return FlowDef( + name="house", + mode="batch", + nodes=[ + NodeDef( + id="scale", + params=dict(params), + requires=[MessageSpec(name="reading", dtype=DType.FLOAT)], + provides=[MessageSpec(name="scaled", dtype=DType.FLOAT)], + ) + ], + ) + + +@pytest.fixture +def store(tmp_path: Path) -> FlowStore: + return FlowStore(tmp_path / "flows") + + +def fingerprint_of(store: FlowStore, flow: FlowDef) -> str: + store.write_flow(flow) + store.write_node_source(flow.name, "scale", SOURCE) + pipeline = FlowController(store).build_run_pipeline( + store.read_flow(flow.name), + state=MemoryState(), + run=RunContext(run_id="r-1"), + ) + return pipeline.nodes[0].fingerprint + + +def test_a_settings_change_is_a_different_node(store: FlowStore): + first = fingerprint_of(store, a_flow(factor=3)) + assert first + assert fingerprint_of(store, a_flow(factor=4)) != first + # And the same flow again is the same node, which is the whole point. + assert fingerprint_of(store, a_flow(factor=3)) == first + + +def test_a_source_change_is_a_different_node(store: FlowStore): + first = fingerprint_of(store, a_flow(factor=3)) + store.write_node_source("house", "scale", SOURCE.replace("*", "+")) + pipeline = FlowController(store).build_run_pipeline( + store.read_flow("house"), state=MemoryState(), run=RunContext(run_id="r-2") + ) + assert pipeline.nodes[0].fingerprint != first + + +def test_a_node_that_opted_out_carries_none(store: FlowStore): + flow = a_flow(factor=3) + flow.nodes[0].cache = False + assert fingerprint_of(store, flow) == "" + + +def test_a_live_pipeline_fingerprints_nothing(store: FlowStore): + """Only a run may reuse a result; a cascade is about what just happened.""" + store.write_flow(a_flow(factor=3)) + store.write_node_source("house", "scale", SOURCE) + controller = FlowController(store) + nodes, _loaded, _initial, _inputs = controller._build_flows( + [(store.read_flow("house"), False)] + ) + assert nodes[0].fingerprint == "" diff --git a/backend/tests/test_cli.py b/backend/tests/test_cli.py index 77fab68..770b79e 100644 --- a/backend/tests/test_cli.py +++ b/backend/tests/test_cli.py @@ -137,3 +137,121 @@ def test_run_syncs_by_default_and_can_be_told_not_to() -> None: assert _parser().parse_args(["run", "train"]).no_sync is False assert _parser().parse_args(["run", "train", "--no-sync"]).no_sync is True + + +def test_a_sweep_is_the_product_of_the_parameters_given() -> None: + """`--param lr=0.1,0.01 --param epochs=1,2` is four runs, typed by the flow.""" + import pytest + + from fluksio.sdk import SyncError + from fluksio.sdk.cli import _grid + + definition = { + "inputs": [ + {"spec": {"name": "lr", "dtype": "float"}}, + {"spec": {"name": "epochs", "dtype": "int"}}, + ] + } + + grid = _grid(definition, ["lr=0.1,0.01", "epochs=1,2"], seed=7) + assert [entry["params"] for entry in grid] == [ + {"lr": 0.1, "epochs": 1}, + {"lr": 0.1, "epochs": 2}, + {"lr": 0.01, "epochs": 1}, + {"lr": 0.01, "epochs": 2}, + ] + assert all(entry["seed"] == 7 for entry in grid) + + with pytest.raises(SyncError, match="not an input of this flow"): + _grid(definition, ["nonesuch=1"], seed=None) + with pytest.raises(SyncError, match="name=value"): + _grid(definition, ["lr"], seed=None) + + +def test_the_local_engine_is_asked_for_rather_than_guessed() -> None: + from fluksio.cli import _parser + + parser = _parser() + assert parser.parse_args(["run", "train"]).local is False + assert parser.parse_args(["run", "train", "--local"]).local is True + assert parser.parse_args(["runs", "--local"]).local is True + assert parser.parse_args(["sweep", "train", "--param", "lr=1"]).local is False + + +def test_a_local_run_always_waits(monkeypatch) -> None: + """The engine is this process, so a run nobody waits for is thrown away.""" + from contextlib import contextmanager + + from fluksio.cli import _parser + from fluksio.sdk import cli + + submitted: dict[str, object] = {} + + class FakeHandle: + id = "run-1" + status = "ok" + result: dict[str, object] = {} + + def wait(self, timeout: float = 0.0) -> "FakeHandle": + submitted["waited"] = True + return self + + class FakeClient: + def get_flow(self, name: str) -> dict[str, object]: + return {"definition": {"inputs": []}} + + def submit(self, flow, params, seed=None, no_cache=False): + submitted["flow"] = flow + submitted["no_cache"] = no_cache + return FakeHandle() + + def run(self, run_id: str) -> dict[str, object]: + return {"nodes": [{"status": "cached"}, {"status": "ok"}]} + + @contextmanager + def fake_engine(): + yield FakeClient() + + monkeypatch.setattr(cli, "_engine_client", fake_engine) + + args = _parser().parse_args(["run", "train", "--local", "--no-sync", "--no-cache"]) + assert cli.cmd_run(args, []) == 0 + assert submitted == {"flow": "train", "no_cache": True, "waited": True} + + +def test_ctrl_c_while_waiting_cancels_the_run(monkeypatch) -> None: + """Interrupting means stop the run, not walk away leaving it going.""" + from contextlib import contextmanager + + from fluksio.cli import _parser + from fluksio.sdk import cli + + cancelled: list[str] = [] + + class FakeHandle: + id = "run-1" + status = "running" + result: dict[str, object] = {} + + def wait(self, timeout: float = 0.0): + raise KeyboardInterrupt + + class FakeClient: + def get_flow(self, name: str) -> dict[str, object]: + return {"definition": {"inputs": []}} + + def submit(self, flow, params, seed=None, no_cache=False): + return FakeHandle() + + def cancel(self, run_id: str) -> None: + cancelled.append(run_id) + + @contextmanager + def fake_engine(): + yield FakeClient() + + monkeypatch.setattr(cli, "_engine_client", fake_engine) + + args = _parser().parse_args(["run", "train", "--local", "--no-sync"]) + assert cli.cmd_run(args, []) == 130 + assert cancelled == ["run-1"] diff --git a/docs/code/api.md b/docs/code/api.md index e5a4f9a..e07cbdb 100644 --- a/docs/code/api.md +++ b/docs/code/api.md @@ -102,12 +102,12 @@ published to. Flows own the namespace; everything else is a client of it. | Method | Path | What | |---|---|---| -| `POST` | `/runs/flows/{name}` | queue one run — `{"params": {...}, "seed": 7, "draft": false}` | +| `POST` | `/runs/flows/{name}` | queue one run — `{"params": {...}, "seed": 7, "draft": false, "no_cache": false}` | | `POST` | `/runs/flows/{name}/sweep` | queue many, sharing a `group_id` | | `GET` | `/runs` | the queryable history: `?flow=`, `?status=`, `?group=`, `?digest=`, `?limit=` | | `GET` | `/runs/{id}` | one run in full: params, result, per-node record, artifacts | | `POST` | `/runs/{id}/cancel` | stop it | -| `GET` | `/runs/{id}/metrics?name=&stride=` | one metric's series, in step order | +| `GET` | `/runs/{id}/metrics?name=&stride=` | one metric's series, in step order; every series of the run without `name` | | `GET` | `/runs/series/compare?ids=a,b,c&metric=` | that metric across several runs | Submitting answers immediately with a `queued` run. Wrong parameters — an diff --git a/docs/code/cli.md b/docs/code/cli.md index fea966a..d658f4f 100644 --- a/docs/code/cli.md +++ b/docs/code/cli.md @@ -12,8 +12,8 @@ should only *run nodes* for an engine elsewhere. It has none of the engine in it. See [Remote workers](workers.md). The command is two things at once: `serve`, `enroll` and `worker` *are* an -installation, while `login`, `sync`, `run` and `runs` talk to one that may be -anywhere. +installation, while `login`, `sync`, `run`, `runs` and `sweep` talk to one that +may be anywhere. ## Where an installation lives @@ -122,8 +122,9 @@ See [Remote workers](workers.md). ## Talking to an engine -The four commands below are the client half: they run wherever you work, and -address an engine over its API rather than being one. +The commands below are the client half: they run wherever you work, and +address an engine over its API rather than being one — except under `--local`, +which boots one inside the command instead. ### `fluksio login` @@ -171,13 +172,34 @@ fluksio run train --lr 0.05 --seed 7 [--wait] Syncs the working directory, then submits a run — so the command after an edit is this one and nothing else. Flags that are not its own are the flow's inputs, typed by what the flow declares them as. `--wait` blocks until the run -finishes and exits non-zero if it failed. +finishes and exits non-zero if it failed. `--follow` waits as well, and prints +the numbers the run reports as they arrive: + +```text + train.loss[14] = 3.40295e-06 +``` + +Ctrl-C while either is waiting cancels the run on the engine rather than only +stopping the watching, and exits 130. `--no-sync` runs what is already on the engine. Worth it in a tight loop where you know nothing changed, since syncing retires the workers and the next call pays its imports again. A directory that declares no flows syncs nothing and says nothing — a flow drawn on the canvas is run the same way. +`--no-cache` executes every node, including one an earlier run already +answered. See [Stage caching](../concepts/runs.md#stage-caching). + +`--local` boots the engine inside this process instead of talking to a served +one, so there is no `fluksio serve` terminal to keep open. It is the same +installation either way — the same `.fluksio`, the same database, artifacts +and run history — so a run made this way and a run made through a served +engine cache against each other. It always waits, because the engine it starts +lives exactly as long as the command. Starting one costs a few seconds of +worker pool and module reconcile, against the ~15 ms of submitting to an +engine that is already up: `--local` is for "I just want to run it", not for a +loop you are iterating in. + ### `fluksio runs` ```sh @@ -185,7 +207,22 @@ fluksio runs [--flow train] [--limit 20] ``` The runs an engine has recorded, newest first: id, status, flow, duration, the -commit of the repository it came from, and its parameters. +commit of the repository it came from, and its parameters. Statuses are +coloured when a terminal is reading the output — `ok` green, `error` red, +`cached` cyan. `--local` reads the same history from an in-process engine, +without one having to be served. + +### `fluksio sweep` + +```sh +fluksio sweep train --param lr=0.1,0.01 --param epochs=10,50 --wait +``` + +Every combination of the parameter lists, submitted as one group — four runs +above, sharing a `group_id` and executing in parallel. Values are typed by the +flow's inputs, the same as `run`'s are, and `--seed`, `--no-sync`, +`--no-cache` and `--local` mean what they do there. `--wait` blocks until all +of them are finished and exits non-zero if any failed. ## What lives in the data directory diff --git a/docs/concepts/runs.md b/docs/concepts/runs.md index d740048..2318f4a 100644 --- a/docs/concepts/runs.md +++ b/docs/concepts/runs.md @@ -66,6 +66,9 @@ curl -X POST $FLUKSIO/runs/flows/train_polymer_gnn/sweep \ }' ``` +From a terminal that is `fluksio sweep train_polymer_gnn --param lr=0.1,0.3`, +which builds the product of the lists you give it and posts the same call. + They share a `group_id`, so `GET /api/v1/runs?group=…` is the sweep, and they execute in parallel. That is safe because **each run has a state backend of its own**: message names are global keys, so two runs of one flow would otherwise @@ -126,7 +129,8 @@ def process(): Every number a node emits is kept as the run's series, stepped by the count of emissions on that message. Read one back with -`GET /api/v1/runs/{id}/metrics?name=.loss`, or compare runs: +`GET /api/v1/runs/{id}/metrics?name=.loss` — or leave `name` off for +every series the run kept — or compare runs: ``` GET /api/v1/runs/series/compare?ids=,,&metric=.loss @@ -168,6 +172,40 @@ one preprocessed input stores it once, and a reference stays valid wherever the store is reachable from. Artifacts a run produced are listed on it and downloadable at `GET /api/v1/artifacts/{digest}`. +## Stage caching + +A run mostly does not redo what an earlier one already did. Before a node +executes it is fingerprinted — a sha256 over its source, its settings and the +values it is about to read — and if some earlier run of that same fingerprint +finished, what that one returned is restored into this run's state and the node +is skipped. It is recorded with the status `cached` and a duration of zero, and +its artifacts are listed on the new run as well, so they stay downloadable from +either. + +The settings go into the key raw, so a secret contributes its `{"$secret": +name}` reference and never its value. An artifact input counts as its content +digest: the same bytes under a different filename are the same input. Each node +on a run carries the `cache_key` it was looked up by. + +The run history *is* the cache; there is no second store. A node's returned +outputs are kept on its run record as canonical JSON, up to 256000 characters — +a node returning more than that is simply not cacheable that run. An entry +whose artifact bytes have since left the store is a miss, not an error. + +Only `python` nodes are cached, and by default all of them are. A built-in node +type or a connector node has side effects and no source to fingerprint, so +neither is ever a candidate. Turn it off for one node with +`@node(..., cache=False)` — the flow document carries it as `cache`, so the +canvas and the API can change it too — or for one run with +`fluksio run --no-cache`, `fluksio sweep --no-cache`, or `"no_cache": true` in +the submission body. + +What a cached node does not bring back is what it emitted on the way. Its +returned outputs are restored; the values it published mid-execution are not, +because those were the story of an execution that is not happening this time. +So a skipped training node contributes no loss curve to the new run — if you +want the curve, that run has to actually train. + ## Objects that cannot be serialized A live model, a `DataLoader`, a JAX-compiled function — these do not cross a diff --git a/docs/getting-started/data-science.md b/docs/getting-started/data-science.md index 126dcc7..d64af60 100644 --- a/docs/getting-started/data-science.md +++ b/docs/getting-started/data-science.md @@ -405,7 +405,24 @@ again. `--lr` is typed by the flow's own inputs, so `0.003` arrives as a float. A parameter you did not declare, or one of the wrong type, is refused before anything executes. It answers immediately with a queued run — training is -measured in hours, so nothing waits for it unless you pass `--wait`. +measured in hours, so nothing waits for it unless you pass `--wait`, or +`--follow`, which waits and prints the numbers as the run reports them. + +No engine has to be up for any of this: `fluksio run --local` boots the +engine inside the command instead, on the same `.fluksio` — the same database, +artifacts and history a served one would use. It costs a few seconds of +startup per invocation against the ~15 ms of submitting to an engine that is +already up, so it is for the run you want now rather than the loop you are +iterating in. + +A second run of a flow you did not change mostly does not execute. Each of +your nodes is fingerprinted by its source, its settings and the values it +reads, and one an earlier run already answered is restored from that run +rather than run again — reported as `cached`, so a flow with nothing left to +do finishes as `(3/3 cached)`. Change `--lr` and only the nodes downstream of +it run. `--no-cache` turns that off for one run, `@node(..., cache=False)` for +one node; the caveat and the details are in +[Stage caching](../concepts/runs.md#stage-caching). From Python, the flow you declared is also the handle to its runs: @@ -436,8 +453,15 @@ was the learning rate on the run that got 94%?". ## Sweep it -A grid search and an ensemble are the same call — you build the list, Fluksio -runs them in parallel: +A grid search and an ensemble are the same submission, run in parallel: + +```sh +fluksio sweep train --param lr=0.001,0.003,0.01 --wait +``` + +Every combination of the lists you give, so a second `--param` is a grid +rather than a second sweep. Where the set you want is not a product, build the +list yourself and post it: ```sh curl -X POST $FLUKSIO/runs/flows/train/sweep -H "Authorization: Bearer $TOKEN" \ @@ -578,7 +602,8 @@ is the same stack. artifacts, sweeps, durability, what happens when your engine dies mid-training - [Writing node code](../code/nodes.md) — generators, settings, what a node may and may not do -- [The command line](../code/cli.md) — `login`, `sync`, `run`, `runs` in full +- [The command line](../code/cli.md) — `login`, `sync`, `run`, `runs`, `sweep` + in full - [Remote workers](../code/workers.md) — send the training node to the GPU box and keep the rest on your laptop - [The flow editor](../interface/flow-editor.md) — once you have a portal, this diff --git a/frontend/src/client/schemas.gen.ts b/frontend/src/client/schemas.gen.ts index 887c68c..95ec3cd 100644 --- a/frontend/src/client/schemas.gen.ts +++ b/frontend/src/client/schemas.gen.ts @@ -1446,6 +1446,11 @@ export const MetricPointSchema = { value: { type: 'number', title: 'Value' + }, + name: { + type: 'string', + title: 'Name', + default: '' } }, type: 'object', @@ -1628,6 +1633,12 @@ export const NodeDef_InputSchema = { title: 'Device Policy', description: 'What to do when no worker carries `device`: wait for one, or run locally anyway.', default: 'require' + }, + cache: { + type: 'boolean', + title: 'Cache', + description: 'Whether a batch run may reuse an earlier execution of this node with the same source, settings and inputs. Turn it off for a function whose answer can change on its own.', + default: true } }, type: 'object', @@ -1717,6 +1728,12 @@ export const NodeDef_OutputSchema = { title: 'Device Policy', description: 'What to do when no worker carries `device`: wait for one, or run locally anyway.', default: 'require' + }, + cache: { + type: 'boolean', + title: 'Cache', + description: 'Whether a batch run may reuse an earlier execution of this node with the same source, settings and inputs. Turn it off for a function whose answer can change on its own.', + default: true } }, type: 'object', @@ -2399,6 +2416,11 @@ export const RunCreateSchema = { type: 'boolean', title: 'Draft', default: false + }, + no_cache: { + type: 'boolean', + title: 'No Cache', + default: false } }, type: 'object', @@ -2551,6 +2573,11 @@ export const RunNodeRowSchema = { logs: { type: 'string', title: 'Logs' + }, + cache_key: { + type: 'string', + title: 'Cache Key', + default: '' } }, type: 'object', @@ -2781,6 +2808,11 @@ export const SweepCreateSchema = { type: 'boolean', title: 'Draft', default: false + }, + no_cache: { + type: 'boolean', + title: 'No Cache', + default: false } }, type: 'object', diff --git a/frontend/src/client/sdk.gen.ts b/frontend/src/client/sdk.gen.ts index d69afe2..2d50f13 100644 --- a/frontend/src/client/sdk.gen.ts +++ b/frontend/src/client/sdk.gen.ts @@ -1820,7 +1820,7 @@ export class RunsService { /** * Read Metrics - * One metric's series, in step order. + * One metric's series, in step order — or every one of them, unnamed. * * ``stride`` thins a long curve down: 3000 steps drawn on a 400-pixel chart * is 3000 points nobody can see. diff --git a/frontend/src/client/types.gen.ts b/frontend/src/client/types.gen.ts index 218fbd6..a822844 100644 --- a/frontend/src/client/types.gen.ts +++ b/frontend/src/client/types.gen.ts @@ -556,6 +556,7 @@ export type MetricPoint = { step: number; ts: number; value: number; + name?: string; }; /** @@ -620,6 +621,10 @@ export type NodeDef_Input = { * What to do when no worker carries `device`: wait for one, or run locally anyway. */ device_policy?: 'require' | 'prefer'; + /** + * Whether a batch run may reuse an earlier execution of this node with the same source, settings and inputs. Turn it off for a function whose answer can change on its own. + */ + cache?: boolean; }; /** @@ -656,6 +661,10 @@ export type NodeDef_Output = { * What to do when no worker carries `device`: wait for one, or run locally anyway. */ device_policy?: 'require' | 'prefer'; + /** + * Whether a batch run may reuse an earlier execution of this node with the same source, settings and inputs. Turn it off for a function whose answer can change on its own. + */ + cache?: boolean; }; /** @@ -864,6 +873,7 @@ export type RunCreate = { }; seed?: (number | null); draft?: boolean; + no_cache?: boolean; }; export type RunDetail = { @@ -902,6 +912,7 @@ export type RunNodeRow = { worker: string; error: string; logs: string; + cache_key?: string; }; export type RunPage = { @@ -982,6 +993,7 @@ export type ShareRequest = { export type SweepCreate = { runs?: Array; draft?: boolean; + no_cache?: boolean; }; export type SweepEntry = { @@ -1606,7 +1618,7 @@ export type RunsCancelRunData = { export type RunsCancelRunResponse = (fluksio__api__routes__runs__RunRow); export type RunsReadMetricsData = { - name: string; + name?: string; runId: string; stride?: number; };