sidecar: add better message to user

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)

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"]

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:

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():

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