From bdca7d2354f320388c9fe41f7522da7b58826976 Mon Sep 17 00:00:00 2001 From: Goffi Date: Wed, 23 Sep 2026 18:19:03 +0200 Subject: [PATCH] tools (common/async_process): add a cleaning mecanism: PID of commands run with `CommandProtocol` are now tracked in memory and stopped/killed on exit. A `pid_file` argument can be use to track the file on disk, which is useful to detect stalled processes after a crash/unclean ending of the backend. Add helper methods to stop and kill those processes. --- libervia/backend/core/main.py | 1 + .../backend/tools/common/async_process.py | 109 ++++++++++++++++-- 2 files changed, 99 insertions(+), 11 deletions(-) diff --git a/libervia/backend/core/main.py b/libervia/backend/core/main.py index bf9ff768..b5ffedc5 100644 --- a/libervia/backend/core/main.py +++ b/libervia/backend/core/main.py @@ -819,6 +819,7 @@ class LiberviaBackend(service.Service): Disconnect all profiles first so profile_disconnected callbacks fire, then unload plugins and storage. """ + await async_process.stop_all() if self.local_shared_path is not None: log.debug("Cleaning shared temporary dir.") shutil.rmtree(self.local_shared_path, ignore_errors=True) diff --git a/libervia/backend/tools/common/async_process.py b/libervia/backend/tools/common/async_process.py index f68d1a22..04635ab0 100644 --- a/libervia/backend/tools/common/async_process.py +++ b/libervia/backend/tools/common/async_process.py @@ -19,17 +19,22 @@ """tools to launch process in a async way (using Twisted)""" -import os.path +import os +import signal from pathlib import Path from typing import Any -from twisted.internet import defer, reactor, protocol +from twisted.internet import defer, protocol, reactor, task from twisted.python.failure import Failure from libervia.backend.core.i18n import _ from libervia.backend.core import exceptions from libervia.backend.core.log import getLogger +from libervia.backend.tools.common import utils log = getLogger(__name__) +# Processes registered for auto-cleaning on shutdown (pid -> CommandProtocol) +_running: dict[int, "CommandProtocol"] = {} + class CommandProtocol(protocol.ProcessProtocol): """handle an external command""" @@ -52,6 +57,8 @@ class CommandProtocol(protocol.ProcessProtocol): self.err_data = [] self.cmd_args: list[str] | None = None self.cmd_kwargs: dict[str, Any] | None = None + self.pid: int | None = None + self.pid_file: Path | None = None @property def command_name(self): @@ -81,6 +88,10 @@ class CommandProtocol(protocol.ProcessProtocol): self.err_data.append(data) def processEnded(self, reason): + if self.pid is not None: + _running.pop(self.pid, None) + if self.pid_file is not None: + self.pid_file.unlink(missing_ok=True) data = b"".join(self.data) if reason.value.exitCode == 0: log.debug(f"{self.command_name!r} command succeed") @@ -113,20 +124,29 @@ class CommandProtocol(protocol.ProcessProtocol): ) @classmethod - def run(cls, *args, **kwargs): + def run( + cls, + *args: str | Path, + auto_clean: bool = True, + pid_file: Path | None = None, + **kwargs: Any, + ) -> defer.Deferred: """Create a new CommandProtocol and execute the given command. - @param *args(unicode): command arguments + @param *args: command arguments if cls.command is specified, it will be the path to the command to execute otherwise, first argument must be the path + @param auto_clean: if True, the process will be stopped on shutdown if not + already finished + @param pid_file: if not None, the process PID is written there (and removed + on exit), to allow cleaning it up after a crash @param **kwargs: can be: - - stdin(unicode, None): data to push to standard input - - verbose(bool): if True stdout and stderr will be logged + - stdin: data to push to standard input + - verbose: if True stdout and stderr will be logged other keyword arguments will be used in reactor.spawnProcess - @return ((D)bytes): stdout in case of success - @raise RuntimeError: command returned a non zero status - stdin and stdout will be given as arguments - + @return: Deferred called with the process stdout on success + @raise CommandException: command returned a non zero status, stdout and + stderr will be attached to the exception """ stdin = kwargs.pop("stdin", None) if stdin is not None: @@ -160,8 +180,75 @@ class CommandProtocol(protocol.ProcessProtocol): # FIXME: `None` doesn't seem to work, despite what documentation says, to be # checked and reported upstream if confirmed. kwargs["env"] = os.environ - reactor.spawnProcess(prot, command, cmd_args, **kwargs) + transport = reactor.spawnProcess(prot, command, cmd_args, **kwargs) + prot.pid = transport.pid + if pid_file is not None: + prot.pid_file = pid_file + prot.pid_file.write_text(f"{prot.pid}\n") + if auto_clean: + _running[prot.pid] = prot return d +def signal_all(sig: signal.Signals) -> None: + """Send a signal to all the registered processes""" + for pid, prot in list(_running.items()): + try: + prot.transport.signalProcess(sig) + except Exception as e: + log.debug(f"Can't send {sig.name} to process {pid}: {e}") + _running.pop(pid, None) + + +async def stop_all(grace: float = 5.0) -> None: + """Stop all processes registered for auto-cleaning. + + Send SIGTERM, wait for the processes to exit, then send SIGKILL to the remaining ones. + + @param grace: time (in seconds) to wait before sending SIGKILL + """ + if not _running: + return + log.info(f"Stopping {len(_running)} external process(es) on shutdown") + signal_all(signal.SIGTERM) + while _running and grace > 0: + await task.deferLater(reactor, 0.5, lambda: None) + grace -= 0.5 + signal_all(signal.SIGKILL) + + +def kill_stale_process(pid_file: Path, expected_cwd: Path) -> None: + """Kill a process recorded in a PID file if it is verified as a previous instance. + + Verification is done by comparing the process working directory with ``expected_cwd`` + (relying on /proc, Linux only). If the process can't be verified, it is left + untouched. + + @param pid_file: path to the PID file + @param expected_cwd: expected working directory of the process + """ + if not utils.is_process_running(pid_file): + pid_file.unlink(missing_ok=True) + return + pid = int(pid_file.read_text().strip()) + try: + cwd = Path(os.readlink(f"/proc/{pid}/cwd")) + except OSError: + log.warning( + f"Can't verify process {pid} (no /proc access), it won't be killed; if it " + "is stale, please kill it manually." + ) + return + if cwd.resolve() != expected_cwd.resolve(): + log.error( + f"Process {pid} recorded in {pid_file} has an unexpected working directory " + f"({cwd!r}), it won't be killed." + ) + pid_file.unlink(missing_ok=True) + return + log.info(f"Killing stale process {pid} left by a previous run.") + os.kill(pid, signal.SIGTERM) + pid_file.unlink(missing_ok=True) + + run = CommandProtocol.run