diff --git a/src/solidlsp/ls_process.py b/src/solidlsp/ls_process.py index 8e8cca2d..e231d967 100644 --- a/src/solidlsp/ls_process.py +++ b/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}") diff --git a/src/solidlsp/util/subprocess_util.py b/src/solidlsp/util/subprocess_util.py index b6912715..8763f745 100644 --- a/src/solidlsp/util/subprocess_util.py +++ b/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()