diff --git a/backend/fluksio/cli.py b/backend/fluksio/cli.py index 1429945..b476ea1 100644 --- a/backend/fluksio/cli.py +++ b/backend/fluksio/cli.py @@ -19,6 +19,7 @@ import argparse import copy import os import secrets +import socket import sys from pathlib import Path from typing import Any @@ -230,6 +231,32 @@ CONCURRENCY_FLAGS = { } +#: 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. +DEFAULT_PORT = 8000 + +#: How far up from it to look before giving up and letting the bind fail. +PORT_TRIES = 20 + + +def _free_port(host: str, start: int) -> int: + """The first port from ``start`` that nothing is listening on. + + Probed with the same address and options uvicorn will bind with, so this + answers the question uvicorn is about to ask rather than a similar one. + """ + for port in range(start, start + PORT_TRIES): + with socket.socket() as probe: + probe.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + try: + probe.bind((host, port)) + except OSError: + continue + return port + return start + + def cmd_serve(args: argparse.Namespace) -> int: data_dir = _data_dir(args.data_dir, args.shared) for flag, name in CONCURRENCY_FLAGS.items(): @@ -267,10 +294,16 @@ def cmd_serve(args: argparse.Namespace) -> int: from fluksio.flow import modules from fluksio.main import app + port = args.port + if port is None: + port = _free_port(args.host, DEFAULT_PORT) + if port != DEFAULT_PORT: + _say(f"Port {DEFAULT_PORT} is in use; serving on {port} instead.") + # The client talks to this engine, and 0.0.0.0 is not an address to talk # to — it is a statement about which interfaces to listen on. reachable = "127.0.0.1" if args.host in ("0.0.0.0", "::", "") else args.host - url = f"http://{reachable}:{args.port}" + url = f"http://{reachable}:{port}" token_path = _sign_in(admin_id, url, data_dir) config = cloud_config.load() @@ -303,7 +336,7 @@ def cmd_serve(args: argparse.Namespace) -> int: uvicorn.run( app, host=args.host, - port=args.port, + port=port, log_level=args.log_level, log_config=_log_config(args.log_level), ) @@ -374,7 +407,14 @@ def _parser() -> argparse.ArgumentParser: serve = subparsers.add_parser("serve", help="run the engine") with_data_dir(serve) serve.add_argument("--host", default="127.0.0.1") - serve.add_argument("--port", type=int, default=8000) + # No default: a port nobody asked for may move when it is taken, and one + # that was asked for may not. + serve.add_argument( + "--port", + type=int, + default=None, + help=f"default {DEFAULT_PORT}, or the next free port when it is in use", + ) serve.add_argument("--log-level", default="info") serve.add_argument("--admin-email", default=None) serve.add_argument("--admin-password", default=None) diff --git a/backend/fluksio/sdk/cli.py b/backend/fluksio/sdk/cli.py index 69b1d56..e203360 100644 --- a/backend/fluksio/sdk/cli.py +++ b/backend/fluksio/sdk/cli.py @@ -115,19 +115,81 @@ def _module_of(path: Path) -> tuple[str, str]: return str(directory), ".".join(reversed(parts)) -def _import(root: str, dotted: str) -> None: +#: Directories a walk never goes into: none of them is a study, and some of +#: them are enormous. +_SKIP_DIRS = frozenset({"__pycache__", "node_modules", "site-packages"}) + + +def _import(root: str, dotted: str, expect: Path | None = None) -> None: if root not in sys.path: sys.path.insert(0, root) - importlib.import_module(dotted) + module = importlib.import_module(dotted) + if expect is None: + return + # Python keeps one module per name, so a second file importing under a + # name already taken is silently the first one — and the generated body + # imports by that name too, so a worker would run the wrong study's code. + actual = getattr(module, "__file__", "") or "" + if actual and Path(actual).resolve() != expect.resolve(): + raise SyncError( + f"two files would both import as '{dotted}':\n" + f" {actual}\n {expect}\n" + "Python keeps one module per name, and a node's generated body " + "imports by that name, so the second would run the first's code. " + "Put an `__init__.py` in each directory — they become " + f"'.{dotted}' and stop colliding — or rename one of the files." + ) -def discover(targets: list[str]) -> list[Flow]: +def _walkable(entry: Path) -> bool: + """Whether a walk should look inside this directory at all.""" + return ( + entry.is_dir() + and not entry.name.startswith(".") + and entry.name not in _SKIP_DIRS + and not (entry / "pyvenv.cfg").exists() + ) + + +def _below(root: Path) -> Iterator[Path]: + """Every module and package under a plain directory, however deep. + + One directory per study — `dev/s1_baseline/study.py` — is a layout people + have, and naming each of them on the command line is bookkeeping the tool + can do. A package is yielded whole and not descended into: its own walk + imports its submodules under the right names. + """ + for entry in sorted(root.iterdir()): + if entry.is_file() and entry.suffix == ".py": + yield entry + elif _walkable(entry): + if (entry / "__init__.py").exists(): + yield entry + else: + yield from _below(entry) + + +def _import_package(directory: Path) -> None: + """A package and every module in it, by their dotted names.""" + root, dotted = _package_of(directory) + _import(root, dotted, directory / "__init__.py") + for info in pkgutil.walk_packages(sys.modules[dotted].__path__, f"{dotted}."): + importlib.import_module(info.name) + + +def discover(targets: list[str], keep_going: bool = False) -> list[Flow]: """Import what was named and hand back the flows it declared. Imported by dotted name with its root on the path, never from a file location: the generated node bodies import the same way, and a module loaded under a different name would generate an import that does not resolve. + + ``keep_going`` warns about a module that will not import instead of + stopping, which is what a *run* wants: a study half-way through an edit + two directories away is not a reason to refuse to run this one. A + collision between two module names is never skipped — it would produce a + node body that imports the wrong file. """ for target in targets: path = Path(target) @@ -136,30 +198,33 @@ def discover(targets: list[str]) -> list[Flow]: continue path = path.resolve() if path.is_file(): - _import(*_module_of(path)) + _import(*_module_of(path), path) continue if (path / "__init__.py").exists(): - root, dotted = _package_of(path) - _import(root, dotted) - package = sys.modules[dotted] - for info in pkgutil.walk_packages(package.__path__, f"{dotted}."): - importlib.import_module(info.name) + _import_package(path) continue - # A plain directory — a repository root, usually. Its own modules, - # and the packages inside it: `myresearch/` beside a `README` is the - # ordinary shape, and naming it explicitly should not be the price of - # keeping your code in a package. - for module in sorted(path.glob("*.py")): - _import(*_module_of(module)) - for child in sorted(path.iterdir()): - if child.name.startswith(".") or not (child / "__init__.py").exists(): - continue - root, dotted = _package_of(child) - _import(root, dotted) - for info in pkgutil.walk_packages( - sys.modules[dotted].__path__, f"{dotted}." - ): - importlib.import_module(info.name) + # A plain directory — a repository root, or a directory of studies. + # Its own modules, the packages inside it, and the same again all the + # way down, so one folder per study needs no naming. + for entry in _below(path): + try: + if entry.is_dir(): + _import_package(entry) + else: + _import(*_module_of(entry), entry) + except SyncError: + # A name collision is never somebody else's problem: it would + # put the wrong file behind a node. + raise + except Exception as exc: + # Importing a module runs it, so this is whatever the study + # does at its top level, and a walk meets every study. + if keep_going: + _say(f"warning: skipped {entry} — {type(exc).__name__}: {exc}") + continue + raise SyncError( + f"{entry} failed to import — {type(exc).__name__}: {exc}" + ) from exc return list(FLOWS.values()) @@ -404,7 +469,7 @@ def _params(definition: dict[str, Any], rest: list[str]) -> dict[str, Any]: return params -def _sync_first(client: Client) -> None: +def _sync_first(client: Client, targets: list[str] | None = None) -> None: """Upload what the working directory declares, before running it. The reason a run exists is usually the edit that came before it, and @@ -412,18 +477,24 @@ def _sync_first(client: Client) -> None: done. So `run` syncs by default — including the worker refresh, which is what makes an edit to your own package take effect at all. + The whole directory, downwards: from a repository root the flow is usually + in a study folder below, and the alternative is syncing by hand and then + running with `--no-sync`. The upload is already a no-op for a flow nothing + changed in, so the cost is importing the other studies — which `--sync` + narrows when that is not free. + A directory that declares nothing is not an error: a flow drawn on the canvas is run the same way, and has nothing to upload. """ try: - flows = discover(["."]) + flows = discover(targets or ["."], keep_going=True) except (ImportError, SyncError) as exc: # Do not fail a run for a module the run may not even need. _say(f"warning: nothing synced — {exc}") return if not flows: return - repo = repo_root(".") + repo = repo_root((targets or ["."])[0]) reports = sync(flows, client, origin=origin_of(repo)) changed = [r for r in reports if not r.unchanged] if changed: @@ -494,7 +565,7 @@ def cmd_run(args: argparse.Namespace, rest: list[str]) -> int: try: with _client_for(args) as client: if not args.no_sync: - _sync_first(client) + _sync_first(client, args.sync) stored = client.get_flow(args.flow) if stored is None: return _fail(f"no flow '{args.flow}' on that engine") @@ -884,7 +955,7 @@ def cmd_sweep(args: argparse.Namespace) -> int: try: with _client_for(args) as client: if not args.no_sync: - _sync_first(client) + _sync_first(client, args.sync) stored = client.get_flow(args.flow) if stored is None: return _fail(f"no flow '{args.flow}' on that engine") @@ -1157,6 +1228,13 @@ def add_parsers(subparsers: Any) -> None: action="store_true", help="run what is already on the engine, without uploading first", ) + parser.add_argument( + "--sync", + action="append", + default=[], + metavar="PATH", + help="what to sync first (default: this directory, downwards)", + ) parser.add_argument( "--no-cache", action="store_true", @@ -1225,6 +1303,13 @@ def add_parsers(subparsers: Any) -> None: action="store_true", help="run what is already on the engine, without uploading first", ) + parser.add_argument( + "--sync", + action="append", + default=[], + metavar="PATH", + help="what to sync first (default: this directory, downwards)", + ) parser.add_argument( "--no-cache", action="store_true", diff --git a/backend/tests/test_cli.py b/backend/tests/test_cli.py index 62e65c5..4cad80f 100644 --- a/backend/tests/test_cli.py +++ b/backend/tests/test_cli.py @@ -388,3 +388,79 @@ def test_an_engine_without_the_route_is_named_rather_than_404() -> None: assert "engine is 0.1.4" in _too_old(Old()) assert "engine is older" in _too_old(Ancient()) assert "pip install -U fluksio" in _too_old(Ancient()) + + +def test_a_study_in_a_subfolder_is_found(tmp_path) -> None: + """One directory per study is a layout; naming each is bookkeeping.""" + from fluksio.sdk.cli import _below + + (tmp_path / "dev" / "s1").mkdir(parents=True) + (tmp_path / "dev" / "s2" / "inner").mkdir(parents=True) + (tmp_path / "results").mkdir() + (tmp_path / "pkg").mkdir() + (tmp_path / ".hidden").mkdir() + (tmp_path / "dev" / "s1" / "study.py").write_text("") + (tmp_path / "dev" / "s2" / "inner" / "probe.py").write_text("") + (tmp_path / "results" / "notes.py").write_text("") + (tmp_path / "pkg" / "__init__.py").write_text("") + (tmp_path / "pkg" / "deep.py").write_text("") + (tmp_path / ".hidden" / "skip.py").write_text("") + + found = {path.relative_to(tmp_path).as_posix() for path in _below(tmp_path)} + + # However deep, plus packages whole — and nothing under a dot directory. + assert found == { + "dev/s1/study.py", + "dev/s2/inner/probe.py", + "results/notes.py", + "pkg", + } + + +def test_two_files_of_one_name_are_refused(tmp_path) -> None: + """Python keeps one module per name, and a node's body imports by it.""" + import pytest + + from fluksio.sdk import SyncError + from fluksio.sdk.cli import _import, _module_of + + for study in ("s1", "s2"): + (tmp_path / study).mkdir() + (tmp_path / study / "study.py").write_text("VALUE = 1\n") + + first = tmp_path / "s1" / "study.py" + second = tmp_path / "s2" / "study.py" + _import(*_module_of(first), first) + with pytest.raises(SyncError, match="both import as 'study'"): + _import(*_module_of(second), second) + + +def test_run_and_sweep_take_what_to_sync() -> None: + from fluksio.cli import _parser + + parser = _parser() + assert parser.parse_args(["sweep", "train", "--sync", "dev/s1"]).sync == ["dev/s1"] + args, _ = parser.parse_known_args(["run", "train", "--sync", "dev/s1"]) + assert args.sync == ["dev/s1"] + # Nothing named means this directory, downwards. + assert parser.parse_args(["sweep", "train"]).sync == [] + + +def test_serve_moves_off_a_port_that_is_taken() -> None: + """A first start should not die on somebody else's dev server.""" + import socket + + from fluksio.cli import DEFAULT_PORT, _free_port, _parser + + with socket.socket() as held: + held.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + held.bind(("127.0.0.1", 0)) + held.listen(1) + taken = held.getsockname()[1] + + assert _free_port("127.0.0.1", taken) == taken + 1 + + # A port that was asked for is not moved off: that is what asking means. + assert _parser().parse_args(["serve"]).port is None + assert _parser().parse_args(["serve", "--port", "9000"]).port == 9000 + assert DEFAULT_PORT == 8000 diff --git a/docs/code/cli.md b/docs/code/cli.md index c5734d2..03c6aa9 100644 --- a/docs/code/cli.md +++ b/docs/code/cli.md @@ -39,11 +39,17 @@ On the first start it creates an admin account and prints its password **once**. Nothing else has to be running: no database server, no message broker, no Docker. +The default port moves out of the way when something already has it — 8001, +8002, and so on — and says which one it took; the URL written to +`client.json` is the one it is actually on. A port you *asked* for is never +moved off: `--port 9000` on a taken 9000 fails, because something else is +there and you named it. + | Option | Default | What it does | |---|---|---| | `--data-dir PATH` | `./.fluksio` (or `$FLUKSIO_HOME`) | where this installation keeps everything | | `--host HOST` | `127.0.0.1` | what to bind | -| `--port PORT` | `8000` | what to listen on | +| `--port PORT` | `8000`, or the next free one | what to listen on | | `--log-level LEVEL` | `info` | uvicorn's log level | | `--admin-email ADDR` | `admin@example.com` | the account created on first run | | `--admin-password PW` | generated | set it instead of having one generated | @@ -152,6 +158,15 @@ each one with a generated import shim per node. A directory that is a package is walked; a dotted name is imported as it stands; nothing is loaded from a file path, because the shim has to import the same way. +A plain directory is walked all the way down, so one folder per study — +`fluksio sync dev` over `dev/s1_baseline/study.py` — needs no naming. Hidden +directories, `__pycache__`, `node_modules` and virtualenvs are left alone. Two +files that would import under the same name are refused rather than +silently collapsed into one: Python keeps one module per name, and a node's +generated body imports by that name, so `dev/s1/study.py` and `dev/s2/study.py` +need an `__init__.py` each — making them `s1.study` and `s2.study` — or +different filenames. + | Flag | What it does | |---|---| | `--dry-run` | print the flow documents and shims, upload nothing | @@ -182,9 +197,16 @@ upgraded its cache is keyed on the whole repository, as it was before. See fluksio run train --lr 0.05 --seed 7 [--wait] ``` -Syncs the working directory, then submits a run — so the command after an edit -is this one and nothing else. Flags that are not its own are the flow's -inputs, typed by what the flow declares them as. `--wait` blocks until the run +Syncs the working directory *and everything under it*, then submits a run — so +the command after an edit is this one and nothing else, from the repository +root as readily as from the study's own folder. A study that will not import +is a warning rather than a stopped run; the upload is already a no-op for a +flow nothing changed in, so what the walk costs is importing the others. +`--sync dev/s1_baseline` (repeatable) narrows it to what you name when that +is not free, and `--no-sync` skips it entirely. + +Flags that are not its own are the flow's inputs, typed by what the flow +declares them as. `--wait` blocks until the run finishes and exits non-zero if it failed. `--follow` waits as well, and prints the numbers the run reports as they arrive: