From 2d3832d5ee58c804e6a9b06e52e13b3f5aace129 Mon Sep 17 00:00:00 2001 From: Ivan Podkidyshev Date: Mon, 28 Sep 2026 16:05:08 +0200 Subject: [PATCH] Support background experiment execution through the API Signed-off-by: Ivan Podkidyshev --- doc/api.rst | 24 ++++++++++++- src/cloudai/api.py | 5 ++- src/cloudai/handlers.py | 75 +++++++++++++++++++++++++++++++++++++++-- tests/test_api.py | 42 +++++++++++++++++++---- 4 files changed, 134 insertions(+), 12 deletions(-) diff --git a/doc/api.rst b/doc/api.rst index e988c95e1..5e0934073 100644 --- a/doc/api.rst +++ b/doc/api.rst @@ -47,6 +47,26 @@ Optional keyword arguments: * ``enable_cache_without_check=True`` uses installed workload components without checking them first, as with the CLI option. +Background execution +-------------------- + +Set ``wait=False`` to launch a separate worker process and return a pending +experiment. The worker continues after the calling process exits. Read the latest +saved snapshot using the returned path: + +.. code-block:: python + + experiment = cloudai.api.run_experiment(scenario, system, wait=False, tests_dir=tests_dir) + # Later, including from another process: + experiment = cloudai.api.get_experiment(experiment.path) + print(experiment.status) + +The result directory contains ``experiment.json``, the worker's ``controller.log``, +and its ``request.json``. +Keep referenced config files available until the worker has read them. The worker +loads the installed CloudAI package and plugins in a fresh Python process; +registrations made only in the caller's memory are not transferred. + Find and read results --------------------- @@ -69,7 +89,9 @@ execution constraints without installing workloads, creating a results directory or querying cluster availability. ``run_experiment`` raises exceptions for invalid configuration and setup errors. -Workload failures appear in the experiment and run statuses. +Workload failures appear in the experiment and run statuses. For background runs, +errors after launch are recorded as a failed experiment, with details in +``controller.log``. Missing result files raise ``FileNotFoundError``. Invalid experiment JSON raises ``pydantic.ValidationError``. Listing also raises for unreadable or invalid result diff --git a/src/cloudai/api.py b/src/cloudai/api.py index 2344217ff..5857d056e 100644 --- a/src/cloudai/api.py +++ b/src/cloudai/api.py @@ -23,6 +23,7 @@ def run_experiment( scenario: str | Path, system: str | Path, + wait: bool = True, *, tests_dir: Path | None = None, hook_dir: Path | None = None, @@ -36,7 +37,8 @@ def run_experiment( Pass TOML content as str or configuration files as Path. Relative test paths require a scenario file. Optional directories have the same meaning as in - the CLI. The call waits for execution to finish. + the CLI. With wait=False, a separate process executes the experiment and + returns an initial pending snapshot; use get_experiment to read later results. Configuration and setup errors raise exceptions. Workload failures are recorded in the returned experiment's status. This function does not invoke @@ -45,6 +47,7 @@ def run_experiment( return cloudai.handlers.run_experiment( scenario, system, + wait, tests_dir=tests_dir, hook_dir=hook_dir, output_dir=output_dir, diff --git a/src/cloudai/handlers.py b/src/cloudai/handlers.py index d476c2395..71716216c 100644 --- a/src/cloudai/handlers.py +++ b/src/cloudai/handlers.py @@ -16,8 +16,12 @@ import copy import datetime +import json import logging +import subprocess +import sys import tempfile +import threading import traceback from contextlib import contextmanager from pathlib import Path @@ -321,6 +325,7 @@ def _configuration_files(system: str | Path, scenario: str | Path | None = None) def run_experiment( scenario: str | Path, system: str | Path, + wait: bool = True, *, tests_dir: Path | None = None, hook_dir: Path | None = None, @@ -329,7 +334,7 @@ def run_experiment( single_sbatch: bool = False, enable_cache_without_check: bool = False, ) -> cloudai.models.output.Experiment: - """Run an experiment from configuration files or TOML text.""" + """Run an experiment synchronously or hand it to an independent worker.""" with _configuration_files(system, scenario) as (system_path, scenario_path): parsed_system, tests, parsed_scenario = load_scenario( scenario_path, system_path, tests_dir=tests_dir, hook_dir=hook_dir @@ -357,8 +362,72 @@ def run_experiment( parsed_system, parsed_scenario, dry_run=dry_run, single_sbatch=single_sbatch, result_dir=result_dir ) output = runner.runner.experiment_output - execute_experiment(runner, tests, enable_cache_without_check) - return output.snapshot() + if wait: + execute_experiment(runner, tests, enable_cache_without_check) + return output.snapshot() + output.experiment.status = "pending" + output.experiment.start = None + output.write() + initial = get_experiment(result_dir) + request = { + "system": str(system_path), + "scenario": str(scenario_path), + "tests_dir": str(tests_dir.resolve()) if tests_dir is not None else None, + "hook_dir": str((hook_dir or HOOK_ROOT).resolve()), + "output_dir": str(parsed_system.output_path.resolve()), + "dry_run": dry_run, + "single_sbatch": single_sbatch, + "enable_cache_without_check": enable_cache_without_check, + } + request_path = result_dir / "request.json" + try: + request_path.write_text(json.dumps(request), encoding="utf-8") + command = [ + sys.executable, + "-c", + "import logging, pathlib, sys; import cloudai.handlers; " + "logging.basicConfig(level=logging.INFO); " + "cloudai.handlers.run_background(pathlib.Path(sys.argv[1]))", + str(request_path), + ] + with (result_dir / "controller.log").open("a", encoding="utf-8") as log: + process = subprocess.Popen( + command, stdin=subprocess.DEVNULL, stdout=log, stderr=subprocess.STDOUT, start_new_session=True + ) + threading.Thread(target=process.wait, daemon=True).start() + except Exception: + runner.runner.finish_output(False) + raise + return initial + + +def run_background(request_path: Path) -> None: + """Execute a persisted request in its worker process.""" + try: + request = json.loads(request_path.read_text(encoding="utf-8")) + system, tests, scenario = load_scenario( + Path(request["scenario"]), + Path(request["system"]), + tests_dir=Path(request["tests_dir"]) if request["tests_dir"] else None, + hook_dir=Path(request["hook_dir"]), + ) + system.output_path = Path(request["output_dir"]) + runner = create_experiment_runner( + system, + scenario, + dry_run=request["dry_run"], + single_sbatch=request["single_sbatch"], + result_dir=request_path.parent, + ) + execute_experiment(runner, tests, request["enable_cache_without_check"]) + except BaseException: + logging.exception("Background experiment failed") + experiment = get_experiment(request_path.parent) + if experiment.status in ("pending", "running"): + cloudai.output.ExperimentOutput(experiment, request_path.parent).finish( + "failed", datetime.datetime.now(datetime.timezone.utc) + ) + raise def get_experiment(exp: str | Path) -> cloudai.models.output.Experiment: diff --git a/tests/test_api.py b/tests/test_api.py index 0d2158672..e3e1606a3 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -63,11 +63,20 @@ def configs(standalone_system: StandaloneSystem) -> tuple[str, str]: return scenario, system -@pytest.mark.parametrize("successful,as_text", [(True, False), (True, True), (False, True)]) +@pytest.mark.parametrize( + "wait,successful,as_text", [(True, True, False), (True, True, True), (True, False, True), (False, True, True)] +) def test_experiment_api( - configs: tuple[str, str], tmp_path: Path, monkeypatch: pytest.MonkeyPatch, successful: bool, as_text: bool + configs: tuple[str, str], + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + wait: bool, + successful: bool, + as_text: bool, ): scenario, system = configs + process = Mock() + monkeypatch.setattr("cloudai.handlers.subprocess.Popen", process) monkeypatch.setattr(StandaloneSystem, "is_job_completed", lambda *_: True) monkeypatch.setattr(StandaloneSystem, "is_job_running", lambda *_: False) monkeypatch.setattr( @@ -97,7 +106,17 @@ def test_experiment_api( system_input = tmp_path / "system.toml" scenario_input.write_text(scenario) system_input.write_text(system) - experiment = cloudai.api.run_experiment(scenario_input, system_input) + experiment = cloudai.api.run_experiment(scenario_input, system_input, wait=wait) + if not wait: + assert experiment.status == "pending" + assert experiment.start is None + assert cloudai.api.get_experiment(experiment.path) == experiment + request_path = Path(experiment.path) / "request.json" + assert process.call_args.args[0][-1] == str(request_path) + cloudai.handlers.run_background(request_path) + experiment = cloudai.api.get_experiment(experiment.path) + else: + process.assert_not_called() assert logging.getLogger().handlers == handlers assert signal.getsignal(signal.SIGINT) == sigint assert experiment.status == ("completed" if successful else "failed") @@ -158,12 +177,14 @@ def test_validate_scenario(configs: tuple[str, str], tmp_path: Path): cloudai.api.get_experiment(tmp_path / "missing") -@pytest.mark.parametrize("failure", ["setup", "submission"]) +@pytest.mark.parametrize("failure", ["setup", "spawn", "worker-parse", "submission"]) def test_failed_experiment_is_saved(configs: tuple[str, str], monkeypatch: pytest.MonkeyPatch, failure: str): scenario, system = configs if failure == "setup": monkeypatch.setattr(StandaloneSystem, "update", Mock(side_effect=RuntimeError("setup failed"))) - else: + elif failure == "spawn": + monkeypatch.setattr("cloudai.handlers.subprocess.Popen", Mock(side_effect=OSError("spawn failed"))) + elif failure == "submission": monkeypatch.setattr( StandaloneRunner, "_submit_test", @@ -173,8 +194,15 @@ def test_failed_experiment_is_saved(configs: tuple[str, str], monkeypatch: pytes ) ), ) - with pytest.raises((RuntimeError, cloudai.core.JobSubmissionError)): - cloudai.api.run_experiment(scenario, system) + if failure == "worker-parse": + monkeypatch.setattr("cloudai.handlers.subprocess.Popen", Mock()) + pending = cloudai.api.run_experiment(scenario, system, wait=False) + (Path(pending.path) / "scenario.toml").write_text("name = [") + with pytest.raises(cloudai.core.TestScenarioParsingError): + cloudai.handlers.run_background(Path(pending.path) / "request.json") + else: + with pytest.raises((RuntimeError, OSError, cloudai.core.JobSubmissionError)): + cloudai.api.run_experiment(scenario, system, wait=failure != "spawn") experiments = cloudai.api.list_experiments(system) assert len(experiments) == 1 failed = cloudai.api.get_experiment(experiments[0][1])