mirrors/hkuds--openharness
1
0
Fork 0
Code Issues Pull Requests Actions 2 Deployments Models Agent Notes Packages Projects Releases Wiki Activity

Fix cron scheduler on Windows

Browse Source
This commit is contained in:
Yudong An
2026-05-11 15:25:01 +08:00
parent 1929ad8051
commit 5b46a2f7b3
2 changed files with 197 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/openharness/services/cron_scheduler.py
+155 -32
View File
@@ -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()
tests/test_services/test_cron_scheduler.py
+42
View File
@@ -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