64 lines
1.9 KiB
Python
64 lines
1.9 KiB
Python
 
|
"""The background worker: the job scheduler plus a heartbeat (beat file and worker_heartbeat row).
|
||
|
|
|
||
|
|
`run(beat_file, build, stop=None, interval=BEAT_SECONDS, settings=None)` blocks until SIGTERM, SIGINT or `stop`.
|
||
|
|
With settings it builds and starts a Scheduler; without (the CLI tests) it only beats the file.
|
||
|
|
"""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import logging
|
||
|
|
import signal
|
||
|
|
import threading
|
||
|
|
from pathlib import Path
|
||
|
|
from typing import TYPE_CHECKING
|
||
|
|
|
||
|
|
from .heartbeat import BEAT_SECONDS, STALE_SECONDS, beat_age, touch
|
||
|
|
|
||
|
|
if TYPE_CHECKING:
|
||
|
|
from ..settings import Settings
|
||
|
|
|
||
|
|
log = logging.getLogger("packetstation.worker")
|
||
|
|
|
||
|
|
__all__ = ["BEAT_SECONDS", "STALE_SECONDS", "beat_age", "run", "touch"]
|
||
|
|
|
||
|
|
|
||
|
|
def run(
|
||
|
|
beat_file: Path,
|
||
|
|
build: str,
|
||
|
|
stop: threading.Event | None = None,
|
||
|
|
interval: float = BEAT_SECONDS,
|
||
|
|
settings: Settings | None = None,
|
||
|
|
) -> int:
|
||
|
|
stop = stop or threading.Event()
|
||
|
|
|
||
|
|
def _stop(signum: int, _frame: object) -> None:
|
||
|
|
log.info("worker stopping on signal %s", signum)
|
||
|
|
stop.set()
|
||
|
|
|
||
|
|
if threading.current_thread() is threading.main_thread():
|
||
|
|
signal.signal(signal.SIGTERM, _stop)
|
||
|
|
signal.signal(signal.SIGINT, _stop)
|
||
|
|
log.info("worker started", extra={"ps": {"build": build, "beat_file": str(beat_file)}})
|
||
|
|
sched = None
|
||
|
|
if settings is not None:
|
||
|
|
from .scheduler import Scheduler
|
||
|
|
|
||
|
|
sched = Scheduler(settings, build=build)
|
||
|
|
sched.start()
|
||
|
|
try:
|
||
|
|
if sched is not None:
|
||
|
|
sched.serve(beat_file, stop, interval)
|
||
|
|
else:
|
||
|
|
while not stop.is_set():
|
||
|
|
try:
|
||
|
|
touch(beat_file)
|
||
|
|
except OSError as e:
|
||
|
|
log.error("can't touch beat file %s: %s", beat_file, e.strerror)
|
||
|
|
log.info("heartbeat", extra={"ps": {"build": build}})
|
||
|
|
stop.wait(interval)
|
||
|
|
finally:
|
||
|
|
if sched is not None:
|
||
|
|
sched.shutdown(wait=True)
|
||
|
|
log.info("worker stopped")
|
||
|
|
return 0
|