diff --git a/examples/qrmi-workload/sampler.py b/examples/qrmi-workload/sampler.py index 847e87b..59f40fa 100644 --- a/examples/qrmi-workload/sampler.py +++ b/examples/qrmi-workload/sampler.py @@ -9,6 +9,7 @@ import argparse import os +import signal import sys import time @@ -73,21 +74,42 @@ def resource(backend, rtype): return QuantumResource(backend, getattr(ResourceType, RESOURCE_TYPES[rtype][0])) -def wait_operational(qr, backend, timeout): - """The backend object, once it reports itself operational.""" +class Unanswered(Exception): + pass + + +def _alarm(signum, frame): + raise Unanswered() + + +def ask(fn, seconds): + """Call fn and give up after seconds. A request that hangs must not hold + the pod, and later a session, with it.""" + signal.signal(signal.SIGALRM, _alarm) + signal.alarm(int(seconds)) + try: + return fn() + finally: + signal.alarm(0) + + +def wait_operational(qr, backend, timeout, each=60): + """The backend object once IBM reports it operational. Asked before a + session is opened, so nothing is held while waiting.""" deadline = time.time() + timeout while True: - dev = get_backend(qr) - status = dev.status() - if status.operational: - print("backend: ", backend, status.status_msg) - return dev + try: + dev = ask(lambda: get_backend(qr), each) + status = ask(dev.status, each) + why = status.status_msg + if status.operational: + print("ready: ", backend, why) + return dev + except Unanswered: + why = "no answer from IBM in {}s".format(each) if time.time() >= deadline: - sys.exit( - "{} not operational after {}s: {}".format( - backend, timeout, status.status_msg - ) - ) + sys.exit("{} not operational after {}s: {}".format(backend, timeout, why)) + print("waiting: ", backend, why) time.sleep(5) @@ -148,6 +170,7 @@ def main(): ) return + qr = resource(backend, rtype) session = os.environ.get("FLUENCE_QUANTUM_SESSION_ID") if os.environ.get("FLUENCE_COORDINATION_ROLE") == "worker": # a worker submits into the scout's session instead of acquiring one @@ -160,7 +183,6 @@ def main(): section("circuit in the scout's session") print("session: ", session) os.environ[backend + "_QRMI_JOB_ACQUISITION_TOKEN"] = session - qr = resource(backend, rtype) run( qr, wait_operational(qr, backend, args.ready_timeout), @@ -170,17 +192,13 @@ def main(): return section("session and circuit, through QRMI") - qr = resource(backend, rtype) + # is the device usable before anything is held + dev = wait_operational(qr, backend, args.ready_timeout) lock = qr.acquire() print("session: ", lock) os.environ[backend + "_QRMI_JOB_ACQUISITION_TOKEN"] = str(lock) try: - run( - qr, - wait_operational(qr, backend, args.ready_timeout), - args.qubits, - args.shots, - ) + run(qr, dev, args.qubits, args.shots) finally: qr.release(lock) print("released:", lock) diff --git a/python/Dockerfile b/python/Dockerfile index 8aa4bac..58dcb7a 100644 --- a/python/Dockerfile +++ b/python/Dockerfile @@ -23,6 +23,6 @@ COPY . /app # container has NONE of these, it patches whatever SDK the user already has. RUN pip install --no-cache-dir ".[all]" && ln -s $(which python3) /usr/bin/python -ENV FLUENCE_TASK_DISCOVERY_TIMEOUT=300000 +ENV FLUENCE_TASK_DISCOVERY_TIMEOUT=600 ENV FLUENCE_POLL_INTERVAL=30 CMD ["fluence-sidecar"] diff --git a/python/fluence/providers/qrmi.py b/python/fluence/providers/qrmi.py index 4853aa0..d6e57c5 100644 --- a/python/fluence/providers/qrmi.py +++ b/python/fluence/providers/qrmi.py @@ -114,7 +114,7 @@ def install_interceptor(self, pod_uid: str) -> bool: return False backend_hint = os.environ.get("FLUXION_BACKEND", "") - published = {"session_id": None, "tier": None} + published = {"session_id": None, "tier": None, "job_id": None} # acquire records the session and tier and otherwise behaves as before if hasattr(QuantumResource, "acquire") and not getattr( @@ -145,6 +145,7 @@ def patched_acquire(self, *args, **kwargs): def patched_task_start(self, *args, **kwargs): job_id = original_task_start(self, *args, **kwargs) + published["job_id"] = str(job_id) try: handoff.publish( job_id=str(job_id), @@ -171,11 +172,19 @@ def patched_task_start(self, *args, **kwargs): original_release = QuantumResource.release def patched_release(self, *args, **kwargs): - if os.environ.get("FLUENCE_COORDINATION_ROLE") == "scout": + # only a scout that published a task has workers using its + # session. A scout that failed before submitting releases at + # once, or the session would stay open forever + if ( + os.environ.get("FLUENCE_COORDINATION_ROLE") == "scout" + and published["job_id"] + ): timeout = int( - os.environ.get("FLUENCE_SESSION_HOLD_TIMEOUT", "0") or 0 + os.environ.get("FLUENCE_SESSION_HOLD_TIMEOUT", "3600") or 3600 + ) + log( + f"[qrmi] holding the session until the worker group is done, {timeout}s at most" ) - log("[qrmi] holding the session until the worker group is done") if handoff.wait_workers_done(timeout, log=log): log("[qrmi] worker group done, releasing the session") else: diff --git a/python/fluence/sidecar.py b/python/fluence/sidecar.py index 11c9401..dd5c2af 100644 --- a/python/fluence/sidecar.py +++ b/python/fluence/sidecar.py @@ -14,6 +14,7 @@ FLUENCE_OBSERVE true for observe mode FLUXION_BACKEND the backend the scheduler chose FLUENCE_TASK_DISCOVERY_TIMEOUT seconds to wait for the task (default 300) + FLUENCE_WORKER_TIMEOUT seconds to hold for the workers (default 3600) FLUENCE_POLL_INTERVAL seconds between polls (default 30) FLUENCE_UNGATE_POSITION ungate at this queue position or closer (default 1) FLUENCE_WORKER_TIMEOUT seconds to wait for the workers after ungating, @@ -65,7 +66,7 @@ def _hold_for_workers(namespace, gang_group, pod_name): """Keep this pod, and with it the scout's session and its allocation, until the worker group has finished. Then write the marker the scout's release() waits for.""" - timeout = int(os.environ.get("FLUENCE_WORKER_TIMEOUT", 0)) + timeout = int(os.environ.get("FLUENCE_WORKER_TIMEOUT", 3600)) log("holding until the worker group finishes") wait_for_pods_done(namespace, gang_group, exclude=pod_name, timeout=timeout) if handoff.mark_workers_done(): diff --git a/python/tests/test_hold.py b/python/tests/test_hold.py index b0dbc3f..a0c3fa2 100644 --- a/python/tests/test_hold.py +++ b/python/tests/test_hold.py @@ -218,14 +218,40 @@ def test_scout_release_gives_up_after_timeout(): QR = _fake_qrmi(released) try: QRMIProvider().install_interceptor("uid") + qr = QR() + qr.task_start() t0 = time.time() - QR().release("tok") + qr.release("tok") assert released == ["tok"] assert time.time() - t0 >= 1 finally: _clean() +def test_scout_that_never_submitted_releases_at_once(): + """A scout that acquired and then failed before submitting has no workers + in its session. Its release must not wait for them, or the session stays + open until the hold times out, which by default used to be never.""" + from fluence.providers.qrmi import QRMIProvider + + released = [] + with tempfile.TemporaryDirectory() as d: + os.environ["FLUENCE_HANDOFF_DIR"] = d + os.environ["FLUENCE_COORDINATION_ROLE"] = "scout" + os.environ.pop("FLUENCE_SESSION_HOLD_TIMEOUT", None) + QR = _fake_qrmi(released) + try: + QRMIProvider().install_interceptor("uid") + qr = QR() + lock = qr.acquire() + t0 = time.time() + qr.release(lock) + assert released == [lock] + assert time.time() - t0 < 1 + finally: + _clean() + + def test_non_scout_release_passes_through(): from fluence.providers.qrmi import QRMIProvider