diff --git a/backend/fluksio/api/routes/runs.py b/backend/fluksio/api/routes/runs.py
index 169bf2b..a49cab8 100644
--- a/backend/fluksio/api/routes/runs.py
+++ b/backend/fluksio/api/routes/runs.py
@@ -378,20 +378,18 @@ RUN_COLUMNS = (
)
-def _selected(
- session: Session,
+def _selection(
flow: str | None,
status: str | None,
group: str | None,
ids: str,
since: datetime | None,
until: datetime | None,
-) -> list[Run]:
- """The runs an export covers, newest first — the filters the list takes.
+) -> Any:
+ """The query behind a selection of runs, newest first.
- Capped: the filters bound a *sensible* request and nothing bounded an
- unfiltered one, so asking for everything read every row of the table into
- memory before a byte was sent. A truncated export says so in a header.
+ Separate from reading it, because the names route asks the same question
+ as a subquery rather than for the rows.
"""
statement = select(Run).order_by(col(Run.created_at).desc())
if flow:
@@ -407,6 +405,25 @@ def _selected(
statement = statement.where(col(Run.created_at) >= _aware(since))
if until:
statement = statement.where(col(Run.created_at) < _aware(until))
+ return statement
+
+
+def _selected(
+ session: Session,
+ flow: str | None,
+ status: str | None,
+ group: str | None,
+ ids: str,
+ since: datetime | None,
+ until: datetime | None,
+) -> list[Run]:
+ """The runs an export covers, newest first — the filters the list takes.
+
+ Capped: the filters bound a *sensible* request and nothing bounded an
+ unfiltered one, so asking for everything read every row of the table into
+ memory before a byte was sent. A truncated export says so in a header.
+ """
+ statement = _selection(flow, status, group, ids, since, until)
return list(session.exec(statement.limit(EXPORT_CAP + 1)))
@@ -614,6 +631,34 @@ def export_runs(
return _stream(format, "runs", columns, chunks(), truncated)
+@router.get("/metrics/names", response_model=list[str])
+def read_metric_names(
+ session: SessionDep,
+ flow: str | None = None,
+ status: str | None = None,
+ group: str | None = None,
+ ids: str = "",
+ since: datetime | None = None,
+ until: datetime | None = None,
+) -> Any:
+ """Every metric name the selected runs recorded.
+
+ A name is flow-qualified — a node of `train` writing `train_loss` records
+ `train.train_loss` — so this is the answer to "what would match". Exact
+ over the whole selection: the client used to read the newest run that had
+ measured anything and take its names for the vocabulary, which missed a
+ name only an older run ever wrote. Declared before `/{run_id}`, or that
+ route would take "metrics" for an id.
+ """
+ chosen = _selection(flow, status, group, ids, since, until).with_only_columns(
+ col(Run.id)
+ )
+ names = session.exec(
+ select(col(RunMetric.name)).where(col(RunMetric.run_id).in_(chosen)).distinct()
+ ).all()
+ return sorted(names)
+
+
@router.get("/{run_id}", response_model=RunDetail)
def read_run(run_id: str, session: SessionDep) -> Any:
"""One run in full: what it was asked, what each node did, what it made."""
diff --git a/backend/fluksio/cli.py b/backend/fluksio/cli.py
index 6e472f2..910e9fd 100644
--- a/backend/fluksio/cli.py
+++ b/backend/fluksio/cli.py
@@ -22,6 +22,7 @@ import os
import secrets
import socket
import sys
+import time
from collections.abc import Callable
from pathlib import Path
from typing import Any
@@ -83,17 +84,19 @@ def _mention_other_instance(data_dir: Path) -> None:
def _mention_undeclared_cards() -> None:
- """Say when a stored flow asks for a card this engine does not have.
+ """Say when a stored flow asks for a card this engine cannot find.
- Cards are declared rather than detected, so an engine told nothing has
- none — and a node asking for one is clamped to zero and runs beside every
- other, which on a GPU is the deadlock the declaration exists to prevent.
- The placer says so once it happens, into the log; this says it while
- somebody is still reading the terminal.
+ NVIDIA's device nodes are counted; anything else has to be declared, so an
+ engine that finds none and is told none has none — and a node asking for
+ one is clamped to zero and runs beside every other, which on a GPU is the
+ deadlock the declaration exists to prevent. The placer says so once it
+ happens, into the log; this says it while somebody is still reading the
+ terminal.
"""
from fluksio.core.config import settings
+ from fluksio.flow.resources import machine_gpus
- if settings.FLOW_GPUS:
+ if settings.FLOW_GPUS or machine_gpus():
return
from fluksio.flow.runs import required_resources
from fluksio.flow.store import FlowStore
@@ -111,8 +114,8 @@ def _mention_undeclared_cards() -> None:
if not asking:
return
named = ", ".join(sorted(asking)[:3]) + (" …" if len(asking) > 3 else "")
- _say(f" Cards 0 declared, but {named} asks for one.")
- _say(" Nothing detects them: `--gpus N` says how many are here.")
+ _say(f" Cards none found, but {named} asks for one.")
+ _say(" `--gpus N` says how many when /dev/nvidia* does not.")
def _warn_if_networked(path: Path) -> None:
@@ -318,6 +321,84 @@ def read_pidfile(data_dir: Path) -> dict[str, int] | None:
return {"pid": pid, "port": port}
+#: How large `serve.log` is allowed to get before it is cut back to nothing.
+#: The dashboard starts the engine with that file as its stdout and reads it
+#: back as the Logs tab, so an engine left running would otherwise fill a disk
+#: with nobody watching.
+LOG_KEEP_BYTES = 5 * 1024 * 1024
+
+#: How often the size is looked at. Cheap — one fstat — and nowhere near a
+#: path anything else takes.
+LOG_CHECK_S = 30.0
+
+
+def _appended_log(fd: int = 1) -> bool:
+ """Whether this descriptor is a file the engine may cut.
+
+ Only a regular file opened for appending: the kernel then puts every write
+ at the new end, where a plain `>` redirect would keep the offset it had and
+ leave a hole the size of what was dropped.
+ """
+ try:
+ import fcntl
+ import stat
+
+ if not stat.S_ISREG(os.fstat(fd).st_mode):
+ return False
+ return bool(fcntl.fcntl(fd, fcntl.F_GETFL) & os.O_APPEND)
+ except (ImportError, OSError, AttributeError):
+ return False
+
+
+def _trim_log(fd: int = 1, limit: int = LOG_KEEP_BYTES) -> bool:
+ """Empty the file behind stdout once it passes `limit`.
+
+ ponytail: a cut, not a rotation — copy the tail aside here when somebody
+ wants yesterday's log. The dashboard's reader already survives it: it
+ notices the file has shrunk under it and reads from the top again.
+ """
+ try:
+ if os.fstat(fd).st_size <= limit:
+ return False
+ os.ftruncate(fd, 0)
+ except OSError:
+ return False
+ return True
+
+
+def _watch_log(fd: int = 1, limit: int = LOG_KEEP_BYTES) -> None:
+ """Keep the engine's own output bounded, whoever started it.
+
+ The dashboard used to cut the file when *it* spawned an engine, which left
+ an adopted one — or one whose screen was closed — writing without a bound.
+ """
+ import threading
+
+ def loop() -> None:
+ while True:
+ time.sleep(LOG_CHECK_S)
+ _trim_log(fd, limit)
+
+ threading.Thread(target=loop, daemon=True).start()
+
+
+def _hold_pidfile(pidfile: Path, serve: Callable[[], None]) -> None:
+ """Run the engine, and take the pidfile down however it ends.
+
+ uvicorn restores the SIGTERM handler it found and re-raises the signal it
+ stopped on, so the default handler would end the process without unwinding
+ — and the file would outlive it. This leaves through the `finally` instead.
+ Ctrl-C already did: the restored handler there raises `KeyboardInterrupt`.
+ """
+ import signal
+
+ signal.signal(signal.SIGTERM, lambda signum, _frame: sys.exit(128 + signum))
+ try:
+ serve()
+ finally:
+ pidfile.unlink(missing_ok=True)
+
+
def _token_for(data_dir: Path) -> str:
"""The credential this directory's last engine wrote, if it wrote one."""
from fluksio.sdk.client import config_path
@@ -516,16 +597,19 @@ def cmd_serve(args: argparse.Namespace) -> int:
# One process: it holds the flow engine, and a second worker would be a
# second engine — duplicated subscriptions, cron ticks and webhooks.
pidfile = write_pidfile(data_dir, port)
- try:
- uvicorn.run(
+ if _appended_log():
+ # Started by the dashboard, with `serve.log` as this process's stdout.
+ _watch_log()
+ _hold_pidfile(
+ pidfile,
+ lambda: uvicorn.run(
app,
host=args.host,
port=port,
log_level=args.log_level,
log_config=_log_config(args.log_level),
- )
- finally:
- pidfile.unlink(missing_ok=True)
+ ),
+ )
return 0
@@ -641,7 +725,7 @@ def _parser() -> argparse.ArgumentParser:
type=_at_least(0),
default=None,
metavar="N",
- help="GPUs on this machine a node may be given (default 0, FLOW_GPUS)",
+ help="GPUs a node may be given (default: /dev/nvidia* counted, FLOW_GPUS)",
)
serve.set_defaults(func=cmd_serve)
diff --git a/backend/fluksio/core/config.py b/backend/fluksio/core/config.py
index c1f344e..cdadf47 100644
--- a/backend/fluksio/core/config.py
+++ b/backend/fluksio/core/config.py
@@ -130,9 +130,9 @@ class Settings(BaseSettings):
# loop answering while the machine is busy. Nodes that declare nothing are
# not accounted against it — they only get its fair share as a thread cap.
FLOW_CPUS: int = 0
- # GPUs on this machine. Not detected, because detecting it means depending
- # on the vendor's tooling: say how many there are and each is held by one
- # node at a time.
+ # GPUs on this machine, each held by one node at a time. 0 counts the
+ # NVIDIA device nodes, the way 0 cores works the core count out; anything
+ # else — another vendor, or keeping a card back — is said outright.
FLOW_GPUS: int = 0
# How long the engine's own metrics, events and run records are kept.
#: The largest body `PUT /artifacts` will take, in bytes. A checkpoint or a
diff --git a/backend/fluksio/flow/placement.py b/backend/fluksio/flow/placement.py
index 3ff24e3..4cc7330 100644
--- a/backend/fluksio/flow/placement.py
+++ b/backend/fluksio/flow/placement.py
@@ -215,12 +215,13 @@ class Placer:
said = [shape[2] for shape in capable if shape[2]]
ram = min(ram, max(said)) if said and ram else ram
if (cpus, gpus, ram) != (wanted.cpus, wanted.gpus, wanted.ram or 0):
- # Cards are not detected, so a machine that has one still reports
- # none until it is told — which reads as "no GPU here" to a node
- # that then runs unserialised beside every other one.
+ # Only NVIDIA's device nodes are counted, so a card behind another
+ # vendor's driver reports as none until it is told — which reads as
+ # "no GPU here" to a node that then runs unserialised beside every
+ # other one.
hint = (
- "; no machine here declares a GPU — `fluksio serve --gpus N` "
- "(or FLOW_GPUS) says how many this one has"
+ "; no machine here has a GPU it knows of — `fluksio serve "
+ "--gpus N` (or FLOW_GPUS) says how many this one has"
if wanted.gpus and not gpus
else ""
)
diff --git a/backend/fluksio/flow/resources.py b/backend/fluksio/flow/resources.py
index 3688f54..27cd6ea 100644
--- a/backend/fluksio/flow/resources.py
+++ b/backend/fluksio/flow/resources.py
@@ -24,6 +24,7 @@ same trust the worker pool already extends to node code.
from __future__ import annotations
+import glob
import logging
import os
import re
@@ -63,6 +64,18 @@ def machine_cpus() -> int:
return max(1, (os.cpu_count() or 1) - ENGINE_RESERVE)
+def machine_gpus() -> int:
+ """Cards a node may be given here, when nobody said.
+
+ The device nodes the NVIDIA driver creates, counted — no vendor tool, so
+ the one dependency does not become two. It answers for the case that
+ actually bit people (a box with cards, an engine told nothing, every GPU
+ node clamped to zero and running at once) and for nothing else: another
+ vendor, or a card this process cannot see, is still `--gpus N`.
+ """
+ return len(glob.glob("/dev/nvidia[0-9]*"))
+
+
@dataclass(frozen=True, slots=True)
class Allocation:
"""What one execution was actually given."""
diff --git a/backend/fluksio/flow/runs.py b/backend/fluksio/flow/runs.py
index b143333..e80b26d 100644
--- a/backend/fluksio/flow/runs.py
+++ b/backend/fluksio/flow/runs.py
@@ -454,6 +454,11 @@ class MetricSink:
which drops what it cannot keep up with, and a training curve with holes in
it is not a result.
+ A batch that has been written announces itself (`on_flush`), which is how a
+ screen watching a run knows there is something new to read. The event
+ carries names rather than numbers: it is a nudge, and the rows are the
+ record. One per batch, so it is nowhere near the per-point path.
+
The step is the count of emissions on that message. A node that publishes
every tenth training step therefore has steps 0, 1, 2 rather than 0, 10,
20 — a faithful x-axis of its own emissions, not of the loop inside it.
@@ -464,10 +469,12 @@ class MetricSink:
run_id: str,
batch: int = METRIC_BATCH,
interval: float = METRIC_FLUSH_S,
+ on_flush: Callable[[list[str]], None] | None = None,
) -> None:
self.run_id = run_id
self._batch = batch
self._interval = interval
+ self._on_flush = on_flush
self._rows: dict[tuple[str, int], RunMetric] = {}
self._steps: dict[str, int] = {}
self._last_flush = time.monotonic()
@@ -508,7 +515,7 @@ class MetricSink:
self._rows.clear()
self._last_flush = time.monotonic()
if rows:
- self._write(rows)
+ self._written(rows)
def flush(self) -> None:
with self._lock:
@@ -516,7 +523,17 @@ class MetricSink:
self._rows.clear()
self._last_flush = time.monotonic()
if rows:
- self._write(rows)
+ self._written(rows)
+
+ def _written(self, rows: list[RunMetric]) -> None:
+ """Write a batch, then say that it is there."""
+ self._write(rows)
+ if self._on_flush is None:
+ return
+ try:
+ self._on_flush(sorted({row.name for row in rows}))
+ except Exception:
+ logger.exception("Announcing a batch of run %s failed", self.run_id)
def _write(self, rows: list[RunMetric]) -> None:
try:
@@ -1207,7 +1224,18 @@ class RunService:
errors += 1
self._record_node(run_id, outcome)
- sink = MetricSink(run_id)
+ sink = MetricSink(
+ run_id,
+ on_flush=lambda names: self._publish_event(
+ {
+ "type": "run_metric",
+ "flow": run.flow,
+ "run": run_id,
+ "names": names,
+ "ts": time.time(),
+ }
+ ),
+ )
try:
flow = self.controller.store.read_flow(run.flow, draft=run.draft)
# Read again, now that it is this run's turn: a sweep queues every
diff --git a/backend/fluksio/main.py b/backend/fluksio/main.py
index c6c2020..9fcb9c1 100644
--- a/backend/fluksio/main.py
+++ b/backend/fluksio/main.py
@@ -39,7 +39,11 @@ from fluksio.flow.plugins import load_plugins
from fluksio.flow.provision import load_provisioners
from fluksio.flow.queue import MemoryWorkQueue, RedisWorkQueue, WorkQueue
from fluksio.flow.remote import RemoteWorkerHub
-from fluksio.flow.resources import ResourceAccountant, fair_share_env
+from fluksio.flow.resources import (
+ ResourceAccountant,
+ fair_share_env,
+ machine_gpus,
+)
from fluksio.flow.runs import RUN_STATE_TTL, RunService, sweep_artifacts
from fluksio.flow.secrets import init_secrets
from fluksio.flow.state import MemoryState, RedisState, StateBackend
@@ -251,8 +255,9 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
volatile = VolatileStore(_volatile_root(), settings.ARTIFACT_VOLATILE_BYTES)
artifacts.volatile = volatile
app.state.artifact_store = artifacts
+ # Zero means "work it out" for both, as it always has for cores.
accountant = ResourceAccountant(
- cpus=settings.FLOW_CPUS, gpus=settings.FLOW_GPUS
+ cpus=settings.FLOW_CPUS, gpus=settings.FLOW_GPUS or machine_gpus()
)
app.state.resources = accountant
# Every machine a node could run on: this one, and whatever attaches.
diff --git a/backend/fluksio/sdk/cli.py b/backend/fluksio/sdk/cli.py
index e6df1ea..242b491 100644
--- a/backend/fluksio/sdk/cli.py
+++ b/backend/fluksio/sdk/cli.py
@@ -999,15 +999,41 @@ def cmd_runs(args: argparse.Namespace) -> int:
return 0
-#: The most runs of one group a retry looks at, which is the list route's own
-#: ceiling. ponytail: a sweep larger than this needs paging, not a bigger number.
-MAX_GROUP = 500
+#: One page of the list route, which is its own ceiling. A sweep larger than
+#: this is read a page at a time rather than asked for in one.
+PAGE = 500
#: What a retry leaves alone. Everything else in a group — error, cancelled,
#: abandoned — is what "the ones that did not make it" means.
KEPT = frozenset({"ok", "cached", "queued", "running"})
+def _group_runs(client: Client, group: str) -> Iterator[dict[str, Any]]:
+ """Every run of a sweep, a page at a time.
+
+ Rows come newest first and `before` is the cursor: handing back the last
+ row's own `created_at` reads the next page whatever landed meanwhile, where
+ an offset would shift under a run submitted between two pages.
+
+ # ponytail: `before` is exclusive, so two runs sharing a timestamp exactly
+ # across a page edge would lose one. Ids are minted per run; the history
+ # has never produced a tie.
+ """
+ before = ""
+ while True:
+ page = list(
+ client.runs(
+ limit=PAGE, group=group, **({"before": before} if before else {})
+ )
+ )
+ yield from page
+ if len(page) < PAGE:
+ return
+ before = str(page[-1].get("created_at") or "")
+ if not before:
+ return
+
+
def cmd_retry(args: argparse.Namespace) -> int:
"""Run the same thing again, one run or a group's unfinished ones."""
try:
@@ -1016,7 +1042,7 @@ def cmd_retry(args: argparse.Namespace) -> int:
if args.group:
ids += [
str(row["id"])
- for row in client.runs(limit=MAX_GROUP, group=args.group)
+ for row in _group_runs(client, args.group)
if str(row["status"]) not in KEPT
]
if not ids:
@@ -1207,39 +1233,15 @@ def _write_rows(rows: list[dict[str, Any]], fmt: str, out: str, hint: str = "")
return 0
-#: How many runs `--list` reads before giving up on finding a metric name.
-#: The names belong to the flow's nodes rather than to a run, so the newest
-#: one that recorded any is the whole vocabulary — the rest are for a
-#: selection whose newest runs failed before they measured anything.
-# ponytail: the first run with names wins; a name only an older run recorded
-# is not listed. A `distinct` over the selection would be exact and is a route
-# of its own.
-LIST_SCAN = 10
-
-
-def metric_names(client: Client, ids: Iterable[str]) -> list[str]:
- """The metric names these runs carry, since a name is flow-qualified.
-
- `train_loss` is recorded as `train.train_loss`, and asking for the bare
- one matches nothing — so this is the answer to "what would match".
- """
- for run_id in list(ids)[:LIST_SCAN]:
- names = sorted({point["name"] for point in client.metrics(run_id)})
- if names:
- return names
- return []
-
-
def _list_names(client: Client, args: argparse.Namespace) -> int:
- filters = _selection(args)
- if "until" in filters:
- # The history spells the same bound `before`, where it is also the
- # cursor a page is taken from.
- filters["before"] = filters.pop("until")
- ids = args.run or [
- row["id"] for row in client.runs(flow=args.flow, limit=LIST_SCAN, **filters)
- ]
- names = metric_names(client, ids)
+ """What `--list` prints: the names a selection recorded, from the engine.
+
+ One `distinct` over the whole selection rather than the newest run that
+ happened to measure anything, so a name only an older run ever wrote is
+ listed too. The selection is the export's own filters, unchanged — this
+ route takes `until` where the history route calls the same bound `before`.
+ """
+ names = client.metric_names(ids=args.run, flow=args.flow or "", **_selection(args))
_say("\n".join(names) if names else "No metrics recorded by these runs.")
return 0
@@ -1275,13 +1277,15 @@ def _export(
)
try:
with _client_for(args, retries=0) as client:
- if getattr(args, "list_names", False):
- return _list_names(client, args)
try:
+ if getattr(args, "list_names", False):
+ return _list_names(client, args)
rows = fetch(client)
except ApiError as exc:
if exc.status != 404:
raise
+ # `--list` asks a route of its own, and it arrived later than
+ # the exports — so an engine without either says the same thing.
return _fail(_too_old(client))
except (SyncError, ApiError) as exc:
return _fail(str(exc))
diff --git a/backend/fluksio/sdk/client.py b/backend/fluksio/sdk/client.py
index bda939e..7c57bfb 100644
--- a/backend/fluksio/sdk/client.py
+++ b/backend/fluksio/sdk/client.py
@@ -390,6 +390,23 @@ class Client:
query["name"] = name
return self._call("GET", f"/runs/{run_id}/metrics", params=query)
+ def metric_names(
+ self, ids: Iterable[str] = (), flow: str = "", **filters: Any
+ ) -> Any:
+ """The metric names a selection of runs recorded, distinct and exact.
+
+ Names are flow-qualified, so `train_loss` is recorded as
+ `train.train_loss` and asking for the bare one matches nothing — this
+ is what says which spellings exist.
+ """
+ query: dict[str, Any] = dict(filters)
+ named = [run_id for run_id in ids if run_id]
+ if named:
+ query["ids"] = ",".join(named)
+ if flow:
+ query["flow"] = flow
+ return self._call("GET", "/runs/metrics/names", params=query)
+
def compare(self, ids: Iterable[str], metric: str, x: str = "") -> Any:
"""One metric across several runs, in the chart widget's series shape.
diff --git a/backend/fluksio/tui/app.py b/backend/fluksio/tui/app.py
index 91141b5..e3ea5ed 100644
--- a/backend/fluksio/tui/app.py
+++ b/backend/fluksio/tui/app.py
@@ -59,9 +59,17 @@ MAX_PICKED = 20
RUNS_SHOWN = 50
#: What the dashboard subscribes to. `message_value` is most of what the bus
-#: carries and none of what this screen draws.
+#: carries and none of what this screen draws. `run_metric` says a run has
+#: written readings, which is the only thing that moves a curve.
KINDS = frozenset(
- {"run_started", "run_finished", "flow_paused", "engine_fatal", "flow_changed"}
+ {
+ "run_started",
+ "run_finished",
+ "flow_paused",
+ "engine_fatal",
+ "flow_changed",
+ "run_metric",
+ }
)
#: How long an event waits for the ones behind it. A sweep starting twenty
@@ -72,17 +80,13 @@ COALESCE_S = 0.25
#: rather than a pipe because the screen is meant to be closed while the
#: engine keeps running: a pipe whose reader has gone breaks the next write,
#: and a node's `print` is one of those writes. It also means the log of an
-#: engine this screen only *adopted* can be read here.
+#: engine this screen only *adopted* can be read here. The engine cuts it back
+#: when it grows (`cli.LOG_KEEP_BYTES`), so the bound holds whoever started it.
LOG_NAME = "serve.log"
#: How much of it to read back when the screen opens.
LOG_TAIL_BYTES = 64 * 1024
-#: What it is truncated to when this screen starts an engine of its own.
-#: ponytail: a size check, not rotation — add rotation when somebody wants
-#: yesterday's log.
-LOG_KEEP_BYTES = 5 * 1024 * 1024
-
#: How long the tail waits when the file has nothing new. Also how long
#: closing the screen waits for that thread.
LOG_POLL_S = 0.25
@@ -299,13 +303,18 @@ class RunsTable(DataTable[str]):
class ServeApp(App[int]):
"""Three tabs: how the engine is, what has run, and what it is saying."""
+ #: The two modals take a border rather than a fill: on the ansi theme the
+ #: surface *is* the terminal's background, so an edge is what stands them
+ #: apart from what they are drawn over.
CSS = """
TabbedContent { height: 1fr; }
TabPane { height: 1fr; padding: 0; }
#overview { padding: 0 1; }
- #enroll { width: 60; height: auto; padding: 1 2; background: $surface; }
+ #enroll { width: 60; height: auto; padding: 1 2; background: $surface;
+ border: round $primary; }
#enroll #note { color: $text-muted; height: auto; }
- #artifacts { width: 80; height: auto; padding: 1 2; background: $surface; }
+ #artifacts { width: 80; height: auto; padding: 1 2; background: $surface;
+ border: round $primary; }
#artifacts DataTable { height: auto; max-height: 14; }
"""
@@ -334,6 +343,8 @@ class ServeApp(App[int]):
#: Set by the socket thread, drained by the coalescing timer — so a
#: cascade of events costs one read of the engine rather than one each.
self.stirred = False
+ #: The same, for readings: a comparison redraws, the panels do not.
+ self.plotted = False
self.stop_stream = threading.Event()
self.stop_log = threading.Event()
#: The credential's mtime when the engine under this screen was
@@ -355,6 +366,12 @@ class ServeApp(App[int]):
yield Footer()
def on_mount(self) -> None:
+ # The screen is drawn inside somebody's terminal and should look like
+ # it: this theme paints no ground of its own and names its colours by
+ # their ANSI slots, so the palette is whatever they have configured —
+ # which is also what the chart's own hues and the status colours have
+ # always been.
+ self.theme = "ansi-dark"
self.title = f"fluksio — {self.data_dir}"
table = self.query_one("#runs", RunsTable)
table.add_column(" ", width=1)
@@ -435,10 +452,10 @@ class ServeApp(App[int]):
# What the credential looked like before the child wrote its own, so
# the wait below cannot mistake the last engine's url for this one's.
self.config_stamp = self._config_stamp()
- path = self.log_path()
- if path.exists() and path.stat().st_size > LOG_KEEP_BYTES:
- path.write_text("")
- handle = path.open("ab", buffering=0)
+ # Opened for appending, which is what lets the engine cut it back when
+ # it grows: every write then lands at the new end rather than at the
+ # offset this process had.
+ handle = self.log_path().open("ab", buffering=0)
try:
self.child = subprocess.Popen( # noqa: S603
child_argv(sys.argv[1:]),
@@ -568,13 +585,20 @@ class ServeApp(App[int]):
# -- the engine's own events ----------------------------------------------
+ def heard(self, event: dict[str, Any]) -> None:
+ """One event, from the socket thread. Two flags, drained on a timer."""
+ if event.get("type") == "run_metric":
+ self.plotted = True
+ else:
+ self.stirred = True
+
@work(thread=True, exclusive=True, group="stream")
def watch_engine(self, url: str, token: str) -> None:
"""The bus, over the websocket the browser subscribes to."""
stream.subscribe(
url,
token,
- lambda event: setattr(self, "stirred", True),
+ self.heard,
self.stop_stream,
kinds=KINDS,
on_error=lambda exc: self.call_from_thread(
@@ -584,14 +608,20 @@ class ServeApp(App[int]):
def drain(self) -> None:
"""Whatever the bus said since the last tick, as one read."""
+ showing = self.screen if isinstance(self.screen, CompareScreen) else None
+ if self.plotted:
+ self.plotted = False
+ # Readings landed. Only the screen drawing them cares.
+ if showing is not None:
+ showing.refresh_live()
if not self.stirred:
return
self.stirred = False
self.refresh_panels()
# A run finishing is exactly when a comparison's numbers become final,
# and the screen showing them cannot hear the bus itself.
- if isinstance(self.screen, CompareScreen):
- self.screen.action_reload()
+ if showing is not None:
+ showing.action_reload()
# -- what the panels show -------------------------------------------------
diff --git a/backend/fluksio/tui/compare.py b/backend/fluksio/tui/compare.py
index 8a3a689..638dc31 100644
--- a/backend/fluksio/tui/compare.py
+++ b/backend/fluksio/tui/compare.py
@@ -19,7 +19,7 @@ from textual.containers import Horizontal
from textual.screen import Screen
from textual.widgets import DataTable, Footer, Header, Select, Static
-from fluksio.sdk.cli import _dur, metric_names
+from fluksio.sdk.cli import _dur
from fluksio.sdk.client import Client
from fluksio.tui.chart import MAX_SERIES, Curves, legend
@@ -33,8 +33,9 @@ PARAM, METRIC = "param.", "metric."
RUNNING = frozenset({"queued", "running"})
#: How often a curve is re-read while a run in the selection is still going.
-#: Metric points are not published on the event bus — the run's sink writes
-#: them — so the socket makes the *table* live and this makes the *curve* live.
+#: The engine announces each batch its sink writes (`run_metric`), which is
+#: what usually wakes this screen; the poll is the heartbeat behind it, and
+#: what makes the screen live against an engine older than that event.
LIVE_S = 1.0
@@ -136,7 +137,7 @@ class CompareScreen(Screen[None]):
def read_table(self) -> None:
try:
rows = self.client.export_runs(ids=self.ids)
- names = metric_names(self.client, self.ids)
+ names = list(self.client.metric_names(ids=self.ids))
except Exception as exc: # noqa: BLE001 — the screen reports it
self.app.call_from_thread(self.show_error, exc)
return
@@ -213,8 +214,7 @@ class CompareScreen(Screen[None]):
*(_cell(row.get(key)) for key in params),
*(_cell(row.get(key)) for key in scores),
)
- # A curve only grows while its run does, and metric points are not on
- # the event bus — so this is the one thing the screen polls for.
+ # A curve only grows while its run does.
self.keep_live(any(str(row.get("status")) in RUNNING for row in rows))
self.read_curves()
@@ -228,9 +228,21 @@ class CompareScreen(Screen[None]):
legend(label for label, _ in lines[:MAX_SERIES])
)
+ def refresh_live(self) -> None:
+ """New readings have landed. Draw them.
+
+ A run opened before it measured anything has no metric picked, and
+ nothing but the table read fills that in — so the whole read is what a
+ first batch needs, and only the curve after that.
+ """
+ if not self.names:
+ self.read_table()
+ else:
+ self.read_curves()
+
def keep_live(self, running: bool) -> None:
if running and self.live is None:
- self.live = self.set_interval(LIVE_S, self.read_curves)
+ self.live = self.set_interval(LIVE_S, self.refresh_live)
elif not running and self.live is not None:
self.live.stop()
self.live = None
diff --git a/backend/tests/api/routes/test_runs.py b/backend/tests/api/routes/test_runs.py
index 8df5a0f..b5a8549 100644
--- a/backend/tests/api/routes/test_runs.py
+++ b/backend/tests/api/routes/test_runs.py
@@ -886,6 +886,64 @@ def plotted():
session.commit()
+def test_the_metric_names_of_a_selection_are_exact(
+ client, superuser_token_headers, plotted
+):
+ """Names are flow-qualified, so this is what says which spellings exist.
+
+ Read over the whole selection rather than off the newest run that measured
+ anything, which is what missed a name only an older run ever wrote — here,
+ the older run's `study.grad` against the newer one's two.
+ """
+ older = new_run_id()
+ with Session(db_engine) as session:
+ session.add(
+ Run(
+ id=older,
+ flow="study",
+ status="ok",
+ created_at=datetime.now(UTC) - timedelta(hours=1),
+ )
+ )
+ session.add(_metric(older, "study.grad", 0, 0.1, 50.0))
+ session.commit()
+
+ try:
+ both = client.get(
+ f"{settings.API_V1_STR}/runs/metrics/names",
+ params={"ids": f"{plotted},{older}"},
+ headers=superuser_token_headers,
+ )
+ assert both.status_code == 200
+ assert both.json() == ["study.epoch", "study.grad", "study.loss"]
+
+ # The same question by filter rather than by name.
+ assert client.get(
+ f"{settings.API_V1_STR}/runs/metrics/names",
+ params={"flow": "study"},
+ headers=superuser_token_headers,
+ ).json() == ["study.epoch", "study.grad", "study.loss"]
+
+ # A selection that recorded nothing has nothing to offer, which is not
+ # an error: a run opened before its first reading is the ordinary case.
+ assert (
+ client.get(
+ f"{settings.API_V1_STR}/runs/metrics/names",
+ params={"ids": "no-such-run"},
+ headers=superuser_token_headers,
+ ).json()
+ == []
+ )
+ finally:
+ with Session(db_engine) as session:
+ for row in session.exec(
+ select(RunMetric).where(col(RunMetric.run_id) == older)
+ ).all():
+ session.delete(row)
+ session.delete(session.get(Run, older))
+ session.commit()
+
+
def test_a_comparison_is_plotted_against_the_step_by_default(
client, superuser_token_headers, plotted
):
diff --git a/backend/tests/flow/test_resources.py b/backend/tests/flow/test_resources.py
index b0e347b..5276028 100644
--- a/backend/tests/flow/test_resources.py
+++ b/backend/tests/flow/test_resources.py
@@ -18,10 +18,32 @@ from fluksio.flow.resources import (
ResourceAccountant,
derive_env,
fair_share_env,
+ machine_gpus,
)
from fluksio.flow.schemas import NodeDef, Resources
+def test_the_cards_are_counted_from_the_devices_the_driver_makes(monkeypatch):
+ """A box with cards and an engine told nothing used to have none of them.
+
+ NVIDIA's own device nodes, counted — no vendor tool, so the one dependency
+ does not become two. The control files beside them are not cards.
+ """
+ from fluksio.flow import resources
+
+ monkeypatch.setattr(
+ resources.glob,
+ "glob",
+ lambda pattern: (
+ ["/dev/nvidia0", "/dev/nvidia1"] if pattern == "/dev/nvidia[0-9]*" else []
+ ),
+ )
+ assert machine_gpus() == 2
+
+ monkeypatch.setattr(resources.glob, "glob", lambda pattern: [])
+ assert machine_gpus() == 0
+
+
def test_what_is_free_is_what_was_handed_out():
accountant = ResourceAccountant(cpus=4, gpus=0)
held = accountant.try_take(cpus=3, gpus=0)
diff --git a/backend/tests/flow/test_runs.py b/backend/tests/flow/test_runs.py
index 73cb2aa..b36a9d7 100644
--- a/backend/tests/flow/test_runs.py
+++ b/backend/tests/flow/test_runs.py
@@ -296,6 +296,30 @@ def test_emissions_reach_the_run_as_a_series_with_a_step_each():
assert sink._steps == {"study.loss": 1}
+def test_a_written_batch_says_it_is_there():
+ """What a screen watching a live run waits on.
+
+ The rows are still the record — this is a nudge carrying names, one per
+ batch rather than one per point, so nothing lands on the hot path.
+ """
+ said = []
+ sink = MetricSink("run-1", batch=1, on_flush=said.append)
+
+ sink.handle("study.train", {"study.loss": 1.0, "study.tag": "ignored"})
+ assert said == [["study.loss"]]
+
+ # Held back until the batch is due, then announced once for all of it.
+ quiet = MetricSink("run-2", batch=10, interval=3600, on_flush=said.append)
+ quiet.handle("study.train", {"study.loss": 1.0, "study.acc": 0.5})
+ assert said == [["study.loss"]]
+ quiet.flush()
+ assert said[-1] == ["study.acc", "study.loss"]
+
+ # Nothing to write is nothing to say.
+ quiet.flush()
+ assert len(said) == 2
+
+
# -----------------------------------------------------------------------------
# What a caller may ask for
# -----------------------------------------------------------------------------
diff --git a/backend/tests/test_cli.py b/backend/tests/test_cli.py
index 07e153d..1ede9f5 100644
--- a/backend/tests/test_cli.py
+++ b/backend/tests/test_cli.py
@@ -414,8 +414,13 @@ def test_an_export_is_parsed_with_its_selection() -> None:
assert runs.format == "jsonl"
-def test_the_metric_names_are_asked_for_rather_than_guessed() -> None:
- """A name is flow-qualified, so `--list` is what says what would match."""
+def test_the_metric_names_are_asked_for_rather_than_guessed(capsys) -> None:
+ """A name is flow-qualified, so `--list` is what says what would match.
+
+ The engine answers it over the whole selection, so a name only an older run
+ ever recorded is listed — which reading the newest run that measured
+ anything could not do.
+ """
from fluksio.cli import _parser
from fluksio.sdk.cli import _list_names
@@ -423,20 +428,19 @@ def test_the_metric_names_are_asked_for_rather_than_guessed() -> None:
assert parser.parse_args(["export", "metrics"]).list_names is False
assert parser.parse_args(["export", "metrics", "--list"]).list_names is True
- class Engine:
- def runs(self, flow="", limit=0, **filters):
- assert filters == {"status": "ok"}
- return [{"id": "r-empty"}, {"id": "r-1"}]
+ asked = {}
- def metrics(self, run_id, name="", stride=1):
- # The newest run failed before it measured anything; the next one
- # carries the vocabulary.
- return [] if run_id == "r-empty" else [{"name": "train.train_loss"}]
+ class Engine:
+ def metric_names(self, ids=(), flow="", **filters):
+ asked.update({"ids": list(ids), "flow": flow, **filters})
+ return ["train.train_loss", "train.val_loss"]
args = parser.parse_args(
["export", "metrics", "--flow", "train", "--status", "ok", "--list"]
)
assert _list_names(Engine(), args) == 0
+ assert asked == {"ids": [], "flow": "train", "status": "ok"}
+ assert capsys.readouterr().out.split() == ["train.train_loss", "train.val_loss"]
def test_a_runs_artifact_is_listed_and_downloaded(tmp_path, monkeypatch) -> None:
@@ -592,14 +596,19 @@ def test_serve_says_when_a_flow_wants_a_card_nobody_declared(
) -> None:
"""The clamp warning goes to the log; this is said while someone is reading.
- Cards are declared rather than detected, so a fresh install that forgets
- `--gpus` clamps a GPU node to zero and runs them all at once.
+ A card behind a driver whose device nodes are not NVIDIA's is not counted,
+ so an install that forgets `--gpus` there clamps a GPU node to zero and
+ runs them all at once. Detection is stubbed either way: whether this test
+ says anything must not depend on the machine running it.
"""
from fluksio import cli
from fluksio.core.config import settings
+ from fluksio.flow import resources
from fluksio.flow.schemas import FlowDef, NodeDef, Resources
from fluksio.flow.store import FlowStore
+ monkeypatch.setattr(resources, "machine_gpus", lambda: 0)
+
store = FlowStore(tmp_path / "flows")
store.write_flow(
FlowDef(
@@ -621,6 +630,13 @@ def test_serve_says_when_a_flow_wants_a_card_nobody_declared(
cli._mention_undeclared_cards()
assert capsys.readouterr().out == ""
+ # Nor when it found them itself, which is the ordinary case on a box with
+ # NVIDIA's driver: nobody has to say what the device nodes already do.
+ monkeypatch.setattr(settings, "FLOW_GPUS", 0)
+ monkeypatch.setattr(resources, "machine_gpus", lambda: 1)
+ cli._mention_undeclared_cards()
+ assert capsys.readouterr().out == ""
+
def test_a_serve_limit_is_refused_as_a_flag_not_as_a_traceback(capsys) -> None:
"""These are written into the environment before the settings are built."""
@@ -745,6 +761,92 @@ def test_a_serving_engine_records_itself_until_it_stops(tmp_path) -> None:
assert read_pidfile(tmp_path) is None
+def test_the_pidfile_goes_with_the_process_however_it_ends(tmp_path) -> None:
+ """A SIGTERM used to leave `serve.pid` behind, which is what a stop sends.
+
+ uvicorn puts back the handler it found and re-raises the signal it stopped
+ on; the default handler ends the process without unwinding, so the `finally`
+ that removes the file never ran. A stale file costs the dashboard a probe
+ and, if a pid is ever recycled, points at a stranger.
+ """
+ import signal
+
+ from fluksio.cli import _hold_pidfile, write_pidfile
+
+ was = signal.getsignal(signal.SIGTERM)
+ try:
+ written = write_pidfile(tmp_path, 8000)
+ assert written.exists()
+ with pytest.raises(SystemExit):
+ _hold_pidfile(written, lambda: signal.raise_signal(signal.SIGTERM))
+ assert not written.exists()
+
+ # And the ordinary way out, which always worked.
+ written = write_pidfile(tmp_path, 8000)
+ _hold_pidfile(written, lambda: None)
+ assert not written.exists()
+ finally:
+ signal.signal(signal.SIGTERM, was)
+
+
+def test_the_engine_cuts_its_own_log_when_it_grows(tmp_path) -> None:
+ """The dashboard cut it only when *it* started the engine, so an adopted
+ one — or one whose screen was closed — wrote without a bound.
+
+ Only an appended file is cut: the kernel then puts the next write at the
+ new end, where a `>` redirect would keep its offset and leave a hole.
+ """
+ from fluksio.cli import _appended_log, _trim_log
+
+ path = tmp_path / "serve.log"
+ with path.open("ab", buffering=0) as handle:
+ fd = handle.fileno()
+ handle.write(b"0123456789")
+ assert _appended_log(fd) is True
+ assert _trim_log(fd, limit=100) is False
+ assert path.stat().st_size == 10
+ assert _trim_log(fd, limit=5) is True
+ assert path.stat().st_size == 0
+ # It keeps writing into the same descriptor afterwards.
+ handle.write(b"after")
+ assert path.read_bytes() == b"after"
+
+ with path.open("wb", buffering=0) as handle:
+ # Not appended, so not this engine's to cut.
+ handle.write(b"0123456789")
+ assert _appended_log(handle.fileno()) is False
+
+
+def test_a_sweep_larger_than_a_page_is_retried_whole() -> None:
+ """`retry --group` read one page of the list route and stopped there.
+
+ Pages come newest first and `before` is the cursor, so the next page is
+ taken from the last row's own timestamp rather than from an offset that
+ shifts under a run submitted meanwhile.
+ """
+ from fluksio.sdk.cli import PAGE, _group_runs
+
+ asked = []
+
+ class Engine:
+ def runs(self, flow="", limit=0, **filters):
+ asked.append(filters)
+ page = 0 if "before" not in filters else 1
+ if page:
+ return [{"id": "last", "created_at": "2026-09-02T00:00:00Z"}]
+ return [
+ {"id": f"r{index}", "created_at": f"2026-09-02T00:00:{index:02d}Z"}
+ for index in range(PAGE)
+ ]
+
+ rows = list(_group_runs(Engine(), "sweep-1"))
+
+ assert len(rows) == PAGE + 1
+ assert asked[0] == {"group": "sweep-1"}
+ # The second page starts where the first ended.
+ assert asked[1]["before"] == rows[PAGE - 1]["created_at"]
+
+
def test_who_holds_the_port_is_told_apart_by_the_token() -> None:
"""Only this directory's own engine may be reported as already up.
diff --git a/docs/code/api.md b/docs/code/api.md
index 6273da2..825a63c 100644
--- a/docs/code/api.md
+++ b/docs/code/api.md
@@ -107,8 +107,10 @@ published to. Flows own the namespace; everything else is a client of it.
| `GET` | `/runs/overview` | one row per flow that has runs, with how many are running or queued |
| `GET` | `/runs/export/metrics?…&name=&stride=&format=` | every selected run's series as one long table: `run, name, step, ts, value` |
| `GET` | `/runs/export/runs?…¶ms=&metrics=&format=` | one row per run: its inputs as columns, its final numbers, its status and provenance |
+| `GET` | `/runs/metrics/names?…` | every metric name the selected runs recorded, distinct; takes the export's own filters |
| `GET` | `/runs/{id}` | one run in full: params, result, per-node record, artifacts |
| `POST` | `/runs/{id}/cancel` | stop it |
+| `POST` | `/runs/{id}/retry` | run the same thing again, as a new run naming this one |
| `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=&x=` | that metric across several runs. `x` is what to plot against: nothing or `step`, `time` (seconds since each run's own first reading), or another metric's name, joined on the step the two share |
diff --git a/docs/code/cli.md b/docs/code/cli.md
index fb499a4..6f5bb2b 100644
--- a/docs/code/cli.md
+++ b/docs/code/cli.md
@@ -77,12 +77,14 @@ Another instance's Fluksio on the port is named, and the move happens as usual.
| `--max-runs N` | 4 | batch runs driven at once (`FLOW_MAX_RUNS`) |
| `--max-cascades N` | 4 | cascades in flight at once (`FLOW_MAX_CASCADES`) |
| `--max-workers N` | 4 | python worker processes (`FLOW_MAX_WORKERS`) |
-| `--gpus N` | 0 | GPUs on this machine a node may be given (`FLOW_GPUS`) |
+| `--gpus N` | counted | GPUs on this machine a node may be given (`FLOW_GPUS`) |
-Cards are declared rather than detected, since asking a vendor's tooling would
-make one dependency two, so a machine with a GPU reports none until `--gpus` says
-otherwise, and a node asking for one is clamped to zero and runs alongside
-every other. `--gpus 1` is what serialises them.
+Cards are counted from NVIDIA's device nodes (`/dev/nvidia0`, `/dev/nvidia1`, …),
+which asks no vendor tool and so keeps the one dependency from becoming two.
+Anything they do not cover — another vendor, or a card this process cannot see
+— reports none until `--gpus` says otherwise, and a node asking for one is then
+clamped to zero and runs alongside every other. `--gpus 1` is what serialises
+them. Passing a number always wins over the count.
`--enroll` with `--portal` is the one-command setup: it pairs before the engine
starts, so the connection is dialled as part of coming up rather than needing a
@@ -147,6 +149,11 @@ which is what lets the screen be closed while the engine keeps running — a
pipe with nobody reading it breaks the next line the engine writes, and a
node's `print` is one of those. It also means the Logs tab has the output of
an engine this screen only adopted, and the scrollback of the one before it.
+The engine cuts the file back to nothing once it passes 5 MB, so the bound
+holds whether or not a screen is open.
+
+The screen takes its colours from the terminal rather than painting its own,
+so it sits inside a light profile as readily as a dark one.
An engine started elsewhere is adopted rather than duplicated, and can be
stopped from here only when it is this instance's own: both the pidfile
@@ -502,7 +509,8 @@ taken leaf by leaf rather than as one blob.
Metric names are flow-qualified (a node of `train` writing `train_loss` records
`train.train_loss`) so `--list` prints the names the selected runs
-carry when the spelling is not obvious.
+carry when the spelling is not obvious. The engine answers it over the whole
+selection, so a name only an older run ever recorded is listed too.
Both take `--flow`, `--run ID` (repeat it), `--group`, `--status`, `--since`,
`--until` and `--local`, and both put the run id on every row: it is the join
diff --git a/docs/code/workers.md b/docs/code/workers.md
index 2655f37..cb01a24 100644
--- a/docs/code/workers.md
+++ b/docs/code/workers.md
@@ -36,7 +36,7 @@ driver in order to run a training step. An engine host already has it, and
| `--parallel` | `1` | how many node calls it will take at once |
| `--artifact-url` | derived from `--url` | where the artifact store is, if not beside the socket |
| `--cpus` | what the job or the machine has | cores to advertise |
-| `--gpus` | what the job says, else none | GPUs to advertise; never probed |
+| `--gpus` | what the job says, else counted | GPUs to advertise |
| `--ram-mb` | what the job or the machine has | memory to advertise, in MB |
| `--max-idle` | never | stop after this many seconds with nothing running |
@@ -51,12 +51,15 @@ the engine schedules against it: a node asking for two cores and a GPU goes to
a machine that has them free, not merely to one carrying the right label.
Cores and memory are read off the machine, or off the batch job that started
-this worker (`SLURM_CPUS_ON_NODE`, `SLURM_MEM_PER_NODE`). **GPUs are never
-probed.** Asking a vendor tool would make the one dependency two, so a GPU is
-something the job says it was given (`SLURM_GPUS_ON_NODE`, `SLURM_JOB_GPUS`,
-or `FLUKSIO_WORKER_GPUS`) or something you say with `--gpus`. A worker that
-reports nothing still attaches and is scheduled by its label alone, as every
-worker was before any of them reported anything.
+this worker (`SLURM_CPUS_ON_NODE`, `SLURM_MEM_PER_NODE`). **What the job says
+it was given always wins for GPUs** (`SLURM_GPUS_ON_NODE`, `SLURM_JOB_GPUS`,
+or `FLUKSIO_WORKER_GPUS`): a node with eight cards may have granted this job
+one, and advertising eight would be a lie the scheduler acts on. With nothing
+said, NVIDIA's device nodes are counted, the same as the engine does for its
+own machine, and `--gpus` overrides either. No vendor tool is asked, which is
+what keeps the one dependency from becoming two. A worker that reports nothing
+still attaches and is scheduled by its label alone, as every worker was before
+any of them reported anything.
The engine tells each call what it may use: thread caps, and the devices it
may see. The worker starts a process per call, so it applies them at the only
diff --git a/docs/getting-started/data-science.md b/docs/getting-started/data-science.md
index db279da..11a02eb 100644
--- a/docs/getting-started/data-science.md
+++ b/docs/getting-started/data-science.md
@@ -298,10 +298,11 @@ nothing else is given that card while it runs.
string whose contents depend on the version you have installed, so writing it
for you would silently replace whatever you had put there.
-The engine has to be told how many cards it has (`fluksio serve --gpus 1`, or
-`FLOW_GPUS`) because detecting them would mean depending on a vendor's
-tooling. Until it is, a node asking for one is quietly given zero and runs
-alongside every other; the log says so the first time it happens.
+The engine counts NVIDIA's device nodes to find its cards, which asks no
+vendor tool. A card those do not cover has to be declared (`fluksio serve
+--gpus 1`, or `FLOW_GPUS`); until it is, a node asking for one is quietly
+given zero and runs alongside every other, and the log says so the first time
+it happens.
Workers are kept warm on purpose, so a library that takes most of the card at
import would hold it after the run finished. The workers that ran on a card
@@ -645,7 +646,7 @@ code digest beside it, so an exported file still says what produced its
numbers. Numbers inside a record are columns of their own
(`metric.final_metrics.train_loss`) and `--metrics` and `--params` take those
dotted paths to narrow the table. Metric names are flow-qualified, so
-`--list` prints the ones a selection carries. `Client.export_metrics()` and
+`--list` prints every one a selection recorded. `Client.export_metrics()` and
`Client.export_runs()` answer the same rows to a notebook, ready for
`pandas.DataFrame`.
diff --git a/docs/reference/configuration.md b/docs/reference/configuration.md
index b40d595..3361680 100644
--- a/docs/reference/configuration.md
+++ b/docs/reference/configuration.md
@@ -127,7 +127,7 @@ warning into a refusal to start.
| `FLOW_MAX_RUNS` | `4` | batch runs driven at once. A different limit from the one above: a run drives a whole graph, and its nodes are bounded by `FLOW_MAX_WORKERS`. This is what a sweep queues behind |
| `FLOW_NODE_TIMEOUT` | `0` | seconds a node may be silent, unless it sets its own; 0 is no limit |
| `FLOW_CPUS` | `0` | cores nodes that declare `resources` may be given; 0 works it out as every core but two, which are what keeps the engine answering while the machine is busy |
-| `FLOW_GPUS` | `0` | GPUs on this machine, each held by one node at a time. Not detected — say how many there are |
+| `FLOW_GPUS` | `0` | GPUs on this machine, each held by one node at a time. 0 counts NVIDIA's device nodes; say a number for anything they miss |
| `OBS_RETENTION_DAYS` | `30` | how long metrics, events and run records are kept |
| `ARTIFACT_GC_INTERVAL_S` | `3600` | how often artifact bytes nothing refers to are swept away; 0 never sweeps |
| `ARTIFACT_GC_GRACE_S` | `3600` | how long a freshly written artifact is spared, whatever refers to it |
diff --git a/frontend/src/components/Flow/useFlowSocket.ts b/frontend/src/components/Flow/useFlowSocket.ts
index 6efd0fa..5519d59 100644
--- a/frontend/src/components/Flow/useFlowSocket.ts
+++ b/frontend/src/components/Flow/useFlowSocket.ts
@@ -109,6 +109,13 @@ type FlowEvent =
group?: string
ts?: number
}
+ | {
+ type: "run_metric"
+ flow: string
+ run: string
+ names: string[]
+ ts?: number
+ }
function socketUrl(): string {
const base = String(OpenAPI.BASE || window.location.origin)
@@ -328,6 +335,14 @@ function connect() {
// live value, so nothing here goes through the live store.
client?.invalidateQueries({ queryKey: runKeys.all })
break
+ case "run_metric":
+ // A batch of readings was written. Only what draws them is refetched —
+ // the run's own detail (its metric names and curve hang below that
+ // key) and any comparison — so a training run reporting every second
+ // does not re-read the list behind it.
+ client?.invalidateQueries({ queryKey: runKeys.detail(message.run) })
+ client?.invalidateQueries({ queryKey: runKeys.compares })
+ break
}
}
diff --git a/frontend/src/components/Runs/queries.ts b/frontend/src/components/Runs/queries.ts
index b94dd72..adfb878 100644
--- a/frontend/src/components/Runs/queries.ts
+++ b/frontend/src/components/Runs/queries.ts
@@ -30,6 +30,7 @@ export const runKeys = {
detail: (id: string) => ["runs", "detail", id] as const,
metrics: (id: string, name: string) =>
["runs", "detail", id, "metrics", name] as const,
+ compares: ["runs", "compare"] as const,
compare: (ids: string[], metric: string, x: string) =>
["runs", "compare", ids.join(","), metric, x] as const,
}
diff --git a/frontend/src/components/UserSettings/RemoteAccess.tsx b/frontend/src/components/UserSettings/RemoteAccess.tsx
index 7024771..56926a6 100644
--- a/frontend/src/components/UserSettings/RemoteAccess.tsx
+++ b/frontend/src/components/UserSettings/RemoteAccess.tsx
@@ -74,7 +74,11 @@ export function RemoteAccess() {
}),
onSuccess: () => {
setCode("")
- showSuccessToast("Connected to the portal")
+ showSuccessToast(
+ status?.enrolled
+ ? "Replaced the connection"
+ : "Connected to the portal",
+ )
invalidate()
},
onError: handleError.bind(showErrorToast),
@@ -104,6 +108,54 @@ export function RemoteAccess() {
if (!status) return null
+ // The same fields either way. A code is redeemed before anything local is
+ // written, so one the portal refuses leaves a working connection working —
+ // which is what makes re-pairing safe to offer beside a live one, rather
+ // than only after a Disconnect.
+ const form = (
+
- Get a code at fluksio.com → Instances → Add instance.
-
-
-
-
-
-
+ form
)}
diff --git a/worker/fluksio_worker/agent.py b/worker/fluksio_worker/agent.py
index dc7ae4b..93f0ec9 100644
--- a/worker/fluksio_worker/agent.py
+++ b/worker/fluksio_worker/agent.py
@@ -28,6 +28,7 @@ from __future__ import annotations
import argparse
import asyncio
import contextlib
+import glob
import json
import logging
import os
@@ -328,18 +329,22 @@ def _detect_ram_mb() -> int | None:
def _detect_gpus() -> int:
- """What this worker was *given*, never what the machine has.
+ """What this worker was *given*, else what the box appears to have.
- Nothing is probed: asking a vendor tool would make the one dependency two,
- and the engine does not probe its own GPUs either. A batch scheduler says
- so in the environment; anywhere else it is ``--gpus``.
+ No vendor tool is asked — that would make the one dependency two. A batch
+ scheduler says what the job was given in the environment and that always
+ wins, since a node with eight cards may have granted this job one. With
+ nothing said, NVIDIA's device nodes are counted, which is what the engine
+ does for its own machine.
"""
for name in ("SLURM_GPUS_ON_NODE", "FLUKSIO_WORKER_GPUS"):
given = os.environ.get(name, "")
if given.isdigit():
return int(given)
listed = os.environ.get("SLURM_JOB_GPUS", "")
- return len([part for part in listed.split(",") if part.strip()])
+ if listed.strip():
+ return len([part for part in listed.split(",") if part.strip()])
+ return len(glob.glob("/dev/nvidia[0-9]*"))
def _inventory(args: argparse.Namespace) -> dict[str, Any]: