Get rid of crashed jobs from pipelines

pyproject.toml

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

src/tomato/daemon/io.py

@@ -23,7 +23,7 @@ def store(daemon: Daemon):
     outfile = datadir / f"tomato_state_{daemon.port}.pkl"
     logger.debug("storing daemon state to '%s'", outfile)
     with outfile.open("wb") as out:
-        pickle.dump(daemon, out, protocol=5)
+        pickle.dump(daemon, out, protocol=4)
 
 
 def load(daemon: Daemon):
@@ -82,9 +82,9 @@ def data_to_pickle(ds: xr.Dataset, path: Path, role: str):
     if path.exists():
         with path.open("rb") as old:
             oldds = pickle.load(old)
-            if oldds is not None:
-                logger.debug("concatenating Dataset with existing data")
-                ds = xr.concat([oldds, ds], dim="uts")
+        if oldds is not None:
+            logger.debug("concatenating Dataset with existing data")
+            ds = xr.concat([oldds, ds], dim="uts")
     logger.debug("dumping Dataset into pickle at '%s'", path)
     with path.open("wb") as out:
-        pickle.dump(ds, out, protocol=5)
+        pickle.dump(ds, out, protocol=4)

src/tomato/daemon/job.py

@@ -19,7 +19,7 @@
 import time
 import argparse
 from importlib import metadata
-from datetime import datetime, timezone
+from datetime import datetime, timezone, timedelta
 from pathlib import Path
 from threading import current_thread, Thread
 import zmq
@@ -35,6 +35,8 @@
 
 logger = logging.getLogger(__name__)
 
+MAX_JOB_NOPID = timedelta(seconds=10)
+
 
 def method_validate(
     method: list[Task],
@@ -143,38 +145,56 @@ def manage_running_pips(pips: dict, dbpath: str, req: zmq.Socket):
     logger.debug(f"{running=}")
     for pip in running:
         job = jobdb.get_job_id(pip.jobid, dbpath)
-        if job.pid is None and job.status in {"qw"}:
-            continue
+
+        if job.pid is None and job.connected_at is not None:
+            # pid is set in the same command as connected_at
+            # unclear how we'd end here
+            logger.error("job status shouldn't be possible: %s", job)
+            pidexists = False
+        elif job.pid is None and job.launched_at is not None:
+            # subprocess was started but job is not (yet) connected
+            td = datetime.now(timezone.utc) - datetime.fromisoformat(job.launched_at)
+            if td > MAX_JOB_NOPID:
+                logger.error("job %d failed to register, aborting", job.id)
+                job.status = "rd"
+                pidexists = False
+            else:
+                continue
         elif job.pid is None:
-            logger.error("we shouldn't be here:")
-            logger.error(f"{pip=}")
-            logger.error(f"{job=}")
+            # subprocess was not yet started
+            logger.warning("job %d failed to start", job.id)
+            # TODO: timeout to be implemented
             continue
-        pidexists = psutil.pid_exists(job.pid)
+        else:
+            pidexists = psutil.pid_exists(job.pid)
         if pidexists:
             pidexists = psutil.Process(job.pid).status() is not psutil.STATUS_ZOMBIE
-        logger.debug(f"{pidexists=}")
+
         reset = False
         update = False
         ready = False
         # running jobs scheduled for killing (status == 'rd') should be killed
-        if pidexists and job.status == "rd":
-            logger.info(f"job {job.id} with pid {job.pid} will be terminated")
-            proc = psutil.Process(pid=job.pid)
-            kill_tomato_job(proc)
-            logger.info(f"job {job.id} with pid {job.pid} was terminated successfully")
-            merge_netcdfs(job)
-            reset = True
+        # jobs that have status == 'rd' but no valid pid should be cleared
+        if job.status == "rd":
+            if pidexists:
+                logger.info(f"job {job.id} with pid {job.pid} will be terminated")
+                proc = psutil.Process(pid=job.pid)
+                kill_tomato_job(proc)
+                logger.info(
+                    f"job {job.id} with pid {job.pid} was terminated successfully"
+                )
+                merge_netcdfs(job)
             update = True
             params = dict(status="cd")
+            reset = True
         # dead jobs marked as running (status == 'r') should be cleared
         elif (not pidexists) and job.status == "r":
             logger.warning(f"the pid {job.pid} of running job {job.id} was not found")
             reset = True
             update = True
             params = dict(status="ce")
         # crashed jobs marked as such (status == 'ce') should also be cleared
-        elif (not pidexists) and job.status == "ce":
+        elif (not pidexists) and job.status in {"ce", "cd"}:
             logger.info(f"the pid {job.pid} of crashed job {job.id} was not found")
             reset = True
         # pipelines of completed jobs should be reset
@@ -229,7 +249,7 @@ def check_queued_jobs(
     return matched
 
 
-def action_queued_jobs(daemon, matched, req):
+def action_queued_jobs(daemon, matched, req, dbpath):
     """
     Function that assigns jobs if a matched pipeline contains the requested sample.
 
@@ -243,30 +263,34 @@ def action_queued_jobs(daemon, matched, req):
                 continue
             elif pip.sampleid != job.payload.sample.name:
                 continue
-            logger.info(f"job {job.id} found a matched & ready pip: {pip.name!r}")
-            params = dict(jobid=job.id, ready=False, name=pip.name)
-            req.send_pyobj(dict(cmd="pipeline", params=params))
-            ret = req.recv_pyobj()
-            if not ret.success:
-                logger.error(f"could not set params {params} on pip: {pip.name!r}")
-                continue
-            else:
-                pip.ready = False
+            logger.info("job %d: found a matched & ready pip '%s'", job.id, pip.name)
 
+            logger.debug("job %d: making job directory", job.id)
             root = Path(daemon.settings["jobs"]["storage"]) / str(job.id)
             os.makedirs(root)
 
+            logger.debug("job %d: storing jobdata.json", job.id)
             jpath = root / "jobdata.json"
             jobargs = {
                 "pipeline": pip.model_dump(),
                 "payload": job.payload.model_dump(),
                 "devices": {dn: dev.model_dump() for dn, dev in daemon.devs.items()},
                 "job": dict(id=job.id, path=str(root)),
             }
-
             with jpath.open("w", encoding="UTF-8") as of:
                 json.dump(jobargs, of, indent=1)
 
+            logger.debug("job %d: reserving pipeline %s", job.id, pip.name)
+            params = dict(jobid=job.id, ready=False, name=pip.name)
+            req.send_pyobj(dict(cmd="pipeline", params=params))
+            ret = req.recv_pyobj()
+            if not ret.success:
+                logger.error("job %d: could not set params %s", job.id, params)
+                continue
+            else:
+                pip.ready = False
+
+            logger.debug("job %d: executing tomato-job", job.id)
             cmd = [
                 "tomato-job",
                 "--port",
@@ -283,7 +307,13 @@ def action_queued_jobs(daemon, matched, req):
                 subprocess.Popen(cmd, creationflags=cfs)
             elif psutil.POSIX:
                 subprocess.Popen(cmd, start_new_session=True)
-            logger.info(f"job {jobid} started on pip: {pip.name!r} and path: {jpath!r}")
+
+            logger.debug("job %d: setting launched_at")
+            params = dict(launched_at=str(datetime.now(timezone.utc)))
+            job = jobdb.update_job_id(jobid, params, dbpath)
+            logger.info(
+                "job %d: launched on pip '%s' and path '%s'", job.id, pip.name, jpath
+            )
             break
 
 
@@ -317,7 +347,7 @@ def manager(port: int, timeout: int = 500):
         matched_pips = check_queued_jobs(
             daemon.pips, daemon.cmps, daemon.drvs, dbpath, context
         )
-        action_queued_jobs(daemon, matched_pips, req)
+        action_queued_jobs(daemon, matched_pips, req, dbpath)
         time.sleep(timeout / 1e3)
     req.close()
     logger.info("instructed to quit")
@@ -411,7 +441,7 @@ def tomato_job() -> None:
     logger.info(f"assigning job {jobid} with pid {pid} into pipeline {pip!r}")
     context = zmq.Context()
 
-    params = dict(pid=pid, status="r", executed_at=str(datetime.now(timezone.utc)))
+    params = dict(pid=pid, status="r", connected_at=str(datetime.now(timezone.utc)))
     job = jobdb.update_job_id(jobid, params, args.dbpath)
 
     output = payload.settings.output

src/tomato/daemon/jobdb.py

@@ -26,7 +26,7 @@ def connect_jobdb(dbpath: str):
 
 
 def jobdb_setup(dbpath: str) -> None:
-    user_version = 1
+    user_version = 2
     conn, cur = connect_jobdb(dbpath)
     logger.debug("attempting to find table 'queue' in '%s'", dbpath)
     cur.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='queue';")
@@ -37,14 +37,18 @@ def jobdb_setup(dbpath: str) -> None:
         curr_version = cur.fetchone()[0]
         assert curr_version == user_version
         # Below is an example of upgrading databases to new user_version:
-        # while curr_version < user_version:
-        #    if curr_version == 0:
-        #        log.info("upgrading table 'queue' from version 0 to 1")
-        #        cur.execute("ALTER TABLE queue ADD COLUMN jobname TEXT;")
-        #        cur.execute("PRAGMA user_version = 1;")
-        #        conn.commit()
-        #    cur.execute("PRAGMA user_version;")
-        #    curr_version = cur.fetchone()[0]
+        while curr_version < user_version:
+            if curr_version == 1:
+                logger.info("upgrading table 'queue' from version 1 to 2")
+                cur.execute(
+                    "ALTER TABLE queue RENAME COLUMN executed_at TO connected_at;"
+                )
+                cur.execute("ALTER TABLE queue ADD COLUMN launched_at TEXT;")
+                cur.execute("UPDATE queue SET launched_at = connected_at;")
+                cur.execute("PRAGMA user_version = 2;")
+                conn.commit()
+            cur.execute("PRAGMA user_version;")
+            curr_version = cur.fetchone()[0]
     else:
         logger.info("creating a new sqlite3 'queue' table at '%s'", dbpath)
         cur.execute(
@@ -55,7 +59,8 @@ def jobdb_setup(dbpath: str) -> None:
             "    pid INTEGER,"
             "    status TEXT NOT NULL,"
             "    submitted_at TEXT NOT NULL,"
-            "    executed_at TEXT,"
+            "    launched_at TEXT,"
+            "    connected_at TEXT,"
             "    completed_at TEXT,"
             "    jobpath TEXT,"
             "    respath TEXT,"
@@ -71,15 +76,16 @@ def insert_job(job: Job, dbpath: str) -> int:
     conn, cur = connect_jobdb(dbpath)
     cur.execute(
         "INSERT INTO queue (payload, jobname, pid, status, submitted_at, "
-        "executed_at, completed_at, jobpath, respath, snappath)"
-        "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?);",
+        "launched_at, connected_at, completed_at, jobpath, respath, snappath)"
+        "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?);",
         (
             pickle.dumps(job.payload),
             job.jobname,
             job.pid,
             job.status,
             job.submitted_at,
-            job.executed_at,
+            job.launched_at,
+            job.connected_at,
             job.completed_at,
             job.jobpath,
             job.respath,

src/tomato/driverinterface_2_1/__init__.py

@@ -339,6 +339,8 @@ def cmp_measure(self, key: Key, **kwargs: dict) -> tuple[bool, str, None]:
         """
         if self.devmap[key].running:
             return (False, f"measurement already running on component {key!r}", None)
+        elif not self.devmap[key].task_list.empty():
+            return (False, f"task list component {key!r} not empty", None)
         else:
             self.devmap[key].task_list.put("measure")
             return (True, f"measurement started on component {key!r}", None)
@@ -397,15 +399,9 @@ def task_stop(
         If there is any cached data, it is returned as a :class:`xarray.Dataset` in the
         :obj:`Reply.data` and the cache is cleared.
         """
-        ret = self.devmap[key].stop_task(**kwargs)
-        if ret is not None:
-            return (False, "failed to stop task", ret)
-        else:
-            ret = self.task_data(key=key)
-            if ret.success:
-                return (True, f"task stopped, {ret.msg}", ret.data)
-            else:
-                return (True, f"task stopped, {ret.msg}", None)
+        self.devmap[key].stop_task(**kwargs)
+        ret = self.task_data(key=key)
+        return (True, f"task stopped, {ret.msg}", ret.data)
 
     @log_errors
     @to_reply

src/tomato/ketchup/__init__.py

@@ -36,7 +36,7 @@
 
 log = logging.getLogger(__name__)
 
-__latest_payload__ = "2.1"
+__latest_payload__ = "2.2"
 
 
 def submit(
@@ -70,7 +70,8 @@ def submit(
     >>> ketchup submit counter_15_0.1.yml -y
     data:
         completed_at: null
-        executed_at: null
+        launched_at: null
+        connected_at: null
         id: 1
         [...]
         status: q
@@ -166,7 +167,8 @@ def status(
     >>> ketchup status 1 -y
     data:
       - completed_at: null
-        executed_at: null
+        launched_at: null
+        connected_at: null
         id: 1
         [...]
         status: qw
@@ -241,7 +243,8 @@ def cancel(
     >>> ketchup cancel 2 -y
     data:
       - completed_at: null
-        executed_at: null
+        launched_at: null
+        connected_at: null
         id: 2
         [...]
         status: cd

src/tomato/models.py

@@ -88,7 +88,8 @@ class Job(BaseModel):
     pid: Optional[int] = None
     status: Literal["q", "qw", "r", "rd", "c", "cd", "ce"] = "q"
     submitted_at: Optional[str] = None
-    executed_at: Optional[str] = None
+    launched_at: Optional[str] = None
+    connected_at: Optional[str] = None
     completed_at: Optional[str] = None
     jobpath: Optional[str] = None
     respath: Optional[str] = None

tests/test_99_example_counter.py

@@ -7,6 +7,7 @@
 import tomato
 import zmq
 import time
+import pickle
 
 from . import utils
 
@@ -35,6 +36,10 @@ def test_counter_npoints_metadata(
     files = os.listdir(os.path.join(".", "Jobs", "1"))
     assert "jobdata.json" in files
     assert "job-1.log" in files
+    with open(os.path.join(".", "Jobs", "1", "counter.pkl"), "rb") as inp:
+        ds = pickle.load(inp)
+        print(f"{ds=}")
+        assert ds["uts"].size == npoints
     if prefix is not None:
         utils.check_npoints_file(f"{prefix}.nc", {"counter": npoints})
 
@@ -108,6 +113,11 @@ def test_counter_multidev(casename, npoints, datadir, stop_tomato_daemon):
     assert "jobdata.json" in files
     assert "job-1.log" in files
     assert os.path.exists("results.1.nc")
+    for k, v in npoints.items():
+        with open(os.path.join(".", "Jobs", "1", f"{k}.pkl"), "rb") as inp:
+            ds = pickle.load(inp)
+        print(f"{ds=}")
+        assert ds["uts"].size == v
     utils.check_npoints_file("results.1.nc", npoints)