From 5b46a2f7b3d1f1b26b06dc99e2278a5553ed5bc0 Mon Sep 17 00:00:00 2001 From: Yudong An Date: Mon, 11 May 2026 15:25:01 +0800 Subject: [PATCH] Fix cron scheduler on Windows --- src/openharness/services/cron_scheduler.py | 187 +++++++++++++++++---- tests/test_services/test_cron_scheduler.py | 42 +++++ 2 files changed, 197 insertions(+), 32 deletions(-) diff --git a/src/openharness/services/cron_scheduler.py b/src/openharness/services/cron_scheduler.py index 3c8d361..3b0a481 100644 --- a/src/openharness/services/cron_scheduler.py +++ b/src/openharness/services/cron_scheduler.py @@ -9,18 +9,22 @@ log. from __future__ import annotations import asyncio +import contextlib import json import logging import os +import subprocess import shlex import signal import sys import time from datetime import datetime, timezone from pathlib import Path -from typing import Any +from types import FrameType +from typing import Any, Callable from openharness.config.paths import get_data_dir, get_logs_dir +from openharness.platforms import get_platform from openharness.services.cron import ( load_cron_jobs, mark_job_run, @@ -98,10 +102,7 @@ def read_pid() -> int | None: pid = int(path.read_text(encoding="utf-8").strip()) except (ValueError, OSError): return None - # Check if process is alive - try: - os.kill(pid, 0) - except OSError: + if not _pid_exists(pid): logger.debug("Removed stale scheduler PID file (pid=%d)", pid) path.unlink(missing_ok=True) return None @@ -131,27 +132,94 @@ def stop_scheduler() -> bool: if pid is None: return False try: - os.kill(pid, signal.SIGTERM) + _terminate_pid(pid) except OSError: remove_pid() return False # Wait briefly for process to exit for _ in range(10): - try: - os.kill(pid, 0) - except OSError: + if not _pid_exists(pid): remove_pid() return True time.sleep(0.2) # Force kill try: - os.kill(pid, signal.SIGKILL) + _kill_pid(pid) except OSError: pass remove_pid() return True +def _pid_exists(pid: int) -> bool: + """Return True when *pid* currently refers to a live process.""" + if pid <= 0: + return False + if get_platform() == "windows": + return _windows_pid_exists(pid) + try: + os.kill(pid, 0) + except ProcessLookupError: + return False + except PermissionError: + return True + except OSError: + return False + return True + + +def _windows_pid_exists(pid: int) -> bool: + """Windows implementation of ``kill(pid, 0)`` without requiring psutil.""" + try: + import ctypes + except Exception: + return _pid_exists_with_kill_zero(pid) + + try: + kernel32 = ctypes.WinDLL("kernel32", use_last_error=True) # type: ignore[attr-defined] + except AttributeError: + return _pid_exists_with_kill_zero(pid) + + synchronize = 0x00100000 + process_query_limited_information = 0x1000 + wait_timeout = 0x00000102 + + handle = kernel32.OpenProcess( + synchronize | process_query_limited_information, + False, + pid, + ) + if not handle: + return ctypes.get_last_error() == 5 # ERROR_ACCESS_DENIED: process exists + try: + return kernel32.WaitForSingleObject(handle, 0) == wait_timeout + finally: + kernel32.CloseHandle(handle) + + +def _pid_exists_with_kill_zero(pid: int) -> bool: + try: + os.kill(pid, 0) + except ProcessLookupError: + return False + except PermissionError: + return True + except OSError: + return False + return True + + +def _terminate_pid(pid: int) -> None: + os.kill(pid, signal.SIGTERM) + + +def _kill_pid(pid: int) -> None: + if get_platform() == "windows": + os.kill(pid, signal.SIGTERM) + return + os.kill(pid, signal.SIGKILL) + + # --------------------------------------------------------------------------- # Job execution # --------------------------------------------------------------------------- @@ -376,13 +444,8 @@ async def run_scheduler_loop(*, once: bool = False) -> None: """Main scheduler loop. Runs until SIGTERM or *once* is True (test mode).""" shutdown = asyncio.Event() - def _on_signal() -> None: - logger.info("Received shutdown signal") - shutdown.set() - loop = asyncio.get_running_loop() - for sig in (signal.SIGTERM, signal.SIGINT): - loop.add_signal_handler(sig, _on_signal) + restore_signals = _install_shutdown_signal_handlers(loop, shutdown) write_pid() logger.info("Cron scheduler started (pid=%d, tick=%ds)", os.getpid(), TICK_INTERVAL_SECONDS) @@ -411,10 +474,45 @@ async def run_scheduler_loop(*, once: bool = False) -> None: except asyncio.TimeoutError: pass finally: + restore_signals() remove_pid() logger.info("Cron scheduler stopped") +def _install_shutdown_signal_handlers( + loop: asyncio.AbstractEventLoop, + shutdown: asyncio.Event, +) -> Callable[[], None]: + """Install portable signal handlers and return a restore callback.""" + previous_handlers: list[tuple[signal.Signals, Any]] = [] + + def _on_signal(signum: int, frame: FrameType | None) -> None: + del frame + logger.info("Received shutdown signal (%s)", signum) + with contextlib.suppress(RuntimeError): + loop.call_soon_threadsafe(shutdown.set) + + signals: list[signal.Signals] = [signal.SIGTERM, signal.SIGINT] + sigbreak = getattr(signal, "SIGBREAK", None) + if sigbreak is not None: + signals.append(sigbreak) + + for sig in signals: + try: + previous = signal.getsignal(sig) + signal.signal(sig, _on_signal) + except (OSError, RuntimeError, ValueError): + continue + previous_handlers.append((sig, previous)) + + def _restore() -> None: + for sig, previous in reversed(previous_handlers): + with contextlib.suppress(OSError, RuntimeError, ValueError): + signal.signal(sig, previous) + + return _restore + + # --------------------------------------------------------------------------- # Daemon entry point (spawned by ``oh cron start``) # --------------------------------------------------------------------------- @@ -432,28 +530,49 @@ def _run_daemon() -> None: def start_daemon() -> int: - """Fork and start the scheduler daemon. Returns the child PID.""" + """Start the scheduler daemon and return its PID.""" existing = read_pid() if existing is not None: raise RuntimeError(f"Scheduler already running (pid={existing})") - pid = os.fork() - if pid > 0: - # Parent — wait a moment for the child to write its PID file - time.sleep(0.3) - return pid + process = _spawn_scheduler_process() + deadline = time.monotonic() + 2.0 + while time.monotonic() < deadline: + pid = read_pid() + if pid is not None: + return pid + if process.poll() is not None: + break + time.sleep(0.1) - # Child — detach - os.setsid() - # Redirect stdio - devnull = os.open(os.devnull, os.O_RDWR) - os.dup2(devnull, 0) - os.dup2(devnull, 1) - os.dup2(devnull, 2) - os.close(devnull) + if process.poll() is not None: + log_file = get_logs_dir() / "cron_scheduler.log" + raise RuntimeError(f"Cron scheduler failed to start; see {log_file}") + return process.pid - _run_daemon() - sys.exit(0) + +def _spawn_scheduler_process() -> subprocess.Popen[bytes]: + """Spawn a detached scheduler subprocess on Unix and Windows.""" + popen_kwargs: dict[str, Any] = { + "stdin": subprocess.DEVNULL, + "stdout": subprocess.DEVNULL, + "stderr": subprocess.DEVNULL, + "close_fds": True, + } + if get_platform() == "windows": + creationflags = getattr(subprocess, "DETACHED_PROCESS", 0) | getattr( + subprocess, + "CREATE_NEW_PROCESS_GROUP", + 0, + ) + if creationflags: + popen_kwargs["creationflags"] = creationflags + else: + popen_kwargs["start_new_session"] = True + return subprocess.Popen( + [sys.executable, "-m", "openharness.services.cron_scheduler"], + **popen_kwargs, + ) def scheduler_status() -> dict[str, Any]: @@ -470,3 +589,7 @@ def scheduler_status() -> dict[str, Any]: "log_file": str(log_path), "history_file": str(get_history_path()), } + + +if __name__ == "__main__": + _run_daemon() diff --git a/tests/test_services/test_cron_scheduler.py b/tests/test_services/test_cron_scheduler.py index 8d0cf90..c5586f1 100644 --- a/tests/test_services/test_cron_scheduler.py +++ b/tests/test_services/test_cron_scheduler.py @@ -4,6 +4,8 @@ from __future__ import annotations from datetime import datetime, timedelta, timezone from pathlib import Path +import subprocess +import sys from unittest.mock import AsyncMock, Mock, patch import pytest @@ -13,6 +15,7 @@ from openharness.services.cron_scheduler import ( append_history, execute_job, load_history, + start_daemon, run_scheduler_loop, ) @@ -169,3 +172,42 @@ class TestSchedulerLoop: entries = load_history(job_name="test-once") assert len(entries) == 1 assert entries[0]["status"] == "success" + + @pytest.mark.asyncio + async def test_once_mode_handles_signal_registration_failures(self) -> None: + """Signal handler setup should not block the scheduler loop on unsupported platforms.""" + with patch("openharness.services.cron_scheduler.signal.signal", side_effect=ValueError()): + await run_scheduler_loop(once=True) + + +class TestDaemonStartup: + def test_start_daemon_uses_portable_process_spawn(self, monkeypatch: pytest.MonkeyPatch) -> None: + """Windows startup should use subprocess spawning instead of fork.""" + monkeypatch.setattr("openharness.services.cron_scheduler.get_platform", lambda: "windows") + monkeypatch.setattr(subprocess, "DETACHED_PROCESS", 0x00000008, raising=False) + monkeypatch.setattr(subprocess, "CREATE_NEW_PROCESS_GROUP", 0x00000200, raising=False) + + fake_process = Mock() + fake_process.pid = 4321 + fake_process.poll.side_effect = [None] + + spawned: dict[str, object] = {} + + def _fake_popen(argv, **kwargs): + spawned["argv"] = argv + spawned["kwargs"] = kwargs + return fake_process + + with patch("openharness.services.cron_scheduler.subprocess.Popen", side_effect=_fake_popen), patch( + "openharness.services.cron_scheduler.read_pid", + side_effect=[None, 4321], + ): + pid = start_daemon() + + assert pid == 4321 + assert spawned["argv"] == [sys.executable, "-m", "openharness.services.cron_scheduler"] + assert spawned["kwargs"]["creationflags"] == 0x00000208 + assert spawned["kwargs"]["stdin"] is subprocess.DEVNULL + assert spawned["kwargs"]["stdout"] is subprocess.DEVNULL + assert spawned["kwargs"]["stderr"] is subprocess.DEVNULL + assert spawned["kwargs"]["close_fds"] is True