Avoid out-of-order task execution (#164)

* attempt to enforce order in task execution

* sad oink

* one more test fix

* make test more robust 2 -> 3

* restrict queue to 1 task.

* MAX_TASK_WAIT

* bugfixes

* snapshot_interval & docs

Author: PeterKraus

docs/source/version.rst

@@ -1,5 +1,29 @@
 Version history
 ===============
+**tomato**-v2.2
+---------------
+..
+    .. image:: https://img.shields.io/static/v1?label=tomato&message=v2.1&color=blue&logo=github
+        :target: https://github.com/dgbowl/tomato/tree/2.1
+    .. image:: https://img.shields.io/static/v1?label=tomato&message=v2.1&color=blue&logo=pypi
+        :target: https://pypi.org/project/tomato/2.1/
+    .. image:: https://img.shields.io/static/v1?label=release%20date&message=2025-06-07&color=red&logo=pypi
+
+.. sectionauthor::
+     Peter Kraus
+
+Developed at the ConCat lab at TU Berlin.
+
+Changes from ``tomato-2.1`` include:
+
+- Fixes many bugs due to the :func:`cmp_measure` function race condition with running tasks.
+- Introduces the "lazy pirate" pattern in ``tomato-job`` processes, which should make jobs more reliable.
+
+- A new ``Payload-2.2``, where the ``settings.snapshot.snapshot_interval`` replaces ``settings.snapshot.frequency``. As a consequence, the ``snapshot_interval`` can be provided as :class:`str`, which will be converted to the number of seconds using :mod:`pint`.
+
+.. codeauthor::
+    Peter Kraus
+
 **tomato**-v2.1
 ---------------
 .. image:: https://img.shields.io/static/v1?label=tomato&message=v2.1&color=blue&logo=github

pyproject.toml

@@ -30,7 +30,7 @@ dependencies = [
     "pyyaml >= 6.0",
     "psutil >= 5.9",
     # "dgbowl_schemas >= 124",
-    "dgbowl_schemas @ git+https://github.com/dgbowl/dgbowl-schemas.git#egg=payload_2.2", 
+    "dgbowl_schemas @ git+https://github.com/dgbowl/dgbowl-schemas.git@payload_2.2",
     "pyzmq >= 25.1",
     "netcdf4 >= 1.7",
     "xarray >= 2024.10.0",

src/tomato/daemon/job.py

@@ -36,6 +36,7 @@
 logger = logging.getLogger(__name__)
 
 MAX_JOB_NOPID = timedelta(seconds=10)
+MAX_TASK_WAIT = 10
 
 
 def method_validate(
@@ -479,7 +480,7 @@ def tomato_job() -> None:
 
 
 def job_thread(
-    tasks: list,
+    tasks: list[Task],
     component: Component,
     device: Device,
     driver: Driver,
@@ -518,23 +519,26 @@ def job_thread(
             taskid += f":{task.task_name!r}"
         thread.current_task = task
         logger.info("%s: processing task", taskid)
+
+        # Hold while start contidions are not met
         while True:
-            time.sleep(1e-1)
             if task.start_with_task_name is None:
-                pass
+                break
             elif task.start_with_task_name in thread.started_task_names:
-                pass
+                break
             else:
                 logger.debug(
                     "%s: waiting for task_name '%s'", taskid, task.start_with_task_name
                 )
-                continue
+                time.sleep(0.1)
+
+        # Hold while component task_list is not ready
+        while True:
             logger.debug(
                 "%s: polling component %s for task readiness", taskid, component.name
             )
-            ret, req = lpp.comm(
-                req, dict(cmd="task_status", params={**kwargs}), **lppargs
-            )
+            msg = dict(cmd="task_status", params={**kwargs})
+            ret, req = lpp.comm(req, msg, **lppargs)
             if ret.success and ret.data["can_submit"]:
                 break
             elif req.closed:
@@ -543,18 +547,56 @@ def job_thread(
             logger.warning(
                 "%s: cannot submit onto component %s, waiting", taskid, component.name
             )
+            time.sleep(0.1)
 
+        # Send task to component
         logger.info("%s: sending task to component %s", taskid, component.name)
+        t0 = time.perf_counter()
         msg = dict(cmd="task_start", params={"task": task, **kwargs})
         ret, req = lpp.comm(req, msg, **lppargs)
         if req.closed:
             thread.crashed = True
             sys.exit()
 
-        t0 = time.perf_counter()
+        # Wait until the correct task is running, or MAX_TASK_WAIT
+        while True:
+            dt = time.perf_counter() - t0
+            msg = dict(cmd="task_status", params={**kwargs})
+            ret, req = lpp.comm(req, msg, **lppargs)
+            if req.closed:
+                thread.crashed = True
+                sys.exit()
+            elif ret.success and ret.data["running"] is False:
+                logger.warning(
+                    "%s: task submitted %f s ago but not yet running", taskid, dt
+                )
+                pass
+            elif ret.success and "task" in ret.data and ret.data["task"] != task:
+                logger.warning(
+                    "%s: task submitted %f s ago but other task running: %s",
+                    taskid,
+                    dt,
+                    ret.data["task"],
+                )
+                pass
+            elif ret.success and "task" in ret.data and ret.data["task"] == task:
+                break
+            elif ret.success and "task" not in ret.data:
+                break
+            if dt > MAX_TASK_WAIT:
+                logger.critical("%s: task was submitted, but is not executed, aborting")
+                thread.crashed = True
+                sys.exit()
+            time.sleep(0.1)
+        logger.info("%s: correct task running on component %s", taskid, component.role)
+
+        # Main task loop
+        tP = time.perf_counter()
         while True:
             tN = time.perf_counter()
-            if tN - t0 > device.pollrate:
+
+            # Poll for data every device.pollrate, save to pickle
+            if tN - tP > device.pollrate:
                 logger.debug("%s: polling task for data", taskid)
                 msg = dict(cmd="task_data", params={**kwargs})
                 ret, req = lpp.comm(req, msg, **lppargs, timeout=5000)
@@ -566,8 +608,9 @@ def job_thread(
                     ds: xr.Dataset = ret.data
                     ds.attrs["tomato_Component"] = component.model_dump_json()
                     data_to_pickle(ds, datapath, role=component.role)
-                t0 += device.pollrate
+                tP += device.pollrate
 
+            # Poll for completion and correct task status
             logger.debug("%s: polling task for completion", taskid)
             msg = dict(cmd="task_status", params={**kwargs})
             ret, req = lpp.comm(req, msg, **lppargs)
@@ -577,10 +620,16 @@ def job_thread(
             elif ret.success and not ret.data["running"]:
                 logger.info("%s: task no longer running, break", taskid)
                 break
+            elif ret.success and "task" in ret.data and ret.data["task"] != task:
+                logger.critical("%s: wront task running, break", taskid)
+                logger.debug("%s: expected task: %s", taskid, task)
+                logger.debug("%s: executed task: %s", taskid, ret.data["task"])
+                break
             elif ret.success is False:
                 logger.critical(f"{ret=}")
                 break
 
+            # Stop task if stop trigger condition met, save to pickle
             if (
                 task.stop_with_task_name is not None
                 and task.stop_with_task_name in thread.started_task_names
@@ -598,7 +647,9 @@ def job_thread(
                     data_to_pickle(ds, datapath, role=component.role)
                 break
 
-            time.sleep(max(1e-1, (device.pollrate - (tN - t0)) / 2))
+            time.sleep(max(1e-1, (device.pollrate - (tN - tP)) / 2))
+
+        # Store final task data, housekeeping.
         logger.info("%s: task fetching final data", taskid)
         msg = dict(cmd="task_data", params={**kwargs})
         ret, req = lpp.comm(req, msg, **lppargs, timeout=5000)
@@ -613,6 +664,7 @@ def job_thread(
         thread.completed_tasks.append(task)
         thread.current_task = None
 
+    # Reset component at the end of the job
     logger.info(
         "%s: all tasks done on component %s, resetting", component.role, component.name
     )
@@ -705,10 +757,10 @@ def job_main_loop(
     started_task_names = set()
     while True:
         tN = time.perf_counter()
-        if snapshot is not None and tN - t0 > snapshot.frequency:
+        if snapshot is not None and tN - t0 > snapshot.snapshot_interval:
             logger.debug("creating snapshot")
             merge_netcdfs(job, snapshot=True)
-            t0 += snapshot.frequency
+            t0 += snapshot.snapshot_interval
 
         # Collect and push task names
         for t in threads.values():

src/tomato/driverinterface_2_1/__init__.py

@@ -379,11 +379,11 @@ def task_status(self, key: Key, **kwargs: dict) -> tuple[bool, str, dict]:
         """
         running = self.devmap[key].running
         can_submit = not self.devmap[key].task_list.full()
-        data = dict(running=running, can_submit=can_submit)
-        if running:
-            return (True, "component has a running task", data)
-        else:
+        data = dict(running=bool(running), can_submit=can_submit, task=running)
+        if running is False:
             return (True, "component is idle", data)
+        else:
+            return (True, "component has a running task", data)
 
     @log_errors
     @to_reply
@@ -589,9 +589,9 @@ def task_runner(self) -> None:
                 thread.do_run = False
                 break
 
-            self.running = True
-            if isinstance(task, Task):
-                try:
+            self.running = task
+            try:
+                if isinstance(task, Task):
                     thread.do_run_task = True
                     self.prepare_task(task=task)
                     t_0 = time.perf_counter()
@@ -614,28 +614,18 @@ def task_runner(self) -> None:
                         task.technique_name,
                         self.key,
                     )
-                except Exception as e:
-                    logger.critical(e, exc_info=True)
-                    thread.do_run = False
-                    break
-            elif task == "measure":
-                try:
+                elif task == "measure":
                     self.do_measure()
                     logger.debug("measurement on component %s is done", self.key)
-                except Exception as e:
-                    logger.critical(e, exc_info=True)
+                else:
+                    logger.critical("Unknown task received: '%s'", task)
                     thread.do_run = False
                     break
-            else:
-                logger.critical("Unknown task received: '%s'", task)
-                thread.do_run = False
-                break
-
-            try:
                 self.task_list.task_done()
-            except ValueError as e:
+            except Exception as e:
                 logger.critical(e, exc_info=True)
-                logger.critical("above error raised on task '%s'", task)
+                thread.do_run = False
+                break
             self.running = False
         logger.warning("task runner is quitting")
         self.running = False

tests/common/psutil_counter_with_task_stop.yml

@@ -14,7 +14,7 @@ method:
   - component_role: "counter"
     technique_name: "random"
     max_duration: 20.0
-    sampling_interval: 1.5
+    sampling_interval: 1.0
     task_params:
         min: 50.0
         max: 100.0

tests/test_05_passata.py

@@ -96,7 +96,7 @@ def test_passata_api_reset_force(datadir, start_tomato_daemon, stop_tomato_daemo
     )
     print(f"{ret=}")
     assert ret.success
-    assert ret.data["running"] is True
+    assert ret.data["running"]
 
     ret = tomato.passata.reset(
         name="example_counter:(example-addr,1)",

tests/test_99_psutil.py

@@ -19,7 +19,7 @@
         ("psutil_1_0.1", {"psutil": 10}),
         ("psutil_counter", {"psutil": 12, "counter": 10}),
         ("psutil_counter_with_task_start", {"psutil": 12, "counter": 10}),
-        ("psutil_counter_with_task_stop", {"psutil": 10, "counter": 2}),
+        ("psutil_counter_with_task_stop", {"psutil": 10, "counter": 3}),
     ],
 )
 def test_psutil_multidev(casename, npoints, datadir, stop_tomato_daemon):