From d48bfd242f52ab994d7e3ab32b6419b5cbfa52b2 Mon Sep 17 00:00:00 2001 From: Dominik Jain Date: Sun, 10 May 2026 22:28:15 +0200 Subject: [PATCH] Refactor LanguageServerProcess, separating the communication interface from the concrete implementation * Rename LanguageServerProcess to LanguageServerInterface (now abstract), - dropping stdio-specific methods - cleaning up documentation, etc. * Move stdio-specific implementation to StdioLanguageServer --- src/solidlsp/ls.py | 15 +- src/solidlsp/ls_process.py | 558 ++++++++++++++++++++----------------- src/solidlsp/ls_request.py | 4 +- 3 files changed, 312 insertions(+), 265 deletions(-) diff --git a/src/solidlsp/ls.py b/src/solidlsp/ls.py index 6bab82aa..dd61e775 100644 --- a/src/solidlsp/ls.py +++ b/src/solidlsp/ls.py @@ -5,7 +5,6 @@ import logging import os import pathlib import shutil -import subprocess import threading from abc import ABC, abstractmethod from collections import defaultdict @@ -25,7 +24,7 @@ from serena.util.text_utils import MatchedConsecutiveLines from solidlsp import ls_types from solidlsp.ls_config import FilenameMatcher, Language, LanguageServerConfig from solidlsp.ls_exceptions import SolidLSPException -from solidlsp.ls_process import LanguageServerProcess +from solidlsp.ls_process import LanguageServerInterface, StdioLanguageServer from solidlsp.ls_types import UnifiedSymbolInformation from solidlsp.ls_utils import FileUtils, PathUtils, TextUtils from solidlsp.lsp_protocol_handler import lsp_types @@ -336,8 +335,7 @@ class LanguageServerDependencyProviderSinglePath(LanguageServerDependencyProvide class SolidLanguageServer(ABC): """ - The LanguageServer class provides a language agnostic interface to the Language Server Protocol. - It is used to communicate with Language Servers of different programming languages. + High-level abstraction for language server interaction, which wraps the underlying low-level LSP interface """ CACHE_FOLDER_NAME = "cache" @@ -551,13 +549,16 @@ class SolidLanguageServer(ABC): self._dependency_provider = self._create_dependency_provider() process_launch_info = self._create_process_launch_info() log.debug(f"Creating language server instance with {language_id=} and process launch info: {process_launch_info}") - self.server = LanguageServerProcess( + self.server = StdioLanguageServer( process_launch_info, language=self.language, determine_log_level=self._determine_log_level, logger=logging_fn, start_independent_lsp_process=config.start_independent_lsp_process, ) + """ + the low-level language server interface + """ self.server.on_any_notification(self._observe_server_notification) # Set up the pathspec matcher for the ignored paths @@ -1208,7 +1209,7 @@ class SolidLanguageServer(ABC): try: log.debug("Sending LSP shutdown request...") # Use a thread to timeout the LSP shutdown call since it can hang - shutdown_thread = threading.Thread(target=self.server.shutdown) + shutdown_thread = threading.Thread(target=self.server.send_shutdown) shutdown_thread.daemon = True shutdown_thread.start() shutdown_thread.join(timeout=2.0) # 2 second timeout for LSP shutdown @@ -3131,7 +3132,7 @@ class SolidLanguageServer(ABC): return self @property - def handler(self) -> LanguageServerProcess: + def handler(self) -> LanguageServerInterface: """Access the underlying language server handler. Useful for advanced operations like sending custom commands diff --git a/src/solidlsp/ls_process.py b/src/solidlsp/ls_process.py index 6319a9d7..a338099b 100644 --- a/src/solidlsp/ls_process.py +++ b/src/solidlsp/ls_process.py @@ -6,10 +6,11 @@ import platform import subprocess import threading import time +from abc import ABC, abstractmethod from collections.abc import Callable from dataclasses import dataclass from queue import Empty, Queue -from typing import Any +from typing import IO, Any, AnyStr import psutil from sensai.util.string import ToStringMixin @@ -92,40 +93,305 @@ class Request(ToStringMixin): raise e -class LanguageServerProcess: +class LanguageServerInterface(ABC): """ - Represents a language server process and provides methods for communicating with it using the + Represents an interface to a language server, providing methods for communicating with it using the Language Server Protocol (LSP). It provides methods for sending requests, responses, and notifications to the server and for registering handlers for requests and notifications from the server. Uses JSON-RPC 2.0 for communication with the server over stdin/stdout. + """ - Attributes: - send: A LspRequest object that can be used to send requests to the server and - await for the responses. - notify: A LspNotification object that can be used to send notifications to the server. - cmd: A string that represents the command to launch the language server process. - process: A subprocess.Popen object that represents the language server process. - request_id: An integer that represents the next available request id for the client. - _pending_requests: A dictionary that maps request ids to Request objects that - store the results or errors of the requests. - on_request_handlers: A dictionary that maps method names to callback functions - that handle requests from the server. - on_notification_handlers: A dictionary that maps method names to callback functions - that handle notifications from the server. - _trace_log_fn: An optional function that takes two strings (source and destination) and - a payload dictionary, and logs the communication between the client and the server. - tasks: A dictionary that maps task ids to asyncio.Task objects that represent - the asynchronous tasks created by the handler. - task_counter: An integer that represents the next available task id for the handler. - loop: An asyncio.AbstractEventLoop object that represents the event loop used by the handler. - start_independent_lsp_process: An optional boolean flag that indicates whether to start the - language server process in an independent process group. Default is `True`. Setting it to - `False` means that the language server process will be in the same process group as the - the current process, and any SIGINT and SIGTERM signals will be sent to both processes. + def __init__( + self, + language: Language, + determine_log_level: Callable[[str], int], + logger: Callable[[str, str, StringDict | str], None] | None = None, + request_timeout: float | None = None, + ) -> None: + """ + :param language: the language + :param determine_log_level: a function for log lines read from stderr, which determines the log level + :param logger: the trace logger function + :param request_timeout: the timeout, in seconds, for all requests sent to the language server. If None, no timeout will be applied. + """ + self.language = language + self._determine_log_level = determine_log_level + self.send = LanguageServerRequest(self) + """ + an object that can be used to send requests to the server + """ + self.notify = LspNotification(self.send_notification) + """ + an object that can be used to send notifications to the server + """ + self._is_shutting_down = False + self.request_id = 1 + """ + the next request id to use for requests + """ + self._pending_requests: dict[Any, Request] = {} + """ + maps request ids to Request objects that store the results or errors of the requests + """ + self.on_request_handlers: dict[str, Callable[[Any], Any]] = {} + self.on_notification_handlers: dict[str, Callable[[Any], None]] = {} + """ + maps method names to callback functions that handle notifications from the server + """ + self._notification_observers: list[Callable[[str, Any], None]] = [] + self._trace_log_fn = logger + self.task_counter = 0 + self._is_stopped = False + self._incoming_messages_queue: Queue[bytes] = Queue() + self._request_timeout = request_timeout + # Add thread locks for shared resources to prevent race conditions + self._request_id_lock = threading.Lock() + self._response_handlers_lock = threading.Lock() + self._tasks_lock = threading.Lock() + + def set_request_timeout(self, timeout: float | None) -> None: + """ + :param timeout: the timeout, in seconds, for all requests sent to the language server. + """ + self._request_timeout = timeout + + @abstractmethod + def is_running(self) -> bool: + """ + Checks whether the language server interface is running + """ + + def start(self) -> None: + """ + Starts communication with the language server + """ + self._is_stopped = False + self._start() + + @abstractmethod + def _start(self) -> None: + """ + Starts the actual communication mechanism with the language server, calling `_handle_body` for each message received from the + language server + """ + + @abstractmethod + def stop(self) -> None: + """ + Stops communication with the language server, freeing resources + """ + + def send_shutdown(self) -> None: + """ + Signals shutdown to the server + """ + self._is_shutting_down = True + log.info("Sending shutdown request to server") + self.send.shutdown() + log.info("Received shutdown response from server") + log.info("Sending exit notification to server") + self.notify.exit() + log.info("Sent exit notification to server") + + def _trace(self, src: str, dest: str, message: str | StringDict) -> None: + """ + Traces LS communication by logging the message with the source and destination of the message + """ + if self._trace_log_fn is not None: + self._trace_log_fn(src, dest, message) + + def _handle_body(self, body: bytes) -> None: + """ + Parse the body text received from the language server process and invoke the appropriate handler + """ + try: + self._receive_payload(json.loads(body)) + except OSError as ex: + log.error(f"Error processing payload: {ex}", exc_info=ex) + except UnicodeDecodeError as ex: + log.error(f"Decoding error for encoding={ENCODING}: {ex}") + except json.JSONDecodeError as ex: + log.error(f"JSON decoding error: {ex}") + + def _receive_payload(self, payload: StringDict) -> None: + """ + Determine if the payload received from server is for a request, response, or notification and invoke the appropriate handler + """ + self._trace("ls", "solidlsp", payload) + try: + if "method" in payload: + if "id" in payload: + self._request_handler(payload) + else: + self._notification_handler(payload) + elif "id" in payload: + self._response_handler(payload) + else: + log.error(f"Unknown payload type: {payload}") + except Exception as err: + log.error(f"Error handling server payload: {err}") + + def send_notification(self, method: str, params: dict | None = None) -> None: + """ + Send notification pertaining to the given method to the server with the given parameters + """ + self._send_payload(make_notification(method, params)) + + def send_response(self, request_id: Any, params: PayloadLike) -> None: + """ + Send response to the given request id to the server with the given parameters + """ + self._send_payload(make_response(request_id, params)) + + def send_error_response(self, request_id: Any, err: LSPError) -> None: + """ + Send error response to the given request id to the server with the given error + """ + self._send_payload(make_error_response(request_id, err)) + + def _cancel_pending_requests(self, exception: Exception) -> None: + """ + Cancel all pending requests by setting their results to an error + """ + with self._response_handlers_lock: + log.info("Cancelling %d pending language server requests", len(self._pending_requests)) + for request in self._pending_requests.values(): + log.info("Cancelling %s", request) + request.on_error(exception) + self._pending_requests.clear() + + def send_request(self, method: str, params: dict | None = None) -> PayloadLike: + """ + Send request to the server, register the request id, and wait for the response + """ + with self._request_id_lock: + request_id = self.request_id + self.request_id += 1 + + request = Request(request_id=request_id, method=method) + log.debug("Starting: %s", request) + + with self._response_handlers_lock: + self._pending_requests[request_id] = request + + self._send_payload(make_request(method, request_id, params)) + + log.debug("Waiting for response to request %s with params:\n%s", method, params) + result = request.get_result(timeout=self._request_timeout) + log.debug("Completed: %s", request) + + if result.is_error(): + raise SolidLSPException(f"Error processing request {method} with params:\n{params}", cause=result.error) from result.error + + log.debug("Returning result:\n%s", result.payload) + return result.payload + + @abstractmethod + def _send_payload(self, payload: StringDict) -> None: + """ + Send the given payload to the server + """ + + def on_request(self, method: str, cb: Callable[[Any], Any]) -> None: + """ + Register the callback function to handle requests from the server to the client for the given method + """ + self.on_request_handlers[method] = cb + + def on_notification(self, method: str, cb: Callable[[Any], None]) -> None: + """ + Register the callback function to handle notifications from the server to the client for the given method + """ + self.on_notification_handlers[method] = cb + + def on_any_notification(self, cb: Callable[[str, Any], None]) -> None: + """ + Register an observer that is invoked for every notification received from the server. + """ + self._notification_observers.append(cb) + + def _response_handler(self, response: StringDict) -> None: + """ + Handle the response received from the server for a request, using the id to determine the request + """ + response_id = response["id"] + with self._response_handlers_lock: + request = self._pending_requests.pop(response_id, None) + if request is None and isinstance(response_id, str) and response_id.isdigit(): + request = self._pending_requests.pop(int(response_id), None) + + if request is None: # need to convert response_id to the right type + log.debug("Request interrupted by user or not found for ID %s", response_id) + return + + if "result" in response and "error" not in response: + request.on_result(response["result"]) + elif "result" not in response and "error" in response: + request.on_error(LSPError.from_lsp(response["error"])) + else: + request.on_error(LSPError(ErrorCodes.InvalidRequest, "")) + + def _request_handler(self, response: StringDict) -> None: + """ + Handle the request received from the server: call the appropriate callback function and return the result + """ + method = response.get("method", "") + params = response.get("params") + request_id = response.get("id") + handler = self.on_request_handlers.get(method) + if not handler: + self.send_error_response( + request_id, + LSPError( + ErrorCodes.MethodNotFound, + f"method '{method}' not handled on client.", + ), + ) + return + try: + self.send_response(request_id, handler(params)) + except LSPError as ex: + self.send_error_response(request_id, ex) + except Exception as ex: + self.send_error_response(request_id, LSPError(ErrorCodes.InternalError, str(ex))) + + def _notification_handler(self, response: StringDict) -> None: + """ + Handle the notification received from the server: call the appropriate callback function + """ + method = response.get("method", "") + params = response.get("params") + + for observer in self._notification_observers: + try: + observer(method, params) + except asyncio.CancelledError: + return + except Exception as ex: + if not self._is_shutting_down: + log.error("Error handling notification observer for method '%s': %s", method, ex, exc_info=ex) + + handler = self.on_notification_handlers.get(method) + if not handler: + log.warning("Unhandled method '%s'", method) + return + try: + handler(params) + except asyncio.CancelledError: + return + except Exception as ex: + if not self._is_shutting_down: + log.error("Error handling notification for method '%s': %s", method, ex, exc_info=ex) + + +class StdioLanguageServer(LanguageServerInterface): + """ + Represents a language server interface where the language server is launched as a subprocess + and communication takes place over the process' stdin/stdout streams. """ def __init__( @@ -137,50 +403,26 @@ class LanguageServerProcess: start_independent_lsp_process: bool = True, request_timeout: float | None = None, ) -> None: - self.language = language - self._determine_log_level = determine_log_level - self.send = LanguageServerRequest(self) - self.notify = LspNotification(self.send_notification) + """ + :param process_launch_info: the information required to launch the language server process + :param language: the language + :param determine_log_level: a function for log lines read from stderr, which determines the log level + :param logger: the trace logger function + :param start_independent_lsp_process: whether to start the language server process in an independent process group + :param request_timeout: the timeout, in seconds, for all requests sent to the language server. If None, no timeout will be applied. + """ + super().__init__(language, determine_log_level, logger, request_timeout) self.process_launch_info = process_launch_info self.process: subprocess.Popen[bytes] | None = None - self._is_shutting_down = False - - self.request_id = 1 - self._pending_requests: dict[Any, Request] = {} - self.on_request_handlers: dict[str, Callable[[Any], Any]] = {} - self.on_notification_handlers: dict[str, Callable[[Any], None]] = {} - self._notification_observers: list[Callable[[str, Any], None]] = [] - self._trace_log_fn = logger - self.tasks: dict[int, Any] = {} - self.task_counter = 0 - self.loop = None self.start_independent_lsp_process = start_independent_lsp_process - self._request_timeout = request_timeout - # Add thread locks for shared resources to prevent race conditions self._stdin_lock = threading.Lock() - self._request_id_lock = threading.Lock() - self._response_handlers_lock = threading.Lock() - self._tasks_lock = threading.Lock() - - def set_request_timeout(self, timeout: float | None) -> None: - """ - :param timeout: the timeout, in seconds, for all requests sent to the language server. - """ - self._request_timeout = timeout def is_running(self) -> bool: - """ - Checks if the language server process is currently running. - """ return self.process is not None and self.process.returncode is None - def start(self) -> None: - """ - Starts the language server process and creates a task to continuously read from its stdout to handle communications - from the server to the client - """ + def _start(self) -> None: child_proc_env = os.environ.copy() child_proc_env.update(self.process_launch_info.env) @@ -295,25 +537,6 @@ class LanguageServerProcess: except Exception: pass - def shutdown(self) -> None: - """ - Perform the shutdown sequence for the client, including sending the shutdown request to the server and notifying it of exit - """ - self._is_shutting_down = True - log.info("Sending shutdown request to server") - self.send.shutdown() - log.info("Received shutdown response from server") - log.info("Sending exit notification to server") - self.notify.exit() - log.info("Sent exit notification to server") - - def _trace(self, src: str, dest: str, message: str | StringDict) -> None: - """ - Traces LS communication by logging the message with the source and destination of the message - """ - if self._trace_log_fn is not None: - self._trace_log_fn(src, dest, message) - def _read_bytes_from_process(self, process, stream, num_bytes) -> bytes: # type: ignore """Read exactly num_bytes from process stdout""" data = b"" @@ -394,92 +617,6 @@ class LanguageServerProcess: else: log.info("Language server stderr reader thread has terminated") - def _handle_body(self, body: bytes) -> None: - """ - Parse the body text received from the language server process and invoke the appropriate handler - """ - try: - self._receive_payload(json.loads(body)) - except OSError as ex: - log.error(f"Error processing payload: {ex}", exc_info=ex) - except UnicodeDecodeError as ex: - log.error(f"Decoding error for encoding={ENCODING}: {ex}") - except json.JSONDecodeError as ex: - log.error(f"JSON decoding error: {ex}") - - def _receive_payload(self, payload: StringDict) -> None: - """ - Determine if the payload received from server is for a request, response, or notification and invoke the appropriate handler - """ - self._trace("ls", "solidlsp", payload) - try: - if "method" in payload: - if "id" in payload: - self._request_handler(payload) - else: - self._notification_handler(payload) - elif "id" in payload: - self._response_handler(payload) - else: - log.error(f"Unknown payload type: {payload}") - except Exception as err: - log.error(f"Error handling server payload: {err}") - - def send_notification(self, method: str, params: dict | None = None) -> None: - """ - Send notification pertaining to the given method to the server with the given parameters - """ - self._send_payload(make_notification(method, params)) - - def send_response(self, request_id: Any, params: PayloadLike) -> None: - """ - Send response to the given request id to the server with the given parameters - """ - self._send_payload(make_response(request_id, params)) - - def send_error_response(self, request_id: Any, err: LSPError) -> None: - """ - Send error response to the given request id to the server with the given error - """ - self._send_payload(make_error_response(request_id, err)) - - def _cancel_pending_requests(self, exception: Exception) -> None: - """ - Cancel all pending requests by setting their results to an error - """ - with self._response_handlers_lock: - log.info("Cancelling %d pending language server requests", len(self._pending_requests)) - for request in self._pending_requests.values(): - log.info("Cancelling %s", request) - request.on_error(exception) - self._pending_requests.clear() - - def send_request(self, method: str, params: dict | None = None) -> PayloadLike: - """ - Send request to the server, register the request id, and wait for the response - """ - with self._request_id_lock: - request_id = self.request_id - self.request_id += 1 - - request = Request(request_id=request_id, method=method) - log.debug("Starting: %s", request) - - with self._response_handlers_lock: - self._pending_requests[request_id] = request - - self._send_payload(make_request(method, request_id, params)) - - log.debug("Waiting for response to request %s with params:\n%s", method, params) - result = request.get_result(timeout=self._request_timeout) - log.debug("Completed: %s", request) - - if result.is_error(): - raise SolidLSPException(f"Error processing request {method} with params:\n{params}", cause=result.error) from result.error - - log.debug("Returning result:\n%s", result.payload) - return result.payload - def _send_payload(self, payload: StringDict) -> None: """ Send the payload to the server by writing to its stdin asynchronously. @@ -498,94 +635,3 @@ class LanguageServerProcess: # Log the error but don't raise to prevent cascading failures log.error(f"Failed to write to stdin: {e}") return - - def on_request(self, method: str, cb: Callable[[Any], Any]) -> None: - """ - Register the callback function to handle requests from the server to the client for the given method - """ - self.on_request_handlers[method] = cb - - def on_notification(self, method: str, cb: Callable[[Any], None]) -> None: - """ - Register the callback function to handle notifications from the server to the client for the given method - """ - self.on_notification_handlers[method] = cb - - def on_any_notification(self, cb: Callable[[str, Any], None]) -> None: - """ - Register an observer that is invoked for every notification received from the server. - """ - self._notification_observers.append(cb) - - def _response_handler(self, response: StringDict) -> None: - """ - Handle the response received from the server for a request, using the id to determine the request - """ - response_id = response["id"] - with self._response_handlers_lock: - request = self._pending_requests.pop(response_id, None) - if request is None and isinstance(response_id, str) and response_id.isdigit(): - request = self._pending_requests.pop(int(response_id), None) - - if request is None: # need to convert response_id to the right type - log.debug("Request interrupted by user or not found for ID %s", response_id) - return - - if "result" in response and "error" not in response: - request.on_result(response["result"]) - elif "result" not in response and "error" in response: - request.on_error(LSPError.from_lsp(response["error"])) - else: - request.on_error(LSPError(ErrorCodes.InvalidRequest, "")) - - def _request_handler(self, response: StringDict) -> None: - """ - Handle the request received from the server: call the appropriate callback function and return the result - """ - method = response.get("method", "") - params = response.get("params") - request_id = response.get("id") - handler = self.on_request_handlers.get(method) - if not handler: - self.send_error_response( - request_id, - LSPError( - ErrorCodes.MethodNotFound, - f"method '{method}' not handled on client.", - ), - ) - return - try: - self.send_response(request_id, handler(params)) - except LSPError as ex: - self.send_error_response(request_id, ex) - except Exception as ex: - self.send_error_response(request_id, LSPError(ErrorCodes.InternalError, str(ex))) - - def _notification_handler(self, response: StringDict) -> None: - """ - Handle the notification received from the server: call the appropriate callback function - """ - method = response.get("method", "") - params = response.get("params") - - for observer in self._notification_observers: - try: - observer(method, params) - except asyncio.CancelledError: - return - except Exception as ex: - if not self._is_shutting_down: - log.error("Error handling notification observer for method '%s': %s", method, ex, exc_info=ex) - - handler = self.on_notification_handlers.get(method) - if not handler: - log.warning("Unhandled method '%s'", method) - return - try: - handler(params) - except asyncio.CancelledError: - return - except Exception as ex: - if not self._is_shutting_down: - log.error("Error handling notification for method '%s': %s", method, ex, exc_info=ex) diff --git a/src/solidlsp/ls_request.py b/src/solidlsp/ls_request.py index 2b2f6b8c..e2a42ced 100644 --- a/src/solidlsp/ls_request.py +++ b/src/solidlsp/ls_request.py @@ -3,11 +3,11 @@ from typing import TYPE_CHECKING, Any, Union from solidlsp.lsp_protocol_handler import lsp_types if TYPE_CHECKING: - from .ls_process import LanguageServerProcess + from .ls_process import LanguageServerInterface class LanguageServerRequest: - def __init__(self, handler: "LanguageServerProcess"): + def __init__(self, handler: "LanguageServerInterface"): self.handler = handler def _send_request(self, method: str, params: Any | None = None) -> Any: