Introduce ManagedSubprocess to encapsulate subprocess lifecycle management

The knowledge that a process launched with start_new_session=True is its own
group leader (and thus that its PGID equals its PID) was previously held by
StdioLanguageServer, i.e. the consumer had to know how the launcher works in
order to terminate the process correctly.

ManagedSubprocess now owns the process group ID alongside the Popen instance,
deriving it at construction time, and provides the termination logic; the
launcher returns instances of it. StdioLanguageServer merely holds one and
accesses the process through its interface.

LanguageServerSubprocessLauncher now returns ManagedSubprocess instances and
is renamed to ManagedSubprocessLauncher accordingly.

src/solidlsp/ls_process.py

@@ -1,9 +1,7 @@
 import asyncio
 import json
 import logging
-import os
 import socket
-import subprocess
 import threading
 import time
 from abc import ABC, abstractmethod
@@ -33,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__)
 
@@ -508,27 +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_group_id: int | None = None
-        """the POSIX process group ID owned by ``self.process``, captured at launch time (see
-        ``_start``); ``None`` on Windows or when the process was not started in its own session"""
-
+        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
-        # start_new_session=True makes the new process its own group leader, so its PGID is its PID.
-        self._process_group_id = process.pid if (self.start_independent_lsp_process and os.name == "posix") else None
+        self._process = process
 
         # Check if process terminated immediately
         if process.returncode is not None:
@@ -551,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
 
@@ -559,20 +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}]",
-                process_group_id=self._process_group_id,
-            )
+            self._process.terminate(timeout=timeout)
         finally:
-            self.process = None
-            self._process_group_id = None
+            self._process = None
 
     @staticmethod
     def _safely_close_pipe(pipe: IO[AnyStr] | None) -> None:
@@ -583,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:
@@ -607,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:
@@ -620,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:
@@ -646,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")
@@ -664,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)
@@ -672,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()