From 4cbc902fcc9de79893779582dac01351d1137c7f Mon Sep 17 00:00:00 2001 From: Aleksey Kladov Date: Sat, 8 Dec 2018 19:30:35 +0300 Subject: grand module rename --- crates/ra_lsp_server/src/main_loop.rs | 495 ++++++++++++++++++++++++++++++++++ 1 file changed, 495 insertions(+) create mode 100644 crates/ra_lsp_server/src/main_loop.rs (limited to 'crates/ra_lsp_server/src/main_loop.rs') diff --git a/crates/ra_lsp_server/src/main_loop.rs b/crates/ra_lsp_server/src/main_loop.rs new file mode 100644 index 000000000..0e1878906 --- /dev/null +++ b/crates/ra_lsp_server/src/main_loop.rs @@ -0,0 +1,495 @@ +mod handlers; +mod subscriptions; + +use std::path::PathBuf; + +use crossbeam_channel::{unbounded, select, Receiver, Sender}; +use gen_lsp_server::{ + handle_shutdown, ErrorCode, RawMessage, RawNotification, RawRequest, RawResponse, +}; +use languageserver_types::NumberOrString; +use ra_analysis::{Canceled, FileId, LibraryData}; +use rayon::{self, ThreadPool}; +use rustc_hash::FxHashSet; +use serde::{de::DeserializeOwned, Serialize}; +use failure::{format_err, bail}; +use failure_derive::Fail; + +use crate::{ + main_loop::subscriptions::Subscriptions, + project_model::{workspace_loader, CargoWorkspace}, + req, + server_world::{ServerWorld, ServerWorldState}, + thread_watcher::Worker, + vfs::{self, FileEvent}, + Result, +}; + +#[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 } + } +} + +#[derive(Debug)] +enum Task { + Respond(RawResponse), + Notify(RawNotification), +} + +pub fn main_loop( + internal_mode: bool, + root: PathBuf, + publish_decorations: bool, + msg_receiver: &Receiver, + msg_sender: &Sender, +) -> Result<()> { + let pool = rayon::ThreadPoolBuilder::new() + .num_threads(4) + .panic_handler(|_| log::error!("thread panicked :(")) + .build() + .unwrap(); + let (task_sender, task_receiver) = unbounded::(); + let (fs_worker, fs_watcher) = vfs::roots_loader(); + let (ws_worker, ws_watcher) = workspace_loader(); + + log::info!("server initialized, serving requests"); + let mut state = ServerWorldState::default(); + + let mut pending_requests = FxHashSet::default(); + let mut subs = Subscriptions::new(); + let main_res = main_loop_inner( + internal_mode, + publish_decorations, + root, + &pool, + msg_sender, + msg_receiver, + task_sender, + task_receiver.clone(), + fs_worker, + ws_worker, + &mut state, + &mut pending_requests, + &mut subs, + ); + + log::info!("waiting for tasks to finish..."); + task_receiver.for_each(|task| on_task(task, msg_sender, &mut pending_requests)); + log::info!("...tasks have finished"); + log::info!("joining threadpool..."); + drop(pool); + log::info!("...threadpool has finished"); + + let fs_res = fs_watcher.stop(); + let ws_res = ws_watcher.stop(); + + main_res?; + fs_res?; + ws_res?; + + Ok(()) +} + +fn main_loop_inner( + internal_mode: bool, + publish_decorations: bool, + ws_root: PathBuf, + pool: &ThreadPool, + msg_sender: &Sender, + msg_receiver: &Receiver, + task_sender: Sender, + task_receiver: Receiver, + fs_worker: Worker)>, + ws_worker: Worker>, + state: &mut ServerWorldState, + pending_requests: &mut FxHashSet, + subs: &mut Subscriptions, +) -> Result<()> { + let (libdata_sender, libdata_receiver) = unbounded(); + ws_worker.send(ws_root.clone()); + fs_worker.send(ws_root.clone()); + loop { + #[derive(Debug)] + enum Event { + Msg(RawMessage), + Task(Task), + Fs(PathBuf, Vec), + Ws(Result), + Lib(LibraryData), + } + log::trace!("selecting"); + let event = select! { + recv(msg_receiver, msg) => match msg { + Some(msg) => Event::Msg(msg), + None => bail!("client exited without shutdown"), + }, + recv(task_receiver, task) => Event::Task(task.unwrap()), + recv(fs_worker.out, events) => match events { + None => bail!("roots watcher died"), + Some((pb, events)) => Event::Fs(pb, events), + } + recv(ws_worker.out, ws) => match ws { + None => bail!("workspace watcher died"), + Some(ws) => Event::Ws(ws), + } + recv(libdata_receiver, data) => Event::Lib(data.unwrap()) + }; + let mut state_changed = false; + match event { + Event::Task(task) => on_task(task, msg_sender, pending_requests), + Event::Fs(root, events) => { + log::info!("fs change, {}, {} events", root.display(), events.len()); + if root == ws_root { + state.apply_fs_changes(events); + } else { + let (files, resolver) = state.events_to_files(events); + let sender = libdata_sender.clone(); + pool.spawn(move || { + let start = ::std::time::Instant::now(); + log::info!("indexing {} ... ", root.display()); + let data = LibraryData::prepare(files, resolver); + log::info!("indexed {:?} {}", start.elapsed(), root.display()); + sender.send(data); + }); + } + state_changed = true; + } + Event::Ws(ws) => match ws { + Ok(ws) => { + let workspaces = vec![ws]; + feedback(internal_mode, "workspace loaded", msg_sender); + for ws in workspaces.iter() { + // Add each library as constant input. If library is + // within the workspace, don't treat it as a library. + // + // HACK: If source roots are nested, pick the outer one. + + let mut roots = ws + .packages() + .filter(|pkg| !pkg.is_member(ws)) + .filter_map(|pkg| { + let root = pkg.root(ws).to_path_buf(); + if root.starts_with(&ws_root) { + None + } else { + Some(root) + } + }) + .collect::>(); + roots.sort_by_key(|it| it.as_os_str().len()); + let unique = roots + .iter() + .enumerate() + .filter(|&(idx, long)| { + !roots[..idx].iter().any(|short| long.starts_with(short)) + }) + .map(|(_idx, root)| root); + + for root in unique { + log::debug!("sending root, {}", root.display()); + fs_worker.send(root.to_owned()); + } + } + state.set_workspaces(workspaces); + state_changed = true; + } + Err(e) => log::warn!("loading workspace failed: {}", e), + }, + Event::Lib(lib) => { + feedback(internal_mode, "library loaded", msg_sender); + state.add_lib(lib); + } + Event::Msg(msg) => match msg { + RawMessage::Request(req) => { + let req = match handle_shutdown(req, msg_sender) { + Some(req) => req, + None => return Ok(()), + }; + match on_request(state, pending_requests, pool, &task_sender, req)? { + None => (), + Some(req) => { + log::error!("unknown request: {:?}", req); + let resp = RawResponse::err( + req.id, + ErrorCode::MethodNotFound as i32, + "unknown request".to_string(), + ); + msg_sender.send(RawMessage::Response(resp)) + } + } + } + RawMessage::Notification(not) => { + on_notification(msg_sender, state, pending_requests, subs, not)?; + state_changed = true; + } + RawMessage::Response(resp) => log::error!("unexpected response: {:?}", resp), + }, + }; + + if state_changed { + update_file_notifications_on_threadpool( + pool, + state.snapshot(), + publish_decorations, + task_sender.clone(), + subs.subscriptions(), + ) + } + } +} + +fn on_task(task: Task, msg_sender: &Sender, pending_requests: &mut FxHashSet) { + match task { + Task::Respond(response) => { + if pending_requests.remove(&response.id) { + msg_sender.send(RawMessage::Response(response)) + } + } + Task::Notify(n) => msg_sender.send(RawMessage::Notification(n)), + } +} + +fn on_request( + world: &mut ServerWorldState, + pending_requests: &mut FxHashSet, + pool: &ThreadPool, + sender: &Sender, + req: RawRequest, +) -> Result> { + let mut pool_dispatcher = PoolDispatcher { + req: Some(req), + res: None, + pool, + world, + sender, + }; + let req = pool_dispatcher + .on::(handlers::handle_syntax_tree)? + .on::(handlers::handle_extend_selection)? + .on::(handlers::handle_find_matching_brace)? + .on::(handlers::handle_join_lines)? + .on::(handlers::handle_on_enter)? + .on::(handlers::handle_on_type_formatting)? + .on::(handlers::handle_document_symbol)? + .on::(handlers::handle_workspace_symbol)? + .on::(handlers::handle_goto_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_folding_range)? + .on::(handlers::handle_signature_help)? + .on::(handlers::handle_hover)? + .on::(handlers::handle_prepare_rename)? + .on::(handlers::handle_rename)? + .on::(handlers::handle_references)? + .finish(); + match req { + Ok(id) => { + let inserted = pending_requests.insert(id); + assert!(inserted, "duplicate request: {}", id); + Ok(None) + } + Err(req) => Ok(Some(req)), + } +} + +fn on_notification( + msg_sender: &Sender, + state: &mut ServerWorldState, + pending_requests: &mut FxHashSet, + 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); + } + }; + pending_requests.remove(&id); + 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))?; + let file_id = state.add_mem_file(path, params.text_document.text); + subs.add_sub(file_id); + 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.change_mem_file(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))?; + let file_id = state.remove_mem_file(path.as_path())?; + subs.remove_sub(file_id); + let params = req::PublishDiagnosticsParams { + uri, + diagnostics: Vec::new(), + }; + let not = RawNotification::new::(¶ms); + msg_sender.send(RawMessage::Notification(not)); + return Ok(()); + } + Err(not) => not, + }; + log::error!("unhandled notification: {:?}", not); + Ok(()) +} + +struct PoolDispatcher<'a> { + req: Option, + res: Option, + pool: &'a ThreadPool, + world: &'a ServerWorldState, + sender: &'a Sender, +} + +impl<'a> PoolDispatcher<'a> { + fn on<'b, R>( + &'b mut self, + f: fn(ServerWorld, R::Params) -> Result, + ) -> Result<&'b mut Self> + where + R: req::Request, + R::Params: DeserializeOwned + Send + 'static, + R::Result: Serialize + 'static, + { + let req = match self.req.take() { + None => return Ok(self), + Some(req) => req, + }; + match req.cast::() { + Ok((id, params)) => { + let world = self.world.snapshot(); + let sender = self.sender.clone(); + self.pool.spawn(move || { + let resp = match f(world, params) { + 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) { + RawResponse::err( + id, + ErrorCode::RequestCancelled as i32, + e.to_string(), + ) + } else { + RawResponse::err( + id, + ErrorCode::InternalError as i32, + format!("{}\n{}", e, e.backtrace()), + ) + } + } + }, + }; + let task = Task::Respond(resp); + sender.send(task); + }); + self.res = Some(id); + } + Err(req) => self.req = Some(req), + } + Ok(self) + } + + fn finish(&mut self) -> ::std::result::Result { + match (self.res.take(), self.req.take()) { + (Some(res), None) => Ok(res), + (None, Some(req)) => Err(req), + _ => unreachable!(), + } + } +} + +fn update_file_notifications_on_threadpool( + pool: &ThreadPool, + world: ServerWorld, + publish_decorations: bool, + sender: Sender, + subscriptions: Vec, +) { + pool.spawn(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)); + } + } + 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)) + } + } + } + } + }); +} + +fn feedback(intrnal_mode: bool, msg: &str, sender: &Sender) { + if !intrnal_mode { + return; + } + let not = RawNotification::new::(&msg.to_string()); + sender.send(RawMessage::Notification(not)); +} + +fn is_canceled(e: &failure::Error) -> bool { + e.downcast_ref::().is_some() +} -- cgit v1.2.3