-
Notifications
You must be signed in to change notification settings - Fork 1.2k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
ruff server no longer hangs after shutdown (#11222)
## Summary Fixes #11207. The server would hang after handling a shutdown request on `IoThreads::join()` because a global sender (`MESSENGER`, used to send `window/showMessage` notifications) would remain allocated even after the event loop finished, which kept the writer I/O thread channel open. To fix this, I've made a few structural changes to `ruff server`. I've wrapped the send/receive channels and thread join handle behind a new struct, `Connection`, which facilitates message sending and receiving, and also runs `IoThreads::join()` after the event loop finishes. To control the number of sender channels, the `Connection` wraps the sender channel in an `Arc` and only allows the creation of a wrapper type, `ClientSender`, which hold a weak reference to this `Arc` instead of direct channel access. The wrapper type implements the channel methods directly to prevent access to the inner channel (which would allow the channel to be cloned). ClientSender's function is analogous to [`WeakSender` in `tokio`](https://docs.rs/tokio/latest/tokio/sync/mpsc/struct.WeakSender.html). Additionally, the receiver channel cannot be accessed directly - the `Connection` only exposes an iterator over it. These changes will guarantee that all channels are closed before the I/O threads are joined. ## Test Plan Repeatedly open and close an editor utilizing `ruff server` while observing the task monitor. The net total amount of open `ruff` instances should be zero once all editor windows have closed. The following logs should also appear after the server is shut down: <img width="835" alt="Screenshot 2024-04-30 at 3 56 22 PM" src="https://github.com/astral-sh/ruff/assets/19577865/404b74f5-ef08-4bb4-9fa2-72e72b946695"> This can be tested on VS Code by changing the settings and then checking `Output`.
- Loading branch information
1 parent
9e69cd6
commit dfbeca5
Showing
5 changed files
with
189 additions
and
53 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,144 @@ | ||
| use lsp_server as lsp; | ||
| use lsp_types::{notification::Notification, request::Request}; | ||
| use std::sync::{Arc, Weak}; | ||
|
|
||
| type ConnectionSender = crossbeam::channel::Sender<lsp::Message>; | ||
| type ConnectionReceiver = crossbeam::channel::Receiver<lsp::Message>; | ||
|
|
||
| /// A builder for `Connection` that handles LSP initialization. | ||
| pub(crate) struct ConnectionInitializer { | ||
| connection: lsp::Connection, | ||
| threads: lsp::IoThreads, | ||
| } | ||
|
|
||
| /// Handles inbound and outbound messages with the client. | ||
| pub(crate) struct Connection { | ||
| sender: Arc<ConnectionSender>, | ||
| receiver: ConnectionReceiver, | ||
| threads: lsp::IoThreads, | ||
| } | ||
|
|
||
| impl ConnectionInitializer { | ||
| /// Create a new LSP server connection over stdin/stdout. | ||
| pub(super) fn stdio() -> Self { | ||
| let (connection, threads) = lsp::Connection::stdio(); | ||
| Self { | ||
| connection, | ||
| threads, | ||
| } | ||
| } | ||
|
|
||
| /// Starts the initialization process with the client by listening for an initialization request. | ||
| /// Returns a request ID that should be passed into `initialize_finish` later, | ||
| /// along with the initialization parameters that were provided. | ||
| pub(super) fn initialize_start( | ||
| &self, | ||
| ) -> crate::Result<(lsp::RequestId, lsp_types::InitializeParams)> { | ||
| let (id, params) = self.connection.initialize_start()?; | ||
| Ok((id, serde_json::from_value(params)?)) | ||
| } | ||
|
|
||
| /// Finishes the initialization process with the client, | ||
| /// returning an initialized `Connection`. | ||
| pub(super) fn initialize_finish( | ||
| self, | ||
| id: lsp::RequestId, | ||
| server_capabilities: &lsp_types::ServerCapabilities, | ||
| name: &str, | ||
| version: &str, | ||
| ) -> crate::Result<Connection> { | ||
| self.connection.initialize_finish( | ||
| id, | ||
| serde_json::json!({ | ||
| "capabilities": server_capabilities, | ||
| "serverInfo": { | ||
| "name": name, | ||
| "version": version | ||
| } | ||
| }), | ||
| )?; | ||
| let Self { | ||
| connection: lsp::Connection { sender, receiver }, | ||
| threads, | ||
| } = self; | ||
| Ok(Connection { | ||
| sender: Arc::new(sender), | ||
| receiver, | ||
| threads, | ||
| }) | ||
| } | ||
| } | ||
|
|
||
| impl Connection { | ||
| /// Make a new `ClientSender` for sending messages to the client. | ||
| pub(super) fn make_sender(&self) -> ClientSender { | ||
| ClientSender { | ||
| weak_sender: Arc::downgrade(&self.sender), | ||
| } | ||
| } | ||
|
|
||
| /// An iterator over incoming messages from the client. | ||
| pub(super) fn incoming(&self) -> crossbeam::channel::Iter<lsp::Message> { | ||
| self.receiver.iter() | ||
| } | ||
|
|
||
| /// Check and respond to any incoming shutdown requests; returns`true` if the server should be shutdown. | ||
| pub(super) fn handle_shutdown(&self, message: &lsp::Message) -> crate::Result<bool> { | ||
| match message { | ||
| lsp::Message::Request(lsp::Request { id, method, .. }) | ||
| if method == lsp_types::request::Shutdown::METHOD => | ||
| { | ||
| self.sender | ||
| .send(lsp::Response::new_ok(id.clone(), ()).into())?; | ||
| tracing::info!("Shutdown request received. Waiting for an exit notification..."); | ||
| match self.receiver.recv_timeout(std::time::Duration::from_secs(30))? { | ||
| lsp::Message::Notification(lsp::Notification { method, .. }) if method == lsp_types::notification::Exit::METHOD => { | ||
| tracing::info!("Exit notification received. Server shutting down..."); | ||
| Ok(true) | ||
| }, | ||
| message => anyhow::bail!("Server received unexpected message {message:?} while waiting for exit notification") | ||
| } | ||
| } | ||
| lsp::Message::Notification(lsp::Notification { method, .. }) | ||
| if method == lsp_types::notification::Exit::METHOD => | ||
| { | ||
| tracing::error!("Server received an exit notification before a shutdown request was sent. Exiting..."); | ||
| Ok(true) | ||
| } | ||
| _ => Ok(false), | ||
| } | ||
| } | ||
|
|
||
| /// Join the I/O threads that underpin this connection. | ||
| /// This is guaranteed to be nearly immediate since | ||
| /// we close the only active channels to these threads prior | ||
| /// to joining them. | ||
| pub(super) fn close(self) -> crate::Result<()> { | ||
| std::mem::drop( | ||
| Arc::into_inner(self.sender) | ||
| .expect("the client sender shouldn't have more than one strong reference"), | ||
| ); | ||
| std::mem::drop(self.receiver); | ||
| self.threads.join()?; | ||
| Ok(()) | ||
| } | ||
| } | ||
|
|
||
| /// A weak reference to an underlying sender channel, used for communication with the client. | ||
| /// If the `Connection` that created this `ClientSender` is dropped, any `send` calls will throw | ||
| /// an error. | ||
| #[derive(Clone, Debug)] | ||
| pub(crate) struct ClientSender { | ||
| weak_sender: Weak<ConnectionSender>, | ||
| } | ||
|
|
||
| // note: additional wrapper functions for senders may be implemented as needed. | ||
| impl ClientSender { | ||
| pub(crate) fn send(&self, msg: lsp::Message) -> crate::Result<()> { | ||
| let Some(sender) = self.weak_sender.upgrade() else { | ||
| anyhow::bail!("The connection with the client has been closed"); | ||
| }; | ||
|
|
||
| Ok(sender.send(msg)?) | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters