1import os
2import time
3from pathlib import Path
4
5from .poncho.manager import Manager as PonchoManager
6from .poncho.printer import Printer
7
8
9class ProcessManager:
10 pidfile: Path
11 log_dir: Path
12
13 def __init__(self):
14 self.pid = os.getpid()
15 self.log_path: Path | None = None
16 self.printer: Printer | None = None
17 self.poncho: PonchoManager | None = None
18
19 # ------------------------------------------------------------------
20 # Class-level pidfile helpers (usable without instantiation)
21 # ------------------------------------------------------------------
22 @classmethod
23 def read_pidfile(cls) -> int | None:
24 """Return the PID recorded in *cls.pidfile* (or ``None``)."""
25 if not cls.pidfile.exists():
26 return None
27
28 try:
29 return int(cls.pidfile.read_text())
30 except (ValueError, OSError):
31 # Corrupted pidfile – remove it so we don't keep trying.
32 cls.rm_pidfile()
33 return None
34
35 @classmethod
36 def rm_pidfile(cls) -> None:
37 if cls.pidfile and cls.pidfile.exists():
38 cls.pidfile.unlink(missing_ok=True) # Python 3.8+
39
40 @classmethod
41 def running_pid(cls) -> int | None:
42 """Return a *running* PID or ``None`` if the process is not alive."""
43 pid = cls.read_pidfile()
44 if pid is None:
45 return None
46
47 try:
48 os.kill(pid, 0) # Does not kill – merely checks for existence.
49 except OSError:
50 cls.rm_pidfile()
51 return None
52
53 return pid
54
55 def write_pidfile(self) -> None:
56 """Create/overwrite the pidfile for *this* process."""
57 self.pidfile.parent.mkdir(parents=True, exist_ok=True)
58 with self.pidfile.open("w+", encoding="utf-8") as f:
59 f.write(str(self.pid))
60
61 def stop_process(self) -> None:
62 """Terminate the process recorded in the pidfile, if it is running."""
63 pid = self.read_pidfile()
64 if pid is None:
65 return
66
67 # Try graceful termination first (SIGTERM)…
68 try:
69 os.kill(pid, 15)
70 except OSError:
71 # Process already gone – ensure we clean up.
72 self.rm_pidfile()
73 self.close()
74 return
75
76 timeout = 10 # seconds
77 start = time.time()
78 while time.time() - start < timeout:
79 try:
80 os.kill(pid, 0)
81 except OSError:
82 break # Process has exited.
83 time.sleep(0.1)
84
85 else: # Still running – force kill.
86 try:
87 os.kill(pid, 9)
88 except OSError:
89 pass
90
91 self.rm_pidfile()
92 self.close()
93
94 # ------------------------------------------------------------------
95 # Logging / Poncho helpers (unchanged)
96 # ------------------------------------------------------------------
97 def prepare_log(self) -> Path:
98 """Create the log directory and return a path for *this* run."""
99 self.log_dir.mkdir(parents=True, exist_ok=True)
100
101 # Keep the 5 most recent log files.
102 logs = sorted(
103 self.log_dir.glob("*.log"),
104 key=lambda p: p.stat().st_mtime,
105 reverse=True,
106 )
107 for old in logs[5:]:
108 old.unlink(missing_ok=True)
109
110 self.log_path = self.log_dir / f"{self.pid}.log"
111 return self.log_path
112
113 def init_poncho(self, print_func) -> PonchoManager: # noqa: D401
114 """Return a :class:`~plain.dev.poncho.manager.Manager` instance."""
115 if self.log_path is None:
116 self.prepare_log()
117
118 self.printer = Printer(print_func, log_file=self.log_path)
119 self.poncho = PonchoManager(printer=self.printer)
120 return self.poncho
121
122 # ------------------------------------------------------------------
123 # Cleanup
124 # ------------------------------------------------------------------
125 def close(self) -> None:
126 if self.printer:
127 self.printer.close()