crates/ra_lsp_server/src/main_loop.rs

   1 mod handlers;
   2 mod subscriptions;
   3
   4 use std::{fmt, path::PathBuf, sync::Arc};
   5
   6 use crossbeam_channel::{select, unbounded, Receiver, RecvError, Sender};
   7 use failure::{bail, format_err};
   8 use failure_derive::Fail;
   9 use gen_lsp_server::{
  10     handle_shutdown, ErrorCode, RawMessage, RawNotification, RawRequest, RawResponse,
  11 };
  12 use lsp_types::NumberOrString;
  13 use ra_ide_api::{Canceled, FileId, LibraryData};
  14 use ra_vfs::VfsTask;
  15 use rustc_hash::FxHashSet;
  16 use serde::{de::DeserializeOwned, Serialize};
  17 use threadpool::ThreadPool;
  18
  19 use crate::{
  20     main_loop::subscriptions::Subscriptions,
  21     project_model::workspace_loader,
  22     req,
  23     server_world::{ServerWorld, ServerWorldState},
  24     Result,
  25 };
  26
  27 #[derive(Debug, Fail)]
  28 #[fail(
  29     display = "Language Server request failed with {}. ({})",
  30     code, message
  31 )]
  32 pub struct LspError {
  33     pub code: i32,
  34     pub message: String,
  35 }
  36
  37 impl LspError {
  38     pub fn new(code: i32, message: String) -> LspError {
  39         LspError { code, message }
  40     }
  41 }
  42
  43 #[derive(Debug)]
  44 enum Task {
  45     Respond(RawResponse),
  46     Notify(RawNotification),
  47 }
  48
  49 const THREADPOOL_SIZE: usize = 8;
  50
  51 pub fn main_loop(
  52     internal_mode: bool,
  53     ws_root: PathBuf,
  54     supports_decorations: bool,
  55     msg_receiver: &Receiver<RawMessage>,
  56     msg_sender: &Sender<RawMessage>,
  57 ) -> Result<()> {
  58     let pool = ThreadPool::new(THREADPOOL_SIZE);
  59     let (task_sender, task_receiver) = unbounded::<Task>();
  60     let (ws_worker, ws_watcher) = workspace_loader();
  61
  62     ws_worker.send(ws_root.clone()).unwrap();
  63     // FIXME: support dynamic workspace loading.
  64     let workspaces = match ws_worker.recv().unwrap() {
  65         Ok(ws) => vec![ws],
  66         Err(e) => {
  67             log::error!("loading workspace failed: {}", e);
  68             Vec::new()
  69         }
  70     };
  71     ws_worker.shutdown();
  72     ws_watcher
  73         .shutdown()
  74         .map_err(|_| format_err!("ws watcher died"))?;
  75     let mut state = ServerWorldState::new(ws_root.clone(), workspaces);
  76
  77     log::info!("server initialized, serving requests");
  78
  79     let mut pending_requests = FxHashSet::default();
  80     let mut subs = Subscriptions::new();
  81     let main_res = main_loop_inner(
  82         internal_mode,
  83         supports_decorations,
  84         &pool,
  85         msg_sender,
  86         msg_receiver,
  87         task_sender,
  88         task_receiver.clone(),
  89         &mut state,
  90         &mut pending_requests,
  91         &mut subs,
  92     );
  93
  94     log::info!("waiting for tasks to finish...");
  95     task_receiver
  96         .into_iter()
  97         .for_each(|task| on_task(task, msg_sender, &mut pending_requests));
  98     log::info!("...tasks have finished");
  99     log::info!("joining threadpool...");
 100     drop(pool);
 101     log::info!("...threadpool has finished");
 102
 103     let vfs = Arc::try_unwrap(state.vfs).expect("all snapshots should be dead");
 104     let vfs_res = vfs.into_inner().shutdown();
 105
 106     main_res?;
 107     vfs_res.map_err(|_| format_err!("fs watcher died"))?;
 108
 109     Ok(())
 110 }
 111
 112 enum Event {
 113     Msg(RawMessage),
 114     Task(Task),
 115     Vfs(VfsTask),
 116     Lib(LibraryData),
 117 }
 118
 119 impl fmt::Debug for Event {
 120     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
 121         let debug_verbose_not = |not: &RawNotification, f: &mut fmt::Formatter| {
 122             f.debug_struct("RawNotification")
 123                 .field("method", &not.method)
 124                 .finish()
 125         };
 126
 127         match self {
 128             Event::Msg(RawMessage::Notification(not)) => {
 129                 if not.is::<req::DidOpenTextDocument>() || not.is::<req::DidChangeTextDocument>() {
 130                     return debug_verbose_not(not, f);
 131                 }
 132             }
 133             Event::Task(Task::Notify(not)) => {
 134                 if not.is::<req::PublishDecorations>() || not.is::<req::PublishDiagnostics>() {
 135                     return debug_verbose_not(not, f);
 136                 }
 137             }
 138             Event::Task(Task::Respond(resp)) => {
 139                 return f
 140                     .debug_struct("RawResponse")
 141                     .field("id", &resp.id)
 142                     .field("error", &resp.error)
 143                     .finish();
 144             }
 145             _ => (),
 146         }
 147         match self {
 148             Event::Msg(it) => fmt::Debug::fmt(it, f),
 149             Event::Task(it) => fmt::Debug::fmt(it, f),
 150             Event::Vfs(it) => fmt::Debug::fmt(it, f),
 151             Event::Lib(it) => fmt::Debug::fmt(it, f),
 152         }
 153     }
 154 }
 155
 156 fn main_loop_inner(
 157     internal_mode: bool,
 158     supports_decorations: bool,
 159     pool: &ThreadPool,
 160     msg_sender: &Sender<RawMessage>,
 161     msg_receiver: &Receiver<RawMessage>,
 162     task_sender: Sender<Task>,
 163     task_receiver: Receiver<Task>,
 164     state: &mut ServerWorldState,
 165     pending_requests: &mut FxHashSet<u64>,
 166     subs: &mut Subscriptions,
 167 ) -> Result<()> {
 168     // We try not to index more than THREADPOOL_SIZE - 3 libraries at the same
 169     // time to always have a thread ready to react to input.
 170     let mut in_flight_libraries = 0;
 171     let mut pending_libraries = Vec::new();
 172
 173     let (libdata_sender, libdata_receiver) = unbounded();
 174     loop {
 175         log::trace!("selecting");
 176         let event = select! {
 177             recv(msg_receiver) -> msg => match msg {
 178                 Ok(msg) => Event::Msg(msg),
 179                 Err(RecvError) => bail!("client exited without shutdown"),
 180             },
 181             recv(task_receiver) -> task => Event::Task(task.unwrap()),
 182             recv(state.vfs.read().task_receiver()) -> task => match task {
 183                 Ok(task) => Event::Vfs(task),
 184                 Err(RecvError) => bail!("vfs died"),
 185             },
 186             recv(libdata_receiver) -> data => Event::Lib(data.unwrap())
 187         };
 188         log::info!("loop_turn = {:?}", event);
 189         let start = std::time::Instant::now();
 190         let mut state_changed = false;
 191         match event {
 192             Event::Task(task) => on_task(task, msg_sender, pending_requests),
 193             Event::Vfs(task) => {
 194                 state.vfs.write().handle_task(task);
 195                 state_changed = true;
 196             }
 197             Event::Lib(lib) => {
 198                 feedback(internal_mode, "library loaded", msg_sender);
 199                 state.add_lib(lib);
 200                 in_flight_libraries -= 1;
 201             }
 202             Event::Msg(msg) => match msg {
 203                 RawMessage::Request(req) => {
 204                     let req = match handle_shutdown(req, msg_sender) {
 205                         Some(req) => req,
 206                         None => return Ok(()),
 207                     };
 208                     match req.cast::<req::CollectGarbage>() {
 209                         Ok((id, ())) => {
 210                             state.collect_garbadge();
 211                             let resp = RawResponse::ok::<req::CollectGarbage>(id, &());
 212                             msg_sender.send(RawMessage::Response(resp)).unwrap()
 213                         }
 214                         Err(req) => {
 215                             match on_request(state, pending_requests, pool, &task_sender, req)? {
 216                                 None => (),
 217                                 Some(req) => {
 218                                     log::error!("unknown request: {:?}", req);
 219                                     let resp = RawResponse::err(
 220                                         req.id,
 221                                         ErrorCode::MethodNotFound as i32,
 222                                         "unknown request".to_string(),
 223                                     );
 224                                     msg_sender.send(RawMessage::Response(resp)).unwrap()
 225                                 }
 226                             }
 227                         }
 228                     }
 229                 }
 230                 RawMessage::Notification(not) => {
 231                     on_notification(msg_sender, state, pending_requests, subs, not)?;
 232                     state_changed = true;
 233                 }
 234                 RawMessage::Response(resp) => log::error!("unexpected response: {:?}", resp),
 235             },
 236         };
 237
 238         pending_libraries.extend(state.process_changes());
 239         while in_flight_libraries < THREADPOOL_SIZE - 3 && !pending_libraries.is_empty() {
 240             let (root, files) = pending_libraries.pop().unwrap();
 241             in_flight_libraries += 1;
 242             let sender = libdata_sender.clone();
 243             pool.execute(move || {
 244                 let start = ::std::time::Instant::now();
 245                 log::info!("indexing {:?} ... ", root);
 246                 let data = LibraryData::prepare(root, files);
 247                 log::info!("indexed {:?} {:?}", start.elapsed(), root);
 248                 sender.send(data).unwrap();
 249             });
 250         }
 251
 252         if state.roots_to_scan == 0 && pending_libraries.is_empty() && in_flight_libraries == 0 {
 253             feedback(internal_mode, "workspace loaded", msg_sender);
 254         }
 255
 256         if state_changed {
 257             update_file_notifications_on_threadpool(
 258                 pool,
 259                 state.snapshot(),
 260                 supports_decorations,
 261                 task_sender.clone(),
 262                 subs.subscriptions(),
 263             )
 264         }
 265         log::info!("loop_turn = {:?}", start.elapsed());
 266     }
 267 }
 268
 269 fn on_task(task: Task, msg_sender: &Sender<RawMessage>, pending_requests: &mut FxHashSet<u64>) {
 270     match task {
 271         Task::Respond(response) => {
 272             if pending_requests.remove(&response.id) {
 273                 msg_sender.send(RawMessage::Response(response)).unwrap();
 274             }
 275         }
 276         Task::Notify(n) => {
 277             msg_sender.send(RawMessage::Notification(n)).unwrap();
 278         }
 279     }
 280 }
 281
 282 fn on_request(
 283     world: &mut ServerWorldState,
 284     pending_requests: &mut FxHashSet<u64>,
 285     pool: &ThreadPool,
 286     sender: &Sender<Task>,
 287     req: RawRequest,
 288 ) -> Result<Option<RawRequest>> {
 289     let mut pool_dispatcher = PoolDispatcher {
 290         req: Some(req),
 291         res: None,
 292         pool,
 293         world,
 294         sender,
 295     };
 296     let req = pool_dispatcher
 297         .on::<req::AnalyzerStatus>(handlers::handle_analyzer_status)?
 298         .on::<req::SyntaxTree>(handlers::handle_syntax_tree)?
 299         .on::<req::ExtendSelection>(handlers::handle_extend_selection)?
 300         .on::<req::FindMatchingBrace>(handlers::handle_find_matching_brace)?
 301         .on::<req::JoinLines>(handlers::handle_join_lines)?
 302         .on::<req::OnEnter>(handlers::handle_on_enter)?
 303         .on::<req::OnTypeFormatting>(handlers::handle_on_type_formatting)?
 304         .on::<req::DocumentSymbolRequest>(handlers::handle_document_symbol)?
 305         .on::<req::WorkspaceSymbol>(handlers::handle_workspace_symbol)?
 306         .on::<req::GotoDefinition>(handlers::handle_goto_definition)?
 307         .on::<req::ParentModule>(handlers::handle_parent_module)?
 308         .on::<req::Runnables>(handlers::handle_runnables)?
 309         .on::<req::DecorationsRequest>(handlers::handle_decorations)?
 310         .on::<req::Completion>(handlers::handle_completion)?
 311         .on::<req::CodeActionRequest>(handlers::handle_code_action)?
 312         .on::<req::CodeLensRequest>(handlers::handle_code_lens)?
 313         .on::<req::FoldingRangeRequest>(handlers::handle_folding_range)?
 314         .on::<req::SignatureHelpRequest>(handlers::handle_signature_help)?
 315         .on::<req::HoverRequest>(handlers::handle_hover)?
 316         .on::<req::PrepareRenameRequest>(handlers::handle_prepare_rename)?
 317         .on::<req::Rename>(handlers::handle_rename)?
 318         .on::<req::References>(handlers::handle_references)?
 319         .on::<req::Formatting>(handlers::handle_formatting)?
 320         .on::<req::DocumentHighlightRequest>(handlers::handle_document_highlight)?
 321         .finish();
 322     match req {
 323         Ok(id) => {
 324             let inserted = pending_requests.insert(id);
 325             assert!(inserted, "duplicate request: {}", id);
 326             Ok(None)
 327         }
 328         Err(req) => Ok(Some(req)),
 329     }
 330 }
 331
 332 fn on_notification(
 333     msg_sender: &Sender<RawMessage>,
 334     state: &mut ServerWorldState,
 335     pending_requests: &mut FxHashSet<u64>,
 336     subs: &mut Subscriptions,
 337     not: RawNotification,
 338 ) -> Result<()> {
 339     let not = match not.cast::<req::Cancel>() {
 340         Ok(params) => {
 341             let id = match params.id {
 342                 NumberOrString::Number(id) => id,
 343                 NumberOrString::String(id) => {
 344                     panic!("string id's not supported: {:?}", id);
 345                 }
 346             };
 347             if pending_requests.remove(&id) {
 348                 let response = RawResponse::err(
 349                     id,
 350                     ErrorCode::RequestCanceled as i32,
 351                     "canceled by client".to_string(),
 352                 );
 353                 msg_sender.send(RawMessage::Response(response)).unwrap()
 354             }
 355             return Ok(());
 356         }
 357         Err(not) => not,
 358     };
 359     let not = match not.cast::<req::DidOpenTextDocument>() {
 360         Ok(params) => {
 361             let uri = params.text_document.uri;
 362             let path = uri
 363                 .to_file_path()
 364                 .map_err(|()| format_err!("invalid uri: {}", uri))?;
 365             if let Some(file_id) = state
 366                 .vfs
 367                 .write()
 368                 .add_file_overlay(&path, params.text_document.text)
 369             {
 370                 subs.add_sub(FileId(file_id.0.into()));
 371             }
 372             return Ok(());
 373         }
 374         Err(not) => not,
 375     };
 376     let not = match not.cast::<req::DidChangeTextDocument>() {
 377         Ok(mut params) => {
 378             let uri = params.text_document.uri;
 379             let path = uri
 380                 .to_file_path()
 381                 .map_err(|()| format_err!("invalid uri: {}", uri))?;
 382             let text = params
 383                 .content_changes
 384                 .pop()
 385                 .ok_or_else(|| format_err!("empty changes"))?
 386                 .text;
 387             state.vfs.write().change_file_overlay(path.as_path(), text);
 388             return Ok(());
 389         }
 390         Err(not) => not,
 391     };
 392     let not = match not.cast::<req::DidCloseTextDocument>() {
 393         Ok(params) => {
 394             let uri = params.text_document.uri;
 395             let path = uri
 396                 .to_file_path()
 397                 .map_err(|()| format_err!("invalid uri: {}", uri))?;
 398             if let Some(file_id) = state.vfs.write().remove_file_overlay(path.as_path()) {
 399                 subs.remove_sub(FileId(file_id.0.into()));
 400             }
 401             let params = req::PublishDiagnosticsParams {
 402                 uri,
 403                 diagnostics: Vec::new(),
 404             };
 405             let not = RawNotification::new::<req::PublishDiagnostics>(&params);
 406             msg_sender.send(RawMessage::Notification(not)).unwrap();
 407             return Ok(());
 408         }
 409         Err(not) => not,
 410     };
 411     log::error!("unhandled notification: {:?}", not);
 412     Ok(())
 413 }
 414
 415 struct PoolDispatcher<'a> {
 416     req: Option<RawRequest>,
 417     res: Option<u64>,
 418     pool: &'a ThreadPool,
 419     world: &'a ServerWorldState,
 420     sender: &'a Sender<Task>,
 421 }
 422
 423 impl<'a> PoolDispatcher<'a> {
 424     fn on<R>(&mut self, f: fn(ServerWorld, R::Params) -> Result<R::Result>) -> Result<&mut Self>
 425     where
 426         R: req::Request,
 427         R::Params: DeserializeOwned + Send + 'static,
 428         R::Result: Serialize + 'static,
 429     {
 430         let req = match self.req.take() {
 431             None => return Ok(self),
 432             Some(req) => req,
 433         };
 434         match req.cast::<R>() {
 435             Ok((id, params)) => {
 436                 let world = self.world.snapshot();
 437                 let sender = self.sender.clone();
 438                 self.pool.execute(move || {
 439                     let resp = match f(world, params) {
 440                         Ok(resp) => RawResponse::ok::<R>(id, &resp),
 441                         Err(e) => match e.downcast::<LspError>() {
 442                             Ok(lsp_error) => {
 443                                 RawResponse::err(id, lsp_error.code, lsp_error.message)
 444                             }
 445                             Err(e) => {
 446                                 if is_canceled(&e) {
 447                                     RawResponse::err(
 448                                         id,
 449                                         ErrorCode::ContentModified as i32,
 450                                         "content modified".to_string(),
 451                                     )
 452                                 } else {
 453                                     RawResponse::err(
 454                                         id,
 455                                         ErrorCode::InternalError as i32,
 456                                         format!("{}\n{}", e, e.backtrace()),
 457                                     )
 458                                 }
 459                             }
 460                         },
 461                     };
 462                     let task = Task::Respond(resp);
 463                     sender.send(task).unwrap();
 464                 });
 465                 self.res = Some(id);
 466             }
 467             Err(req) => self.req = Some(req),
 468         }
 469         Ok(self)
 470     }
 471
 472     fn finish(&mut self) -> ::std::result::Result<u64, RawRequest> {
 473         match (self.res.take(), self.req.take()) {
 474             (Some(res), None) => Ok(res),
 475             (None, Some(req)) => Err(req),
 476             _ => unreachable!(),
 477         }
 478     }
 479 }
 480
 481 fn update_file_notifications_on_threadpool(
 482     pool: &ThreadPool,
 483     world: ServerWorld,
 484     publish_decorations: bool,
 485     sender: Sender<Task>,
 486     subscriptions: Vec<FileId>,
 487 ) {
 488     pool.execute(move || {
 489         for file_id in subscriptions {
 490             match handlers::publish_diagnostics(&world, file_id) {
 491                 Err(e) => {
 492                     if !is_canceled(&e) {
 493                         log::error!("failed to compute diagnostics: {:?}", e);
 494                     }
 495                 }
 496                 Ok(params) => {
 497                     let not = RawNotification::new::<req::PublishDiagnostics>(&params);
 498                     sender.send(Task::Notify(not)).unwrap();
 499                 }
 500             }
 501             if publish_decorations {
 502                 match handlers::publish_decorations(&world, file_id) {
 503                     Err(e) => {
 504                         if !is_canceled(&e) {
 505                             log::error!("failed to compute decorations: {:?}", e);
 506                         }
 507                     }
 508                     Ok(params) => {
 509                         let not = RawNotification::new::<req::PublishDecorations>(&params);
 510                         sender.send(Task::Notify(not)).unwrap();
 511                     }
 512                 }
 513             }
 514         }
 515     });
 516 }
 517
 518 fn feedback(intrnal_mode: bool, msg: &str, sender: &Sender<RawMessage>) {
 519     if !intrnal_mode {
 520         return;
 521     }
 522     let not = RawNotification::new::<req::InternalFeedback>(&msg.to_string());
 523     sender.send(RawMessage::Notification(not)).unwrap();
 524 }
 525
 526 fn is_canceled(e: &failure::Error) -> bool {
 527     e.downcast_ref::<Canceled>().is_some()
 528 }