perf(standalone): detect job completion by polling the process handle

src/cloudai/systems/standalone/standalone_job.py

@@ -15,7 +15,8 @@
 # limitations under the License.
 
 import datetime
-from dataclasses import dataclass
+import subprocess
+from dataclasses import dataclass, field
 
 from cloudai.core import BaseJob
 
@@ -26,3 +27,5 @@ class StandaloneJob(BaseJob):
 
     start: datetime.datetime | None = None
     finish: datetime.datetime | None = None
+    process: subprocess.Popen | None = field(default=None, compare=False)
+    """``None`` when this process did not launch the job; completion falls back to a probe."""

src/cloudai/systems/standalone/standalone_runner.py

@@ -78,12 +78,14 @@ def _submit_test(self, tr: TestRun) -> StandaloneJob:
         logging.info(f"Executing command for test {tr.name}: {exec_cmd}")
         job_id = 0
         start = None
+        process = None
         if self.mode == "run":
             start = datetime.datetime.now(datetime.timezone.utc)
-            pid = self.cmd_shell.execute(exec_cmd).pid
+            process = self.cmd_shell.execute(exec_cmd)
+            pid = process.pid
             job_id = pid
             if job_id is None:
                 raise JobIdRetrievalError(
                     test_name=str(tr.name), command=exec_cmd, stdout="", stderr="", message="Failed to retrieve job ID."
                 )
-        return StandaloneJob(tr, id=job_id, start=start)
+        return StandaloneJob(tr, id=job_id, start=start, process=process)

src/cloudai/systems/standalone/standalone_system.py

@@ -15,10 +15,14 @@
 # limitations under the License.
 
 import logging
+import os
+from typing import cast
 
 from cloudai.core import BaseJob, System
 from cloudai.util import CommandShell
 
+from .standalone_job import StandaloneJob
+
 
 class StandaloneSystem(System):
     """
@@ -49,15 +53,42 @@ def is_job_running(self, job: BaseJob) -> bool:
         Returns:
             bool: True if the job is running, False otherwise.
         """
-        command = f"ps -p {job.id}"
-        logging.debug(f"Checking job status with command: {command}")
-        stdout = self.cmd_shell.execute(command).communicate()[0]
-
-        # Check if the job's PID is in the ps output
-        is_running = str(job.id) in stdout
-        logging.debug(f"Job {job.id} running status: {is_running}")
+        # Poll the handle when we own it. Shelling out to `ps` spawns two processes per
+        # check and the check runs on every monitor tick, so on a fast backend it cost more
+        # than the job being waited for. Polling also reaps the child, so a finished process
+        # cannot linger as a zombie and keep reading as running.
+        process = cast(StandaloneJob, job).process
+        if process is not None:
+            is_running = process.poll() is None
+            logging.debug(f"Job {job.id} running status: {is_running}")
+            return is_running
+
+        # No handle: the job was not launched by this process. Probe with signal 0, which
+        # checks for the pid without delivering anything.
+        try:
+            pid = int(job.id)
+        except (TypeError, ValueError):
+            logging.debug(f"Job {job.id} running status: False")
+            return False
+
+        # 0 and negatives are not pids to os.kill: 0 addresses this process group and -1
+        # every process the user may signal, so both would probe successfully and report a
+        # job that does not exist as running.
+        if pid <= 0:
+            logging.debug(f"Job {job.id} running status: False")
+            return False
 
-        return is_running
+        try:
+            os.kill(pid, 0)
+        except ProcessLookupError:
+            logging.debug(f"Job {job.id} running status: False")
+            return False
+        except PermissionError:
+            # The process exists, this user just may not signal it. Reporting it complete
+            # would drop a live job from monitoring.
+            pass
+        logging.debug(f"Job {job.id} running status: True")
+        return True
 
     def is_job_completed(self, job: BaseJob) -> bool:
         """

tests/systems/standalone/test_system.py

@@ -65,30 +65,78 @@ def standalone_job(standalone_system, mock_test):
     return StandaloneJob(mock_test, id=12345)
 
 
-@pytest.mark.parametrize(
-    "ps_output, expected_result",
-    [
-        ("12345\n", True),  # Job is running, PID is in ps output
-        ("", False),  # Job is not running, ps output is empty
-    ],
-)
-@patch("cloudai.util.CommandShell.execute")
-def test_is_job_running(mock_execute, standalone_system, standalone_job, ps_output, expected_result):
+@pytest.mark.parametrize("returncode, expected_result", [(None, True), (0, False), (1, False)])
+def test_is_job_running_polls_the_process_handle(standalone_system, mock_test, returncode, expected_result):
+    """Completion is read from the handle we own, not discovered by running `ps`.
+
+    Replaces a test that mocked ``CommandShell.execute`` and asserted the pid appeared in
+    mocked ``ps`` stdout. That assertion described the old implementation -- it could only
+    pass while completion was detected by shelling out -- so it could not survive this
+    change. ``Popen.poll()`` returns ``None`` while the child runs and its exit status once
+    it has finished.
     """
-    Test if a job is running using a mocked CommandShell.
+    process = MagicMock()
+    process.poll.return_value = returncode
+    job = StandaloneJob(mock_test, id=12345, process=process)
 
-    Args:
-        mock_execute (MagicMock): Mocked CommandShell execute method.
-        standalone_system (StandaloneSystem): Instance of the system under test.
-        standalone_job (StandaloneJob): Job instance to check.
-        ps_output (str): Mocked output of the ps command.
-        expected_result (bool): Expected result for the job running status.
+    assert standalone_system.is_job_running(job) is expected_result
+    process.poll.assert_called()
+
+
+def test_is_job_running_spawns_no_subprocess(standalone_system, mock_test):
+    """The point of the change: a completion check must not cost a process.
+
+    The old path spawned a shell and a ``ps`` on every monitor tick. With the poll interval
+    driven down for a fast backend, that cost more than the job being waited for.
     """
-    mock_process = MagicMock()
-    mock_process.communicate.return_value = (ps_output, "")
-    mock_execute.return_value = mock_process
+    process = MagicMock()
+    process.poll.return_value = None
+    job = StandaloneJob(mock_test, id=12345, process=process)
+
+    with patch("cloudai.util.CommandShell.execute") as mock_execute:
+        assert standalone_system.is_job_running(job) is True
+
+    mock_execute.assert_not_called()
+
+
+@pytest.mark.parametrize("alive, expected_result", [(True, True), (False, False)])
+def test_is_job_running_falls_back_to_a_signal_probe(standalone_system, mock_test, alive, expected_result):
+    """A job with no handle was not launched by this process; probe instead of giving up."""
+    job = StandaloneJob(mock_test, id=12345, process=None)
+
+    def fake_kill(pid: int, sig: int) -> None:
+        assert sig == 0, "probe must not deliver a signal"
+        if not alive:
+            raise ProcessLookupError
+
+    with patch("os.kill", side_effect=fake_kill):
+        assert standalone_system.is_job_running(job) is expected_result
+
+
+def test_is_job_running_handles_a_non_numeric_id(standalone_system, mock_test):
+    """Dry-run and reconstructed jobs can carry an id that is not a pid."""
+    job = StandaloneJob(mock_test, id="not-a-pid", process=None)
+
+    assert standalone_system.is_job_running(job) is False
+
+
+def test_is_job_running_treats_an_unsignalable_process_as_running(standalone_system, mock_test):
+    """EPERM means the process exists; calling it complete would drop a live job."""
+    job = StandaloneJob(mock_test, id=12345, process=None)
+
+    with patch("os.kill", side_effect=PermissionError):
+        assert standalone_system.is_job_running(job) is True
+
+
+@pytest.mark.parametrize("job_id", [0, -1])
+def test_is_job_running_rejects_non_positive_ids_without_probing(standalone_system, mock_test, job_id):
+    """os.kill would address a process group for these, and report success."""
+    job = StandaloneJob(mock_test, id=job_id, process=None)
+
+    with patch("os.kill") as mock_kill:
+        assert standalone_system.is_job_running(job) is False
 
-    assert standalone_system.is_job_running(standalone_job) == expected_result
+    mock_kill.assert_not_called()
 
 
 @patch("cloudai.util.CommandShell.execute")