From c3a58cdadd8be85b79d773122e807862a3da3a2f Mon Sep 17 00:00:00 2001 From: Blaž Hrastnik Date: Tue, 31 Aug 2021 16:03:06 +0900 Subject: lsp: Refactor capabilities as an async OnceCell First step in making LSP init asynchronous --- helix-lsp/src/lib.rs | 29 ++++++++++++++++++++++++++--- 1 file changed, 26 insertions(+), 3 deletions(-) (limited to 'helix-lsp/src/lib.rs') diff --git a/helix-lsp/src/lib.rs b/helix-lsp/src/lib.rs index 72606b70..a118239f 100644 --- a/helix-lsp/src/lib.rs +++ b/helix-lsp/src/lib.rs @@ -312,17 +312,40 @@ impl Registry { Entry::Vacant(entry) => { // initialize a new client let id = self.counter.fetch_add(1, Ordering::Relaxed); - let (mut client, incoming) = Client::start( + let (client, incoming) = Client::start( &config.command, &config.args, serde_json::from_str(language_config.config.as_deref().unwrap_or("")).ok(), id, )?; - // TODO: run this async without blocking - futures_executor::block_on(client.initialize())?; s_incoming.push(UnboundedReceiverStream::new(incoming)); let client = Arc::new(client); + let _client = client.clone(); + let initialize = tokio::spawn(async move { + use futures_util::TryFutureExt; + + let value = _client + .capabilities + .get_or_try_init(|| { + _client + .initialize() + .map_ok(|response| response.capabilities) + }) + .await; + + value.expect("failed to initialize capabilities"); + + // next up, notify + _client + .notify::(lsp::InitializedParams {}) + .await + .unwrap(); + }); + + // TODO: remove this block + futures_executor::block_on(initialize).map_err(|_| anyhow::anyhow!("bail"))?; + entry.insert((id, client.clone())); Ok(client) } -- cgit v1.2.3-70-g09d2 From 5a558e0d8e20eb5b5d474e0f27fd51f4c633dd80 Mon Sep 17 00:00:00 2001 From: Blaž Hrastnik Date: Tue, 31 Aug 2021 16:48:59 +0900 Subject: lsp: Delay requests & notifications until initialization is complete --- helix-lsp/src/client.rs | 15 +++++--- helix-lsp/src/lib.rs | 11 +++--- helix-lsp/src/transport.rs | 91 ++++++++++++++++++++++++++++++++++++++-------- 3 files changed, 91 insertions(+), 26 deletions(-) (limited to 'helix-lsp/src/lib.rs') diff --git a/helix-lsp/src/client.rs b/helix-lsp/src/client.rs index 87078c69..02cd5747 100644 --- a/helix-lsp/src/client.rs +++ b/helix-lsp/src/client.rs @@ -9,13 +9,16 @@ use lsp_types as lsp; use serde_json::Value; use std::future::Future; use std::process::Stdio; -use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{ + atomic::{AtomicU64, Ordering}, + Arc, +}; use tokio::{ io::{BufReader, BufWriter}, process::{Child, Command}, sync::{ mpsc::{channel, UnboundedReceiver, UnboundedSender}, - OnceCell, + Notify, OnceCell, }, }; @@ -31,12 +34,13 @@ pub struct Client { } impl Client { + #[allow(clippy::type_complexity)] pub fn start( cmd: &str, args: &[String], config: Option, id: usize, - ) -> Result<(Self, UnboundedReceiver<(usize, Call)>)> { + ) -> Result<(Self, UnboundedReceiver<(usize, Call)>, Arc)> { let process = Command::new(cmd) .args(args) .stdin(Stdio::piped()) @@ -53,7 +57,8 @@ impl Client { let reader = BufReader::new(process.stdout.take().expect("Failed to open stdout")); let stderr = BufReader::new(process.stderr.take().expect("Failed to open stderr")); - let (server_rx, server_tx) = Transport::start(reader, writer, stderr, id); + let (server_rx, server_tx, initialize_notify) = + Transport::start(reader, writer, stderr, id); let client = Self { id, @@ -65,7 +70,7 @@ impl Client { config, }; - Ok((client, server_rx)) + Ok((client, server_rx, initialize_notify)) } pub fn id(&self) -> usize { diff --git a/helix-lsp/src/lib.rs b/helix-lsp/src/lib.rs index a118239f..3a761ad0 100644 --- a/helix-lsp/src/lib.rs +++ b/helix-lsp/src/lib.rs @@ -312,7 +312,7 @@ impl Registry { Entry::Vacant(entry) => { // initialize a new client let id = self.counter.fetch_add(1, Ordering::Relaxed); - let (client, incoming) = Client::start( + let (client, incoming, initialize_notify) = Client::start( &config.command, &config.args, serde_json::from_str(language_config.config.as_deref().unwrap_or("")).ok(), @@ -322,9 +322,9 @@ impl Registry { let client = Arc::new(client); let _client = client.clone(); - let initialize = tokio::spawn(async move { + // Initialize the client asynchronously + tokio::spawn(async move { use futures_util::TryFutureExt; - let value = _client .capabilities .get_or_try_init(|| { @@ -341,10 +341,9 @@ impl Registry { .notify::(lsp::InitializedParams {}) .await .unwrap(); - }); - // TODO: remove this block - futures_executor::block_on(initialize).map_err(|_| anyhow::anyhow!("bail"))?; + initialize_notify.notify_one(); + }); entry.insert((id, client.clone())); Ok(client) diff --git a/helix-lsp/src/transport.rs b/helix-lsp/src/transport.rs index 9353de20..071c5b93 100644 --- a/helix-lsp/src/transport.rs +++ b/helix-lsp/src/transport.rs @@ -1,7 +1,7 @@ use crate::{Error, Result}; use anyhow::Context; use jsonrpc_core as jsonrpc; -use log::{debug, error, info, warn}; +use log::{error, info}; use serde::{Deserialize, Serialize}; use serde_json::Value; use std::collections::HashMap; @@ -11,7 +11,7 @@ use tokio::{ process::{ChildStderr, ChildStdin, ChildStdout}, sync::{ mpsc::{unbounded_channel, Sender, UnboundedReceiver, UnboundedSender}, - Mutex, + Mutex, Notify, }, }; @@ -51,9 +51,11 @@ impl Transport { ) -> ( UnboundedReceiver<(usize, jsonrpc::Call)>, UnboundedSender, + Arc, ) { let (client_tx, rx) = unbounded_channel(); let (tx, client_rx) = unbounded_channel(); + let notify = Arc::new(Notify::new()); let transport = Self { id, @@ -64,9 +66,14 @@ impl Transport { tokio::spawn(Self::recv(transport.clone(), server_stdout, client_tx)); tokio::spawn(Self::err(transport.clone(), server_stderr)); - tokio::spawn(Self::send(transport, server_stdin, client_rx)); - - (rx, tx) + tokio::spawn(Self::send( + transport, + server_stdin, + client_rx, + notify.clone(), + )); + + (rx, tx, notify) } async fn recv_server_message( @@ -82,7 +89,8 @@ impl Transport { // debug!("<- header {:?}", buffer); - if header.is_empty() { + if buffer == "\r\n" { + // look for an empty CRLF line break; } @@ -99,7 +107,8 @@ impl Transport { // Workaround: Some non-conformant language servers will output logging and other garbage // into the same stream as JSON-RPC messages. This can also happen from shell scripts that spawn // the server. Skip such lines and log a warning. - warn!("Failed to parse header: {:?}", header); + + // warn!("Failed to parse header: {:?}", header); } } } @@ -261,15 +270,67 @@ impl Transport { transport: Arc, mut server_stdin: BufWriter, mut client_rx: UnboundedReceiver, + initialize_notify: Arc, ) { - while let Some(msg) = client_rx.recv().await { - match transport - .send_payload_to_server(&mut server_stdin, msg) - .await - { - Ok(_) => {} - Err(err) => { - error!("err: <- {:?}", err); + let mut pending_messages: Vec = Vec::new(); + let mut is_pending = true; + + // Determine if a message is allowed to be sent early + fn is_initialize(payload: &Payload) -> bool { + use lsp_types::{ + notification::{Initialized, Notification}, + request::{Initialize, Request}, + }; + match payload { + Payload::Request { + value: jsonrpc::MethodCall { method, .. }, + .. + } if method == Initialize::METHOD => true, + Payload::Notification(jsonrpc::Notification { method, .. }) + if method == Initialized::METHOD => + { + true + } + _ => false, + } + } + + // TODO: events that use capabilities need to do the right thing + + loop { + tokio::select! { + biased; + _ = initialize_notify.notified() => { // TODO: notified is technically not cancellation safe + // server successfully initialized + is_pending = false; + // drain the pending queue and send payloads to server + for msg in pending_messages.drain(..) { + log::info!("Draining pending message {:?}", msg); + match transport.send_payload_to_server(&mut server_stdin, msg).await { + Ok(_) => {} + Err(err) => { + error!("err: <- {:?}", err); + } + } + } + } + msg = client_rx.recv() => { + if let Some(msg) = msg { + if is_pending && !is_initialize(&msg) { + log::info!("Language server not initialized, delaying request"); + pending_messages.push(msg); + } else { + match transport.send_payload_to_server(&mut server_stdin, msg).await { + Ok(_) => {} + Err(err) => { + error!("err: <- {:?}", err); + } + } + } + } else { + // channel closed + break; + } } } } -- cgit v1.2.3-70-g09d2 From 48fd4843fc4a28bfd05ea01ef0d10f4ea816db20 Mon Sep 17 00:00:00 2001 From: Blaž Hrastnik Date: Thu, 2 Sep 2021 11:10:00 +0900 Subject: lsp: Outdated comment --- helix-lsp/src/lib.rs | 26 -------------------------- 1 file changed, 26 deletions(-) (limited to 'helix-lsp/src/lib.rs') diff --git a/helix-lsp/src/lib.rs b/helix-lsp/src/lib.rs index 3a761ad0..e10c107b 100644 --- a/helix-lsp/src/lib.rs +++ b/helix-lsp/src/lib.rs @@ -437,32 +437,6 @@ impl LspProgressMap { } } -// REGISTRY = HashMap>> -// spawn one server per language type, need to spawn one per workspace if server doesn't support -// workspaces -// -// could also be a client per root dir -// -// storing a copy of Option>> on Document would make the LSP client easily -// accessible during edit/save callbacks -// -// the event loop needs to process all incoming streams, maybe we can just have that be a separate -// task that's continually running and store the state on the client, then use read lock to -// retrieve data during render -// -> PROBLEM: how do you trigger an update on the editor side when data updates? -// -// -> The data updates should pull all events until we run out so we don't frequently re-render -// -// -// v2: -// -// there should be a registry of lsp clients, one per language type (or workspace). -// the clients should lazy init on first access -// the client.initialize() should be called async and we buffer any requests until that completes -// there needs to be a way to process incoming lsp messages from all clients. -// -> notifications need to be dispatched to wherever -// -> requests need to generate a reply and travel back to the same lsp! - #[cfg(test)] mod tests { use super::{lsp, util::*, OffsetEncoding}; -- cgit v1.2.3-70-g09d2 From 46f3c69f06cc55f36bcc6244a9f96c2481836dea Mon Sep 17 00:00:00 2001 From: Blaž Hrastnik Date: Thu, 2 Sep 2021 13:55:08 +0900 Subject: lsp: Don't send notifications until initialize completes Then send open events for all documents with the LSP attached. --- helix-lsp/src/lib.rs | 98 +++++++++++++++++++++---------------------- helix-lsp/src/transport.rs | 29 ++++++++++++- helix-term/src/application.rs | 31 ++++++++++++++ helix-view/src/editor.rs | 5 ++- 4 files changed, 111 insertions(+), 52 deletions(-) (limited to 'helix-lsp/src/lib.rs') diff --git a/helix-lsp/src/lib.rs b/helix-lsp/src/lib.rs index e10c107b..7357c885 100644 --- a/helix-lsp/src/lib.rs +++ b/helix-lsp/src/lib.rs @@ -226,6 +226,8 @@ impl MethodCall { #[derive(Debug, PartialEq, Clone)] pub enum Notification { + // we inject this notification to signal the LSP is ready + Initialized, PublishDiagnostics(lsp::PublishDiagnosticsParams), ShowMessage(lsp::ShowMessageParams), LogMessage(lsp::LogMessageParams), @@ -237,6 +239,7 @@ impl Notification { use lsp::notification::Notification as _; let notification = match method { + lsp::notification::Initialized::METHOD => Self::Initialized, lsp::notification::PublishDiagnostics::METHOD => { let params: lsp::PublishDiagnosticsParams = params .parse() @@ -294,7 +297,7 @@ impl Registry { } } - pub fn get_by_id(&mut self, id: usize) -> Option<&Client> { + pub fn get_by_id(&self, id: usize) -> Option<&Client> { self.inner .values() .find(|(client_id, _)| client_id == &id) @@ -302,55 +305,52 @@ impl Registry { } pub fn get(&mut self, language_config: &LanguageConfiguration) -> Result> { - if let Some(config) = &language_config.language_server { - // avoid borrow issues - let inner = &mut self.inner; - let s_incoming = &mut self.incoming; - - match inner.entry(language_config.scope.clone()) { - Entry::Occupied(entry) => Ok(entry.get().1.clone()), - Entry::Vacant(entry) => { - // initialize a new client - let id = self.counter.fetch_add(1, Ordering::Relaxed); - let (client, incoming, initialize_notify) = Client::start( - &config.command, - &config.args, - serde_json::from_str(language_config.config.as_deref().unwrap_or("")).ok(), - id, - )?; - s_incoming.push(UnboundedReceiverStream::new(incoming)); - let client = Arc::new(client); - - let _client = client.clone(); - // Initialize the client asynchronously - tokio::spawn(async move { - use futures_util::TryFutureExt; - let value = _client - .capabilities - .get_or_try_init(|| { - _client - .initialize() - .map_ok(|response| response.capabilities) - }) - .await; - - value.expect("failed to initialize capabilities"); - - // next up, notify - _client - .notify::(lsp::InitializedParams {}) - .await - .unwrap(); - - initialize_notify.notify_one(); - }); - - entry.insert((id, client.clone())); - Ok(client) - } + let config = match &language_config.language_server { + Some(config) => config, + None => return Err(Error::LspNotDefined), + }; + + match self.inner.entry(language_config.scope.clone()) { + Entry::Occupied(entry) => Ok(entry.get().1.clone()), + Entry::Vacant(entry) => { + // initialize a new client + let id = self.counter.fetch_add(1, Ordering::Relaxed); + let (client, incoming, initialize_notify) = Client::start( + &config.command, + &config.args, + serde_json::from_str(language_config.config.as_deref().unwrap_or("")).ok(), + id, + )?; + self.incoming.push(UnboundedReceiverStream::new(incoming)); + let client = Arc::new(client); + + // Initialize the client asynchronously + let _client = client.clone(); + tokio::spawn(async move { + use futures_util::TryFutureExt; + let value = _client + .capabilities + .get_or_try_init(|| { + _client + .initialize() + .map_ok(|response| response.capabilities) + }) + .await; + + value.expect("failed to initialize capabilities"); + + // next up, notify + _client + .notify::(lsp::InitializedParams {}) + .await + .unwrap(); + + initialize_notify.notify_one(); + }); + + entry.insert((id, client.clone())); + Ok(client) } - } else { - Err(Error::LspNotDefined) } } diff --git a/helix-lsp/src/transport.rs b/helix-lsp/src/transport.rs index 071c5b93..cf7e66a8 100644 --- a/helix-lsp/src/transport.rs +++ b/helix-lsp/src/transport.rs @@ -64,11 +64,16 @@ impl Transport { let transport = Arc::new(transport); - tokio::spawn(Self::recv(transport.clone(), server_stdout, client_tx)); + tokio::spawn(Self::recv( + transport.clone(), + server_stdout, + client_tx.clone(), + )); tokio::spawn(Self::err(transport.clone(), server_stderr)); tokio::spawn(Self::send( transport, server_stdin, + client_tx, client_rx, notify.clone(), )); @@ -269,6 +274,7 @@ impl Transport { async fn send( transport: Arc, mut server_stdin: BufWriter, + mut client_tx: UnboundedSender<(usize, jsonrpc::Call)>, mut client_rx: UnboundedReceiver, initialize_notify: Arc, ) { @@ -303,6 +309,22 @@ impl Transport { _ = initialize_notify.notified() => { // TODO: notified is technically not cancellation safe // server successfully initialized is_pending = false; + + use lsp_types::notification::Notification; + // Hack: inject an initialized notification so we trigger code that needs to happen after init + let notification = ServerMessage::Call(jsonrpc::Call::Notification(jsonrpc::Notification { + jsonrpc: None, + + method: lsp_types::notification::Initialized::METHOD.to_string(), + params: jsonrpc::Params::None, + })); + match transport.process_server_message(&mut client_tx, notification).await { + Ok(_) => {} + Err(err) => { + error!("err: <- {:?}", err); + } + } + // drain the pending queue and send payloads to server for msg in pending_messages.drain(..) { log::info!("Draining pending message {:?}", msg); @@ -317,6 +339,11 @@ impl Transport { msg = client_rx.recv() => { if let Some(msg) = msg { if is_pending && !is_initialize(&msg) { + // ignore notifications + if let Payload::Notification(_) = msg { + continue; + } + log::info!("Language server not initialized, delaying request"); pending_messages.push(msg); } else { diff --git a/helix-term/src/application.rs b/helix-term/src/application.rs index d3b65a4f..e21c5504 100644 --- a/helix-term/src/application.rs +++ b/helix-term/src/application.rs @@ -275,6 +275,37 @@ impl Application { }; match notification { + Notification::Initialized => { + let language_server = + match self.editor.language_servers.get_by_id(server_id) { + Some(language_server) => language_server, + None => { + warn!("can't find language server with id `{}`", server_id); + return; + } + }; + + let docs = self.editor.documents().filter(|doc| { + doc.language_server().map(|server| server.id()) == Some(server_id) + }); + + // trigger textDocument/didOpen for docs that are already open + for doc in docs { + // TODO: extract and share with editor.open + let language_id = doc + .language() + .and_then(|s| s.split('.').last()) // source.rust + .map(ToOwned::to_owned) + .unwrap_or_default(); + + tokio::spawn(language_server.text_document_did_open( + doc.url().unwrap(), + doc.version(), + doc.text(), + language_id, + )); + } + } Notification::PublishDiagnostics(params) => { let path = params.uri.to_file_path().unwrap(); let doc = self.editor.document_by_path_mut(&path); diff --git a/helix-view/src/editor.rs b/helix-view/src/editor.rs index c8abd5b5..3d2d4a87 100644 --- a/helix-view/src/editor.rs +++ b/helix-view/src/editor.rs @@ -255,20 +255,21 @@ impl Editor { .and_then(|language| self.language_servers.get(language).ok()); if let Some(language_server) = language_server { - doc.set_language_server(Some(language_server.clone())); - let language_id = doc .language() .and_then(|s| s.split('.').last()) // source.rust .map(ToOwned::to_owned) .unwrap_or_default(); + // TODO: this now races with on_init code if the init happens too quickly tokio::spawn(language_server.text_document_did_open( doc.url().unwrap(), doc.version(), doc.text(), language_id, )); + + doc.set_language_server(Some(language_server)); } let id = self.documents.insert(doc); -- cgit v1.2.3-70-g09d2 From ef532e0c0df3e9f8bf4ac5af74b54f32b7ea2728 Mon Sep 17 00:00:00 2001 From: Kirawi Date: Wed, 15 Sep 2021 01:58:06 -0400 Subject: log errors produced when trying to initialize the LSP (#746) --- helix-lsp/src/lib.rs | 10 +++++++++- helix-view/src/editor.rs | 12 ++++++++---- 2 files changed, 17 insertions(+), 5 deletions(-) (limited to 'helix-lsp/src/lib.rs') diff --git a/helix-lsp/src/lib.rs b/helix-lsp/src/lib.rs index 7357c885..35cff754 100644 --- a/helix-lsp/src/lib.rs +++ b/helix-lsp/src/lib.rs @@ -318,7 +318,15 @@ impl Registry { let (client, incoming, initialize_notify) = Client::start( &config.command, &config.args, - serde_json::from_str(language_config.config.as_deref().unwrap_or("")).ok(), + serde_json::from_str(language_config.config.as_deref().unwrap_or("")) + .map_err(|e| { + log::error!( + "LSP Config, {}, in `languages.toml` for `{}`", + e, + language_config.scope() + ) + }) + .ok(), id, )?; self.incoming.push(UnboundedReceiverStream::new(incoming)); diff --git a/helix-view/src/editor.rs b/helix-view/src/editor.rs index 52a0060c..a3d0d032 100644 --- a/helix-view/src/editor.rs +++ b/helix-view/src/editor.rs @@ -249,10 +249,14 @@ impl Editor { let mut doc = Document::open(&path, None, Some(&self.theme), Some(&self.syn_loader))?; // try to find a language server based on the language name - let language_server = doc - .language - .as_ref() - .and_then(|language| self.language_servers.get(language).ok()); + let language_server = doc.language.as_ref().and_then(|language| { + self.language_servers + .get(language) + .map_err(|e| { + log::error!("Failed to get LSP, {}, for `{}`", e, language.scope()) + }) + .ok() + }); if let Some(language_server) = language_server { let language_id = doc -- cgit v1.2.3-70-g09d2