mod handlers; mod subscriptions; pub(crate) mod pending_requests; use std::{fmt, path::PathBuf, sync::Arc, time::Instant}; use crossbeam_channel::{select, unbounded, Receiver, RecvError, Sender}; use failure::{bail, format_err}; use failure_derive::Fail; use gen_lsp_server::{ handle_shutdown, ErrorCode, RawMessage, RawNotification, RawRequest, RawResponse, }; use lsp_types::NumberOrString; use ra_ide_api::{Canceled, FileId, LibraryData}; use ra_vfs::VfsTask; use serde::{de::DeserializeOwned, Serialize}; use threadpool::ThreadPool; use ra_prof::profile; use crate::{ main_loop::{ subscriptions::Subscriptions, pending_requests::{PendingRequests, PendingRequest}, }, project_model::workspace_loader, req, world::{WorldSnapshot, WorldState}, Result, InitializationOptions, }; const THREADPOOL_SIZE: usize = 8; const MAX_IN_FLIGHT_LIBS: usize = THREADPOOL_SIZE - 3; #[derive(Debug, Fail)] #[fail(display = "Language Server request failed with {}. ({})", code, message)] pub struct LspError { pub code: i32, pub message: String, } impl LspError { pub fn new(code: i32, message: String) -> LspError { LspError { code, message } } } pub fn main_loop( ws_roots: Vec, options: InitializationOptions, msg_receiver: &Receiver, msg_sender: &Sender, ) -> Result<()> { // FIXME: support dynamic workspace loading. let workspaces = { let ws_worker = workspace_loader(); let mut loaded_workspaces = Vec::new(); for ws_root in &ws_roots { ws_worker.sender().send(ws_root.clone()).unwrap(); match ws_worker.receiver().recv().unwrap() { Ok(ws) => loaded_workspaces.push(ws), Err(e) => { log::error!("loading workspace failed: {}", e); show_message( req::MessageType::Error, format!("rust-analyzer failed to load workspace: {}", e), msg_sender, ); } } } loaded_workspaces }; let mut state = WorldState::new(ws_roots, workspaces); let pool = ThreadPool::new(THREADPOOL_SIZE); let (task_sender, task_receiver) = unbounded::(); let mut pending_requests = PendingRequests::default(); log::info!("server initialized, serving requests"); let main_res = main_loop_inner( options, &pool, msg_sender, msg_receiver, task_sender, task_receiver.clone(), &mut state, &mut pending_requests, ); log::info!("waiting for tasks to finish..."); task_receiver .into_iter() .for_each(|task| on_task(task, msg_sender, &mut pending_requests, &mut state)); log::info!("...tasks have finished"); log::info!("joining threadpool..."); drop(pool); log::info!("...threadpool has finished"); let vfs = Arc::try_unwrap(state.vfs).expect("all snapshots should be dead"); drop(vfs); main_res } #[derive(Debug)] enum Task { Respond(RawResponse), Notify(RawNotification), } enum Event { Msg(RawMessage), Task(Task), Vfs(VfsTask), Lib(LibraryData), } impl fmt::Debug for Event { fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { let debug_verbose_not = |not: &RawNotification, f: &mut fmt::Formatter| { f.debug_struct("RawNotification").field("method", ¬.method).finish() }; match self { Event::Msg(RawMessage::Notification(not)) => { if not.is::() || not.is::() { return debug_verbose_not(not, f); } } Event::Task(Task::Notify(not)) => { if not.is::() || not.is::() { return debug_verbose_not(not, f); } } Event::Task(Task::Respond(resp)) => { return f .debug_struct("RawResponse") .field("id", &resp.id) .field("error", &resp.error) .finish(); } _ => (), } match self { Event::Msg(it) => fmt::Debug::fmt(it, f), Event::Task(it) => fmt::Debug::fmt(it, f), Event::Vfs(it) => fmt::Debug::fmt(it, f), Event::Lib(it) => fmt::Debug::fmt(it, f), } } } fn main_loop_inner( options: InitializationOptions, pool: &ThreadPool, msg_sender: &Sender, msg_receiver: &Receiver, task_sender: Sender, task_receiver: Receiver, state: &mut WorldState, pending_requests: &mut PendingRequests, ) -> Result<()> { let mut subs = Subscriptions::default(); // We try not to index more than MAX_IN_FLIGHT_LIBS libraries at the same // time to always have a thread ready to react to input. let mut in_flight_libraries = 0; let mut pending_libraries = Vec::new(); let mut send_workspace_notification = true; let (libdata_sender, libdata_receiver) = unbounded(); loop { log::trace!("selecting"); let event = select! { recv(msg_receiver) -> msg => match msg { Ok(msg) => Event::Msg(msg), Err(RecvError) => bail!("client exited without shutdown"), }, recv(task_receiver) -> task => Event::Task(task.unwrap()), recv(state.vfs.read().task_receiver()) -> task => match task { Ok(task) => Event::Vfs(task), Err(RecvError) => bail!("vfs died"), }, recv(libdata_receiver) -> data => Event::Lib(data.unwrap()) }; 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 = pool.queued_count(); if queue_count > 0 { log::info!("queued count = {}", queue_count); } let mut state_changed = false; match event { Event::Task(task) => { on_task(task, msg_sender, pending_requests, state); state.maybe_collect_garbage(); } Event::Vfs(task) => { state.vfs.write().handle_task(task); state_changed = true; } Event::Lib(lib) => { state.add_lib(lib); state.maybe_collect_garbage(); in_flight_libraries -= 1; } Event::Msg(msg) => match msg { RawMessage::Request(req) => { let req = match handle_shutdown(req, msg_sender) { Some(req) => req, None => return Ok(()), }; on_request( state, pending_requests, pool, &task_sender, msg_sender, loop_start, req, )? } RawMessage::Notification(not) => { on_notification(msg_sender, state, pending_requests, &mut subs, not)?; state_changed = true; } RawMessage::Response(resp) => log::error!("unexpected response: {:?}", resp), }, }; pending_libraries.extend(state.process_changes()); while in_flight_libraries < MAX_IN_FLIGHT_LIBS && !pending_libraries.is_empty() { let (root, files) = pending_libraries.pop().unwrap(); in_flight_libraries += 1; let sender = libdata_sender.clone(); pool.execute(move || { log::info!("indexing {:?} ... ", root); let _p = profile(&format!("indexed {:?}", root)); let data = LibraryData::prepare(root, files); sender.send(data).unwrap(); }); } if send_workspace_notification && state.roots_to_scan == 0 && pending_libraries.is_empty() && in_flight_libraries == 0 { let n_packages: usize = state.workspaces.iter().map(|it| it.count()).sum(); if options.show_workspace_loaded { let msg = format!("workspace loaded, {} rust packages", n_packages); show_message(req::MessageType::Info, msg, msg_sender); } // Only send the notification first time send_workspace_notification = false; } if state_changed { update_file_notifications_on_threadpool( pool, state.snapshot(), options.publish_decorations, task_sender.clone(), subs.subscriptions(), ) } } } fn on_task( task: Task, msg_sender: &Sender, pending_requests: &mut PendingRequests, state: &mut WorldState, ) { match task { Task::Respond(response) => { if let Some(completed) = pending_requests.finish(response.id) { log::info!("handled req#{} in {:?}", completed.id, completed.duration); state.complete_request(completed); msg_sender.send(response.into()).unwrap(); } } Task::Notify(n) => { msg_sender.send(n.into()).unwrap(); } } } fn on_request( world: &mut WorldState, pending_requests: &mut PendingRequests, pool: &ThreadPool, sender: &Sender, msg_sender: &Sender, request_received: Instant, req: RawRequest, ) -> Result<()> { let mut pool_dispatcher = PoolDispatcher { req: Some(req), pool, world, sender, msg_sender, pending_requests, 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_find_matching_brace(s.snapshot(), p) })? .on::(handlers::handle_analyzer_status)? .on::(handlers::handle_syntax_tree)? .on::(handlers::handle_extend_selection)? .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_parent_module)? .on::(handlers::handle_runnables)? .on::(handlers::handle_decorations)? .on::(handlers::handle_completion)? .on::(handlers::handle_code_action)? .on::(handlers::handle_code_lens)? .on::(handlers::handle_code_lens_resolve)? .on::(handlers::handle_folding_range)? .on::(handlers::handle_signature_help)? .on::(handlers::handle_hover)? .on::(handlers::handle_prepare_rename)? .on::(handlers::handle_rename)? .on::(handlers::handle_references)? .on::(handlers::handle_formatting)? .on::(handlers::handle_document_highlight)? .finish(); Ok(()) } fn on_notification( msg_sender: &Sender, state: &mut WorldState, pending_requests: &mut PendingRequests, subs: &mut Subscriptions, not: RawNotification, ) -> Result<()> { let not = match not.cast::() { Ok(params) => { let id = match params.id { NumberOrString::Number(id) => id, NumberOrString::String(id) => { panic!("string id's not supported: {:?}", id); } }; if pending_requests.cancel(id) { let response = RawResponse::err( id, ErrorCode::RequestCanceled as i32, "canceled by client".to_string(), ); msg_sender.send(response.into()).unwrap() } return Ok(()); } Err(not) => not, }; let not = match not.cast::() { Ok(params) => { let uri = params.text_document.uri; let path = uri.to_file_path().map_err(|()| format_err!("invalid uri: {}", uri))?; if let Some(file_id) = state.vfs.write().add_file_overlay(&path, params.text_document.text) { subs.add_sub(FileId(file_id.0.into())); } return Ok(()); } Err(not) => not, }; let not = match not.cast::() { Ok(mut params) => { let uri = params.text_document.uri; let path = uri.to_file_path().map_err(|()| format_err!("invalid uri: {}", uri))?; let text = params.content_changes.pop().ok_or_else(|| format_err!("empty changes"))?.text; state.vfs.write().change_file_overlay(path.as_path(), text); return Ok(()); } Err(not) => not, }; let not = match not.cast::() { Ok(params) => { let uri = params.text_document.uri; let path = uri.to_file_path().map_err(|()| format_err!("invalid uri: {}", uri))?; if let Some(file_id) = state.vfs.write().remove_file_overlay(path.as_path()) { subs.remove_sub(FileId(file_id.0.into())); } let params = req::PublishDiagnosticsParams { uri, diagnostics: Vec::new() }; let not = RawNotification::new::(¶ms); msg_sender.send(not.into()).unwrap(); return Ok(()); } Err(not) => not, }; log::error!("unhandled notification: {:?}", not); Ok(()) } struct PoolDispatcher<'a> { req: Option, pool: &'a ThreadPool, world: &'a mut WorldState, pending_requests: &'a mut PendingRequests, msg_sender: &'a Sender, 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 WorldState, R::Params) -> Result, ) -> Result<&mut Self> where R: req::Request + 'static, R::Params: DeserializeOwned + Send + 'static, R::Result: Serialize + 'static, { let (id, params) = match self.parse::() { Some(it) => it, None => { return Ok(self); } }; let result = f(self.world, params); let task = result_to_task::(id, result); on_task(task, self.msg_sender, self.pending_requests, self.world); Ok(self) } /// Dispatches the request onto thread pool fn on(&mut self, f: fn(WorldSnapshot, R::Params) -> Result) -> Result<&mut Self> where R: req::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.pool.execute({ let world = self.world.snapshot(); let sender = self.sender.clone(); move || { let result = f(world, params); let task = result_to_task::(id, result); sender.send(task).unwrap(); } }); Ok(self) } fn parse(&mut self) -> Option<(u64, R::Params)> where R: req::Request + 'static, R::Params: DeserializeOwned + Send + 'static, { let req = self.req.take()?; let (id, params) = match req.cast::() { Ok(it) => it, Err(req) => { self.req = Some(req); return None; } }; self.pending_requests.start(PendingRequest { id, method: R::METHOD.to_string(), received: self.request_received, }); Some((id, params)) } fn finish(&mut self) { match self.req.take() { None => (), Some(req) => { log::error!("unknown request: {:?}", req); let resp = RawResponse::err( req.id, ErrorCode::MethodNotFound as i32, "unknown request".to_string(), ); self.msg_sender.send(resp.into()).unwrap(); } } } } fn result_to_task(id: u64, result: Result) -> Task where R: req::Request + 'static, R::Params: DeserializeOwned + Send + 'static, R::Result: Serialize + 'static, { let response = match result { Ok(resp) => RawResponse::ok::(id, &resp), Err(e) => match e.downcast::() { Ok(lsp_error) => RawResponse::err(id, lsp_error.code, lsp_error.message), Err(e) => { if is_canceled(&e) { // FIXME: When https://github.com/Microsoft/vscode-languageserver-node/issues/457 // gets fixed, we can return the proper response. // This works around the issue where "content modified" error would continuously // show an message pop-up in VsCode // RawResponse::err( // id, // ErrorCode::ContentModified as i32, // "content modified".to_string(), // ) RawResponse { id, result: Some(serde_json::to_value(&()).unwrap()), error: None, } } else { RawResponse::err( id, ErrorCode::InternalError as i32, format!("{}\n{}", e, e.backtrace()), ) } } }, }; Task::Respond(response) } fn update_file_notifications_on_threadpool( pool: &ThreadPool, world: WorldSnapshot, publish_decorations: bool, sender: Sender, subscriptions: Vec, ) { pool.execute(move || { for file_id in subscriptions { match handlers::publish_diagnostics(&world, file_id) { Err(e) => { if !is_canceled(&e) { log::error!("failed to compute diagnostics: {:?}", e); } } Ok(params) => { let not = RawNotification::new::(¶ms); sender.send(Task::Notify(not)).unwrap(); } } if publish_decorations { match handlers::publish_decorations(&world, file_id) { Err(e) => { if !is_canceled(&e) { log::error!("failed to compute decorations: {:?}", e); } } Ok(params) => { let not = RawNotification::new::(¶ms); sender.send(Task::Notify(not)).unwrap(); } } } } }); } fn show_message(typ: req::MessageType, message: impl Into, sender: &Sender) { let message = message.into(); let params = req::ShowMessageParams { typ, message }; let not = RawNotification::new::(¶ms); sender.send(not.into()).unwrap(); } fn is_canceled(e: &failure::Error) -> bool { e.downcast_ref::().is_some() }