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

@@ -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)

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