diff --git a/csaxs_bec/bec_ipython_client/plugins/flomni/flomni.py b/csaxs_bec/bec_ipython_client/plugins/flomni/flomni.py index 7fc83f1..46cdddc 100644 --- a/csaxs_bec/bec_ipython_client/plugins/flomni/flomni.py +++ b/csaxs_bec/bec_ipython_client/plugins/flomni/flomni.py @@ -5,6 +5,7 @@ import os import random import subprocess import time +import uuid from pathlib import Path import h5py @@ -1659,6 +1660,41 @@ class _TomoQueueProxy: jobs[index].update(kwargs) self._save(jobs) + def update_by_id(self, job_id: str, **kwargs) -> bool: + """Update the job whose ``id`` matches ``job_id``. + + Returns True if a matching job was found and updated, False otherwise + (e.g. the job was deleted concurrently). Addressing a job by its stable + id rather than its list index makes status writes safe against queue + edits made while the queue runs: a reorder, or a delete of some *other* + job, can no longer misdirect a status write to the wrong entry. + """ + jobs = self._load() + for job in jobs: + if job.get("id") == job_id: + job.update(kwargs) + self._save(jobs) + return True + return False + + def ensure_ids(self) -> list: + """Assign a stable id to any job that lacks one and persist the result. + + Back-compat shim for queues written before jobs carried an ``id`` field + (and for any job added by older code). Called once at the top of + tomo_queue_execute() so the pick-next executor can address every job by + id. Returns the (possibly healed) job list. + """ + jobs = self._load() + changed = False + for job in jobs: + if not job.get("id"): + job["id"] = uuid.uuid4().hex + changed = True + if changed: + self._save(jobs) + return jobs + def clear(self) -> None: self._save([]) @@ -3559,6 +3595,7 @@ class Flomni( params = {name: getattr(self, name) for name in self._TOMO_QUEUE_PARAM_NAMES} index = len(self._tomo_queue_proxy) job = { + "id": uuid.uuid4().hex, "label": label or f"job_{index + 1}", "params": params, "status": "pending", @@ -3613,9 +3650,29 @@ class Flomni( ``tomo_scan()`` -- or, for a job that didn't run to completion last time, ``tomo_scan_resume()``, so it picks back up mid-scan instead of restarting from subtomo 1 / angle 0. Jobs already marked "done" - are skipped on the next call, so simply calling tomo_queue_execute() - again resumes from where it stopped (e.g. after fixing a hardware - issue). + are skipped, so simply calling tomo_queue_execute() again resumes + from where it stopped (e.g. after fixing a hardware issue). + + Pick-next execution model + ------------------------- + The queue is re-read from its global var at the start of every job + (not snapshotted once up front), and each job is addressed by its + stable ``id``, not by its list index. This is what makes it safe to + edit the queue *while it runs*: appending a job, deleting a pending + job, or reordering the pending tail all take effect at the next job + boundary, and a reorder/delete can never misdirect a status write to + the wrong entry (an index-based loop could). Legacy jobs written + before the ``id`` field existed are healed once up front by + ensure_ids(). + + Resume-before-fresh ordering: if any job is in a resumable state, it + is always run before any fresh "pending" job, regardless of list + order. This preserves the single-resumable-job invariant -- tomo + progress lives in one shared ``progress`` global var, so starting a + fresh tomo_scan() before resuming an interrupted one would overwrite + the very state tomo_scan_resume() needs. (Under the old strictly + in-order loop this couldn't happen; allowing reorder makes it + possible, hence the explicit guard.) A job is considered not-yet-complete (and so gets resumed rather than restarted) if its status is "incomplete" (a Python exception @@ -3632,51 +3689,85 @@ class Flomni( yourself to recover from a crash (bypassing the queue), that scan is now actually finished even though the queue still has the job marked "incomplete" or "running" -- mark it done yourself before - calling this again, or it will be re-run from scratch: - flomni._tomo_queue_proxy.update(job_index, status="done") + calling this again, or it will be re-run from scratch. Address the + job by its id (indices can shift if the queue is reordered): + flomni._tomo_queue_proxy.update_by_id(job_id, status="done") Args: - start_index: Queue index to start from. Defaults to 0, but jobs - already marked "done" are skipped automatically either way. + start_index: Position in the *current* queue order below which + fresh "pending" jobs are ignored. Defaults to 0. Resumable + jobs are always run regardless of start_index (a crashed + job must be recovered wherever it sits). Jobs already marked + "done" are skipped automatically either way. Note that with + reordering active this is applied to the queue's current + order at each pick. """ + self._tomo_queue_proxy.ensure_ids() jobs = self._tomo_queue_proxy.as_list() if not jobs: print("Tomo queue is empty.") return + runnable = [ + job + for idx, job in enumerate(jobs) + if job["status"] != "done" + and (job["status"] in ("incomplete", "running") or idx >= start_index) + ] + if not runnable: + print("No pending tomo queue jobs to run.") + return + if not self.OMNYTools.yesno( - f"Starting automatic execution of {len(jobs) - start_index} queued tomo scan(s) on" + f"Starting automatic execution of {len(runnable)} queued tomo scan(s) on" f" sample '{self.sample_name}'. OK?", "y", ): print("Aborted.") return - for i in range(start_index, len(jobs)): - job = jobs[i] - if job["status"] == "done": - continue + while True: + jobs = self._tomo_queue_proxy.as_list() + # Resume any in-flight/interrupted job first, before starting any + # fresh one, so the shared progress var is never clobbered. Search + # the whole list (not the start_index slice): a crashed job must be + # recovered wherever it sits. + job = next((j for j in jobs if j["status"] in ("incomplete", "running")), None) + if job is None: + job = next( + ( + j + for idx, j in enumerate(jobs) + if idx >= start_index and j["status"] == "pending" + ), + None, + ) + if job is None: + break + + job_id = job["id"] + label = job["label"] resume_job = job["status"] in ("incomplete", "running") - print(f"\n=== Tomo queue job {i + 1}/{len(jobs)}: {job['label']} ===") + print(f"\n=== Tomo queue job: {label} ===") for name, value in job["params"].items(): setattr(self, name, value) - self._tomo_queue_proxy.update(i, status="running") + self._tomo_queue_proxy.update_by_id(job_id, status="running") try: if resume_job: self.tomo_scan_resume() else: self.tomo_scan() except Exception as exc: - self._tomo_queue_proxy.update(i, status="incomplete") - print(f"Tomo queue job {i} ({job['label']}) did not complete: {exc}") + self._tomo_queue_proxy.update_by_id(job_id, status="incomplete") + print(f"Tomo queue job '{label}' did not complete: {exc}") print( "Queue paused. Fix the issue and call tomo_queue_execute() " "again to resume from this job." ) raise - self._tomo_queue_proxy.update(i, status="done") + self._tomo_queue_proxy.update_by_id(job_id, status="done") print("\nTomo queue finished -- all jobs done.") diff --git a/csaxs_bec/device_configs/simulated_omny/simulated_flomni.yaml b/csaxs_bec/device_configs/simulated_omny/simulated_flomni.yaml index 06b1ce4..7591e75 100644 --- a/csaxs_bec/device_configs/simulated_omny/simulated_flomni.yaml +++ b/csaxs_bec/device_configs/simulated_omny/simulated_flomni.yaml @@ -180,7 +180,7 @@ fsamroy: - 365 port: 8084 sign: -1 - sim_velocity: 25 + sim_velocity: 250 # 10x hardware sim_initial_position: 0 enabled: true onFailure: buffer diff --git a/pyproject.toml b/pyproject.toml index e413433..838ac6d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -38,6 +38,7 @@ dev = [ "pylint", "pytest", "pytest-random-order", + "pytest-bec-e2e", "ophyd_devices", "bec_server", ]