From ee3e7e229792c0b3c3f24d62a115b72b9c894962 Mon Sep 17 00:00:00 2001 From: vsoch Date: Mon, 19 Oct 2026 13:23:55 -0700 Subject: [PATCH] scout: add support for an exit marker Signed-off-by: vsoch --- .github/workflows/sidecar-build-deploy.yaml | 1 + python/fluence/handoff.py | 43 +++++++++++- python/fluence/interceptor.py | 13 ++++ python/fluence/providers/braket.py | 4 ++ python/fluence/sidecar.py | 15 +++++ python/tests/test_rehearsal.py | 75 +++++++++++++++++++++ 6 files changed, 150 insertions(+), 1 deletion(-) create mode 100644 python/tests/test_rehearsal.py diff --git a/.github/workflows/sidecar-build-deploy.yaml b/.github/workflows/sidecar-build-deploy.yaml index 8c31e40..5087f9e 100644 --- a/.github/workflows/sidecar-build-deploy.yaml +++ b/.github/workflows/sidecar-build-deploy.yaml @@ -51,6 +51,7 @@ jobs: python tests/test_fluence.py python tests/test_qrmi.py python tests/test_hold.py + python tests/test_rehearsal.py build-deploy: runs-on: ubuntu-latest diff --git a/python/fluence/handoff.py b/python/fluence/handoff.py index e8aa1c0..282abbc 100644 --- a/python/fluence/handoff.py +++ b/python/fluence/handoff.py @@ -67,13 +67,19 @@ def read(path=None) -> "dict | None": def wait(timeout, interval=5, path=None, log=None) -> "dict | None": - """Poll for the record until it appears or timeout seconds pass.""" + """Poll for the record until it appears, the scout exits without one, or + timeout seconds pass.""" deadline = time.time() + timeout announced = False + exited = os.path.join(os.path.dirname(handoff_path(path)), EXITED_FILENAME) while time.time() < deadline: record = read(path) if record is not None: return record + if os.path.exists(exited): + if log: + log("the scout exited without publishing a task") + return None if log and not announced: log(f"waiting for the scout to publish its job id at {handoff_path(path)}") announced = True @@ -81,6 +87,41 @@ def wait(timeout, interval=5, path=None, log=None) -> "dict | None": return None +# the interceptor writes this when the scout's interpreter exits with no task +# published, so a rehearsal with no credentials does not leave the workers +# gated until the discovery timeout +EXITED_FILENAME = "scout-exited" + + +def exited_path(path=None) -> str: + if path: + return path + d = os.environ.get(HANDOFF_DIR_ENV, DEFAULT_HANDOFF_DIR) + return os.path.join(d, EXITED_FILENAME) + + +def mark_scout_exited(path=None) -> bool: + """Write the marker, unless a task was published. Never raises.""" + try: + if read() is not None: + return False + dest = exited_path(path) + os.makedirs(os.path.dirname(dest), exist_ok=True) + fd, tmp = tempfile.mkstemp( + dir=os.path.dirname(dest), prefix=".exited.", suffix=".tmp" + ) + with os.fdopen(fd, "w") as f: + json.dump({"exited_at": time.time()}, f) + os.replace(tmp, dest) + return True + except Exception: + return False + + +def scout_exited(path=None) -> bool: + return os.path.exists(exited_path(path)) + + # the sidecar writes this once the workers are done and the scout's release waits for it DONE_FILENAME = "workers-done" diff --git a/python/fluence/interceptor.py b/python/fluence/interceptor.py index 67399d7..4da861b 100644 --- a/python/fluence/interceptor.py +++ b/python/fluence/interceptor.py @@ -12,6 +12,7 @@ from __future__ import annotations +import atexit import os @@ -38,5 +39,17 @@ def install() -> None: ) +def _mark_exit() -> None: + # tell the sidecar the application ended, so a scout that never submitted + # releases its workers instead of leaving them gated + try: + from fluence import handoff + + handoff.mark_scout_exited() + except Exception: + pass + + # Install on import. install() +atexit.register(_mark_exit) diff --git a/python/fluence/providers/braket.py b/python/fluence/providers/braket.py index ac0edbc..6744887 100644 --- a/python/fluence/providers/braket.py +++ b/python/fluence/providers/braket.py @@ -16,6 +16,7 @@ import os import time +from fluence import handoff from fluence.providers.base import ( Provider, Task, @@ -176,6 +177,9 @@ def find_my_task(self, pod_uid, backend, timeout): set() ) # regions that returned AccessDenied, skip them after the first hit while time.time() < deadline: + if handoff.scout_exited(): + log("[braket] the scout exited without a task, giving up the search") + return None for region in regions: if region in denied: continue diff --git a/python/fluence/sidecar.py b/python/fluence/sidecar.py index 1356434..c4d97c3 100644 --- a/python/fluence/sidecar.py +++ b/python/fluence/sidecar.py @@ -101,6 +101,21 @@ def main(): log(f"resolved provider: {provider.name}") task = provider.find_my_task(pod_uid, backend, discovery_timeout) + if task is None and handoff.scout_exited() and not observe: + # a rehearsal, or an application that never submitted. Release the + # workers with no session and hold for them as usual + log( + "the scout finished without submitting a task, releasing the workers with no session" + ) + ungate_pods( + wait_for_gated_pods( + namespace, gang_group, exclude=pod_name, timeout=ungate_timeout + ), + "", + namespace, + ) + _hold_for_workers(namespace, gang_group, pod_name) + sys.exit(0) if task is None: log("ERROR: could not discover quantum task") if not observe: diff --git a/python/tests/test_rehearsal.py b/python/tests/test_rehearsal.py new file mode 100644 index 0000000..69527a1 --- /dev/null +++ b/python/tests/test_rehearsal.py @@ -0,0 +1,75 @@ +""" +Tests for the scout exit marker. A scout that ends without submitting a task, +a rehearsal with no credentials, leaves a marker that ends the sidecar's wait +for the handoff at once, so the workers are released instead of gated until +the discovery timeout. + +Run: PYTHONPATH=python python3 python/tests/test_rehearsal.py +""" + +import os +import sys +import tempfile +import time + + +def test_marker_roundtrip(): + from fluence import handoff + + with tempfile.TemporaryDirectory() as d: + p = os.path.join(d, "scout-exited") + assert not handoff.scout_exited(p) + assert handoff.mark_scout_exited(p) + assert handoff.scout_exited(p) + + +def test_no_marker_when_a_task_was_published(): + from fluence import handoff + + with tempfile.TemporaryDirectory() as d: + os.environ["FLUENCE_HANDOFF_DIR"] = d + try: + assert handoff.publish("job-1") + assert not handoff.mark_scout_exited() + assert not handoff.scout_exited() + finally: + os.environ.pop("FLUENCE_HANDOFF_DIR", None) + + +def test_wait_returns_at_once_after_the_scout_exits(): + from fluence import handoff + + with tempfile.TemporaryDirectory() as d: + os.environ["FLUENCE_HANDOFF_DIR"] = d + try: + handoff.mark_scout_exited() + start = time.time() + assert handoff.wait(30, interval=1) is None + assert time.time() - start < 5 + finally: + os.environ.pop("FLUENCE_HANDOFF_DIR", None) + + +def test_wait_still_times_out_without_a_marker(): + from fluence import handoff + + with tempfile.TemporaryDirectory() as d: + start = time.time() + assert handoff.wait(1, interval=1, path=os.path.join(d, "task.json")) is None + assert time.time() - start >= 1 + + +if __name__ == "__main__": + fns = [ + v for k, v in sorted(globals().items()) if k.startswith("test_") and callable(v) + ] + failed = 0 + for fn in fns: + try: + fn() + print(f"PASS {fn.__name__}") + except Exception as e: + failed += 1 + print(f"FAIL {fn.__name__}: {e}") + print(f"\n{len(fns) - failed}/{len(fns)} passed") + sys.exit(1 if failed else 0)