diff --git a/backend/fluksio/sdk/__init__.py b/backend/fluksio/sdk/__init__.py index 2a64d95..bdffc07 100644 --- a/backend/fluksio/sdk/__init__.py +++ b/backend/fluksio/sdk/__init__.py @@ -14,6 +14,7 @@ have to resolve inside a worker, where ``fluksio`` is the reporter module. from __future__ import annotations import ast +import copy import inspect import json import re @@ -27,6 +28,7 @@ from fluksio.sdk.imports import digest_of, reached __all__ = [ "FLOWS", + "Declared", "Flow", "Port", "SyncError", @@ -750,18 +752,27 @@ class Flow: ], "outputs": self.outputs, } - if origin is not None: - declared_in = _relative(self.declared_in, str(origin.get("repo") or "")) - doc["origin"] = origin | { - "declared_in": declared_in, - "module": self.module if declared_in else "", - } - return doc + return _stamped(doc, origin, self.declared_in, self.module) def shims(self) -> dict[str, str]: """One generated node body per node, keyed by node id.""" return {spec.id: _shim(spec) for spec in self.nodes} + def declaration(self) -> dict[str, Any]: + """Everything `sync` takes of this flow, as JSON a process can hand on. + + The document is worked out here, where the code was imported, and the + origin is stamped onto it later by :class:`Declared`: only the process + that runs `sync` knows which repository it is stamping with. + """ + return { + "name": self.name, + "document": self.document(), + "shims": self.shims(), + "declared_in": self.declared_in, + "module": self.module, + } + def submit( self, *, seed: int | None = None, client: Any = None, **params: Any ) -> Any: @@ -783,6 +794,47 @@ class Flow: return f"Flow({self.name!r}, nodes={[n.id for n in self.nodes]})" +def _stamped( + doc: dict[str, Any], origin: dict[str, Any] | None, declared_in: str, module: str +) -> dict[str, Any]: + """Where the flow came from, added to its document.""" + if origin is not None: + inside = _relative(declared_in, str(origin.get("repo") or "")) + doc["origin"] = origin | { + "declared_in": inside, + "module": module if inside else "", + } + return doc + + +class Declared: + """A flow another process imported: what `sync` needs, without the code. + + Stands where a :class:`Flow` does for :func:`fluksio.sdk.client.sync`, so + `fluksio sync` can read declarations in a child process and never import + the module itself. Built from :meth:`Flow.declaration`. + """ + + def __init__(self, data: dict[str, Any]) -> None: + self.name: str = data["name"] + self._data = data + + def document(self, origin: dict[str, Any] | None = None) -> dict[str, Any]: + return _stamped( + copy.deepcopy(self._data["document"]), + origin, + self._data["declared_in"], + self._data["module"], + ) + + def shims(self) -> dict[str, str]: + shims: dict[str, str] = dict(self._data["shims"]) + return shims + + def __repr__(self) -> str: + return f"Declared({self.name!r})" + + def build(target: Flow, origin: dict[str, Any] | None = None) -> dict[str, Any]: """The flow document for a flow — checked, without a version.""" return target.document(origin) diff --git a/backend/fluksio/sdk/cli.py b/backend/fluksio/sdk/cli.py index d7002fd..dd64ee2 100644 --- a/backend/fluksio/sdk/cli.py +++ b/backend/fluksio/sdk/cli.py @@ -42,6 +42,7 @@ from fluksio.sdk.client import ( repo_root, sync, ) +from fluksio.sdk.declare import ImportFailed, declare, declare_module __all__ = ["add_parsers", "discover"] @@ -199,6 +200,10 @@ def discover(targets: list[str], keep_going: bool = False) -> list[Flow]: 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. + + Imports in *this* process. The commands go through + :func:`fluksio.sdk.declare.declare`, which runs this in a child so the + command line never holds what the modules open. """ for target in targets: path = Path(target) @@ -336,9 +341,14 @@ def cmd_login(args: argparse.Namespace) -> int: def cmd_sync(args: argparse.Namespace) -> int: targets = args.targets or ["."] try: - flows = discover(targets) + flows = declare(targets) except (ImportError, SyncError) as exc: return _fail(str(exc)) + except ImportFailed as exc: + # What an uncaught exception from the module printed, back when the + # import happened in this process. + print(exc.trace, file=sys.stderr, end="") + return _fail(str(exc)) if not flows: return _fail( f"no flows declared in {', '.join(targets)} — a flow is a `Flow(...)` " @@ -553,7 +563,8 @@ def _sync_first( A flow with no such record — never synced, or an engine older than the field — falls back to the whole directory, downwards, skipping what the repository ignores; a study there that will not import is a warning. - `--sync` names the targets outright. + `--sync` names the targets outright. Either way the import happens in a + child process (see :mod:`fluksio.sdk.declare`), not in this one. 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. @@ -561,13 +572,12 @@ def _sync_first( declared = None if targets else _declaration(client, flow) try: if declared is None: - flows = discover(targets or ["."], keep_going=True) + flows = declare(targets or ["."], keep_going=True) else: # Under the name it was synced with, so the node bodies it # generates are the ones already stored. - _import(*declared) - flows = list(FLOWS.values()) - except SyncError as exc: + flows = declare_module(*declared) + except (SyncError, ImportFailed) as exc: # Do not fail a run for a module the run may not even need. _say(f"warning: nothing synced — {exc}") return diff --git a/backend/fluksio/sdk/client.py b/backend/fluksio/sdk/client.py index c1ec83e..07d7b5b 100644 --- a/backend/fluksio/sdk/client.py +++ b/backend/fluksio/sdk/client.py @@ -18,7 +18,7 @@ from pathlib import Path from typing import Any from urllib.parse import quote -from fluksio.sdk import MARKER, Flow, SyncError +from fluksio.sdk import MARKER, Declared, Flow, SyncError __all__ = [ "Client", @@ -784,7 +784,7 @@ def ignored(directory: Path) -> frozenset[Path]: def sync( - flows: Iterable[Flow], + flows: Iterable[Flow | Declared], client: Client, *, repo: str = "", @@ -813,7 +813,7 @@ def sync( def _sync_one( - target: Flow, + target: Flow | Declared, client: Client, origin: dict[str, Any], *, @@ -908,7 +908,7 @@ def _moved_code(stored: dict[str, Any] | None, document: dict[str, Any]) -> list def _refuse_on_drift( - client: Client, target: Flow, definition: dict[str, Any] + client: Client, target: Flow | Declared, definition: dict[str, Any] ) -> dict[str, str]: """Stop before overwriting work that was done somewhere else. diff --git a/backend/fluksio/sdk/declare.py b/backend/fluksio/sdk/declare.py new file mode 100644 index 0000000..e89e4d2 --- /dev/null +++ b/backend/fluksio/sdk/declare.py @@ -0,0 +1,117 @@ +"""Read the flows a module declares, in a process of its own. + +Importing a module runs it. A research module that sets up a GPU framework at +its top then opens a CUDA context in whichever process imported it — and for +`fluksio run --follow` that process is the command line, which stays alive for +the length of the run and holds the card without the `resources.env` the node +was declared with. So the command line never imports the module: it starts a +short-lived child, the child imports and declares, writes down what `sync` +needs of each flow, and exits. Whatever the import opened goes with it. + +The request and the answer are files in a directory of their own, not a pipe +or the command line. The child's output stays the user's: +a `print` at the top of a study, a warning, a skipped module in a walk, all +appear on the terminal as they did when the import happened here. + +``python -m fluksio.sdk.declare`` is the child. It runs under the same +interpreter and with the same ``sys.path`` as the command, so the module +resolves its imports exactly as it did in-process. +""" + +from __future__ import annotations + +import json +import subprocess +import sys +import tempfile +import traceback +from pathlib import Path +from typing import Any + +from fluksio.sdk import FLOWS, Declared, SyncError + +__all__ = ["ImportFailed", "declare", "declare_module"] + + +class ImportFailed(Exception): + """A module raised something other than a refusal while it was imported. + + ``str()`` is what `fluksio` printed for it when the import happened in + process, ``Type: message``; the child's traceback is kept for the command + that wants to show it. + """ + + def __init__(self, kind: str, message: str, trace: str = "") -> None: + super().__init__(f"{kind}: {message}") + self.trace = trace + + +#: What the child raised, by name, for the ones a command already handles by +#: type. Anything else is an :class:`ImportFailed`. +_RAISED: dict[str, type[Exception]] = { + "SyncError": SyncError, + "ImportError": ImportError, + "ModuleNotFoundError": ModuleNotFoundError, +} + + +def declare(targets: list[str], keep_going: bool = False) -> list[Declared]: + """What importing these targets declares — see ``cli.discover`` for the rules.""" + return _ask({"targets": targets, "keep_going": keep_going}, ", ".join(targets)) + + +def declare_module(root: str, dotted: str, expect: Path) -> list[Declared]: + """What importing one module, by name and from this root, declares.""" + return _ask({"module": [root, dotted, str(expect)]}, dotted) + + +def _ask(request: dict[str, Any], what: str) -> list[Declared]: + request["path"] = sys.path + with tempfile.TemporaryDirectory() as scratch: + (Path(scratch) / "request.json").write_text(json.dumps(request)) + done = subprocess.run([sys.executable, "-m", __name__, scratch]) + try: + reply = json.loads((Path(scratch) / "answer.json").read_text()) + except (OSError, ValueError): + raise SyncError( + f"importing {what} ended without an answer " + f"(exit status {done.returncode}); its output is above" + ) from None + error = reply.get("error") + if error: + raised = _RAISED.get(error["type"]) + if raised is not None: + raise raised(error["message"]) + raise ImportFailed(error["type"], error["message"], error["traceback"]) + return [Declared(entry) for entry in reply["flows"]] + + +def main(scratch: Path) -> None: + """The child: read the request, import, declare, write the answer down.""" + request = json.loads((scratch / "request.json").read_text()) + # Imported here, not above: `cli` is the one place the walk lives, and it + # imports this module for the other half. + from fluksio.sdk.cli import _import, discover + + sys.path[:] = request["path"] + try: + if "module" in request: + root, dotted, expect = request["module"] + _import(root, dotted, Path(expect)) + flows = list(FLOWS.values()) + else: + flows = discover(request["targets"], request["keep_going"]) + reply: dict[str, Any] = {"flows": [flow.declaration() for flow in flows]} + except Exception as exc: + reply = { + "error": { + "type": type(exc).__name__, + "message": str(exc), + "traceback": traceback.format_exc(), + } + } + (scratch / "answer.json").write_text(json.dumps(reply)) + + +if __name__ == "__main__": + main(Path(sys.argv[1])) diff --git a/backend/tests/api/routes/test_sync.py b/backend/tests/api/routes/test_sync.py index 0dbb977..ddacfc0 100644 --- a/backend/tests/api/routes/test_sync.py +++ b/backend/tests/api/routes/test_sync.py @@ -8,7 +8,7 @@ import pytest from fastapi.testclient import TestClient from fluksio.core.config import settings -from fluksio.sdk import FLOWS, MARKER, SyncError +from fluksio.sdk import FLOWS, MARKER, Declared, SyncError from fluksio.sdk.client import ApiError, Client, sync EXAMPLES = Path(__file__).parents[4] / "examples" @@ -74,6 +74,18 @@ def test_sync_stores_a_runnable_flow_with_its_origin(api, flows): assert "from myresearch.train import fit" in source +def test_a_flow_read_by_another_process_syncs_as_the_flow_does(api, flows): + """`fluksio sync` imports in a child and uploads what it handed back.""" + carried = Declared(flows["train"].declaration()) + + sync([carried], api, origin=ORIGIN) + stored = api.get_flow("train")["definition"] + assert stored["origin"]["declared_in"] == "myresearch/pipeline.py" + + # Nothing differs from what the flow itself would have stored. + assert sync([flows["train"]], api, origin=ORIGIN)[0].unchanged + + def test_a_node_carries_the_modules_its_code_reaches(api, flows): """The shim names one function; the cache has to key on what it calls.""" sync([flows["train"]], api, origin=ORIGIN) diff --git a/backend/tests/sdk/test_declare.py b/backend/tests/sdk/test_declare.py new file mode 100644 index 0000000..9fc9119 --- /dev/null +++ b/backend/tests/sdk/test_declare.py @@ -0,0 +1,141 @@ +"""Reading a module's flows in a child, so the command line never imports it.""" + +import os +import sys +from pathlib import Path + +import pytest + +from fluksio.sdk import FLOWS, SyncError, cli +from fluksio.sdk.declare import ImportFailed, declare, declare_module + +STUDY = """\ +import os + +from fluksio import Flow, Port, node + +with open({marker!r}, "a") as marker: + marker.write(str(os.getpid()) + "\\n") + + +@node(provides=Port("out", "int")) +def work(): + return 1 + + +repro = Flow("repro", nodes=[work], outputs=["out"]) +""" + + +def _study(tmp_path: Path, body: str | None = None) -> Path: + path = tmp_path / "study.py" + path.write_text(body or STUDY.format(marker=str(tmp_path / "marker"))) + return path + + +def test_the_module_is_imported_by_a_child_and_not_here(tmp_path): + """What opens a CUDA context at import must not do it in the command line.""" + path = _study(tmp_path) + + found = declare_module(str(tmp_path), "study", path) + + assert [flow.name for flow in found] == ["repro"] + pids = (tmp_path / "marker").read_text().split() + assert pids and str(os.getpid()) not in pids + assert "study" not in sys.modules and "repro" not in FLOWS + assert str(tmp_path) not in sys.path + + +def test_a_declared_flow_is_what_the_flow_itself_would_sync(tmp_path): + """Same document, same shims, same origin: sync cannot tell them apart.""" + path = _study(tmp_path) + origin = {"kind": "python", "repo": str(tmp_path), "commit": "", "dirty": False} + + (declared,) = declare([str(path.parent)]) + try: + (local,) = cli.discover([str(path)]) + assert declared.document(origin) == local.document(origin) + assert declared.shims() == local.shims() + assert declared.document(origin)["origin"]["declared_in"] == "study.py" + finally: + FLOWS.pop("repro", None) + sys.modules.pop("study", None) + sys.path[:] = [entry for entry in sys.path if entry != str(tmp_path)] + + +def test_what_a_module_raises_reaches_the_user_as_it_did_in_process(tmp_path): + path = _study(tmp_path, "raise ValueError('boom')\n") + with pytest.raises(ImportFailed, match="^ValueError: boom$") as raised: + declare_module(str(tmp_path), "study", path) + assert "study.py" in raised.value.trace + + path.write_text("import not_a_module_anywhere\n") + with pytest.raises(ModuleNotFoundError, match="not_a_module_anywhere"): + declare_module(str(tmp_path), "study", path) + + # A refusal from the declaration itself keeps its own type, and message. + path.write_text( + "from fluksio import Flow\nFlow('Not A Name', nodes=[])\n", + ) + with pytest.raises(SyncError, match="flow name"): + declare_module(str(tmp_path), "study", path) + + +def test_a_child_that_ends_without_an_answer_is_named(tmp_path): + path = _study(tmp_path, "import os\nos._exit(3)\n") + with pytest.raises(SyncError, match="exit status 3"): + declare_module(str(tmp_path), "study", path) + + +def test_a_walk_keeps_going_past_a_study_that_will_not_import(tmp_path, capfd): + (tmp_path / "bad.py").write_text("raise RuntimeError('half an edit')\n") + _study(tmp_path) + + found = declare([str(tmp_path)], keep_going=True) + + assert [flow.name for flow in found] == ["repro"] + assert "warning: skipped" in capfd.readouterr().out + with pytest.raises(SyncError, match="failed to import"): + declare([str(tmp_path)]) + + +def test_sync_prints_what_the_child_declared(tmp_path, capsys): + from fluksio.cli import _parser + + path = _study(tmp_path) + args = _parser().parse_args(["sync", "--dry-run", str(path.parent)]) + + assert cli.cmd_sync(args) == 0 + out = capsys.readouterr().out + assert "=== flow repro" in out + assert '"declared_in": "study.py"' in out + assert "=== repro.work" in out + + +def test_a_failing_import_is_reported_by_sync_and_warned_about_by_run( + tmp_path, monkeypatch, capsys +): + from fluksio.cli import _parser + + path = _study(tmp_path, "raise ValueError('boom')\n") + + # Found by a walk, it is the walk's sentence. + assert cli.cmd_sync(_parser().parse_args(["sync", str(tmp_path)])) == 1 + assert "study.py failed to import — ValueError: boom" in capsys.readouterr().err + + # Named outright, it is what the module raised, with the child's traceback + # where an uncaught exception used to print its own. + assert cli.cmd_sync(_parser().parse_args(["sync", str(path)])) == 1 + seen = capsys.readouterr() + assert "fluksio: ValueError: boom" in seen.err + assert "Traceback" in seen.err and "study.py" in seen.err + + class Engine: + def get_flow(self, name): + return { + "definition": {"origin": {"declared_in": "study.py", "module": "study"}} + } + + monkeypatch.chdir(tmp_path) + cli._sync_first(Engine(), [], "repro") + assert "warning: nothing synced — ValueError: boom" in capsys.readouterr().out diff --git a/backend/tests/test_cli.py b/backend/tests/test_cli.py index 67d79f5..5416d93 100644 --- a/backend/tests/test_cli.py +++ b/backend/tests/test_cli.py @@ -687,10 +687,12 @@ def test_run_imports_the_module_that_declared_the_flow(tmp_path, monkeypatch): imported: list[tuple] = [] walked: list[list[str]] = [] - monkeypatch.setattr(sdk_cli, "_import", lambda *args: imported.append(args)) + monkeypatch.setattr( + sdk_cli, "declare_module", lambda *args: imported.append(args) or [] + ) monkeypatch.setattr( sdk_cli, - "discover", + "declare", lambda targets, keep_going=False: walked.append(targets) or [], ) diff --git a/docs/code/cli.md b/docs/code/cli.md index 944dda4..1de324e 100644 --- a/docs/code/cli.md +++ b/docs/code/cli.md @@ -161,6 +161,11 @@ With no target, it searches the current directory. Directory searches skip hidden folders, virtual environments, `__pycache__` and `node_modules`. File paths are not targets: use `study`, not `study.py`. +The modules are imported in a short-lived child process, so work done at +import time, such as opening a GPU context, ends with it and does not stay +in the command. The child's output and its errors appear as they would from +the command itself. + The generated node bodies import your functions. Keep the project available to each machine running them; sync does not copy your package to a remote worker. Uploading source requires a superuser token. @@ -198,10 +203,11 @@ that shares its name with a flag of `run`, such as `timeout`: fluksio run train --wait --timeout 600 -- timeout=30 lr=0.05 ``` -After the first sync, `run` imports the module that declared the flow. If no -declaration is recorded, it searches the working directory and skips paths -covered by `.gitignore`. Import errors in that fallback search produce -warnings. Use `--sync PATH` to choose a target explicitly. +After the first sync, `run` imports the module that declared the flow, in a +child process as `sync` does. If no declaration is recorded, it searches the +working directory and skips paths covered by `.gitignore`. Import errors in +that fallback search produce warnings. Use `--sync PATH` to choose a target +explicitly, or `--no-sync` to skip importing altogether. | Flag | Purpose | |---|---|