Merge pull request #1823 from AmirF194/fix/1818-process-group-cleanup-without-enumeration

src/solidlsp/ls_process.py

@@ -2,7 +2,6 @@ import asyncio
 import json
 import logging
 import socket
-import subprocess
 import threading
 import time
 from abc import ABC, abstractmethod
@@ -32,6 +31,7 @@ from solidlsp.lsp_protocol_handler.server import (
     make_response,
 )
 from solidlsp.util import subprocess_util
+from solidlsp.util.subprocess_util import ManagedSubprocess
 
 log = logging.getLogger(__name__)
 
@@ -507,22 +507,21 @@ class StdioLanguageServer(LanguageServerInterface):
         """
         super().__init__(ls_id, determine_log_level, logger, request_timeout)
 
-        self.process_launch_info = process_launch_info
-        self.process: subprocess.Popen[bytes] | None = None
-        self.start_independent_lsp_process = start_independent_lsp_process
-
+        self._process_launch_info = process_launch_info
+        self._process: ManagedSubprocess[bytes] | None = None
+        self._start_independent_lsp_process = start_independent_lsp_process
         self._stdin_lock = threading.Lock()
 
     def is_running(self) -> bool:
-        return self.process is not None and self.process.returncode is None
+        return self._process is not None and self._process.returncode is None
 
     def _start(self) -> None:
-        log.info("Starting language server process via command: %s", self.process_launch_info.cmd)
+        log.info("Starting language server process via command: %s", self._process_launch_info.cmd)
 
-        process = subprocess_util.LanguageServerSubprocessLauncher.get_instance().launch(
-            self.process_launch_info, start_new_session=self.start_independent_lsp_process
+        process = subprocess_util.ManagedSubprocessLauncher.get_instance().launch(
+            self._process_launch_info, name=f"LS[{self.ls_id.value}]", start_new_session=self._start_independent_lsp_process
         )
-        self.process = process
+        self._process = process
 
         # Check if process terminated immediately
         if process.returncode is not None:
@@ -545,7 +544,7 @@ class StdioLanguageServer(LanguageServerInterface):
         ).start()
 
     def _stop(self, timeout: float) -> None:
-        if self.process is None:
+        if self._process is None:
             log.debug("Server process is None, cannot shutdown.")
             return
 
@@ -553,16 +552,14 @@ class StdioLanguageServer(LanguageServerInterface):
             # send LSP shutdown and close stdin to signal no more input
             try:
                 self._send_shutdown_in_thread()
-                self._safely_close_pipe(self.process.stdin)
+                self._safely_close_pipe(self._process.stdin)
             except Exception as e:
                 log.debug(f"Exception during graceful shutdown: {e}")
                 # Ignore errors here, we are proceeding to terminate anyway.
             # terminate the process
-            subprocess_util.terminate_process_tree_with_kill_fallback(
-                self.process, terminate_timeout=timeout, process_name=f"LS[{self.ls_id.value}]"
-            )
+            self._process.terminate(timeout=timeout)
         finally:
-            self.process = None
+            self._process = None
 
     @staticmethod
     def _safely_close_pipe(pipe: IO[AnyStr] | None) -> None:
@@ -573,7 +570,7 @@ class StdioLanguageServer(LanguageServerInterface):
             except Exception:
                 pass
 
-    def _read_bytes_from_process(self, process, stream, num_bytes) -> bytes:
+    def _read_bytes_from_process(self, process: ManagedSubprocess[bytes], stream: IO[bytes], num_bytes: int) -> bytes:
         """Read exactly num_bytes from process stdout"""
         data = b""
         while len(data) < num_bytes:
@@ -597,10 +594,10 @@ class StdioLanguageServer(LanguageServerInterface):
         """
         exception: Exception | None = None
         try:
-            while self.process and self.process.stdout:
-                if self.process.poll() is not None:  # process has terminated
+            while self._process and self._process.stdout:
+                if self._process.poll() is not None:  # process has terminated
                     break
-                line = self.process.stdout.readline()
+                line = self._process.stdout.readline()
                 if not line:
                     continue
                 try:
@@ -610,10 +607,10 @@ class StdioLanguageServer(LanguageServerInterface):
                 if num_bytes is None:
                     continue
                 while line and line.strip():
-                    line = self.process.stdout.readline()
+                    line = self._process.stdout.readline()
                 if not line:
                     continue
-                body = self._read_bytes_from_process(self.process, self.process.stdout, num_bytes)
+                body = self._read_bytes_from_process(self._process, self._process.stdout, num_bytes)
 
                 self._handle_body(body)
         except LanguageServerTerminatedException as e:
@@ -636,11 +633,11 @@ class StdioLanguageServer(LanguageServerInterface):
         Continuously read from the language server process stderr and log the messages
         """
         try:
-            while self.process and self.process.stderr:
-                if self.process.poll() is not None:
+            while self._process and self._process.stderr:
+                if self._process.poll() is not None:
                     # process has terminated
                     break
-                line = self.process.stderr.readline()
+                line = self._process.stderr.readline()
                 if not line:
                     continue
                 line_str = line.decode(ENCODING, errors="replace")
@@ -654,7 +651,7 @@ class StdioLanguageServer(LanguageServerInterface):
             log.info("Language server stderr reader thread has terminated")
 
     def _send_payload(self, payload: StringDict) -> None:
-        if not self.process or not self.process.stdin:
+        if not self._process or not self._process.stdin:
             return
         self._trace("solidlsp", "ls", payload)
         msg = create_message(payload)
@@ -662,8 +659,8 @@ class StdioLanguageServer(LanguageServerInterface):
         # Use lock to prevent concurrent writes to stdin that cause buffer corruption
         with self._stdin_lock:
             try:
-                self.process.stdin.writelines(msg)
-                self.process.stdin.flush()
+                self._process.stdin.writelines(msg)
+                self._process.stdin.flush()
             except (BrokenPipeError, ConnectionResetError, OSError) as e:
                 # Log the error but don't raise to prevent cascading failures
                 log.error(f"Failed to write to stdin: {e}")

src/solidlsp/util/subprocess_util.py

@@ -6,10 +6,11 @@ import signal
 import subprocess
 import threading
 from collections.abc import Callable
-from typing import TYPE_CHECKING, Any, cast
+from typing import IO, TYPE_CHECKING, Any, Generic, TypeVar, cast
 
 import oslex
 import psutil
+from sensai.util.string import ToStringMixin
 
 if TYPE_CHECKING:
     import ctypes
@@ -17,6 +18,7 @@ if TYPE_CHECKING:
     from solidlsp.lsp_protocol_handler.server import ProcessLaunchInfo
 
 log = logging.getLogger(__name__)
+TStream = TypeVar("TStream", bound=str | bytes)
 
 
 def subprocess_kwargs() -> dict:
@@ -70,9 +72,73 @@ def convert_shell_cmd(cmd: str | list[str]) -> str:
     return oslex.join(cmd) if isinstance(cmd, list) else cmd
 
 
-class LanguageServerSubprocessLauncher:
+class ManagedSubprocess(Generic[TStream], ToStringMixin):
     """
-    Launcher for language server subprocesses, which are started for stdio-based communication.
+    Represents a subprocess.Popen instance with additional lifecycle management, including the ability to terminate the process
+    and its children gracefully, with a fallback to forceful termination if necessary.
+    """
+
+    def __init__(self, popen: subprocess.Popen[TStream], name: str, start_new_session: bool) -> None:
+        """
+        :param popen: the subprocess.Popen instance representing the launched process
+        :param start_new_session: whether the process was launched in its own session, i.e. whether it is the
+            leader of its own process group
+        """
+        self._popen = popen
+        self._name = name
+
+        # a process launched with start_new_session=True is its own group leader, so its PGID is its PID
+        self._process_group_id = popen.pid if (start_new_session and os.name == "posix") else None
+
+    def _tostring_includes(self) -> list[str]:
+        return ["_name"]
+
+    @property
+    def stdin(self) -> "IO[TStream] | None":
+        return self._popen.stdin
+
+    @property
+    def stdout(self) -> "IO[TStream] | None":
+        return self._popen.stdout
+
+    @property
+    def stderr(self) -> "IO[TStream] | None":
+        return self._popen.stderr
+
+    @property
+    def returncode(self) -> int | None:
+        """
+        :return: the exit code of the process as of the most recent status check (see :meth:`poll`),
+            or None if the process was not yet known to have terminated
+        """
+        return self._popen.returncode
+
+    def poll(self) -> int | None:
+        """
+        Checks whether the process has terminated, updating :attr:`returncode` accordingly.
+
+        :return: the exit code of the process or None if it is still running
+        """
+        return self._popen.poll()
+
+    def terminate(self, timeout: float) -> None:
+        """
+        Terminates the process and its children, forcefully killing them if they do not exit in time.
+
+        :param timeout: the time to wait for the process to terminate gracefully before killing it
+        :param process_name: the name of the process (used for logging purposes); should start with a capital letter
+        """
+        terminate_process_tree_with_kill_fallback(
+            self._popen,
+            terminate_timeout=timeout,
+            process_name=self._name,
+            process_group_id=self._process_group_id,
+        )
+
+
+class ManagedSubprocessLauncher:
+    """
+    Launcher for managed subprocesses (see :class:`ManagedSubprocess`), which are started for stdio-based communication.
     It is home to the concern of launching a subprocess with well-defined lifecycle properties,
     ensuring, in particular, that a launched subprocess cannot outlive this process (insofar as
     the platform allows) -- even if the subprocess is started in its own session (see
@@ -85,16 +151,16 @@ class LanguageServerSubprocessLauncher:
     _PR_SET_PDEATHSIG = 1
     """the PR_SET_PDEATHSIG option value for prctl(2)"""
 
-    _instance: "LanguageServerSubprocessLauncher | None" = None
+    _instance: "ManagedSubprocessLauncher | None" = None
     _instance_lock = threading.Lock()
 
     def __init__(self) -> None:
         self._libc = self._load_libc()
-        self._spawner: "LanguageServerSubprocessLauncher._PDeathSigSpawner | None" = None
+        self._spawner: "ManagedSubprocessLauncher._PDeathSigSpawner | None" = None
         self._spawner_lock = threading.Lock()
 
     @classmethod
-    def get_instance(cls) -> "LanguageServerSubprocessLauncher":
+    def get_instance(cls) -> "ManagedSubprocessLauncher":
         if cls._instance is None:
             with cls._instance_lock:
                 if cls._instance is None:
@@ -115,7 +181,7 @@ class LanguageServerSubprocessLauncher:
             return ctypes.CDLL(None)
         except OSError as e:
             log.warning(
-                "Could not load libc (%s); language server processes will not be protected against "
+                "Could not load libc (%s); subprocesses will not be protected against "
                 "orphaning if this process is killed without a chance to shut down cleanly",
                 e,
             )
@@ -130,11 +196,12 @@ class LanguageServerSubprocessLauncher:
         if self._libc is not None:
             self._libc.prctl(self._PR_SET_PDEATHSIG, signal.SIGTERM)
 
-    def launch(self, process_launch_info: "ProcessLaunchInfo", start_new_session: bool) -> subprocess.Popen[bytes]:
+    def launch(self, process_launch_info: "ProcessLaunchInfo", name: str, start_new_session: bool) -> ManagedSubprocess[bytes]:
         """
-        Launches a language server process from ``process_launch_info``.
+        Launches a subprocess from ``process_launch_info``.
 
         :param process_launch_info: the command, environment and working directory to launch with
+        :param name: the name of the process (used for logging purposes); should start with a capital letter
         :param start_new_session: whether to start the process in its own session (own process
             group, detached from ours)
         """
@@ -158,8 +225,8 @@ class LanguageServerSubprocessLauncher:
         if use_pdeathsig:
             kwargs["preexec_fn"] = self._set_pdeathsig_on_parent_exit
 
-        def do_popen() -> subprocess.Popen[bytes]:
-            return cast(
+        def do_popen() -> ManagedSubprocess[bytes]:
+            popen = cast(
                 "subprocess.Popen[bytes]",
                 subprocess.Popen(
                     cmd,
@@ -172,6 +239,7 @@ class LanguageServerSubprocessLauncher:
                     **kwargs,
                 ),
             )
+            return ManagedSubprocess(popen, name, start_new_session)
 
         # perform the actual Popen call, funneling the fork() through the dedicated spawner thread
         # when pdeathsig applies
@@ -192,7 +260,7 @@ class LanguageServerSubprocessLauncher:
         """
 
         def __init__(self) -> None:
-            self._queue: queue.Queue[tuple[Callable[[], subprocess.Popen], "queue.Queue"]] = queue.Queue()
+            self._queue: queue.Queue[tuple[Callable[[], ManagedSubprocess[bytes]], "queue.Queue"]] = queue.Queue()
             self._thread = threading.Thread(target=self._run, name="solidlsp-pdeathsig-spawner", daemon=True)
             self._thread.start()
 
@@ -204,7 +272,7 @@ class LanguageServerSubprocessLauncher:
                 except BaseException as e:
                     result_queue.put((e, None))
 
-        def spawn(self, func: Callable[[], subprocess.Popen]) -> subprocess.Popen:
+        def spawn(self, func: Callable[[], ManagedSubprocess[bytes]]) -> ManagedSubprocess[bytes]:
             result_queue: queue.Queue = queue.Queue(maxsize=1)
             self._queue.put((func, result_queue))
             error, process = result_queue.get()
@@ -246,7 +314,33 @@ def _signal_process_tree(process: subprocess.Popen[bytes], terminate: bool = Tru
         signal_process(process)
 
 
-def terminate_process_tree_with_kill_fallback(process: subprocess.Popen, terminate_timeout: float, process_name: str = "Process") -> None:
+def _signal_process_group(pgid: int, terminate: bool = True) -> None:
+    """
+    Sends a signal to every process in the given POSIX process group by group ID, without
+    enumerating the system process table. Requires the caller to know the group already exists
+    and is owned by us (see ``terminate_process_tree_with_kill_fallback``'s ``process_group_id``).
+
+    :param pgid: the process group ID to signal
+    :param terminate: if True, signal terminate (SIGTERM), otherwise signal kill (SIGKILL)
+    """
+    sig = signal.SIGTERM if terminate else signal.SIGKILL
+    try:
+        os.killpg(pgid, sig)
+    except ProcessLookupError:
+        # the group is already gone; nothing left to signal
+        pass
+    except PermissionError as e:
+        log.warning(f"Permission denied signaling process group {pgid} with {sig.name}: {e}")
+    except Exception as e:
+        log.warning(f"Unexpected error signaling process group {pgid} with {sig.name}: {e}")
+
+
+def terminate_process_tree_with_kill_fallback(
+    process: subprocess.Popen,
+    terminate_timeout: float,
+    process_name: str = "Process",
+    process_group_id: int | None = None,
+) -> None:
     """
     Attempts to terminate the given process and its children by signaling them to terminate,
     and if that fails (i.e. they don't exit within the given timeout), forcefully kills them.
@@ -256,9 +350,23 @@ def terminate_process_tree_with_kill_fallback(process: subprocess.Popen, termina
     :param process: the process to terminate
     :param terminate_timeout: the time to wait for the process to terminate gracefully before killing it
     :param process_name: the name of the process (used for logging purposes); should start with capital letter
+    :param process_group_id: if given, the POSIX process group ID that ``process`` leads (i.e. it was
+        launched with ``start_new_session=True``, so its PGID equals its PID). When set, cleanup
+        signals the whole group directly via ``os.killpg`` instead of walking the process tree with
+        ``psutil``, which requires system-wide process-table enumeration (``sysctl(KERN_PROC_ALL)`` on
+        macOS) that can be denied even for processes we started and own. Only pass this for a process
+        that was started in its own session: signaling the group of a process that shares ours would
+        also signal us.
     """
     log.debug(f"Terminating process {process.pid}, current status: {process.poll()}")
-    _signal_process_tree(process, terminate=True)
+
+    def signal_tree(terminate: bool) -> None:
+        if process_group_id is not None:
+            _signal_process_group(process_group_id, terminate=terminate)
+        else:
+            _signal_process_tree(process, terminate=terminate)
+
+    signal_tree(terminate=True)
     try:
         log.debug(f"Waiting for process {process.pid} to terminate...")
         exit_code = process.wait(timeout=terminate_timeout)
@@ -266,7 +374,7 @@ def terminate_process_tree_with_kill_fallback(process: subprocess.Popen, termina
     except subprocess.TimeoutExpired:
         # If termination failed, forcefully kill the process
         log.warning(f"{process_name} (pid={process.pid}) termination timed out, killing process forcefully...")
-        _signal_process_tree(process, terminate=False)
+        signal_tree(terminate=False)
         try:
             exit_code = process.wait(timeout=2.0)
             log.info(f"{process_name} killed successfully with exit code {exit_code}.")