stroblme
5c7a8773ad
The command line no longer imports a research module, so import-time GPU work ends with a short-lived child instead of holding a CUDA context for the length of a followed run. fluksio.sdk.declare starts the child, which writes each flow's document and shims as JSON; Declared stands in for Flow in sync(). Errors keep their type and message, a child's traceback reaches the user.
118 lines
4.4 KiB
Python
118 lines
4.4 KiB
Python
"""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]))
|