From 292f11b40c16a0942974d7b8242f442f2b974d2f Mon Sep 17 00:00:00 2001 From: Rutayan Patro Date: Tue, 22 Sep 2026 20:38:32 +0000 Subject: [PATCH] perf(standalone): detect job completion by polling the process handle is_job_running shelled out to `ps -p ` and matched the pid as a substring of the output. The runner already owns the Popen it launched and then discarded it, keeping only the pid, so the handle had to be rediscovered externally on every check. That check runs on every monitor tick. Each one spawns two processes, a shell and a ps, costing about 10 ms; measured on a proxy-model DSE run it fired roughly seven times per trial against ~25 ms of actual work. Polling the handle is a single syscall. StandaloneJob now carries the Popen for the process it represents, which is where the handle belongs: its lifetime is the job's, so there is no registry to keep in step and nothing to leak. The runner passes the object it was already creating. Two secondary defects go with it. `str(job.id) in stdout` matches the pid anywhere in ps output, including inside another column, so an unrelated process could read as the job still running. And a finished-but-unreaped child stays visible to ps as a zombie, which reads as running forever; poll() reaps it. A job with no handle -- dry run, or reconstructed from prior state -- falls back to probing with signal 0, which tests for the pid without delivering anything, and treats a non-numeric id as not running. Replaces test_is_job_running, which mocked CommandShell.execute and asserted the pid appeared in mocked ps stdout. That assertion described the implementation being removed and could not survive the change; it is substituted by tests that poll the handle, assert no subprocess is spawned, cover the fallback probe, and cover a non-numeric id. kill() still uses the shell and is left alone; routing it through the handle is a separate change. --- .../systems/standalone/standalone_job.py | 5 +- .../systems/standalone/standalone_runner.py | 6 +- .../systems/standalone/standalone_system.py | 47 ++++++++-- tests/systems/standalone/test_system.py | 88 ++++++++++++++----- 4 files changed, 115 insertions(+), 31 deletions(-) diff --git a/src/cloudai/systems/standalone/standalone_job.py b/src/cloudai/systems/standalone/standalone_job.py index 66e3a3d63..1ed121788 100644 --- a/src/cloudai/systems/standalone/standalone_job.py +++ b/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.""" diff --git a/src/cloudai/systems/standalone/standalone_runner.py b/src/cloudai/systems/standalone/standalone_runner.py index f8a0df2a7..13a3b2f63 100644 --- a/src/cloudai/systems/standalone/standalone_runner.py +++ b/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) diff --git a/src/cloudai/systems/standalone/standalone_system.py b/src/cloudai/systems/standalone/standalone_system.py index dca505f93..90797a856 100644 --- a/src/cloudai/systems/standalone/standalone_system.py +++ b/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: """ diff --git a/tests/systems/standalone/test_system.py b/tests/systems/standalone/test_system.py index 1b18ca707..502e53e3a 100644 --- a/tests/systems/standalone/test_system.py +++ b/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")