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")