diff --git a/backend/fluksio/cli.py b/backend/fluksio/cli.py index b476ea1..237ca09 100644 --- a/backend/fluksio/cli.py +++ b/backend/fluksio/cli.py @@ -21,6 +21,7 @@ import os import secrets import socket import sys +from collections.abc import Callable from pathlib import Path from typing import Any @@ -228,9 +229,26 @@ CONCURRENCY_FLAGS = { "max_workers": "FLOW_MAX_WORKERS", "max_cascades": "FLOW_MAX_CASCADES", "max_runs": "FLOW_MAX_RUNS", + "gpus": "FLOW_GPUS", } +def _at_least(minimum: int) -> Callable[[str], int]: + """A flag's value, checked here rather than by the settings. + + These are written into the environment before the settings are built, so a + number they refuse dies inside a pydantic import with no flag named in it. + """ + + def parse(text: str) -> int: + value = int(text) + if value < minimum: + raise argparse.ArgumentTypeError(f"is {value}, needs at least {minimum}") + return value + + return parse + + #: What `serve` listens on when nobody says. Taken often enough — another #: engine, another framework's dev server — that dying on it is the first #: thing a zero-config start would hit. @@ -426,25 +444,32 @@ def _parser() -> argparse.ArgumentParser: ) serve.add_argument( "--max-runs", - type=int, + type=_at_least(1), default=None, metavar="N", help="batch runs driven at once (default 4, FLOW_MAX_RUNS)", ) serve.add_argument( "--max-cascades", - type=int, + type=_at_least(1), default=None, metavar="N", help="cascades in flight at once (default 4, FLOW_MAX_CASCADES)", ) serve.add_argument( "--max-workers", - type=int, + type=_at_least(1), default=None, metavar="N", help="python worker processes (default 4, FLOW_MAX_WORKERS)", ) + serve.add_argument( + "--gpus", + type=_at_least(0), + default=None, + metavar="N", + help="GPUs on this machine a node may be given (default 0, FLOW_GPUS)", + ) serve.set_defaults(func=cmd_serve) enroll = subparsers.add_parser( diff --git a/backend/fluksio/flow/placement.py b/backend/fluksio/flow/placement.py index 9b70dc6..da1111a 100644 --- a/backend/fluksio/flow/placement.py +++ b/backend/fluksio/flow/placement.py @@ -215,9 +215,18 @@ 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. + hint = ( + "; no machine here declares a GPU — `fluksio serve --gpus N` " + "(or FLOW_GPUS) says how many this one has" + if wanted.gpus and not gpus + else "" + ) logger.warning( "%s asked for %d cpu(s), %d gpu(s) and %s MB; " - "the largest machine here can give %d, %d and %s", + "the largest machine here can give %d, %d and %s%s", node or "a node", wanted.cpus, wanted.gpus, @@ -225,6 +234,7 @@ class Placer: cpus, gpus, ram or "no stated", + hint, ) return cpus, gpus, ram diff --git a/backend/tests/flow/test_placement.py b/backend/tests/flow/test_placement.py index d5edeca..73dabf5 100644 --- a/backend/tests/flow/test_placement.py +++ b/backend/tests/flow/test_placement.py @@ -129,7 +129,7 @@ def test_a_preferred_label_falls_back_here_and_is_still_accounted(loop): assert placer.local.snapshot()["cpus"]["free"] == 1 -def test_asking_for_more_than_anything_has_gets_what_there_is(loop): +def test_asking_for_more_than_anything_has_gets_what_there_is(loop, caplog): """A flow written on a cluster still has to run on a laptop.""" placer = placer_over(cpus=2) @@ -138,6 +138,10 @@ def test_asking_for_more_than_anything_has_gets_what_there_is(loop): assert allocation.cpus == 2 assert allocation.gpus == () + # Cards are declared, not detected, so a machine that has one reads as + # having none until it is told — and the warning is where that is noticed. + assert "fluksio serve --gpus" in caplog.text + def test_what_is_clamped_to_is_a_machine_that_exists(loop): """Each dimension taken separately can describe a machine nobody has. diff --git a/backend/tests/test_cli.py b/backend/tests/test_cli.py index 8ef1eaa..802ce09 100644 --- a/backend/tests/test_cli.py +++ b/backend/tests/test_cli.py @@ -520,6 +520,24 @@ def test_two_files_of_one_name_are_refused(tmp_path) -> None: _import(*_module_of(second), second) +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.""" + import pytest + + from fluksio.cli import _parser + + parser = _parser() + assert parser.parse_args(["serve", "--max-workers", "2"]).max_workers == 2 + # A machine may genuinely have no card, so zero is a number here. + assert parser.parse_args(["serve", "--gpus", "0"]).gpus == 0 + assert parser.parse_args(["serve"]).gpus is None + + for flag, value in (("--max-workers", "0"), ("--gpus", "-1")): + with pytest.raises(SystemExit): + parser.parse_args(["serve", flag, value]) + assert "at least" in capsys.readouterr().err + + def test_run_and_sweep_take_what_to_sync() -> None: from fluksio.cli import _parser