A node's worker label is worker, so a device is only a paired one
- NodeDef `device`/`device_policy` become `worker`/`worker_policy`, the issue `ignored_device` becomes `ignored_worker`, and the SDK's `@node` takes `worker=` and `worker_policy=` - FlowStore._migrate renames the keys in published and draft documents once, bumping their version; a queued run's `needs` reads the old key - placement passes the label as `label`, client regenerated - docs: worker placement wording, MCP issuer without DOMAIN, device rows in the CLI data directory and the API table, device links in compose comments Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 5.5
parent
f8011d2a8a
commit
a9d453c9d9
+1
-1
@@ -5,7 +5,7 @@
|
||||
# Domain Traefik routes on and acquires TLS certificates for.
|
||||
DOMAIN=localhost
|
||||
|
||||
# The address people and devices reach this instance at: panel links, mails
|
||||
# The address people and devices reach this instance at: device links, mails
|
||||
# and the MCP issuer are built from it.
|
||||
FRONTEND_HOST=http://localhost:5173
|
||||
|
||||
|
||||
+1
-1
@@ -23,7 +23,7 @@ from fluksio import Port, node
|
||||
@node(
|
||||
requires=["dataset", Port("lr", "float")],
|
||||
provides=[Port("loss", "float", stream=True), Port("weights", "artifact")],
|
||||
device="gpu", device_policy="prefer",
|
||||
worker="gpu", worker_policy="prefer",
|
||||
)
|
||||
def fit(dataset, lr, epochs=25):
|
||||
for epoch in range(epochs):
|
||||
|
||||
@@ -745,13 +745,13 @@ async def save_node_source(
|
||||
_follow_header, controller, name, node_id, source.code, drop
|
||||
)
|
||||
node_def = next((n for n in definition.nodes if n.id == node_id), None)
|
||||
device = (
|
||||
node_def.device
|
||||
if node_def is not None and node_def.device_policy == "require"
|
||||
worker = (
|
||||
node_def.worker
|
||||
if node_def is not None and node_def.worker_policy == "require"
|
||||
else None
|
||||
)
|
||||
error = await run_in_threadpool(
|
||||
controller.compile_check, name, node_id, source.code, device
|
||||
controller.compile_check, name, node_id, source.code, worker
|
||||
)
|
||||
return NodeSourceSaved(
|
||||
id=f"{name}.{node_id}",
|
||||
|
||||
@@ -83,7 +83,7 @@ class WorkerInfo(BaseModel):
|
||||
|
||||
class TokenRequest(BaseModel):
|
||||
name: str = Field(min_length=1, max_length=255)
|
||||
#: What a node's `device` may match on this worker; it cannot add to them.
|
||||
#: What a node's `worker` may match on this worker; it cannot add to them.
|
||||
labels: list[str] = Field(default_factory=list)
|
||||
|
||||
|
||||
|
||||
@@ -175,8 +175,8 @@ class LoadedNode:
|
||||
#: Published, and with no body of its own — so what runs is the new-node
|
||||
#: template, which returns nothing at all.
|
||||
missing_source: bool = False
|
||||
#: The device this node names although its type never runs on a worker.
|
||||
ignored_device: str = ""
|
||||
#: The worker this node names although its type never runs on one.
|
||||
ignored_worker: str = ""
|
||||
|
||||
|
||||
@dataclass
|
||||
@@ -482,7 +482,7 @@ class FlowController:
|
||||
# Without a pool, python nodes are compiled and run in this process —
|
||||
# which is what the tests do, and what a bare `Pipeline` has always done.
|
||||
self.workers = workers
|
||||
# Workers on other hosts. A node without a device never touches it.
|
||||
# Workers on other hosts. A node naming no worker never touches it.
|
||||
self.remote = remote
|
||||
# This machine's own books. The pool sizes its fair share off them.
|
||||
self.resources = resources
|
||||
@@ -1005,7 +1005,7 @@ class FlowController:
|
||||
def _placed(
|
||||
self,
|
||||
wanted: Resources,
|
||||
device: str | None,
|
||||
label: str | None,
|
||||
policy: str,
|
||||
owner: str,
|
||||
local: str,
|
||||
@@ -1034,7 +1034,7 @@ class FlowController:
|
||||
|
||||
def call(**kwargs: Any) -> Any:
|
||||
with placer.claim( # type: ignore[union-attr]
|
||||
wanted, device=device, policy=policy, node=node_id, run=run_id
|
||||
wanted, label=label, policy=policy, node=node_id, run=run_id
|
||||
) as (target, allocation):
|
||||
env = derive_env(wanted, allocation, local=target.worker is None)
|
||||
if target.worker is None:
|
||||
@@ -1100,8 +1100,8 @@ class FlowController:
|
||||
node_type = NODE_TYPES.get(node_def.type)
|
||||
if node_type is None:
|
||||
raise ValueError(f"Unknown node type '{node_def.type}'")
|
||||
if node_def.device and not node_type.has_source:
|
||||
entry.ignored_device = node_def.device
|
||||
if node_def.worker and not node_type.has_source:
|
||||
entry.ignored_worker = node_def.worker
|
||||
|
||||
params = resolve_params(node_def.params)
|
||||
|
||||
@@ -1153,7 +1153,7 @@ class FlowController:
|
||||
else:
|
||||
# The code never runs here: it is loaded in a worker, and
|
||||
# the node calls that worker instead of a local function.
|
||||
# A node bound to a device is loaded on *that* machine —
|
||||
# A node bound to a worker is loaded on *that* machine —
|
||||
# one importing torch is correct on the GPU box and a
|
||||
# missing module here, so checking it here would fail a
|
||||
# node that is fine.
|
||||
@@ -1170,17 +1170,17 @@ class FlowController:
|
||||
entry.error = str(exc)
|
||||
return entry
|
||||
remote_only = (
|
||||
node_def.device
|
||||
and node_def.device_policy == "require"
|
||||
node_def.worker
|
||||
and node_def.worker_policy == "require"
|
||||
and self.remote is not None
|
||||
)
|
||||
if remote_only and self.remote is not None:
|
||||
problem = self.remote.compile(
|
||||
node_def.device or "", owner, local, code
|
||||
node_def.worker or "", owner, local, code
|
||||
)
|
||||
elif self._runs_elsewhere(wanted):
|
||||
# Asks for more than this machine has, so it will run on
|
||||
# one that has it. Same reason as a device: checking the
|
||||
# one that has it. Same reason as a worker label: checking the
|
||||
# import here would fail a node that is fine there.
|
||||
problem = None
|
||||
else:
|
||||
@@ -1199,12 +1199,12 @@ class FlowController:
|
||||
if wanted is not None and self.placer is not None:
|
||||
# Says how much of a machine it takes, so which machine
|
||||
# and how much of it are one decision — including when
|
||||
# it also names a device, which used to mean the two
|
||||
# it also names a worker, which used to mean the two
|
||||
# answers disagreed and nothing was accounted at all.
|
||||
function = self._placed(
|
||||
wanted,
|
||||
node_def.device,
|
||||
node_def.device_policy,
|
||||
node_def.worker,
|
||||
node_def.worker_policy,
|
||||
owner,
|
||||
local,
|
||||
code,
|
||||
@@ -1213,11 +1213,11 @@ class FlowController:
|
||||
run_id=run.run_id if run else "",
|
||||
on_event=emissions.handle,
|
||||
)
|
||||
elif node_def.device and self.remote is not None:
|
||||
# A device and nothing about size: the label alone
|
||||
elif node_def.worker and self.remote is not None:
|
||||
# A worker and nothing about size: the label alone
|
||||
# decides, least busy first, as it always has.
|
||||
function = self.remote.proxy(
|
||||
node_def.device,
|
||||
node_def.worker,
|
||||
owner,
|
||||
local,
|
||||
code,
|
||||
@@ -1235,7 +1235,7 @@ class FlowController:
|
||||
run_id=run.run_id if run else "",
|
||||
on_event=emissions.handle,
|
||||
)
|
||||
if node_def.device_policy == "prefer"
|
||||
if node_def.worker_policy == "prefer"
|
||||
else None
|
||||
),
|
||||
)
|
||||
@@ -1442,11 +1442,11 @@ class FlowController:
|
||||
)
|
||||
|
||||
def compile_check(
|
||||
self, flow: str, node_id: str, code: str, device: str | None = None
|
||||
self, flow: str, node_id: str, code: str, worker: str | None = None
|
||||
) -> str | None:
|
||||
"""Does this source load? Returns what to show the author, or None.
|
||||
|
||||
Loaded on the machine that will run it: a node bound to a device is
|
||||
Loaded on the machine that will run it: a node bound to a worker is
|
||||
checked against that worker's packages, because a missing import here
|
||||
says nothing about whether it is missing there.
|
||||
|
||||
@@ -1454,8 +1454,8 @@ class FlowController:
|
||||
the published one — so ``keep=False``: compile it to answer the
|
||||
question, and leave the worker holding what it is serving calls from.
|
||||
"""
|
||||
if device and self.remote is not None:
|
||||
return self.remote.compile(device, flow, node_id, code)
|
||||
if worker and self.remote is not None:
|
||||
return self.remote.compile(worker, flow, node_id, code)
|
||||
if self.workers is not None:
|
||||
return self.workers.compile(flow, node_id, code, keep=False)
|
||||
try:
|
||||
@@ -2082,21 +2082,21 @@ def _collect_issues(
|
||||
for entry in loaded.values()
|
||||
if entry.missing_source
|
||||
]
|
||||
# Only code travels to a worker, so a device on any other node is a field
|
||||
# Only code travels to a worker, so a worker on any other node is a field
|
||||
# nothing reads. Saying so beats a setting that looks like it took.
|
||||
issues += [
|
||||
ValidationIssue(
|
||||
code="ignored_device",
|
||||
code="ignored_worker",
|
||||
message=(
|
||||
f"Node '{entry.id.rpartition('.')[2]}' names device "
|
||||
f"'{entry.ignored_device}', but only a Function node runs on a "
|
||||
f"Node '{entry.id.rpartition('.')[2]}' names worker "
|
||||
f"'{entry.ignored_worker}', but only a Function node runs on a "
|
||||
"worker. The setting is ignored."
|
||||
),
|
||||
flow=entry.flow,
|
||||
node=entry.id,
|
||||
)
|
||||
for entry in loaded.values()
|
||||
if entry.ignored_device
|
||||
if entry.ignored_worker
|
||||
]
|
||||
return issues
|
||||
|
||||
|
||||
@@ -47,7 +47,7 @@ MANUAL_RUN_PREFIX = "manual-"
|
||||
# neither the brain graph nor the health summary treats them as a fault.
|
||||
# Lives here rather than beside the controller so `ValidationIssue` can carry
|
||||
# the distinction itself, and every reader gets it for free.
|
||||
ADVISORY_ISSUES = frozenset({"unauthenticated_hook", "ignored_device"})
|
||||
ADVISORY_ISSUES = frozenset({"unauthenticated_hook", "ignored_worker"})
|
||||
|
||||
|
||||
class ValidationIssue(BaseModel):
|
||||
@@ -62,7 +62,7 @@ class ValidationIssue(BaseModel):
|
||||
"self_loop_needs_initial",
|
||||
"missing_source",
|
||||
"node_unhealthy",
|
||||
"ignored_device",
|
||||
"ignored_worker",
|
||||
"unknown_device",
|
||||
"device_port",
|
||||
]
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
"""Which machine a node runs on, and the waiting when none of them has room.
|
||||
|
||||
The engine used to answer this twice, in two ways that could not see each
|
||||
other. A node with a ``device`` went to a worker carrying that label, chosen by
|
||||
other. A node with a ``worker`` went to a worker carrying that label, chosen by
|
||||
which of them had least in flight — a machine's *name*, never its size. A node
|
||||
with ``resources`` was accounted against the engine's own cores, and only the
|
||||
engine's: declaring both meant the second answer silently won and nothing was
|
||||
@@ -63,15 +63,15 @@ class Target:
|
||||
#: None is the engine itself, which runs nodes in its own worker pool.
|
||||
worker: RemoteWorker | None = None
|
||||
|
||||
def carries(self, device: str) -> bool:
|
||||
"""Whether a node asking for this device may run here.
|
||||
def carries(self, label: str) -> bool:
|
||||
"""Whether a node asking for this label may run here.
|
||||
|
||||
The engine carries no label: a label names a machine somebody attached
|
||||
for the purpose, and the engine is where the node would have run anyway.
|
||||
"""
|
||||
if self.worker is None:
|
||||
return False
|
||||
return device == self.name or device in self.worker.labels
|
||||
return label == self.name or label in self.worker.labels
|
||||
|
||||
|
||||
@dataclass
|
||||
@@ -138,7 +138,7 @@ class Placer:
|
||||
cpus: int,
|
||||
gpus: int,
|
||||
ram_mb: int,
|
||||
device: str | None,
|
||||
label: str | None,
|
||||
policy: str,
|
||||
) -> list[Target]:
|
||||
"""The machines that could run this node, best first."""
|
||||
@@ -146,8 +146,8 @@ class Placer:
|
||||
remote = [target for target in targets if target.worker is not None]
|
||||
engine = [target for target in targets if target.worker is None]
|
||||
|
||||
if device:
|
||||
labelled = [target for target in remote if target.carries(device)]
|
||||
if label:
|
||||
labelled = [target for target in remote if target.carries(label)]
|
||||
fitting = self._fitting(labelled, cpus, gpus, ram_mb)
|
||||
if fitting or policy != "prefer":
|
||||
# `require` waits for the labelled machine even when the engine
|
||||
@@ -174,7 +174,7 @@ class Placer:
|
||||
|
||||
# -- what can actually be granted ------------------------------------------
|
||||
|
||||
def _shapes(self, device: str | None, policy: str) -> list[tuple[int, int, int]]:
|
||||
def _shapes(self, label: str | None, policy: str) -> list[tuple[int, int, int]]:
|
||||
"""Every machine's size, as something to measure a request against.
|
||||
|
||||
A provisioner's job shapes count: a machine it can start on demand is
|
||||
@@ -182,18 +182,18 @@ class Placer:
|
||||
"""
|
||||
shapes = []
|
||||
for target in self.targets():
|
||||
if device and not target.carries(device):
|
||||
if label and not target.carries(label):
|
||||
# A `prefer` node may end up here, so the engine still counts.
|
||||
if policy == "require" or target.worker is not None:
|
||||
continue
|
||||
books = target.accountant
|
||||
shapes.append((books.cpus, books.gpus, books.ram_mb or 0))
|
||||
for provisioner in self.provisioners:
|
||||
shapes.extend(provisioner.shapes(device))
|
||||
shapes.extend(provisioner.shapes(label))
|
||||
return shapes or [(self.local.cpus, self.local.gpus, self.local.ram_mb or 0)]
|
||||
|
||||
def _clamp(
|
||||
self, wanted: Resources, node: str, device: str | None, policy: str
|
||||
self, wanted: Resources, node: str, label: str | None, policy: str
|
||||
) -> tuple[int, int, int]:
|
||||
"""What can be granted somewhere, which may be less than was asked.
|
||||
|
||||
@@ -206,7 +206,7 @@ class Placer:
|
||||
measured only against machines that carry that many.
|
||||
"""
|
||||
ram = wanted.ram or 0
|
||||
shapes = self._shapes(device, policy)
|
||||
shapes = self._shapes(label, policy)
|
||||
gpus = min(wanted.gpus, max(shape[1] for shape in shapes))
|
||||
capable = [shape for shape in shapes if shape[1] >= gpus]
|
||||
cpus = min(wanted.cpus, max(shape[0] for shape in capable))
|
||||
@@ -246,7 +246,7 @@ class Placer:
|
||||
self,
|
||||
wanted: Resources,
|
||||
*,
|
||||
device: str | None = None,
|
||||
label: str | None = None,
|
||||
policy: str = "require",
|
||||
node: str = "",
|
||||
run: str = "",
|
||||
@@ -259,8 +259,8 @@ class Placer:
|
||||
|
||||
with self._condition:
|
||||
while chosen is None:
|
||||
cpus, gpus, ram = self._clamp(wanted, node, device, policy)
|
||||
for target in self._candidates(cpus, gpus, ram, device, policy):
|
||||
cpus, gpus, ram = self._clamp(wanted, node, label, policy)
|
||||
for target in self._candidates(cpus, gpus, ram, label, policy):
|
||||
got = target.accountant.try_take(cpus, gpus, ram)
|
||||
if got is not None:
|
||||
chosen, allocation = target, got
|
||||
@@ -269,9 +269,9 @@ class Placer:
|
||||
break
|
||||
if not ticket:
|
||||
ticket = self._announce(
|
||||
node, run, waited_from, self._reason(cpus, gpus, device)
|
||||
node, run, waited_from, self._reason(cpus, gpus, label)
|
||||
)
|
||||
self._provision(cpus, gpus, ram, device)
|
||||
self._provision(cpus, gpus, ram, label)
|
||||
self._condition.wait(timeout=WAKE_S)
|
||||
if ticket:
|
||||
self._waiting.pop(ticket, None)
|
||||
@@ -291,21 +291,21 @@ class Placer:
|
||||
# timeout, a cancellation, or the worker going away mid-call.
|
||||
chosen.accountant.give_back(allocation)
|
||||
|
||||
def _reason(self, cpus: int, gpus: int, device: str | None) -> str:
|
||||
def _reason(self, cpus: int, gpus: int, label: str | None) -> str:
|
||||
parts = []
|
||||
if gpus:
|
||||
parts.append(f"{gpus} gpu(s)")
|
||||
parts.append(f"{cpus} cpu(s)")
|
||||
wants = " and ".join(parts)
|
||||
if device:
|
||||
return f"waiting for a worker labelled '{device}' with {wants} free"
|
||||
if label:
|
||||
return f"waiting for a worker labelled '{label}' with {wants} free"
|
||||
return f"waiting for {wants}"
|
||||
|
||||
def _provision(self, cpus: int, gpus: int, ram_mb: int, device: str | None) -> None:
|
||||
def _provision(self, cpus: int, gpus: int, ram_mb: int, label: str | None) -> None:
|
||||
"""Ask for a machine, if something here can start one."""
|
||||
for provisioner in self.provisioners:
|
||||
if provisioner.covers(cpus, gpus, ram_mb, device):
|
||||
provisioner.provision(cpus, gpus, ram_mb, device)
|
||||
if provisioner.covers(cpus, gpus, ram_mb, label):
|
||||
provisioner.provision(cpus, gpus, ram_mb, label)
|
||||
return
|
||||
|
||||
def provision_for(self, needs: dict[str, Any] | None) -> None:
|
||||
@@ -321,7 +321,7 @@ class Placer:
|
||||
int(needs.get("cpus") or 1),
|
||||
int(needs.get("gpus") or 0),
|
||||
int(needs.get("ram_mb") or 0),
|
||||
needs.get("device") or None,
|
||||
needs.get("worker") or None,
|
||||
)
|
||||
|
||||
def _announce(self, node: str, run: str, since: float, reason: str) -> int:
|
||||
@@ -362,15 +362,15 @@ class Placer:
|
||||
cpus = int(needs.get("cpus") or 1)
|
||||
gpus = int(needs.get("gpus") or 0)
|
||||
ram = int(needs.get("ram_mb") or 0)
|
||||
device = needs.get("device") or None
|
||||
label = needs.get("worker") or None
|
||||
for target in self.targets():
|
||||
if device and not target.carries(device):
|
||||
if label and not target.carries(label):
|
||||
continue
|
||||
if target.accountant.fits(cpus, gpus, ram):
|
||||
# Something here could take it. Busy is queueing, not waiting.
|
||||
return None
|
||||
if not any(
|
||||
provisioner.covers(cpus, gpus, ram, device)
|
||||
provisioner.covers(cpus, gpus, ram, label)
|
||||
for provisioner in self.provisioners
|
||||
):
|
||||
# Nothing attached fits and nothing can be started, so the node
|
||||
|
||||
@@ -91,7 +91,7 @@ class SlurmProfile:
|
||||
cpus: int = 1
|
||||
gpus: int = 0
|
||||
ram_mb: int = 0
|
||||
#: What a worker started this way advertises, so a node bound to a device
|
||||
#: What a worker started this way advertises, so a node naming one of them
|
||||
#: can be the reason one is started.
|
||||
labels: list[str] = field(default_factory=list)
|
||||
#: ``#SBATCH`` lines, verbatim: partition, gres, walltime, account.
|
||||
|
||||
@@ -125,7 +125,7 @@ class RemoteWorker:
|
||||
self._lock = threading.Lock()
|
||||
self._gone = False
|
||||
# Source this worker has already loaded, by digest. A rebuild asks
|
||||
# every device-bound node whether it compiles, and without this that
|
||||
# every worker-bound node whether it compiles, and without this that
|
||||
# is a network round trip each time — which is how attaching a GPU box
|
||||
# makes rebuilding the flows slow enough to fail a health check. A
|
||||
# worker that reconnects is a new object, so this empties with it.
|
||||
@@ -462,7 +462,7 @@ class RemoteWorkerHub:
|
||||
fallback: Callable[..., Any] | None = None,
|
||||
env: dict[str, str] | None = None,
|
||||
) -> Callable[..., Any]:
|
||||
"""What a node with a device runs instead of its own function.
|
||||
"""What a node bound to a worker runs instead of its own function.
|
||||
|
||||
``fallback`` is the local pool's proxy, used when the node only prefers
|
||||
the label rather than requiring it.
|
||||
|
||||
@@ -181,7 +181,7 @@ def _runs_on_a_worker(node: NodeDef) -> bool:
|
||||
"""Whether this node's body could go to a worker at all.
|
||||
|
||||
Only a node whose source travels can: a connector is an entry point loaded
|
||||
in this process, so a device on one is a field nothing reads. Holding a run
|
||||
in this process, so a worker on one is a field nothing reads. Holding a run
|
||||
for a worker that could never take that node is a run that never starts.
|
||||
"""
|
||||
node_type = NODE_TYPES.get(node.type)
|
||||
@@ -191,17 +191,17 @@ def _runs_on_a_worker(node: NodeDef) -> bool:
|
||||
def required_labels(flow: FlowDef) -> list[str]:
|
||||
"""Worker labels this flow cannot run without.
|
||||
|
||||
Only the nodes that *require* their device. One that merely prefers it runs
|
||||
Only the nodes that *require* their worker. One that merely prefers it runs
|
||||
on the engine when no such worker is attached, so holding the whole run
|
||||
back to wait for one would be waiting for something it does not need.
|
||||
"""
|
||||
return sorted(
|
||||
{
|
||||
node.device
|
||||
node.worker
|
||||
for node in flow.nodes
|
||||
if node.device
|
||||
and node.device.strip()
|
||||
and node.device_policy == "require"
|
||||
if node.worker
|
||||
and node.worker.strip()
|
||||
and node.worker_policy == "require"
|
||||
and _runs_on_a_worker(node)
|
||||
}
|
||||
)
|
||||
@@ -212,11 +212,11 @@ def required_resources(flow: FlowDef) -> dict[str, Any] | None:
|
||||
|
||||
Dimension by dimension rather than per node, which is deliberately blunt:
|
||||
it answers "could this run start at all", not "in what order". A node that
|
||||
only *prefers* its device is left out — it runs here when nothing carries
|
||||
only *prefers* its worker is left out — it runs here when nothing carries
|
||||
the label, so it is never what a run is waiting for.
|
||||
"""
|
||||
cpus, gpus, ram = 0, 0, 0
|
||||
device = ""
|
||||
worker = ""
|
||||
for node in flow.nodes:
|
||||
wanted = node.resources
|
||||
if wanted is None or not _runs_on_a_worker(node):
|
||||
@@ -224,11 +224,11 @@ def required_resources(flow: FlowDef) -> dict[str, Any] | None:
|
||||
cpus = max(cpus, wanted.cpus)
|
||||
gpus = max(gpus, wanted.gpus)
|
||||
ram = max(ram, wanted.ram or 0)
|
||||
if node.device and node.device_policy == "require":
|
||||
device = node.device
|
||||
if node.worker and node.worker_policy == "require":
|
||||
worker = node.worker
|
||||
if not (cpus or gpus or ram):
|
||||
return None
|
||||
return {"cpus": cpus, "gpus": gpus, "ram_mb": ram, "device": device}
|
||||
return {"cpus": cpus, "gpus": gpus, "ram_mb": ram, "worker": worker}
|
||||
|
||||
|
||||
#: What a run's output is called from outside it: ``@run:<id>.<output>``.
|
||||
@@ -1310,6 +1310,9 @@ class RunService:
|
||||
run = session.get(Run, run_id)
|
||||
needed = list(run.labels) if run else []
|
||||
needs = dict(run.needs) if run and run.needs else None
|
||||
if needs and "device" in needs:
|
||||
# Queued before the label was called `worker`.
|
||||
needs["worker"] = needs.pop("device")
|
||||
hub = self.controller.remote
|
||||
available = hub.labels() | {w.name for w in hub.workers()} if hub else set()
|
||||
# A node that only prefers its label runs locally instead, so it is not
|
||||
|
||||
@@ -177,7 +177,7 @@ class NodeDef(BaseModel):
|
||||
"inherits the engine default."
|
||||
),
|
||||
)
|
||||
device: str | None = Field(
|
||||
worker: str | None = Field(
|
||||
default=None,
|
||||
description=(
|
||||
"Label of the worker this node's code must run on, such as 'gpu'. "
|
||||
@@ -185,10 +185,10 @@ class NodeDef(BaseModel):
|
||||
"attached worker carries waits rather than failing."
|
||||
),
|
||||
)
|
||||
device_policy: Literal["require", "prefer"] = Field(
|
||||
worker_policy: Literal["require", "prefer"] = Field(
|
||||
default="require",
|
||||
description=(
|
||||
"What to do when no worker carries `device`: wait for one, or run "
|
||||
"What to do when no worker carries `worker`: wait for one, or run "
|
||||
"locally anyway."
|
||||
),
|
||||
)
|
||||
|
||||
@@ -172,8 +172,10 @@ class FlowStore:
|
||||
"""Rewrite node types and settings stored under a name they no longer have.
|
||||
|
||||
On 2026-10-15 the node type ``trigger`` became ``debounce`` and the mqtt
|
||||
setting ``publish_queue_size`` became ``send_queue_size``, with no alias
|
||||
left for the old names. Each published and draft document is read as
|
||||
setting ``publish_queue_size`` became ``send_queue_size``. On 2026-10-16
|
||||
a node's ``device`` and ``device_policy`` became ``worker`` and
|
||||
``worker_policy``, so that a device is only ever a paired one. No alias
|
||||
is left for the old names. Each published and draft document is read as
|
||||
plain JSON and only the matching node changes, so a field this version
|
||||
does not model survives, and the old document stays in git. The version
|
||||
goes up by one, so an editor still holding the old document gets a
|
||||
@@ -192,25 +194,36 @@ class FlowStore:
|
||||
if n.get("type") == "mqtt"
|
||||
and "publish_queue_size" in (n.get("params") or {})
|
||||
]
|
||||
placed = [
|
||||
n for n in data["nodes"] if "device" in n or "device_policy" in n
|
||||
]
|
||||
except Exception:
|
||||
# Unreadable to the engine too, which reports it when it loads.
|
||||
continue
|
||||
if not nodes and not queues:
|
||||
if not nodes and not queues and not placed:
|
||||
continue
|
||||
for node in nodes:
|
||||
node["type"] = "debounce"
|
||||
for params in queues:
|
||||
params["send_queue_size"] = params.pop("publish_queue_size")
|
||||
for node in placed:
|
||||
for old in ("device", "device_policy"):
|
||||
if old in node:
|
||||
node[old.replace("device", "worker")] = node.pop(old)
|
||||
data["version"] = data.get("version", 1) + 1
|
||||
path.write_text(json.dumps(data, indent=2, ensure_ascii=False) + "\n")
|
||||
changed.append(f"{path.parent.name}/{path.name}")
|
||||
if changed:
|
||||
logger.warning(
|
||||
"Renamed node type 'trigger' to 'debounce' and mqtt "
|
||||
"'publish_queue_size' to 'send_queue_size' in %s",
|
||||
"Renamed node type 'trigger' to 'debounce', mqtt "
|
||||
"'publish_queue_size' to 'send_queue_size' and a node's "
|
||||
"'device' to 'worker' in %s",
|
||||
", ".join(changed),
|
||||
)
|
||||
self._commit("Rename 'trigger' nodes and mqtt 'publish_queue_size'")
|
||||
self._commit(
|
||||
"Rename 'trigger' nodes, mqtt 'publish_queue_size' and a "
|
||||
"node's 'device'"
|
||||
)
|
||||
|
||||
# -------------------------------------------------------------------------
|
||||
# git
|
||||
|
||||
@@ -398,7 +398,7 @@ class Run(SQLModel, table=True):
|
||||
status_reason: str = Field(default="", max_length=1024)
|
||||
#: Worker labels its nodes need, so a run with nowhere to go can say so.
|
||||
labels: list[str] = Field(sa_column=Column(JSON), default_factory=list)
|
||||
#: The largest machine its nodes ask for, as cpus, gpus, ram_mb and device.
|
||||
#: The largest machine its nodes ask for, as cpus, gpus, ram_mb and worker.
|
||||
#: Read while the run is queued, to tell waiting for a machine from having
|
||||
#: nowhere to run at all.
|
||||
needs: dict[str, Any] | None = Field(sa_column=Column(JSON), default=None)
|
||||
|
||||
@@ -193,8 +193,8 @@ class NodeSpec:
|
||||
settings: dict[str, Any],
|
||||
title: str,
|
||||
timeout: float | None,
|
||||
device: str | None,
|
||||
device_policy: str,
|
||||
worker: str | None,
|
||||
worker_policy: str,
|
||||
cache: bool = True,
|
||||
resources: dict[str, Any] | None = None,
|
||||
) -> None:
|
||||
@@ -206,8 +206,8 @@ class NodeSpec:
|
||||
self.settings = settings
|
||||
self.title = title
|
||||
self.timeout = timeout
|
||||
self.device = device
|
||||
self.device_policy = device_policy
|
||||
self.worker = worker
|
||||
self.worker_policy = worker_policy
|
||||
self.cache = cache
|
||||
self.resources = resources
|
||||
|
||||
@@ -238,8 +238,8 @@ class NodeSpec:
|
||||
settings={**self.settings, **settings},
|
||||
title=self.title,
|
||||
timeout=self.timeout,
|
||||
device=self.device,
|
||||
device_policy=self.device_policy,
|
||||
worker=self.worker,
|
||||
worker_policy=self.worker_policy,
|
||||
cache=self.cache,
|
||||
resources=self.resources,
|
||||
)
|
||||
@@ -270,8 +270,8 @@ def node(
|
||||
settings: dict[str, Any] | None = None,
|
||||
title: str = "",
|
||||
timeout: float | None = None,
|
||||
device: str | None = None,
|
||||
device_policy: str = "require",
|
||||
worker: str | None = None,
|
||||
worker_policy: str = "require",
|
||||
cache: bool = True,
|
||||
resources: dict[str, Any] | str | None = None,
|
||||
) -> Callable[[F], F]:
|
||||
@@ -282,8 +282,8 @@ def node(
|
||||
shorthand for a ``float`` port of that name.
|
||||
|
||||
Parameters with a default that are not ports become the node's settings, so
|
||||
the canvas can tune them without editing code. ``device`` picks the worker
|
||||
the code runs on — ``"gpu"``, say — and ``device_policy="prefer"`` runs it
|
||||
the canvas can tune them without editing code. ``worker`` picks the worker
|
||||
the code runs on — ``"gpu"``, say — and ``worker_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
|
||||
@@ -312,8 +312,8 @@ def node(
|
||||
thing as ``{"flavor": "gpu-small"}`` and stands in for cpus, gpus and ram.
|
||||
Sizes accept ``"2G"``; ``duration_s`` accepts ``"30m"``.
|
||||
"""
|
||||
if device_policy not in ("require", "prefer"):
|
||||
raise SyncError("device_policy is 'require' or 'prefer'")
|
||||
if worker_policy not in ("require", "prefer"):
|
||||
raise SyncError("worker_policy is 'require' or 'prefer'")
|
||||
if isinstance(resources, str):
|
||||
resources = {"flavor": resources}
|
||||
if resources is not None:
|
||||
@@ -341,8 +341,8 @@ def node(
|
||||
settings=dict(settings or {}),
|
||||
title=title,
|
||||
timeout=timeout,
|
||||
device=device,
|
||||
device_policy=device_policy,
|
||||
worker=worker,
|
||||
worker_policy=worker_policy,
|
||||
cache=cache,
|
||||
resources=dict(resources) if resources is not None else None,
|
||||
)
|
||||
@@ -547,8 +547,8 @@ def _check(spec: NodeSpec, mode: str) -> dict[str, Any]:
|
||||
"requires": [port.spec() for port in spec.requires],
|
||||
"provides": [port.spec() for port in spec.provides],
|
||||
"timeout": spec.timeout,
|
||||
"device": spec.device,
|
||||
"device_policy": spec.device_policy,
|
||||
"worker": spec.worker,
|
||||
"worker_policy": spec.worker_policy,
|
||||
"cache": spec.cache,
|
||||
"resources": spec.resources,
|
||||
}
|
||||
|
||||
@@ -125,8 +125,8 @@ def test_a_published_node_with_no_body_is_reported(store: FlowStore):
|
||||
assert [i for i in controller.issues if i.code == "missing_source"] == []
|
||||
|
||||
|
||||
def test_a_device_on_a_node_without_code_is_reported(store: FlowStore):
|
||||
"""Only code travels to a worker, so the device would be ignored silently."""
|
||||
def test_a_worker_on_a_node_without_code_is_reported(store: FlowStore):
|
||||
"""Only code travels to a worker, so the worker would be ignored silently."""
|
||||
store.write_flow(
|
||||
FlowDef(
|
||||
name="heating",
|
||||
@@ -134,7 +134,7 @@ def test_a_device_on_a_node_without_code_is_reported(store: FlowStore):
|
||||
NodeDef(
|
||||
id="relay",
|
||||
type="change",
|
||||
device="gpu",
|
||||
worker="gpu",
|
||||
provides=[MessageSpec(name="temp")],
|
||||
)
|
||||
],
|
||||
@@ -143,7 +143,7 @@ def test_a_device_on_a_node_without_code_is_reported(store: FlowStore):
|
||||
controller = FlowController(store)
|
||||
|
||||
asyncio.run(controller.reload())
|
||||
(issue,) = [i for i in controller.issues if i.code == "ignored_device"]
|
||||
(issue,) = [i for i in controller.issues if i.code == "ignored_worker"]
|
||||
assert issue.node == "heating.relay"
|
||||
assert issue.advisory
|
||||
|
||||
|
||||
@@ -108,7 +108,7 @@ def test_a_required_label_waits_for_its_machine_rather_than_running_here(loop):
|
||||
placer = placer_over(hub)
|
||||
worker, _ = attach(hub, loop, labels=("gpu",))
|
||||
|
||||
with placer.claim(Resources(cpus=2), device="gpu", policy="require") as (
|
||||
with placer.claim(Resources(cpus=2), label="gpu", policy="require") as (
|
||||
target,
|
||||
_,
|
||||
):
|
||||
@@ -120,7 +120,7 @@ def test_a_preferred_label_falls_back_here_and_is_still_accounted(loop):
|
||||
hub = RemoteWorkerHub()
|
||||
placer = placer_over(hub)
|
||||
|
||||
with placer.claim(Resources(cpus=3), device="gpu", policy="prefer") as (
|
||||
with placer.claim(Resources(cpus=3), label="gpu", policy="prefer") as (
|
||||
target,
|
||||
allocation,
|
||||
):
|
||||
|
||||
@@ -231,7 +231,7 @@ def test_a_run_held_before_it_starts_still_gets_a_machine_asked_for(monkeypatch)
|
||||
hpc = cluster()
|
||||
placer = Placer(local=ResourceAccountant(cpus=4, gpus=0))
|
||||
placer.provisioners = [hpc]
|
||||
needs = {"cpus": 2, "gpus": 1, "ram_mb": 0, "device": ""}
|
||||
needs = {"cpus": 2, "gpus": 1, "ram_mb": 0, "worker": ""}
|
||||
|
||||
assert placer.satisfiable(needs) == "1 gpu(s) and 2 cpu(s)"
|
||||
placer.provision_for(needs)
|
||||
|
||||
@@ -378,23 +378,23 @@ def test_the_digest_identifies_the_inputs_not_their_order():
|
||||
assert digest_of({"a": 1}, 3) != digest_of({"a": 1}, 4)
|
||||
|
||||
|
||||
def test_labels_come_from_the_nodes_that_ask_for_a_device():
|
||||
def test_labels_come_from_the_nodes_that_ask_for_a_worker():
|
||||
flow = double_flow()
|
||||
assert required_labels(flow) == []
|
||||
flow.nodes[0].device = "gpu"
|
||||
flow.nodes[0].worker = "gpu"
|
||||
assert required_labels(flow) == ["gpu"]
|
||||
|
||||
|
||||
def test_a_preferred_device_does_not_hold_a_run_back():
|
||||
def test_a_preferred_worker_does_not_hold_a_run_back():
|
||||
flow = double_flow()
|
||||
flow.nodes[0].device = "gpu"
|
||||
flow.nodes[0].device_policy = "prefer"
|
||||
flow.nodes[0].worker = "gpu"
|
||||
flow.nodes[0].worker_policy = "prefer"
|
||||
# It runs on the engine when no such worker is attached, so waiting for
|
||||
# one would be waiting for something the run does not need.
|
||||
assert required_labels(flow) == []
|
||||
|
||||
|
||||
def test_a_device_on_a_node_no_worker_could_run_is_not_waited_for():
|
||||
def test_a_worker_on_a_node_no_worker_could_run_is_not_waited_for():
|
||||
"""A connector is an entry point loaded here, so a worker cannot take it.
|
||||
|
||||
Holding the run for one is a run that never starts — and it looked exactly
|
||||
@@ -402,7 +402,7 @@ def test_a_device_on_a_node_no_worker_could_run_is_not_waited_for():
|
||||
"""
|
||||
flow = double_flow()
|
||||
flow.nodes[0].type = "mqtt"
|
||||
flow.nodes[0].device = "gpu"
|
||||
flow.nodes[0].worker = "gpu"
|
||||
assert required_labels(flow) == []
|
||||
|
||||
|
||||
@@ -416,7 +416,7 @@ def test_what_a_run_needs_is_the_largest_single_machine_it_asks_for():
|
||||
"cpus": 8,
|
||||
"gpus": 1,
|
||||
"ram_mb": 0,
|
||||
"device": "",
|
||||
"worker": "",
|
||||
}
|
||||
|
||||
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
from datetime import UTC, datetime
|
||||
@@ -173,3 +174,22 @@ def test_opening_the_store_renames_the_mqtt_publish_queue_size(store: FlowStore)
|
||||
assert published.nodes[0].params == {"send_queue_size": 8}
|
||||
assert published.version == 2
|
||||
assert commit_count(store) == commits + 1
|
||||
|
||||
|
||||
def test_opening_the_store_renames_a_nodes_device_to_worker(store: FlowStore):
|
||||
"""`device` names a paired device now, so a node's worker label moves over."""
|
||||
store.write_flow(FlowDef(name="lab", nodes=[NodeDef(id="fit")]))
|
||||
path = store.root / "lab" / "flow.json"
|
||||
data = json.loads(path.read_text())
|
||||
node = data["nodes"][0]
|
||||
del node["worker"], node["worker_policy"]
|
||||
node.update(device="gpu", device_policy="prefer")
|
||||
path.write_text(json.dumps(data))
|
||||
|
||||
published = FlowStore(store.root).read_flow("lab")
|
||||
|
||||
assert (published.nodes[0].worker, published.nodes[0].worker_policy) == (
|
||||
"gpu",
|
||||
"prefer",
|
||||
)
|
||||
assert published.version == 2
|
||||
|
||||
@@ -17,8 +17,8 @@ def prepare(limit=8):
|
||||
@node(
|
||||
requires=["dataset", Port("lr", "float")],
|
||||
provides=[Port("loss", "float", stream=True), Port("weights", "artifact")],
|
||||
device="gpu",
|
||||
device_policy="prefer",
|
||||
worker="gpu",
|
||||
worker_policy="prefer",
|
||||
)
|
||||
def fit(dataset, lr, epochs=3):
|
||||
for step in range(epochs):
|
||||
@@ -68,12 +68,12 @@ def test_defaults_become_settings_and_ports_do_not():
|
||||
assert {p.port for p in nodes["fit"].requires} == {"dataset", "lr"}
|
||||
|
||||
|
||||
def test_device_travels_to_the_node():
|
||||
def test_worker_travels_to_the_node():
|
||||
definition = FlowDef.model_validate(a_flow().document())
|
||||
node_def = next(n for n in definition.nodes if n.id == "fit")
|
||||
|
||||
assert node_def.device == "gpu"
|
||||
assert node_def.device_policy == "prefer"
|
||||
assert node_def.worker == "gpu"
|
||||
assert node_def.worker_policy == "prefer"
|
||||
|
||||
|
||||
def test_a_required_port_takes_the_type_it_is_provided_as():
|
||||
|
||||
+1
-1
@@ -23,7 +23,7 @@ services:
|
||||
volumes:
|
||||
- data:/data
|
||||
environment:
|
||||
# The address people and devices reach this instance at. Panel links and
|
||||
# The address people and devices reach this instance at. Device links and
|
||||
# mails are built from it.
|
||||
FRONTEND_HOST: http://localhost:8080
|
||||
# Schedules run in this timezone, such as Europe/Berlin.
|
||||
|
||||
+4
-4
@@ -126,7 +126,7 @@ HTML report lands in `backend/htmlcov`. Nothing has to be running — mail is pa
|
||||
`tests/`, and the flow engine falls back to `MemoryState` while `REDIS_HOST` is empty.
|
||||
`tests/__init__.py` pins the suite's own environment before anything imports the settings: a
|
||||
SQLite file in a temp directory (deleted per session, so a run never touches development
|
||||
data), its own superuser, and `ENVIRONMENT=local` with `DOMAIN=localhost`, so a checkout
|
||||
data), its own superuser, and `ENVIRONMENT=local`, so a checkout
|
||||
configured for a deployment cannot drag that deployment's configuration into the run.
|
||||
|
||||
**Frontend (e2e).** `make test-frontend` runs the suite against a stack of its own, because
|
||||
@@ -155,9 +155,9 @@ in once into `playwright/.auth/user.json`, which the browser projects reuse; `mo
|
||||
`mobile.spec.ts` only, on a Pixel 5 at 393px.
|
||||
|
||||
The origin comes from `PLAYWRIGHT_BASE_URL`, with the API on it unless `PLAYWRIGHT_API_URL`
|
||||
names another, deliberately never from `VITE_API_URL` — that one belongs to the app build and in a deployment checkout it
|
||||
names the deployment, which would mean driving a browser at the local stack while sending
|
||||
teardown `DELETE`s to the live instance. `tests/guard.ts` runs first and refuses the whole
|
||||
names another, deliberately never from `.env`. In a deployment checkout that file names the
|
||||
deployment, which would mean driving a browser at the local stack while sending teardown
|
||||
`DELETE`s to the live instance. `tests/guard.ts` runs first and refuses the whole
|
||||
run when either origin resolves outside loopback or the private ranges, because the suite
|
||||
creates and deletes flows, dashboards and users. `PLAYWRIGHT_ALLOW_PUBLIC=1` overrides it.
|
||||
|
||||
|
||||
@@ -10,7 +10,7 @@ services:
|
||||
ports: !override
|
||||
- "${APP_PORT:-8080}:8000"
|
||||
environment:
|
||||
# The address the instance gives out: the link the Panels dialog shows a
|
||||
# The address the instance gives out: the link a device's page shows a
|
||||
# screen and the one its emails carry. app.${DOMAIN} resolves on this host
|
||||
# only, so `make dev-lan` passes the published port on the LAN address.
|
||||
- FRONTEND_HOST=${LAN_URL:-http://app.${DOMAIN:-localhost}}
|
||||
|
||||
+2
-2
@@ -18,8 +18,8 @@ issuer of the client's tokens.
|
||||
|
||||
The client registers, opens browser authorization, and receives an OAuth
|
||||
token after you approve access. The issuer and browser redirects use
|
||||
`DOMAIN`, `ENVIRONMENT`, and `FRONTEND_HOST`, so configure those for the
|
||||
addresses your client and browser can reach.
|
||||
`FRONTEND_HOST` and `ENVIRONMENT`, so configure those for the addresses your
|
||||
client and browser can reach.
|
||||
|
||||
Manage registered clients under **Admin → Agents**. Revoking a client
|
||||
invalidates its access. MCP uses a separate signing key at
|
||||
|
||||
@@ -195,6 +195,7 @@ on its own, so this is for reclaiming space sooner.
|
||||
| `POST` | `/devices/pair` | start device pairing, optionally with `{name, board, firmware}`; unauthenticated |
|
||||
| `GET` | `/devices/pair/{code}` | poll pairing with its `secret`; answers `device` (and `panel` for older firmware); unauthenticated |
|
||||
| `GET` | `/devices/pending` | devices waiting for a code, as they describe themselves |
|
||||
| `GET` | `/devices/pair/{code}/device` | what is waiting on a code, before it is approved |
|
||||
| `POST` | `/devices/{id}/pair` | adopt by `{code}`, or by `{request}` from the pending list for a request from your network |
|
||||
| `POST` | `/devices/{id}/unpair` | revoke the device credential |
|
||||
|
||||
|
||||
+2
-1
@@ -340,7 +340,8 @@ The main files and directories are:
|
||||
|---|---|
|
||||
| `client.json` | Local API URL and token |
|
||||
| `fluksio.db` | SQLite database: accounts, runs, metrics and operations history |
|
||||
| `flows/` | Git repository of flows, node sources, dashboards and module requirements |
|
||||
| `flows/` | Git repository of flows, node sources, dashboards, devices (`_devices/`) and module requirements |
|
||||
| `devices.json` | Device credential nonces, outside the flow repository |
|
||||
| `artifacts/` | Files stored by content digest |
|
||||
| `user-venv/` | Managed node environment, when used |
|
||||
| `secrets.enc` | Encrypted credentials, outside the flow repository |
|
||||
|
||||
+2
-2
@@ -324,10 +324,10 @@ waiting for that deadline.
|
||||
|
||||
## Running a node somewhere else
|
||||
|
||||
Set a device label to select an attached worker:
|
||||
Set a worker label to select an attached worker:
|
||||
|
||||
```json
|
||||
{"id": "train", "device": "gpu", "device_policy": "require", "timeout": 7200}
|
||||
{"id": "train", "worker": "gpu", "worker_policy": "require", "timeout": 7200}
|
||||
```
|
||||
|
||||
`require` is the default and waits for a worker with that label. `prefer` uses
|
||||
|
||||
@@ -76,16 +76,16 @@ Set these fields on a node through `PUT /api/v1/flows/{name}`:
|
||||
```json
|
||||
{
|
||||
"id": "train",
|
||||
"device": "gpu",
|
||||
"device_policy": "require",
|
||||
"worker": "gpu",
|
||||
"worker_policy": "require",
|
||||
"timeout": 7200
|
||||
}
|
||||
```
|
||||
|
||||
This is a node fragment; keep the node's other fields and the rest of the flow
|
||||
when updating it. `device` and `device_policy` are not in the node panel yet.
|
||||
when updating it. `worker` and `worker_policy` are not in the node panel yet.
|
||||
|
||||
| `device_policy` | If no matching worker is attached |
|
||||
| `worker_policy` | If no matching worker is attached |
|
||||
|---|---|
|
||||
| `require` (default) | queues until a matching worker is available |
|
||||
| `prefer` | falls back to the engine |
|
||||
|
||||
@@ -137,7 +137,7 @@ The canvas reports issues on the affected nodes:
|
||||
| `missing_source` | a Function node has no saved code |
|
||||
| `node_error` | code or configuration could not load |
|
||||
| `unauthenticated_hook` | a webhook or served websocket has no shared secret; advisory |
|
||||
| `ignored_device` | a node other than a Function node names a device, which nothing reads; advisory |
|
||||
| `ignored_worker` | a node other than a Function node names a worker, which nothing reads; advisory |
|
||||
| `unknown_device` | a Device node names no device, or one that does not exist |
|
||||
| `device_port` | a device port is wired in two places, or against the device's report the wrong way or type; a port the report does not list is advisory |
|
||||
| `node_unhealthy` | a loaded node reports a runtime problem, such as a lost connection, or a device being offline |
|
||||
|
||||
@@ -157,7 +157,7 @@ artifact and rebuild the object in the next node.
|
||||
## Running a node somewhere else
|
||||
|
||||
Attach a [remote worker](../code/workers.md) to the engine and declare the
|
||||
worker's label as the node's `device`. `device_policy="require"` waits for
|
||||
worker's label as the node's `worker`. `worker_policy="require"` waits for
|
||||
that label; `"prefer"` uses the local engine when none is attached.
|
||||
|
||||
The selected machine needs your imported packages. Code-defined flows also
|
||||
|
||||
@@ -322,10 +322,10 @@ face, and the editor marks it **Poor fit for round**.
|
||||
| Gesture | Effect |
|
||||
|---|---|
|
||||
| Swipe left / right | next / previous face of this dashboard; stops at either end |
|
||||
| Swipe up / down | next / previous dashboard of the panel (`/panel/{id}` only); stops at either end |
|
||||
| Swipe up / down | next / previous dashboard of the device (`/panel/{id}` only); stops at either end |
|
||||
| Page dot | that face |
|
||||
| PageDown / PageUp | next / previous face |
|
||||
| Shift+PageDown / Shift+PageUp | next / previous dashboard of the panel |
|
||||
| Shift+PageDown / Shift+PageUp | next / previous dashboard of the device |
|
||||
| Arrow keys | turn the ring, not the page |
|
||||
|
||||
A swipe carries the page with the finger ([Swiping](#swiping)); a drag that
|
||||
@@ -423,7 +423,7 @@ angle, which is what the browser's reveal amounts to.
|
||||
|
||||
### Swiping
|
||||
|
||||
Faces are swiped sideways and a panel's dashboards up and down, with the page
|
||||
Faces are swiped sideways and a device's dashboards up and down, with the page
|
||||
under the finger, as a tile view moves.
|
||||
|
||||
| Part | Value |
|
||||
|
||||
@@ -68,7 +68,7 @@ yielding regularly can run longer than its timeout. Empty or `0` means no
|
||||
limit, the default.
|
||||
|
||||
**Resources** selects a shared worker pool, a named flavor or custom CPU,
|
||||
memory and GPU requirements. Placement by `device` and `device_policy` is
|
||||
memory and GPU requirements. Placement by `worker` and `worker_policy` is
|
||||
configured through the API or flow JSON; see [Remote workers](../code/workers.md).
|
||||
|
||||
### Code and reuse
|
||||
|
||||
@@ -92,7 +92,7 @@ proxy on that route must support WebSockets.
|
||||
the portal to be online. It also deletes the users the portal let in; the
|
||||
account that enrolled and every superuser stay.
|
||||
- To replace the pairing, use **Connect with a new code**. Remote users and
|
||||
panels may need to pair again.
|
||||
devices paired through the portal may need to pair again.
|
||||
|
||||
If the instance is removed at the portal, or its code is renewed there, the
|
||||
status reads **Removed at the portal** and the instance stops dialling. Connect
|
||||
|
||||
+1
-1
@@ -31,7 +31,7 @@
|
||||
- [The fluksio command](/code/cli/): serve, enroll, worker, and the data directory
|
||||
- [Writing node code](/code/nodes/): process(), generators, fluksio.emit, artifacts
|
||||
- [The HTTP API](/code/api/): auth, flows, messages, runs, observability, websocket
|
||||
- [Remote workers](/code/workers/): fluksio-worker, device labels, artifacts across hosts
|
||||
- [Remote workers](/code/workers/): fluksio-worker, worker labels, artifacts across hosts
|
||||
- [Agents over MCP](/code/agents/): the tool list and the security model
|
||||
- [Writing a connector](/code/connectors/): packaging a device integration as a node type
|
||||
|
||||
|
||||
@@ -24,8 +24,8 @@ def train_curve(lr: float, epochs: int) -> list[float]:
|
||||
Port("weights", "artifact"),
|
||||
Port("final_loss", "float"),
|
||||
],
|
||||
device="gpu",
|
||||
device_policy="prefer",
|
||||
worker="gpu",
|
||||
worker_policy="prefer",
|
||||
timeout=600,
|
||||
resources={"cpus": 2},
|
||||
)
|
||||
@@ -34,8 +34,8 @@ def fit(dataset, lr, epochs=25):
|
||||
|
||||
Yielding is the reporting: each one publishes on the `loss` port the
|
||||
instant it happens, and the run keeps every value as a series — which is
|
||||
why there is no `log_metric()` to call. `device="gpu"` with
|
||||
`device_policy="prefer"` sends this to a worker carrying that label when
|
||||
why there is no `log_metric()` to call. `worker="gpu"` with
|
||||
`worker_policy="prefer"` sends this to a worker carrying that label when
|
||||
one is attached, and runs it here when none is.
|
||||
|
||||
`resources` is what a real training node needs and the rest of a flow does
|
||||
|
||||
@@ -2606,7 +2606,7 @@ export const NodeDef_InputSchema = {
|
||||
title: 'Timeout',
|
||||
description: "Seconds this node's code may be silent before it is stopped. A yield or an emit resets the clock, and the first call's imports are not charged to it. 0 disables the limit: the node runs until it finishes, and only a dead worker fails the call. Empty inherits the engine default."
|
||||
},
|
||||
device: {
|
||||
worker: {
|
||||
anyOf: [
|
||||
{
|
||||
type: 'string'
|
||||
@@ -2615,14 +2615,14 @@ export const NodeDef_InputSchema = {
|
||||
type: 'null'
|
||||
}
|
||||
],
|
||||
title: 'Device',
|
||||
title: 'Worker',
|
||||
description: "Label of the worker this node's code must run on, such as 'gpu'. Empty means the engine's own workers. A run needing a label no attached worker carries waits rather than failing."
|
||||
},
|
||||
device_policy: {
|
||||
worker_policy: {
|
||||
type: 'string',
|
||||
enum: ['require', 'prefer'],
|
||||
title: 'Device Policy',
|
||||
description: 'What to do when no worker carries `device`: wait for one, or run locally anyway.',
|
||||
title: 'Worker Policy',
|
||||
description: 'What to do when no worker carries `worker`: wait for one, or run locally anyway.',
|
||||
default: 'require'
|
||||
},
|
||||
cache: {
|
||||
@@ -2726,7 +2726,7 @@ export const NodeDef_OutputSchema = {
|
||||
title: 'Timeout',
|
||||
description: "Seconds this node's code may be silent before it is stopped. A yield or an emit resets the clock, and the first call's imports are not charged to it. 0 disables the limit: the node runs until it finishes, and only a dead worker fails the call. Empty inherits the engine default."
|
||||
},
|
||||
device: {
|
||||
worker: {
|
||||
anyOf: [
|
||||
{
|
||||
type: 'string'
|
||||
@@ -2735,14 +2735,14 @@ export const NodeDef_OutputSchema = {
|
||||
type: 'null'
|
||||
}
|
||||
],
|
||||
title: 'Device',
|
||||
title: 'Worker',
|
||||
description: "Label of the worker this node's code must run on, such as 'gpu'. Empty means the engine's own workers. A run needing a label no attached worker carries waits rather than failing."
|
||||
},
|
||||
device_policy: {
|
||||
worker_policy: {
|
||||
type: 'string',
|
||||
enum: ['require', 'prefer'],
|
||||
title: 'Device Policy',
|
||||
description: 'What to do when no worker carries `device`: wait for one, or run locally anyway.',
|
||||
title: 'Worker Policy',
|
||||
description: 'What to do when no worker carries `worker`: wait for one, or run locally anyway.',
|
||||
default: 'require'
|
||||
},
|
||||
cache: {
|
||||
@@ -4727,7 +4727,7 @@ export const ValidationIssueSchema = {
|
||||
properties: {
|
||||
code: {
|
||||
type: 'string',
|
||||
enum: ['cycle', 'unconnected_input', 'missing_initial_value', 'node_error', 'unauthenticated_hook', 'self_loop_needs_initial', 'missing_source', 'node_unhealthy', 'ignored_device', 'unknown_device', 'device_port'],
|
||||
enum: ['cycle', 'unconnected_input', 'missing_initial_value', 'node_error', 'unauthenticated_hook', 'self_loop_needs_initial', 'missing_source', 'node_unhealthy', 'ignored_worker', 'unknown_device', 'device_port'],
|
||||
title: 'Code'
|
||||
},
|
||||
message: {
|
||||
|
||||
@@ -887,11 +887,11 @@ export type NodeDef_Input = {
|
||||
/**
|
||||
* Label of the worker this node's code must run on, such as 'gpu'. Empty means the engine's own workers. A run needing a label no attached worker carries waits rather than failing.
|
||||
*/
|
||||
device?: (string | null);
|
||||
worker?: (string | null);
|
||||
/**
|
||||
* What to do when no worker carries `device`: wait for one, or run locally anyway.
|
||||
* What to do when no worker carries `worker`: wait for one, or run locally anyway.
|
||||
*/
|
||||
device_policy?: 'require' | 'prefer';
|
||||
worker_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.
|
||||
*/
|
||||
@@ -913,9 +913,9 @@ export type NodeDef_Input = {
|
||||
};
|
||||
|
||||
/**
|
||||
* What to do when no worker carries `device`: wait for one, or run locally anyway.
|
||||
* What to do when no worker carries `worker`: wait for one, or run locally anyway.
|
||||
*/
|
||||
export type device_policy = 'require' | 'prefer';
|
||||
export type worker_policy = 'require' | 'prefer';
|
||||
|
||||
/**
|
||||
* A node as stored: identity, configuration and ports.
|
||||
@@ -941,11 +941,11 @@ export type NodeDef_Output = {
|
||||
/**
|
||||
* Label of the worker this node's code must run on, such as 'gpu'. Empty means the engine's own workers. A run needing a label no attached worker carries waits rather than failing.
|
||||
*/
|
||||
device?: (string | null);
|
||||
worker?: (string | null);
|
||||
/**
|
||||
* What to do when no worker carries `device`: wait for one, or run locally anyway.
|
||||
* What to do when no worker carries `worker`: wait for one, or run locally anyway.
|
||||
*/
|
||||
device_policy?: 'require' | 'prefer';
|
||||
worker_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.
|
||||
*/
|
||||
@@ -1514,7 +1514,7 @@ export type ValidationError = {
|
||||
* Something wrong with a flow — a fault, or merely advisory.
|
||||
*/
|
||||
export type ValidationIssue = {
|
||||
code: 'cycle' | 'unconnected_input' | 'missing_initial_value' | 'node_error' | 'unauthenticated_hook' | 'self_loop_needs_initial' | 'missing_source' | 'node_unhealthy' | 'ignored_device' | 'unknown_device' | 'device_port';
|
||||
code: 'cycle' | 'unconnected_input' | 'missing_initial_value' | 'node_error' | 'unauthenticated_hook' | 'self_loop_needs_initial' | 'missing_source' | 'node_unhealthy' | 'ignored_worker' | 'unknown_device' | 'device_port';
|
||||
message: string;
|
||||
flow?: string;
|
||||
nodes?: Array<(string)>;
|
||||
@@ -1524,7 +1524,7 @@ export type ValidationIssue = {
|
||||
advisory?: boolean;
|
||||
};
|
||||
|
||||
export type code = 'cycle' | 'unconnected_input' | 'missing_initial_value' | 'node_error' | 'unauthenticated_hook' | 'self_loop_needs_initial' | 'missing_source' | 'node_unhealthy' | 'ignored_device' | 'unknown_device' | 'device_port';
|
||||
export type code = 'cycle' | 'unconnected_input' | 'missing_initial_value' | 'node_error' | 'unauthenticated_hook' | 'self_loop_needs_initial' | 'missing_source' | 'node_unhealthy' | 'ignored_worker' | 'unknown_device' | 'device_port';
|
||||
|
||||
export type ValidationResult = {
|
||||
issues?: Array<ValidationIssue>;
|
||||
|
||||
@@ -23,7 +23,7 @@ import { isLight, parsePalette, type Roles, rolesOf } from "./ui/core/theme"
|
||||
*
|
||||
* A bound setting is read the way a widget reads a value — the same live store
|
||||
* and the same socket — so a panel's own settings messages are part of what
|
||||
* its credential is entitled to (`flow/panels.py`, `messages_for`).
|
||||
* its credential is entitled to (`flow/devices.py`, `messages_for`).
|
||||
*/
|
||||
|
||||
/**
|
||||
|
||||
@@ -22,10 +22,9 @@ export const firstSuperuserPassword = getEnvVar("FIRST_SUPERUSER_PASSWORD")
|
||||
* The two origins a run talks to: the app in the browser, and the API for the
|
||||
* setup and teardown around a spec.
|
||||
*
|
||||
* Deliberately *not* taken from `VITE_API_URL`, even though `../../.env` above
|
||||
* has one. That variable belongs to the app build, and in a checkout configured
|
||||
* for a deployment it names the deployment — so a suite reading it would drive
|
||||
* a browser at the local stack while sending its `DELETE`s to the live
|
||||
* Deliberately *not* taken from `../../.env` above, which in a checkout
|
||||
* configured for a deployment names the deployment. A suite reading it would
|
||||
* drive a browser at the local stack while sending its `DELETE`s to the live
|
||||
* instance. Both origins come from the same place instead: whatever the browser
|
||||
* is pointed at is what teardown may write to.
|
||||
*/
|
||||
|
||||
@@ -463,7 +463,7 @@ def main(argv: list[str] | None = None) -> int:
|
||||
parser.add_argument(
|
||||
"--labels",
|
||||
default="",
|
||||
help="comma-separated, e.g. gpu,cuda12 — what a node's device matches",
|
||||
help="comma-separated, e.g. gpu,cuda12 — what a node's worker matches",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--python",
|
||||
|
||||
Reference in New Issue
Block a user