From 34243e690221264bbc7980b9b7377f761d16963d Mon Sep 17 00:00:00 2001 From: stroblme Date: Wed, 28 Oct 2026 08:07:35 +0100 Subject: [PATCH] SDK: artifacts work outside a node, run --follow prints status and queue waits, fluksio cancel save_artifact/load_artifact outside a node use the worker's own store code in $FLUKSIO_ARTIFACT_DIR or /fluksio-artifacts, so a node can be called by hand. --follow prints status changes and each node's resource wait; the placer's waiting snapshot now carries the run. fluksio cancel ... --- backend/fluksio/__init__.py | 44 +++++++++--- backend/fluksio/api/routes/workers.py | 2 + backend/fluksio/flow/placement.py | 4 +- backend/fluksio/sdk/cli.py | 63 ++++++++++++++++- backend/fluksio/sdk/client.py | 5 ++ backend/tests/flow/test_placement.py | 4 +- backend/tests/sdk/test_build.py | 37 ++++++++++ backend/tests/test_cli.py | 97 +++++++++++++++++++++++++++ docs/code/cli.md | 14 +++- docs/code/nodes.md | 6 +- docs/getting-started/data-science.md | 10 +-- examples/myresearch/__init__.py | 9 ++- frontend/src/client/schemas.gen.ts | 5 ++ frontend/src/client/types.gen.ts | 1 + 14 files changed, 274 insertions(+), 27 deletions(-) diff --git a/backend/fluksio/__init__.py b/backend/fluksio/__init__.py index 388539f..33ed817 100644 --- a/backend/fluksio/__init__.py +++ b/backend/fluksio/__init__.py @@ -11,9 +11,10 @@ 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. +node imports may well declare them at its top. Everywhere else the functions +below stand in, so a node called directly — in a test, in a notebook — behaves +like the real one: ``emit`` does nothing and the artifact helpers use a local +directory. 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 @@ -21,6 +22,7 @@ second of start-up cost would both begin here. """ import logging +import os from importlib.metadata import PackageNotFoundError from importlib.metadata import version as _version from typing import Any @@ -46,11 +48,6 @@ __all__ = [ 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. @@ -63,16 +60,41 @@ def emit(**ports: Any) -> None: logger.debug("fluksio.emit(%s) outside a node", ", ".join(ports)) +def _local_store() -> Any: + """The worker's own artifact code, pointed at a directory of this machine. + + ``$FLUKSIO_ARTIFACT_DIR`` when it is set, which is how a worker finds the + engine's store, and a directory in the temporary location otherwise. The + references that come out are the ones a node gets inside the engine, so + what a node returns can be handed to the next one by hand. + """ + import tempfile + from pathlib import Path + + from fluksio_worker.worker_main import ARTIFACT_DIR_ENV, _Reporter + + os.environ.setdefault( + ARTIFACT_DIR_ENV, str(Path(tempfile.gettempdir()) / "fluksio-artifacts") + ) + return _Reporter("fluksio") + + 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")) + """Put bytes or a file in the artifact store and return a reference to them. + + Outside a node the store is a local directory, see :func:`_local_store`. + """ + reference: dict[str, Any] = _local_store().save_artifact( + source, name, media_type, volatile + ) + return reference 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")) + return str(_local_store().load_artifact(ref)) diff --git a/backend/fluksio/api/routes/workers.py b/backend/fluksio/api/routes/workers.py index 9c06631..ca024b0 100644 --- a/backend/fluksio/api/routes/workers.py +++ b/backend/fluksio/api/routes/workers.py @@ -134,6 +134,8 @@ class ResourceLevel(BaseModel): class WaitingNode(BaseModel): node: str + #: The run it belongs to, so a client can tell its own wait from another's. + run: str = "" reason: str seconds: float diff --git a/backend/fluksio/flow/placement.py b/backend/fluksio/flow/placement.py index f35d2fe..8fcb839 100644 --- a/backend/fluksio/flow/placement.py +++ b/backend/fluksio/flow/placement.py @@ -79,6 +79,7 @@ class _Waiting: node: str since: float reason: str + run: str = "" class Placer: @@ -333,7 +334,7 @@ class Placer: """ self._ticket += 1 ticket = self._ticket - self._waiting[ticket] = _Waiting(node=node, since=since, reason=reason) + self._waiting[ticket] = _Waiting(node=node, since=since, reason=reason, run=run) logger.info("%s is %s", node or "a node", reason) if self.events is not None: self.events.publish( @@ -407,6 +408,7 @@ class Placer: "waiting": [ { "node": entry.node, + "run": entry.run, "reason": entry.reason, "seconds": round(time.monotonic() - entry.since, 1), } diff --git a/backend/fluksio/sdk/cli.py b/backend/fluksio/sdk/cli.py index 23fc960..d7002fd 100644 --- a/backend/fluksio/sdk/cli.py +++ b/backend/fluksio/sdk/cli.py @@ -585,8 +585,38 @@ def _sync_first( _say(f"synced {', '.join(r.flow for r in changed)}") +def _waits(client: Client, run_id: str) -> list[tuple[str, str]]: + """What this run's nodes are queued for on a machine, as (node, reason). + + A run that says nothing for an hour is either hung or waiting its turn, and + this is the half of that answer a metric cannot give. Best effort: an + engine that does not account for resources, or is too old to say whose a + wait is, simply has nothing to tell. + """ + try: + waiting = client.resources().get("waiting") or [] + except (httpx.HTTPError, ApiError): + return [] + return [ + (str(entry.get("node", "")), str(entry.get("reason", ""))) + for entry in waiting + if entry.get("run") == run_id + ] + + +def _state(handle: RunHandle) -> tuple[str, str]: + """Its status, and why — the reason only while it is true. + + The engine leaves "waiting for a worker" on the row after the run starts. + """ + return handle.status, handle.status_reason if handle.status == "queued" else "" + + def _follow(client: Client, handle: RunHandle, poll: float = 1.0) -> None: - """Print a run's numbers as they arrive, until it is over. + """Print what a run does as it does it, until it is over. + + Its numbers, a change of status, and what a queued node is waiting for, so + a run that reports nothing is not a silent one. 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 @@ -594,6 +624,8 @@ def _follow(client: Client, handle: RunHandle, poll: float = 1.0) -> None: before the numbers, so the last batch is never the one that gets missed. """ seen: set[tuple[str, int]] = set() + told: set[tuple[str, str]] = set() + state = _state(handle) failures = 0 while True: try: @@ -610,6 +642,16 @@ def _follow(client: Client, handle: RunHandle, poll: float = 1.0) -> None: time.sleep(poll) continue failures = 0 + if not done: + now = _state(handle) + if now != state: + state = now + _say(f" {_status(now[0])}" + (f" {now[1]}" if now[1] else "")) + if handle.status == "running": + for wait in _waits(client, handle.id): + if wait not in told: + told.add(wait) + _say(f" {wait[0]} {wait[1]}") for point in points: mark = (str(point.get("name", "")), int(point.get("step", -1))) if mark in seen: @@ -1114,6 +1156,20 @@ def cmd_retry(args: argparse.Namespace) -> int: return 0 +def cmd_cancel(args: argparse.Namespace) -> int: + """Stop runs by id, which `run` without `--wait` leaves no other way to do.""" + try: + with _client_for(args) as client: + for run_id in args.run_id: + made = client.cancel(run_id) + _say(f"{run_id} {_status(str(made.get('status') or 'cancelled'))}") + except (SyncError, ApiError) as exc: + return _fail(str(exc)) + except httpx.HTTPError as exc: + return _unreachable(exc) + return 0 + + def _artifacts(client: Client, args: argparse.Namespace) -> int: """List a run's files, or write one of them here.""" handle = RunHandle(client, args.run_id, client.run(args.run_id)) @@ -1524,6 +1580,11 @@ def add_parsers(subparsers: Any) -> None: with_engine(parser) parser.set_defaults(func=cmd_retry) + parser = subparsers.add_parser("cancel", help="stop runs that are still going") + parser.add_argument("run_id", nargs="+", help="the runs to stop") + with_engine(parser) + parser.set_defaults(func=cmd_cancel) + parser = subparsers.add_parser( "artifacts", help="the files a run produced; name one to download it" ) diff --git a/backend/fluksio/sdk/client.py b/backend/fluksio/sdk/client.py index fa71835..c1ec83e 100644 --- a/backend/fluksio/sdk/client.py +++ b/backend/fluksio/sdk/client.py @@ -577,6 +577,11 @@ class RunHandle: def status(self) -> str: return str(self._row.get("status", "")) + @property + def status_reason(self) -> str: + """Why it is in that status, such as what a queued run is waiting for.""" + return str(self._row.get("status_reason") or "") + @property def done(self) -> bool: return self.status in DONE diff --git a/backend/tests/flow/test_placement.py b/backend/tests/flow/test_placement.py index fe2b61e..f54e167 100644 --- a/backend/tests/flow/test_placement.py +++ b/backend/tests/flow/test_placement.py @@ -176,7 +176,7 @@ def test_a_node_waits_for_a_busy_machine_and_says_so(loop): started = threading.Event() def second() -> None: - with placer.claim(Resources(cpus=2), node="study.b"): + with placer.claim(Resources(cpus=2), node="study.b", run="r2"): started.set() with placer.claim(Resources(cpus=2), node="study.a"): @@ -185,6 +185,8 @@ def test_a_node_waits_for_a_busy_machine_and_says_so(loop): # Long enough to have run if nothing was holding it back. assert not started.wait(0.3) assert placer.snapshot()["waiting"][0]["node"] == "study.b" + # Whose wait it is, so a client following one run reads only its own. + assert placer.snapshot()["waiting"][0]["run"] == "r2" assert events[0]["type"] == "node_queued" assert "2 cpu(s)" in events[0]["detail"] diff --git a/backend/tests/sdk/test_build.py b/backend/tests/sdk/test_build.py index 65e1178..380540d 100644 --- a/backend/tests/sdk/test_build.py +++ b/backend/tests/sdk/test_build.py @@ -1,5 +1,7 @@ """What the decorators declare, and what the engine would be given.""" +from pathlib import Path + import pytest from fluksio.flow.schemas import FlowDef @@ -396,3 +398,38 @@ def test_a_flow_records_the_module_that_declared_it(): # A repository the file is not in has no path to give it. elsewhere = flow.document({"kind": "python", "repo": "/elsewhere"})["origin"] assert (elsewhere["declared_in"], elsewhere["module"]) == ("", "") + + +def test_artifacts_round_trip_outside_a_node(tmp_path, monkeypatch): + """A node that returns an artifact can be smoke-tested by calling it.""" + import fluksio + + monkeypatch.setenv("FLUKSIO_ARTIFACT_DIR", str(tmp_path)) + + ref = fluksio.save_artifact(b"weights", "w.bin", "application/x-test") + + assert ref["name"] == "w.bin" + assert ref["size"] == 7 + assert ref["media_type"] == "application/x-test" + assert ref["digest"].startswith("sha256:") + assert Path(fluksio.load_artifact(ref)).read_bytes() == b"weights" + assert str(tmp_path) in fluksio.load_artifact(ref) + + source = tmp_path / "plot.png" + source.write_bytes(b"png") + assert fluksio.save_artifact(source)["name"] == "plot.png" + + +def test_the_local_artifact_store_defaults_to_the_temporary_directory(monkeypatch): + import tempfile + + import fluksio + + # Set and removed again, so the variable the default writes is undone too. + monkeypatch.setenv("FLUKSIO_ARTIFACT_DIR", "") + monkeypatch.delenv("FLUKSIO_ARTIFACT_DIR") + ref = fluksio.save_artifact(b"x" * 3) + + assert fluksio.load_artifact(ref).startswith( + str(Path(tempfile.gettempdir()) / "fluksio-artifacts") + ) diff --git a/backend/tests/test_cli.py b/backend/tests/test_cli.py index c6826c8..67d79f5 100644 --- a/backend/tests/test_cli.py +++ b/backend/tests/test_cli.py @@ -309,6 +309,103 @@ def test_ctrl_c_while_waiting_cancels_the_run(monkeypatch) -> None: assert cancelled == ["run-1"] +def test_follow_says_what_a_run_that_reports_nothing_is_doing(capsys) -> None: + """Queued for a worker, then waiting for a card: both are said once.""" + from fluksio.sdk import cli + from fluksio.sdk.client import ApiError + + held = "Waiting for a worker with 1 gpu(s)" + rows = iter( + [ + {"status": "queued", "status_reason": held}, + {"status": "queued", "status_reason": held}, + # The engine leaves the reason on the row once the run starts. + {"status": "running", "status_reason": held}, + {"status": "running", "status_reason": held}, + {"status": "running", "status_reason": held}, + {"status": "ok", "status_reason": ""}, + ] + ) + + class FakeHandle: + id = "r1" + status = "queued" + status_reason = "" + + @property + def done(self) -> bool: + return self.status == "ok" + + def refresh(self) -> "FakeHandle": + row = next(rows) + self.status, self.status_reason = row["status"], row["status_reason"] + return self + + asked: list[int] = [] + + class FakeClient: + def metrics(self, run_id: str) -> list: + return [] + + def resources(self) -> dict: + asked.append(1) + if len(asked) == 3: + raise ApiError(503, "Resources are not accounted here") + return { + "waiting": [ + { + "node": "train.fit", + "run": "r1", + "reason": "waiting for 1 gpu(s)", + }, + { + "node": "train.fit", + "run": "r2", + "reason": "waiting for 9 gpu(s)", + }, + ] + } + + cli._follow(FakeClient(), FakeHandle(), poll=0) + + assert capsys.readouterr().out.splitlines() == [ + f" queued {held}", + " running", + " train.fit waiting for 1 gpu(s)", + ] + + +def test_cancel_stops_the_runs_it_is_given(monkeypatch, capsys) -> None: + from contextlib import contextmanager + + from fluksio.cli import _parser + from fluksio.sdk import cli + from fluksio.sdk.client import ApiError + + cancelled: list[str] = [] + + class FakeClient: + def cancel(self, run_id: str) -> dict[str, object]: + if run_id == "gone": + raise ApiError(404, "No such run") + cancelled.append(run_id) + return {"status": "ok" if run_id == "done" else "cancelled"} + + @contextmanager + def fake_client(_args): + yield FakeClient() + + monkeypatch.setattr(cli, "_client_for", fake_client) + + args = _parser().parse_args(["cancel", "r1", "done"]) + assert cli.cmd_cancel(args) == 0 + assert cancelled == ["r1", "done"] + # A run already past its last node says how it ended, not that it stopped. + assert capsys.readouterr().out.splitlines() == ["r1 cancelled", "done ok"] + + assert cli.cmd_cancel(_parser().parse_args(["cancel", "gone"])) == 1 + + def test_a_failed_run_prints_what_each_node_said(monkeypatch, capsys) -> None: """Otherwise the traceback is stored and nothing at a terminal shows it.""" from contextlib import contextmanager diff --git a/docs/code/cli.md b/docs/code/cli.md index e4a0250..944dda4 100644 --- a/docs/code/cli.md +++ b/docs/code/cli.md @@ -206,7 +206,7 @@ warnings. Use `--sync PATH` to choose a target explicitly. | Flag | Purpose | |---|---| | `--wait` | Wait for completion; exit non-zero if the run fails | -| `--follow` | Wait and print emitted metrics | +| `--follow` | Wait and print emitted metrics, status changes and what a queued node waits for | | `--timeout SECONDS` | Stop waiting after this long; leave the run active | | `--seed N` | Record the seed and fill a declared input named `seed` | | `--defaults` | Use declared input defaults without prompting | @@ -217,7 +217,8 @@ warnings. Use `--sync PATH` to choose a target explicitly. With no parameters at a terminal, the command prompts for each input. Press Enter to keep its default. Scripted commands do not prompt. -Ctrl-C while waiting cancels the run and exits with status 130. +Ctrl-C while waiting cancels the run and exits with status 130. A run started +without `--wait` is stopped with [`fluksio cancel`](#fluksio-cancel). A seed does not initialize your libraries' random number generators. Declare a `seed` input and pass it to your code if it controls randomness. @@ -256,6 +257,15 @@ fluksio runs --flow train --limit 10 Lists recent runs, newest first. Omit `--flow` for all flows. `--local` reads the local instance. +### `fluksio cancel` + +```sh +fluksio cancel [ ...] +``` + +Stops the runs and their active node processes. A run already past its last +node is left as it finished, and its status is printed. + ### `fluksio retry` ```sh diff --git a/docs/code/nodes.md b/docs/code/nodes.md index 7fda621..5564022 100644 --- a/docs/code/nodes.md +++ b/docs/code/nodes.md @@ -110,8 +110,10 @@ def process(): ``` Another node can read it with `fluksio.load_artifact(report)`, which returns -a local file path. These two helpers require a running node; for tests, -keep the underlying calculation in ordinary Python functions. +a local file path. Called directly, outside a node, the two helpers store the +bytes in `$FLUKSIO_ARTIFACT_DIR` or, when that is unset, in `fluksio-artifacts` +under the temporary directory. The reference has the same shape as inside the +engine, so a test can pass what one node returns to the next. Artifacts are stored by their SHA-256 digest, so identical bytes share storage. Use [`fluksio artifacts`](cli.md#fluksio-artifacts) to list and download a diff --git a/docs/getting-started/data-science.md b/docs/getting-started/data-science.md index bd36357..47fed76 100644 --- a/docs/getting-started/data-science.md +++ b/docs/getting-started/data-science.md @@ -113,7 +113,8 @@ fluksio run train --lr 0.1 --epochs 20 --follow `sync` uploads the flow and generates node code that imports your functions. Keep the project files available to the engine. Later `run` commands sync -edits before submitting; `--follow` waits and prints metrics as they arrive. +edits before submitting. `--follow` waits and prints metrics as they arrive, +along with status changes and what a queued node is waiting for. This run should finish with `final_loss` around `0.1353` and `score` around `0.8647`. @@ -155,9 +156,10 @@ Pass either list to `pandas.DataFrame` if you use pandas. ## Bring your own experiment -Replace the toy calculation in `fit` with your training code. Keep ordinary -Python helpers separate from artifact storage so you can still test them -without an engine: `save_artifact` and `load_artifact` work only inside nodes. +Replace the toy calculation in `fit` with your training code. You can call a +node directly to test it without an engine: `save_artifact` and +`load_artifact` then use a folder under the temporary directory, and `emit` +does nothing. The public repository's `examples/myresearch/` package shows this pattern across several files. diff --git a/examples/myresearch/__init__.py b/examples/myresearch/__init__.py index 32537b1..c41bb31 100644 --- a/examples/myresearch/__init__.py +++ b/examples/myresearch/__init__.py @@ -2,9 +2,8 @@ Each node is a thin wrapper over a plain function: `prepare` stores what `make_rows` made, `fit` yields what `train_curve` computed, `evaluate` reports -what `score` worked out. That split is the point — `fluksio.save_artifact` and -`load_artifact` only work *inside* a node, so a body calling them cannot be run -by hand, while `python -m myresearch.pipeline` runs the arithmetic beside them -with no engine anywhere. `fluksio sync examples/myresearch` is what turns the -wrappers into a flow. +what `score` worked out. That split is the point — the arithmetic stays plain +Python, so `python -m myresearch.pipeline` runs it with no engine anywhere, +while the wrappers own the artifact storage. `fluksio sync examples/myresearch` +is what turns the wrappers into a flow. """ diff --git a/frontend/src/client/schemas.gen.ts b/frontend/src/client/schemas.gen.ts index 1d5e0d1..c740387 100644 --- a/frontend/src/client/schemas.gen.ts +++ b/frontend/src/client/schemas.gen.ts @@ -5002,6 +5002,11 @@ export const WaitingNodeSchema = { type: 'string', title: 'Node' }, + run: { + type: 'string', + title: 'Run', + default: '' + }, reason: { type: 'string', title: 'Reason' diff --git a/frontend/src/client/types.gen.ts b/frontend/src/client/types.gen.ts index 538ac14..d00fe4d 100644 --- a/frontend/src/client/types.gen.ts +++ b/frontend/src/client/types.gen.ts @@ -1580,6 +1580,7 @@ export type ValidationResult = { export type WaitingNode = { node: string; + run?: string; reason: string; seconds: number; };