diff --git a/crates/flycheck/src/lib.rs b/crates/flycheck/src/lib.rs index 9e8205ae790..4dcab7a61f9 100644 --- a/crates/flycheck/src/lib.rs +++ b/crates/flycheck/src/lib.rs @@ -120,7 +120,13 @@ fn new( ) -> FlycheckActor { FlycheckActor { sender, config, workspace_root, last_update_req: None, check_process: None } } - + fn next_event(&self, inbox: &Receiver) -> Option { + let check_chan = self.check_process.as_ref().map(|(chan, _thread)| chan); + select! { + recv(inbox) -> msg => msg.ok().map(Event::Restart), + recv(check_chan.unwrap_or(&never())) -> msg => Some(Event::CheckEvent(msg.ok())), + } + } fn run(&mut self, inbox: Receiver) { // If we rerun the thread, we need to discard the previous check results first self.send(Message::ClearDiagnostics); @@ -167,15 +173,6 @@ fn run(&mut self, inbox: Receiver) { } } } - - fn next_event(&self, inbox: &Receiver) -> Option { - let check_chan = self.check_process.as_ref().map(|(chan, _thread)| chan); - select! { - recv(inbox) -> msg => msg.ok().map(Event::Restart), - recv(check_chan.unwrap_or(&never())) -> msg => Some(Event::CheckEvent(msg.ok())), - } - } - fn should_recheck(&mut self) -> bool { if let Some(_last_update_req) = &self.last_update_req { // We currently only request an update on save, as we need up to diff --git a/crates/rust-analyzer/src/dispatch.rs b/crates/rust-analyzer/src/dispatch.rs new file mode 100644 index 00000000000..5fdbed8ef39 --- /dev/null +++ b/crates/rust-analyzer/src/dispatch.rs @@ -0,0 +1,176 @@ +//! A visitor for downcasting arbitrary request (JSON) into a specific type. +use std::{panic, time::Instant}; + +use serde::{de::DeserializeOwned, Serialize}; + +use crate::{ + global_state::{GlobalState, GlobalStateSnapshot}, + lsp_utils::is_canceled, + main_loop::Task, + LspError, Result, +}; + +pub(crate) struct RequestDispatcher<'a> { + pub(crate) req: Option, + pub(crate) global_state: &'a mut GlobalState, + pub(crate) request_received: Instant, +} + +impl<'a> RequestDispatcher<'a> { + /// Dispatches the request onto the current thread + pub(crate) fn on_sync( + &mut self, + f: fn(&mut GlobalState, R::Params) -> Result, + ) -> Result<&mut Self> + where + R: lsp_types::request::Request + 'static, + R::Params: DeserializeOwned + panic::UnwindSafe + 'static, + R::Result: Serialize + 'static, + { + let (id, params) = match self.parse::() { + Some(it) => it, + None => { + return Ok(self); + } + }; + let world = panic::AssertUnwindSafe(&mut *self.global_state); + let task = panic::catch_unwind(move || { + let result = f(world.0, params); + result_to_task::(id, result) + }) + .map_err(|_| format!("sync task {:?} panicked", R::METHOD))?; + self.global_state.on_task(task); + Ok(self) + } + + /// Dispatches the request onto thread pool + pub(crate) fn on( + &mut self, + f: fn(GlobalStateSnapshot, R::Params) -> Result, + ) -> Result<&mut Self> + where + R: lsp_types::request::Request + 'static, + R::Params: DeserializeOwned + Send + 'static, + R::Result: Serialize + 'static, + { + let (id, params) = match self.parse::() { + Some(it) => it, + None => { + return Ok(self); + } + }; + + self.global_state.task_pool.0.spawn({ + let world = self.global_state.snapshot(); + move || { + let result = f(world, params); + result_to_task::(id, result) + } + }); + + Ok(self) + } + + pub(crate) fn finish(&mut self) { + match self.req.take() { + None => (), + Some(req) => { + log::error!("unknown request: {:?}", req); + let resp = lsp_server::Response::new_err( + req.id, + lsp_server::ErrorCode::MethodNotFound as i32, + "unknown request".to_string(), + ); + self.global_state.send(resp.into()); + } + } + } + + fn parse(&mut self) -> Option<(lsp_server::RequestId, R::Params)> + where + R: lsp_types::request::Request + 'static, + R::Params: DeserializeOwned + 'static, + { + let req = self.req.take()?; + let (id, params) = match req.extract::(R::METHOD) { + Ok(it) => it, + Err(req) => { + self.req = Some(req); + return None; + } + }; + self.global_state + .req_queue + .incoming + .register(id.clone(), (R::METHOD, self.request_received)); + Some((id, params)) + } +} + +fn result_to_task(id: lsp_server::RequestId, result: Result) -> Task +where + R: lsp_types::request::Request + 'static, + R::Params: DeserializeOwned + 'static, + R::Result: Serialize + 'static, +{ + let response = match result { + Ok(resp) => lsp_server::Response::new_ok(id, &resp), + Err(e) => match e.downcast::() { + Ok(lsp_error) => lsp_server::Response::new_err(id, lsp_error.code, lsp_error.message), + Err(e) => { + if is_canceled(&*e) { + lsp_server::Response::new_err( + id, + lsp_server::ErrorCode::ContentModified as i32, + "content modified".to_string(), + ) + } else { + lsp_server::Response::new_err( + id, + lsp_server::ErrorCode::InternalError as i32, + e.to_string(), + ) + } + } + }, + }; + Task::Respond(response) +} + +pub(crate) struct NotificationDispatcher<'a> { + pub(crate) not: Option, + pub(crate) global_state: &'a mut GlobalState, +} + +impl<'a> NotificationDispatcher<'a> { + pub(crate) fn on( + &mut self, + f: fn(&mut GlobalState, N::Params) -> Result<()>, + ) -> Result<&mut Self> + where + N: lsp_types::notification::Notification + 'static, + N::Params: DeserializeOwned + Send + 'static, + { + let not = match self.not.take() { + Some(it) => it, + None => return Ok(self), + }; + let params = match not.extract::(N::METHOD) { + Ok(it) => it, + Err(not) => { + self.not = Some(not); + return Ok(self); + } + }; + f(self.global_state, params)?; + Ok(self) + } + + pub(crate) fn finish(&mut self) { + if let Some(not) = &self.not { + if !not.method.starts_with("$/") { + log::error!("unhandled notification: {:?}", not); + } + } + } +} diff --git a/crates/rust-analyzer/src/global_state.rs b/crates/rust-analyzer/src/global_state.rs index de6b9568668..56d50c78997 100644 --- a/crates/rust-analyzer/src/global_state.rs +++ b/crates/rust-analyzer/src/global_state.rs @@ -5,7 +5,7 @@ use std::{convert::TryFrom, sync::Arc}; -use crossbeam_channel::{unbounded, Receiver}; +use crossbeam_channel::{unbounded, Receiver, Sender}; use flycheck::{FlycheckConfig, FlycheckHandle}; use lsp_types::Url; use parking_lot::RwLock; @@ -22,6 +22,7 @@ line_endings::LineEndings, main_loop::{ReqQueue, Task}, request_metrics::{LatestRequests, RequestMetrics}, + show_message, thread_pool::TaskPool, to_proto::url_from_abs_path, Result, @@ -66,6 +67,7 @@ fn default() -> Self { /// snapshot of the file systems, and `analysis_host`, which stores our /// incremental salsa database. pub(crate) struct GlobalState { + sender: Sender, pub(crate) config: Config, pub(crate) task_pool: (TaskPool, Receiver), pub(crate) analysis_host: AnalysisHost, @@ -95,6 +97,7 @@ pub(crate) struct GlobalStateSnapshot { impl GlobalState { pub(crate) fn new( + sender: Sender, workspaces: Vec, lru_capacity: Option, config: Config, @@ -162,6 +165,7 @@ pub(crate) fn new( }; let mut res = GlobalState { + sender, config, task_pool, analysis_host, @@ -252,6 +256,19 @@ pub(crate) fn collect_garbage(&mut self) { pub(crate) fn complete_request(&mut self, request: RequestMetrics) { self.latest_requests.write().record(request) } + + pub(crate) fn send(&mut self, message: lsp_server::Message) { + self.sender.send(message).unwrap() + } + pub(crate) fn show_message(&mut self, typ: lsp_types::MessageType, message: String) { + show_message(typ, message, &self.sender) + } +} + +impl Drop for GlobalState { + fn drop(&mut self) { + self.analysis_host.request_cancellation() + } } impl GlobalStateSnapshot { diff --git a/crates/rust-analyzer/src/lib.rs b/crates/rust-analyzer/src/lib.rs index ca788dd3cf9..d503fe96e2d 100644 --- a/crates/rust-analyzer/src/lib.rs +++ b/crates/rust-analyzer/src/lib.rs @@ -19,6 +19,7 @@ macro_rules! eprintln { mod global_state; mod main_loop; +mod dispatch; mod handlers; mod caps; mod cargo_target_spec; diff --git a/crates/rust-analyzer/src/lsp_utils.rs b/crates/rust-analyzer/src/lsp_utils.rs index 14adb8ae733..35917030c8d 100644 --- a/crates/rust-analyzer/src/lsp_utils.rs +++ b/crates/rust-analyzer/src/lsp_utils.rs @@ -1,12 +1,13 @@ //! Utilities for LSP-related boilerplate code. use std::{error::Error, ops::Range}; -use crate::from_proto; use crossbeam_channel::Sender; use lsp_server::{Message, Notification}; use ra_db::Canceled; use ra_ide::LineIndex; -use serde::{de::DeserializeOwned, Serialize}; +use serde::Serialize; + +use crate::from_proto; pub fn show_message( typ: lsp_types::MessageType, @@ -29,14 +30,6 @@ pub(crate) fn notification_is( notification.method == N::METHOD } -pub(crate) fn notification_cast(notification: Notification) -> Result -where - N: lsp_types::notification::Notification, - N::Params: DeserializeOwned, -{ - notification.extract(N::METHOD) -} - pub(crate) fn notification_new(params: N::Params) -> Notification where N: lsp_types::notification::Notification, diff --git a/crates/rust-analyzer/src/main_loop.rs b/crates/rust-analyzer/src/main_loop.rs index 1a9c5ee2cef..c2f43df1d0b 100644 --- a/crates/rust-analyzer/src/main_loop.rs +++ b/crates/rust-analyzer/src/main_loop.rs @@ -5,27 +5,26 @@ time::{Duration, Instant}, }; -use crossbeam_channel::{never, select, RecvError, Sender}; -use lsp_server::{Connection, ErrorCode, Notification, Request, RequestId, Response}; -use lsp_types::{request::Request as _, NumberOrString}; +use crossbeam_channel::{never, select, Receiver}; +use lsp_server::{Connection, Notification, Request, Response}; +use lsp_types::{notification::Notification as _, request::Request as _}; use ra_db::VfsPath; use ra_ide::{Canceled, FileId}; use ra_prof::profile; use ra_project_model::{PackageRoot, ProjectWorkspace}; -use serde::{de::DeserializeOwned, Serialize}; use crate::{ config::{Config, FilesWatcher, LinkedProject}, diagnostics::DiagnosticTask, + dispatch::{NotificationDispatcher, RequestDispatcher}, from_proto, - global_state::{file_id_to_url, GlobalState, GlobalStateSnapshot, Status}, + global_state::{file_id_to_url, GlobalState, Status}, handlers, lsp_ext, lsp_utils::{ - apply_document_changes, is_canceled, notification_cast, notification_is, notification_new, - show_message, + apply_document_changes, is_canceled, notification_is, notification_new, show_message, }, request_metrics::RequestMetrics, - LspError, Result, + Result, }; pub fn main_loop(config: Config, connection: Connection) -> Result<()> { @@ -50,7 +49,7 @@ pub fn main_loop(config: Config, connection: Connection) -> Result<()> { SetThreadPriority(thread, thread_priority_above_normal); } - let mut global_state = { + let global_state = { let workspaces = { if config.linked_projects.is_empty() && config.notifications.cargo_toml_not_found { show_message( @@ -113,40 +112,346 @@ pub fn main_loop(config: Config, connection: Connection) -> Result<()> { connection.sender.send(request.into()).unwrap(); } - GlobalState::new(workspaces, config.lru_capacity, config, req_queue) + GlobalState::new( + connection.sender.clone(), + workspaces, + config.lru_capacity, + config, + req_queue, + ) }; log::info!("server initialized, serving requests"); - { - loop { - log::trace!("selecting"); - let event = select! { - recv(&connection.receiver) -> msg => match msg { - Ok(msg) => Event::Lsp(msg), - Err(RecvError) => return Err("client exited without shutdown".into()), - }, - recv(&global_state.task_pool.1) -> task => Event::Task(task.unwrap()), - recv(global_state.task_receiver) -> task => match task { - Ok(task) => Event::Vfs(task), - Err(RecvError) => return Err("vfs died".into()), - }, - recv(global_state.flycheck.as_ref().map_or(&never(), |it| &it.1)) -> task => match task { - Ok(task) => Event::Flycheck(task), - Err(RecvError) => return Err("check watcher died".into()), - }, - }; - if let Event::Lsp(lsp_server::Message::Request(req)) = &event { - if connection.handle_shutdown(&req)? { - break; - }; - } - assert!(!global_state.vfs.read().0.has_changes()); - loop_turn(&connection, &mut global_state, event)?; - assert!(!global_state.vfs.read().0.has_changes()); + global_state.run(connection.receiver)?; + Ok(()) +} + +impl GlobalState { + fn next_event(&self, inbox: &Receiver) -> Option { + select! { + recv(inbox) -> msg => + msg.ok().map(Event::Lsp), + + recv(self.task_pool.1) -> task => + Some(Event::Task(task.unwrap())), + + recv(self.task_receiver) -> task => + Some(Event::Vfs(task.unwrap())), + + recv(self.flycheck.as_ref().map_or(&never(), |it| &it.1)) -> task => + Some(Event::Flycheck(task.unwrap())), } } - global_state.analysis_host.request_cancellation(); - Ok(()) + + fn run(mut self, inbox: Receiver) -> Result<()> { + while let Some(event) = self.next_event(&inbox) { + let loop_start = Instant::now(); + // NOTE: don't count blocking select! call as a loop-turn time + let _p = profile("main_loop_inner/loop-turn"); + + log::info!("loop turn = {:?}", event); + let queue_count = self.task_pool.0.len(); + if queue_count > 0 { + log::info!("queued count = {}", queue_count); + } + + let mut became_ready = false; + match event { + Event::Lsp(msg) => match msg { + lsp_server::Message::Request(req) => self.on_request(loop_start, req)?, + lsp_server::Message::Notification(not) => { + if not.method == lsp_types::notification::Exit::METHOD { + return Ok(()); + } + self.on_notification(not)?; + } + lsp_server::Message::Response(resp) => { + let handler = self.req_queue.outgoing.complete(resp.id.clone()); + handler(&mut self, resp) + } + }, + Event::Task(task) => { + self.on_task(task); + self.maybe_collect_garbage(); + } + Event::Vfs(task) => match task { + vfs::loader::Message::Loaded { files } => { + let vfs = &mut self.vfs.write().0; + for (path, contents) in files { + let path = VfsPath::from(path); + if !self.mem_docs.contains(&path) { + vfs.set_file_contents(path, contents) + } + } + } + vfs::loader::Message::Progress { n_total, n_done } => { + let state = if n_done == 0 { + Progress::Begin + } else if n_done < n_total { + Progress::Report + } else { + assert_eq!(n_done, n_total); + self.status = Status::Ready; + became_ready = true; + Progress::End + }; + report_progress( + &mut self, + "roots scanned", + state, + Some(format!("{}/{}", n_done, n_total)), + Some(percentage(n_done, n_total)), + ) + } + }, + Event::Flycheck(task) => on_check_task(task, &mut self)?, + } + + let state_changed = self.process_changes(); + if became_ready { + if let Some(flycheck) = &self.flycheck { + flycheck.0.update(); + } + } + + if self.status == Status::Ready && (state_changed || became_ready) { + let subscriptions = self + .mem_docs + .iter() + .map(|path| self.vfs.read().0.file_id(&path).unwrap()) + .collect::>(); + + self.update_file_notifications_on_threadpool(subscriptions); + } + + let loop_duration = loop_start.elapsed(); + if loop_duration > Duration::from_millis(100) { + log::error!("overly long loop turn: {:?}", loop_duration); + if env::var("RA_PROFILE").is_ok() { + self.show_message( + lsp_types::MessageType::Error, + format!("overly long loop turn: {:?}", loop_duration), + ) + } + } + } + Err("client exited without proper shutdown sequence")? + } + + fn on_request(&mut self, request_received: Instant, req: Request) -> Result<()> { + RequestDispatcher { req: Some(req), global_state: self, request_received } + .on_sync::(|s, ()| Ok(s.collect_garbage()))? + .on_sync::(|s, p| handlers::handle_join_lines(s.snapshot(), p))? + .on_sync::(|s, p| handlers::handle_on_enter(s.snapshot(), p))? + .on_sync::(|_, ()| Ok(()))? + .on_sync::(|s, p| { + handlers::handle_selection_range(s.snapshot(), p) + })? + .on_sync::(|s, p| { + handlers::handle_matching_brace(s.snapshot(), p) + })? + .on::(handlers::handle_analyzer_status)? + .on::(handlers::handle_syntax_tree)? + .on::(handlers::handle_expand_macro)? + .on::(handlers::handle_parent_module)? + .on::(handlers::handle_runnables)? + .on::(handlers::handle_inlay_hints)? + .on::(handlers::handle_code_action)? + .on::(handlers::handle_resolve_code_action)? + .on::(handlers::handle_hover)? + .on::(handlers::handle_on_type_formatting)? + .on::(handlers::handle_document_symbol)? + .on::(handlers::handle_workspace_symbol)? + .on::(handlers::handle_goto_definition)? + .on::(handlers::handle_goto_implementation)? + .on::(handlers::handle_goto_type_definition)? + .on::(handlers::handle_completion)? + .on::(handlers::handle_code_lens)? + .on::(handlers::handle_code_lens_resolve)? + .on::(handlers::handle_folding_range)? + .on::(handlers::handle_signature_help)? + .on::(handlers::handle_prepare_rename)? + .on::(handlers::handle_rename)? + .on::(handlers::handle_references)? + .on::(handlers::handle_formatting)? + .on::( + handlers::handle_document_highlight, + )? + .on::( + handlers::handle_call_hierarchy_prepare, + )? + .on::( + handlers::handle_call_hierarchy_incoming, + )? + .on::( + handlers::handle_call_hierarchy_outgoing, + )? + .on::(handlers::handle_semantic_tokens)? + .on::( + handlers::handle_semantic_tokens_range, + )? + .on::(handlers::handle_ssr)? + .finish(); + Ok(()) + } + fn on_notification(&mut self, not: Notification) -> Result<()> { + NotificationDispatcher { not: Some(not), global_state: self } + .on::(|this, params| { + let id: lsp_server::RequestId = match params.id { + lsp_types::NumberOrString::Number(id) => id.into(), + lsp_types::NumberOrString::String(id) => id.into(), + }; + if let Some(response) = this.req_queue.incoming.cancel(id) { + this.send(response.into()); + } + Ok(()) + })? + .on::(|this, params| { + if let Ok(path) = from_proto::vfs_path(¶ms.text_document.uri) { + if !this.mem_docs.insert(path.clone()) { + log::error!("duplicate DidOpenTextDocument: {}", path) + } + this.vfs + .write() + .0 + .set_file_contents(path, Some(params.text_document.text.into_bytes())); + } + Ok(()) + })? + .on::(|this, params| { + if let Ok(path) = from_proto::vfs_path(¶ms.text_document.uri) { + assert!(this.mem_docs.contains(&path)); + let vfs = &mut this.vfs.write().0; + let file_id = vfs.file_id(&path).unwrap(); + let mut text = String::from_utf8(vfs.file_contents(file_id).to_vec()).unwrap(); + apply_document_changes(&mut text, params.content_changes); + vfs.set_file_contents(path, Some(text.into_bytes())) + } + Ok(()) + })? + .on::(|this, params| { + if let Ok(path) = from_proto::vfs_path(¶ms.text_document.uri) { + if !this.mem_docs.remove(&path) { + log::error!("orphan DidCloseTextDocument: {}", path) + } + if let Some(path) = path.as_path() { + this.loader.invalidate(path.to_path_buf()); + } + } + let params = lsp_types::PublishDiagnosticsParams { + uri: params.text_document.uri, + diagnostics: Vec::new(), + version: None, + }; + let not = notification_new::(params); + this.send(not.into()); + Ok(()) + })? + .on::(|this, _params| { + if let Some(flycheck) = &this.flycheck { + flycheck.0.update(); + } + Ok(()) + })? + .on::(|this, _params| { + // As stated in https://github.com/microsoft/language-server-protocol/issues/676, + // this notification's parameters should be ignored and the actual config queried separately. + let request = this.req_queue.outgoing.register( + lsp_types::request::WorkspaceConfiguration::METHOD.to_string(), + lsp_types::ConfigurationParams { + items: vec![lsp_types::ConfigurationItem { + scope_uri: None, + section: Some("rust-analyzer".to_string()), + }], + }, + |this, resp| { + log::debug!("config update response: '{:?}", resp); + let Response { error, result, .. } = resp; + + match (error, result) { + (Some(err), _) => { + log::error!("failed to fetch the server settings: {:?}", err) + } + (None, Some(configs)) => { + if let Some(new_config) = configs.get(0) { + let mut config = this.config.clone(); + config.update(&new_config); + this.update_configuration(config); + } + } + (None, None) => log::error!( + "received empty server settings response from the client" + ), + } + }, + ); + this.send(request.into()); + + return Ok(()); + })? + .on::(|this, params| { + for change in params.changes { + if let Ok(path) = from_proto::abs_path(&change.uri) { + this.loader.invalidate(path); + } + } + Ok(()) + })? + .finish(); + Ok(()) + } + pub(crate) fn on_task(&mut self, task: Task) { + match task { + Task::Respond(response) => { + if let Some((method, start)) = self.req_queue.incoming.complete(response.id.clone()) + { + let duration = start.elapsed(); + log::info!("handled req#{} in {:?}", response.id, duration); + self.complete_request(RequestMetrics { + id: response.id.clone(), + method: method.to_string(), + duration, + }); + self.send(response.into()); + } + } + Task::Diagnostics(tasks) => { + tasks.into_iter().for_each(|task| on_diagnostic_task(task, self)) + } + Task::Unit => (), + } + } + fn update_file_notifications_on_threadpool(&mut self, subscriptions: Vec) { + log::trace!("updating notifications for {:?}", subscriptions); + if self.config.publish_diagnostics { + let snapshot = self.snapshot(); + let subscriptions = subscriptions.clone(); + self.task_pool.0.spawn(move || { + let diagnostics = subscriptions + .into_iter() + .filter_map(|file_id| { + handlers::publish_diagnostics(&snapshot, file_id) + .map_err(|err| { + if !is_canceled(&*err) { + log::error!("failed to compute diagnostics: {:?}", err); + } + () + }) + .ok() + }) + .collect::>(); + Task::Diagnostics(diagnostics) + }) + } + self.task_pool.0.spawn({ + let subs = subscriptions; + let snap = self.snapshot(); + move || { + snap.analysis.prime_caches(subs).unwrap_or_else(|_: Canceled| ()); + Task::Unit + } + }); + } } #[derive(Debug)] @@ -199,333 +504,10 @@ fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { pub(crate) type ReqQueue = lsp_server::ReqQueue<(&'static str, Instant), ReqHandler>; const DO_NOTHING: ReqHandler = |_, _| (); -fn loop_turn(connection: &Connection, global_state: &mut GlobalState, event: Event) -> Result<()> { - let loop_start = Instant::now(); - - // NOTE: don't count blocking select! call as a loop-turn time - let _p = profile("main_loop_inner/loop-turn"); - log::info!("loop turn = {:?}", event); - let queue_count = global_state.task_pool.0.len(); - if queue_count > 0 { - log::info!("queued count = {}", queue_count); - } - - let mut became_ready = false; - match event { - Event::Task(task) => { - on_task(task, &connection.sender, global_state); - global_state.maybe_collect_garbage(); - } - Event::Vfs(task) => match task { - vfs::loader::Message::Loaded { files } => { - let vfs = &mut global_state.vfs.write().0; - for (path, contents) in files { - let path = VfsPath::from(path); - if !global_state.mem_docs.contains(&path) { - vfs.set_file_contents(path, contents) - } - } - } - vfs::loader::Message::Progress { n_total, n_done } => { - let state = if n_done == 0 { - Progress::Begin - } else if n_done < n_total { - Progress::Report - } else { - assert_eq!(n_done, n_total); - global_state.status = Status::Ready; - became_ready = true; - Progress::End - }; - report_progress( - global_state, - &connection.sender, - "roots scanned", - state, - Some(format!("{}/{}", n_done, n_total)), - Some(percentage(n_done, n_total)), - ) - } - }, - Event::Flycheck(task) => on_check_task(task, global_state, &connection.sender)?, - Event::Lsp(msg) => match msg { - lsp_server::Message::Request(req) => { - on_request(global_state, &connection.sender, loop_start, req)? - } - lsp_server::Message::Notification(not) => { - on_notification(&connection.sender, global_state, not)?; - } - lsp_server::Message::Response(resp) => { - let handler = global_state.req_queue.outgoing.complete(resp.id.clone()); - handler(global_state, resp) - } - }, - }; - - let state_changed = global_state.process_changes(); - - if became_ready { - if let Some(flycheck) = &global_state.flycheck { - flycheck.0.update(); - } - } - - if global_state.status == Status::Ready && (state_changed || became_ready) { - let subscriptions = global_state - .mem_docs - .iter() - .map(|path| global_state.vfs.read().0.file_id(&path).unwrap()) - .collect::>(); - - update_file_notifications_on_threadpool(global_state, subscriptions.clone()); - global_state.task_pool.0.spawn({ - let subs = subscriptions; - let snap = global_state.snapshot(); - move || { - snap.analysis.prime_caches(subs).unwrap_or_else(|_: Canceled| ()); - Task::Unit - } - }); - } - - let loop_duration = loop_start.elapsed(); - if loop_duration > Duration::from_millis(100) { - log::error!("overly long loop turn: {:?}", loop_duration); - if env::var("RA_PROFILE").is_ok() { - show_message( - lsp_types::MessageType::Error, - format!("overly long loop turn: {:?}", loop_duration), - &connection.sender, - ); - } - } - - Ok(()) -} - -fn on_task(task: Task, msg_sender: &Sender, global_state: &mut GlobalState) { - match task { - Task::Respond(response) => { - if let Some((method, start)) = - global_state.req_queue.incoming.complete(response.id.clone()) - { - let duration = start.elapsed(); - log::info!("handled req#{} in {:?}", response.id, duration); - global_state.complete_request(RequestMetrics { - id: response.id.clone(), - method: method.to_string(), - duration, - }); - msg_sender.send(response.into()).unwrap(); - } - } - Task::Diagnostics(tasks) => { - tasks.into_iter().for_each(|task| on_diagnostic_task(task, msg_sender, global_state)) - } - Task::Unit => (), - } -} - -fn on_request( - global_state: &mut GlobalState, - msg_sender: &Sender, - request_received: Instant, - req: Request, -) -> Result<()> { - let mut pool_dispatcher = - PoolDispatcher { req: Some(req), global_state, msg_sender, request_received }; - pool_dispatcher - .on_sync::(|s, ()| Ok(s.collect_garbage()))? - .on_sync::(|s, p| handlers::handle_join_lines(s.snapshot(), p))? - .on_sync::(|s, p| handlers::handle_on_enter(s.snapshot(), p))? - .on_sync::(|s, p| { - handlers::handle_selection_range(s.snapshot(), p) - })? - .on_sync::(|s, p| handlers::handle_matching_brace(s.snapshot(), p))? - .on::(handlers::handle_analyzer_status)? - .on::(handlers::handle_syntax_tree)? - .on::(handlers::handle_expand_macro)? - .on::(handlers::handle_parent_module)? - .on::(handlers::handle_runnables)? - .on::(handlers::handle_inlay_hints)? - .on::(handlers::handle_code_action)? - .on::(handlers::handle_resolve_code_action)? - .on::(handlers::handle_hover)? - .on::(handlers::handle_on_type_formatting)? - .on::(handlers::handle_document_symbol)? - .on::(handlers::handle_workspace_symbol)? - .on::(handlers::handle_goto_definition)? - .on::(handlers::handle_goto_implementation)? - .on::(handlers::handle_goto_type_definition)? - .on::(handlers::handle_completion)? - .on::(handlers::handle_code_lens)? - .on::(handlers::handle_code_lens_resolve)? - .on::(handlers::handle_folding_range)? - .on::(handlers::handle_signature_help)? - .on::(handlers::handle_prepare_rename)? - .on::(handlers::handle_rename)? - .on::(handlers::handle_references)? - .on::(handlers::handle_formatting)? - .on::(handlers::handle_document_highlight)? - .on::(handlers::handle_call_hierarchy_prepare)? - .on::( - handlers::handle_call_hierarchy_incoming, - )? - .on::( - handlers::handle_call_hierarchy_outgoing, - )? - .on::(handlers::handle_semantic_tokens)? - .on::( - handlers::handle_semantic_tokens_range, - )? - .on::(handlers::handle_ssr)? - .finish(); - Ok(()) -} - -fn on_notification( - msg_sender: &Sender, - global_state: &mut GlobalState, - not: Notification, -) -> Result<()> { - let not = match notification_cast::(not) { - Ok(params) => { - let id: RequestId = match params.id { - NumberOrString::Number(id) => id.into(), - NumberOrString::String(id) => id.into(), - }; - if let Some(response) = global_state.req_queue.incoming.cancel(id) { - msg_sender.send(response.into()).unwrap() - } - return Ok(()); - } - Err(not) => not, - }; - let not = match notification_cast::(not) { - Ok(params) => { - if let Ok(path) = from_proto::vfs_path(¶ms.text_document.uri) { - if !global_state.mem_docs.insert(path.clone()) { - log::error!("duplicate DidOpenTextDocument: {}", path) - } - global_state - .vfs - .write() - .0 - .set_file_contents(path, Some(params.text_document.text.into_bytes())); - } - return Ok(()); - } - Err(not) => not, - }; - let not = match notification_cast::(not) { - Ok(params) => { - if let Ok(path) = from_proto::vfs_path(¶ms.text_document.uri) { - assert!(global_state.mem_docs.contains(&path)); - let vfs = &mut global_state.vfs.write().0; - let file_id = vfs.file_id(&path).unwrap(); - let mut text = String::from_utf8(vfs.file_contents(file_id).to_vec()).unwrap(); - apply_document_changes(&mut text, params.content_changes); - vfs.set_file_contents(path, Some(text.into_bytes())) - } - return Ok(()); - } - Err(not) => not, - }; - let not = match notification_cast::(not) { - Ok(params) => { - if let Ok(path) = from_proto::vfs_path(¶ms.text_document.uri) { - if !global_state.mem_docs.remove(&path) { - log::error!("orphan DidCloseTextDocument: {}", path) - } - if let Some(path) = path.as_path() { - global_state.loader.invalidate(path.to_path_buf()); - } - } - let params = lsp_types::PublishDiagnosticsParams { - uri: params.text_document.uri, - diagnostics: Vec::new(), - version: None, - }; - let not = notification_new::(params); - msg_sender.send(not.into()).unwrap(); - return Ok(()); - } - Err(not) => not, - }; - let not = match notification_cast::(not) { - Ok(_params) => { - if let Some(flycheck) = &global_state.flycheck { - flycheck.0.update(); - } - return Ok(()); - } - Err(not) => not, - }; - let not = match notification_cast::(not) { - Ok(_) => { - // As stated in https://github.com/microsoft/language-server-protocol/issues/676, - // this notification's parameters should be ignored and the actual config queried separately. - let request = global_state.req_queue.outgoing.register( - lsp_types::request::WorkspaceConfiguration::METHOD.to_string(), - lsp_types::ConfigurationParams { - items: vec![lsp_types::ConfigurationItem { - scope_uri: None, - section: Some("rust-analyzer".to_string()), - }], - }, - |global_state, resp| { - log::debug!("config update response: '{:?}", resp); - let Response { error, result, .. } = resp; - - match (error, result) { - (Some(err), _) => { - log::error!("failed to fetch the server settings: {:?}", err) - } - (None, Some(configs)) => { - if let Some(new_config) = configs.get(0) { - let mut config = global_state.config.clone(); - config.update(&new_config); - global_state.update_configuration(config); - } - } - (None, None) => { - log::error!("received empty server settings response from the client") - } - } - }, - ); - msg_sender.send(request.into())?; - - return Ok(()); - } - Err(not) => not, - }; - let not = match notification_cast::(not) { - Ok(params) => { - for change in params.changes { - if let Ok(path) = from_proto::abs_path(&change.uri) { - global_state.loader.invalidate(path) - } - } - return Ok(()); - } - Err(not) => not, - }; - if not.method.starts_with("$/") { - return Ok(()); - } - log::error!("unhandled notification: {:?}", not); - Ok(()) -} - -fn on_check_task( - task: flycheck::Message, - global_state: &mut GlobalState, - msg_sender: &Sender, -) -> Result<()> { +fn on_check_task(task: flycheck::Message, global_state: &mut GlobalState) -> Result<()> { match task { flycheck::Message::ClearDiagnostics => { - on_diagnostic_task(DiagnosticTask::ClearCheck, msg_sender, global_state) + on_diagnostic_task(DiagnosticTask::ClearCheck, global_state) } flycheck::Message::AddDiagnostic { workspace_root, diagnostic } => { @@ -550,7 +532,6 @@ fn on_check_task( diag.diagnostic, diag.fixes.into_iter().map(|it| it.into()).collect(), ), - msg_sender, global_state, ) } @@ -563,26 +544,22 @@ fn on_check_task( flycheck::Progress::End => (Progress::End, None), }; - report_progress(global_state, msg_sender, "cargo check", state, message, None); + report_progress(global_state, "cargo check", state, message, None); } }; Ok(()) } -fn on_diagnostic_task( - task: DiagnosticTask, - msg_sender: &Sender, - state: &mut GlobalState, -) { - let subscriptions = state.diagnostics.handle_task(task); +fn on_diagnostic_task(task: DiagnosticTask, global_state: &mut GlobalState) { + let subscriptions = global_state.diagnostics.handle_task(task); for file_id in subscriptions { - let url = file_id_to_url(&state.vfs.read().0, file_id); - let diagnostics = state.diagnostics.diagnostics_for(file_id).cloned().collect(); + let url = file_id_to_url(&global_state.vfs.read().0, file_id); + let diagnostics = global_state.diagnostics.diagnostics_for(file_id).cloned().collect(); let params = lsp_types::PublishDiagnosticsParams { uri: url, diagnostics, version: None }; let not = notification_new::(params); - msg_sender.send(not.into()).unwrap(); + global_state.send(not.into()); } } @@ -599,7 +576,6 @@ fn percentage(done: usize, total: usize) -> f64 { fn report_progress( global_state: &mut GlobalState, - sender: &Sender, title: &str, state: Progress, message: Option, @@ -616,7 +592,7 @@ fn report_progress( lsp_types::WorkDoneProgressCreateParams { token: token.clone() }, DO_NOTHING, ); - sender.send(work_done_progress_create.into()).unwrap(); + global_state.send(work_done_progress_create.into()); lsp_types::WorkDoneProgress::Begin(lsp_types::WorkDoneProgressBegin { title: title.into(), @@ -641,155 +617,5 @@ fn report_progress( token, value: lsp_types::ProgressParamsValue::WorkDone(work_done_progress), }); - sender.send(notification.into()).unwrap(); -} - -struct PoolDispatcher<'a> { - req: Option, - global_state: &'a mut GlobalState, - msg_sender: &'a Sender, - request_received: Instant, -} - -impl<'a> PoolDispatcher<'a> { - /// Dispatches the request onto the current thread - fn on_sync( - &mut self, - f: fn(&mut GlobalState, R::Params) -> Result, - ) -> Result<&mut Self> - where - R: lsp_types::request::Request + 'static, - R::Params: DeserializeOwned + panic::UnwindSafe + 'static, - R::Result: Serialize + 'static, - { - let (id, params) = match self.parse::() { - Some(it) => it, - None => { - return Ok(self); - } - }; - let world = panic::AssertUnwindSafe(&mut *self.global_state); - let task = panic::catch_unwind(move || { - let result = f(world.0, params); - result_to_task::(id, result) - }) - .map_err(|_| format!("sync task {:?} panicked", R::METHOD))?; - on_task(task, self.msg_sender, self.global_state); - Ok(self) - } - - /// Dispatches the request onto thread pool - fn on( - &mut self, - f: fn(GlobalStateSnapshot, R::Params) -> Result, - ) -> Result<&mut Self> - where - R: lsp_types::request::Request + 'static, - R::Params: DeserializeOwned + Send + 'static, - R::Result: Serialize + 'static, - { - let (id, params) = match self.parse::() { - Some(it) => it, - None => { - return Ok(self); - } - }; - - self.global_state.task_pool.0.spawn({ - let world = self.global_state.snapshot(); - move || { - let result = f(world, params); - result_to_task::(id, result) - } - }); - - Ok(self) - } - - fn parse(&mut self) -> Option<(RequestId, R::Params)> - where - R: lsp_types::request::Request + 'static, - R::Params: DeserializeOwned + 'static, - { - let req = self.req.take()?; - let (id, params) = match req.extract::(R::METHOD) { - Ok(it) => it, - Err(req) => { - self.req = Some(req); - return None; - } - }; - self.global_state - .req_queue - .incoming - .register(id.clone(), (R::METHOD, self.request_received)); - Some((id, params)) - } - - fn finish(&mut self) { - match self.req.take() { - None => (), - Some(req) => { - log::error!("unknown request: {:?}", req); - let resp = Response::new_err( - req.id, - ErrorCode::MethodNotFound as i32, - "unknown request".to_string(), - ); - self.msg_sender.send(resp.into()).unwrap(); - } - } - } -} - -fn result_to_task(id: RequestId, result: Result) -> Task -where - R: lsp_types::request::Request + 'static, - R::Params: DeserializeOwned + 'static, - R::Result: Serialize + 'static, -{ - let response = match result { - Ok(resp) => Response::new_ok(id, &resp), - Err(e) => match e.downcast::() { - Ok(lsp_error) => Response::new_err(id, lsp_error.code, lsp_error.message), - Err(e) => { - if is_canceled(&*e) { - Response::new_err( - id, - ErrorCode::ContentModified as i32, - "content modified".to_string(), - ) - } else { - Response::new_err(id, ErrorCode::InternalError as i32, e.to_string()) - } - } - }, - }; - Task::Respond(response) -} - -fn update_file_notifications_on_threadpool( - global_state: &mut GlobalState, - subscriptions: Vec, -) { - log::trace!("updating notifications for {:?}", subscriptions); - if global_state.config.publish_diagnostics { - let snapshot = global_state.snapshot(); - global_state.task_pool.0.spawn(move || { - let diagnostics = subscriptions - .into_iter() - .filter_map(|file_id| { - handlers::publish_diagnostics(&snapshot, file_id) - .map_err(|err| { - if !is_canceled(&*err) { - log::error!("failed to compute diagnostics: {:?}", err); - } - () - }) - .ok() - }) - .collect::>(); - Task::Diagnostics(diagnostics) - }) - } + global_state.send(notification.into()); } diff --git a/crates/vfs-notify/src/lib.rs b/crates/vfs-notify/src/lib.rs index 68fdb8cb09c..25ba8d7982d 100644 --- a/crates/vfs-notify/src/lib.rs +++ b/crates/vfs-notify/src/lib.rs @@ -82,7 +82,12 @@ fn new(sender: loader::Sender) -> NotifyActor { watcher_receiver, } } - + fn next_event(&self, receiver: &Receiver) -> Option { + select! { + recv(receiver) -> it => it.ok().map(Event::Message), + recv(&self.watcher_receiver) -> it => Some(Event::NotifyEvent(it.unwrap())), + } + } fn run(mut self, inbox: Receiver) { while let Some(event) = self.next_event(&inbox) { log::debug!("vfs-notify event: {:?}", event); @@ -154,12 +159,6 @@ fn run(mut self, inbox: Receiver) { } } } - fn next_event(&self, receiver: &Receiver) -> Option { - select! { - recv(receiver) -> it => it.ok().map(Event::Message), - recv(&self.watcher_receiver) -> it => Some(Event::NotifyEvent(it.unwrap())), - } - } fn load_entry( &mut self, entry: loader::Entry,