aboutsummaryrefslogtreecommitdiff
path: root/crates/ra_lsp_server/src/main_loop.rs
diff options
context:
space:
mode:
authorAleksey Kladov <[email protected]>2020-02-18 12:30:40 +0000
committerGitHub <[email protected]>2020-02-18 12:30:40 +0000
commitd05480a178b132e62b8aff8986a8cb3dd3a89c0b (patch)
tree0fc36373073a66c2bbd6c7cfae6cb734527d847f /crates/ra_lsp_server/src/main_loop.rs
parent2768476e491d985317b08230824f96e6718f338a (diff)
parent865759925be6b72f7ef39124ed0e4c86c0412a69 (diff)
Merge pull request #3216 from matklad/rename-to-rust-analyzer
rename binary to rust-analyzer
Diffstat (limited to 'crates/ra_lsp_server/src/main_loop.rs')
-rw-r--r--crates/ra_lsp_server/src/main_loop.rs889
1 files changed, 0 insertions, 889 deletions
diff --git a/crates/ra_lsp_server/src/main_loop.rs b/crates/ra_lsp_server/src/main_loop.rs
deleted file mode 100644
index 67d8a5f6f..000000000
--- a/crates/ra_lsp_server/src/main_loop.rs
+++ /dev/null
@@ -1,889 +0,0 @@
1//! The main loop of `ra_lsp_server` responsible for dispatching LSP
2//! requests/replies and notifications back to the client.
3
4mod handlers;
5mod subscriptions;
6pub(crate) mod pending_requests;
7
8use std::{
9 env,
10 error::Error,
11 fmt, panic,
12 path::PathBuf,
13 sync::Arc,
14 time::{Duration, Instant},
15};
16
17use crossbeam_channel::{select, unbounded, RecvError, Sender};
18use lsp_server::{Connection, ErrorCode, Message, Notification, Request, RequestId, Response};
19use lsp_types::{ClientCapabilities, NumberOrString};
20use ra_cargo_watch::{url_from_path_with_drive_lowercasing, CheckOptions, CheckTask};
21use ra_ide::{Canceled, FeatureFlags, FileId, LibraryData, SourceRootId};
22use ra_prof::profile;
23use ra_vfs::{VfsFile, VfsTask, Watch};
24use relative_path::RelativePathBuf;
25use rustc_hash::FxHashSet;
26use serde::{de::DeserializeOwned, Serialize};
27use threadpool::ThreadPool;
28
29use crate::{
30 diagnostics::DiagnosticTask,
31 main_loop::{
32 pending_requests::{PendingRequest, PendingRequests},
33 subscriptions::Subscriptions,
34 },
35 req,
36 world::{Options, WorldSnapshot, WorldState},
37 Result, ServerConfig,
38};
39
40#[derive(Debug)]
41pub struct LspError {
42 pub code: i32,
43 pub message: String,
44}
45
46impl LspError {
47 pub fn new(code: i32, message: String) -> LspError {
48 LspError { code, message }
49 }
50}
51
52impl fmt::Display for LspError {
53 fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
54 write!(f, "Language Server request failed with {}. ({})", self.code, self.message)
55 }
56}
57
58impl Error for LspError {}
59
60pub fn main_loop(
61 ws_roots: Vec<PathBuf>,
62 client_caps: ClientCapabilities,
63 config: ServerConfig,
64 connection: Connection,
65) -> Result<()> {
66 log::info!("server_config: {:#?}", config);
67
68 // Windows scheduler implements priority boosts: if thread waits for an
69 // event (like a condvar), and event fires, priority of the thread is
70 // temporary bumped. This optimization backfires in our case: each time the
71 // `main_loop` schedules a task to run on a threadpool, the worker threads
72 // gets a higher priority, and (on a machine with fewer cores) displaces the
73 // main loop! We work-around this by marking the main loop as a
74 // higher-priority thread.
75 //
76 // https://docs.microsoft.com/en-us/windows/win32/procthread/scheduling-priorities
77 // https://docs.microsoft.com/en-us/windows/win32/procthread/priority-boosts
78 // https://github.com/rust-analyzer/rust-analyzer/issues/2835
79 #[cfg(windows)]
80 unsafe {
81 use winapi::um::processthreadsapi::*;
82 let thread = GetCurrentThread();
83 let thread_priority_above_normal = 1;
84 SetThreadPriority(thread, thread_priority_above_normal);
85 }
86
87 let mut loop_state = LoopState::default();
88 let mut world_state = {
89 let feature_flags = {
90 let mut ff = FeatureFlags::default();
91 for (flag, value) in config.feature_flags {
92 if ff.set(flag.as_str(), value).is_err() {
93 log::error!("unknown feature flag: {:?}", flag);
94 show_message(
95 req::MessageType::Error,
96 format!("unknown feature flag: {:?}", flag),
97 &connection.sender,
98 );
99 }
100 }
101 ff
102 };
103 log::info!("feature_flags: {:#?}", feature_flags);
104
105 // FIXME: support dynamic workspace loading.
106 let workspaces = {
107 let mut loaded_workspaces = Vec::new();
108 for ws_root in &ws_roots {
109 let workspace = ra_project_model::ProjectWorkspace::discover_with_sysroot(
110 ws_root.as_path(),
111 config.with_sysroot,
112 &config.cargo_features,
113 );
114 match workspace {
115 Ok(workspace) => loaded_workspaces.push(workspace),
116 Err(e) => {
117 log::error!("loading workspace failed: {}", e);
118 if let Some(ra_project_model::CargoTomlNotFoundError(_)) = e.downcast_ref()
119 {
120 if !feature_flags.get("notifications.cargo-toml-not-found") {
121 continue;
122 }
123 }
124 show_message(
125 req::MessageType::Error,
126 format!("rust-analyzer failed to load workspace: {}", e),
127 &connection.sender,
128 );
129 }
130 }
131 }
132 loaded_workspaces
133 };
134
135 let globs = config
136 .exclude_globs
137 .iter()
138 .map(|glob| crate::vfs_glob::Glob::new(glob))
139 .collect::<std::result::Result<Vec<_>, _>>()?;
140
141 if config.use_client_watching {
142 let registration_options = req::DidChangeWatchedFilesRegistrationOptions {
143 watchers: workspaces
144 .iter()
145 .flat_map(|ws| ws.to_roots())
146 .filter(|root| root.is_member())
147 .map(|root| format!("{}/**/*.rs", root.path().display()))
148 .map(|glob_pattern| req::FileSystemWatcher { glob_pattern, kind: None })
149 .collect(),
150 };
151 let registration = req::Registration {
152 id: "file-watcher".to_string(),
153 method: "workspace/didChangeWatchedFiles".to_string(),
154 register_options: Some(serde_json::to_value(registration_options).unwrap()),
155 };
156 let params = req::RegistrationParams { registrations: vec![registration] };
157 let request =
158 request_new::<req::RegisterCapability>(loop_state.next_request_id(), params);
159 connection.sender.send(request.into()).unwrap();
160 }
161
162 let options = {
163 let text_document_caps = client_caps.text_document.as_ref();
164 Options {
165 publish_decorations: config.publish_decorations,
166 supports_location_link: text_document_caps
167 .and_then(|it| it.definition)
168 .and_then(|it| it.link_support)
169 .unwrap_or(false),
170 line_folding_only: text_document_caps
171 .and_then(|it| it.folding_range.as_ref())
172 .and_then(|it| it.line_folding_only)
173 .unwrap_or(false),
174 max_inlay_hint_length: config.max_inlay_hint_length,
175 cargo_watch: CheckOptions {
176 enable: config.cargo_watch_enable,
177 args: config.cargo_watch_args,
178 command: config.cargo_watch_command,
179 all_targets: config.cargo_watch_all_targets,
180 },
181 rustfmt_args: config.rustfmt_args,
182 }
183 };
184
185 WorldState::new(
186 ws_roots,
187 workspaces,
188 config.lru_capacity,
189 &globs,
190 Watch(!config.use_client_watching),
191 options,
192 feature_flags,
193 )
194 };
195
196 let pool = ThreadPool::default();
197 let (task_sender, task_receiver) = unbounded::<Task>();
198 let (libdata_sender, libdata_receiver) = unbounded::<LibraryData>();
199
200 log::info!("server initialized, serving requests");
201 {
202 let task_sender = task_sender;
203 let libdata_sender = libdata_sender;
204 loop {
205 log::trace!("selecting");
206 let event = select! {
207 recv(&connection.receiver) -> msg => match msg {
208 Ok(msg) => Event::Msg(msg),
209 Err(RecvError) => Err("client exited without shutdown")?,
210 },
211 recv(task_receiver) -> task => Event::Task(task.unwrap()),
212 recv(world_state.task_receiver) -> task => match task {
213 Ok(task) => Event::Vfs(task),
214 Err(RecvError) => Err("vfs died")?,
215 },
216 recv(libdata_receiver) -> data => Event::Lib(data.unwrap()),
217 recv(world_state.check_watcher.task_recv) -> task => match task {
218 Ok(task) => Event::CheckWatcher(task),
219 Err(RecvError) => Err("check watcher died")?,
220 }
221 };
222 if let Event::Msg(Message::Request(req)) = &event {
223 if connection.handle_shutdown(&req)? {
224 break;
225 };
226 }
227 loop_turn(
228 &pool,
229 &task_sender,
230 &libdata_sender,
231 &connection,
232 &mut world_state,
233 &mut loop_state,
234 event,
235 )?;
236 }
237 }
238 world_state.analysis_host.request_cancellation();
239 log::info!("waiting for tasks to finish...");
240 task_receiver.into_iter().for_each(|task| {
241 on_task(task, &connection.sender, &mut loop_state.pending_requests, &mut world_state)
242 });
243 libdata_receiver.into_iter().for_each(drop);
244 log::info!("...tasks have finished");
245 log::info!("joining threadpool...");
246 drop(pool);
247 log::info!("...threadpool has finished");
248
249 let vfs = Arc::try_unwrap(world_state.vfs).expect("all snapshots should be dead");
250 drop(vfs);
251
252 Ok(())
253}
254
255#[derive(Debug)]
256enum Task {
257 Respond(Response),
258 Notify(Notification),
259 Diagnostic(DiagnosticTask),
260}
261
262enum Event {
263 Msg(Message),
264 Task(Task),
265 Vfs(VfsTask),
266 Lib(LibraryData),
267 CheckWatcher(CheckTask),
268}
269
270impl fmt::Debug for Event {
271 fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
272 let debug_verbose_not = |not: &Notification, f: &mut fmt::Formatter| {
273 f.debug_struct("Notification").field("method", &not.method).finish()
274 };
275
276 match self {
277 Event::Msg(Message::Notification(not)) => {
278 if notification_is::<req::DidOpenTextDocument>(not)
279 || notification_is::<req::DidChangeTextDocument>(not)
280 {
281 return debug_verbose_not(not, f);
282 }
283 }
284 Event::Task(Task::Notify(not)) => {
285 if notification_is::<req::PublishDecorations>(not)
286 || notification_is::<req::PublishDiagnostics>(not)
287 {
288 return debug_verbose_not(not, f);
289 }
290 }
291 Event::Task(Task::Respond(resp)) => {
292 return f
293 .debug_struct("Response")
294 .field("id", &resp.id)
295 .field("error", &resp.error)
296 .finish();
297 }
298 _ => (),
299 }
300 match self {
301 Event::Msg(it) => fmt::Debug::fmt(it, f),
302 Event::Task(it) => fmt::Debug::fmt(it, f),
303 Event::Vfs(it) => fmt::Debug::fmt(it, f),
304 Event::Lib(it) => fmt::Debug::fmt(it, f),
305 Event::CheckWatcher(it) => fmt::Debug::fmt(it, f),
306 }
307 }
308}
309
310#[derive(Debug, Default)]
311struct LoopState {
312 next_request_id: u64,
313 pending_responses: FxHashSet<RequestId>,
314 pending_requests: PendingRequests,
315 subscriptions: Subscriptions,
316 // We try not to index more than MAX_IN_FLIGHT_LIBS libraries at the same
317 // time to always have a thread ready to react to input.
318 in_flight_libraries: usize,
319 pending_libraries: Vec<(SourceRootId, Vec<(FileId, RelativePathBuf, Arc<String>)>)>,
320 workspace_loaded: bool,
321}
322
323impl LoopState {
324 fn next_request_id(&mut self) -> RequestId {
325 self.next_request_id += 1;
326 let res: RequestId = self.next_request_id.into();
327 let inserted = self.pending_responses.insert(res.clone());
328 assert!(inserted);
329 res
330 }
331}
332
333fn loop_turn(
334 pool: &ThreadPool,
335 task_sender: &Sender<Task>,
336 libdata_sender: &Sender<LibraryData>,
337 connection: &Connection,
338 world_state: &mut WorldState,
339 loop_state: &mut LoopState,
340 event: Event,
341) -> Result<()> {
342 let loop_start = Instant::now();
343
344 // NOTE: don't count blocking select! call as a loop-turn time
345 let _p = profile("main_loop_inner/loop-turn");
346 log::info!("loop turn = {:?}", event);
347 let queue_count = pool.queued_count();
348 if queue_count > 0 {
349 log::info!("queued count = {}", queue_count);
350 }
351
352 match event {
353 Event::Task(task) => {
354 on_task(task, &connection.sender, &mut loop_state.pending_requests, world_state);
355 world_state.maybe_collect_garbage();
356 }
357 Event::Vfs(task) => {
358 world_state.vfs.write().handle_task(task);
359 }
360 Event::Lib(lib) => {
361 world_state.add_lib(lib);
362 world_state.maybe_collect_garbage();
363 loop_state.in_flight_libraries -= 1;
364 }
365 Event::CheckWatcher(task) => on_check_task(task, world_state, task_sender)?,
366 Event::Msg(msg) => match msg {
367 Message::Request(req) => on_request(
368 world_state,
369 &mut loop_state.pending_requests,
370 pool,
371 task_sender,
372 &connection.sender,
373 loop_start,
374 req,
375 )?,
376 Message::Notification(not) => {
377 on_notification(
378 &connection.sender,
379 world_state,
380 &mut loop_state.pending_requests,
381 &mut loop_state.subscriptions,
382 not,
383 )?;
384 }
385 Message::Response(resp) => {
386 let removed = loop_state.pending_responses.remove(&resp.id);
387 if !removed {
388 log::error!("unexpected response: {:?}", resp)
389 }
390 }
391 },
392 };
393
394 let mut state_changed = false;
395 if let Some(changes) = world_state.process_changes() {
396 state_changed = true;
397 loop_state.pending_libraries.extend(changes);
398 }
399
400 let max_in_flight_libs = pool.max_count().saturating_sub(2).max(1);
401 while loop_state.in_flight_libraries < max_in_flight_libs
402 && !loop_state.pending_libraries.is_empty()
403 {
404 let (root, files) = loop_state.pending_libraries.pop().unwrap();
405 loop_state.in_flight_libraries += 1;
406 let sender = libdata_sender.clone();
407 pool.execute(move || {
408 log::info!("indexing {:?} ... ", root);
409 let data = LibraryData::prepare(root, files);
410 sender.send(data).unwrap();
411 });
412 }
413
414 if !loop_state.workspace_loaded
415 && world_state.roots_to_scan == 0
416 && loop_state.pending_libraries.is_empty()
417 && loop_state.in_flight_libraries == 0
418 {
419 loop_state.workspace_loaded = true;
420 let n_packages: usize = world_state.workspaces.iter().map(|it| it.n_packages()).sum();
421 if world_state.feature_flags().get("notifications.workspace-loaded") {
422 let msg = format!("workspace loaded, {} rust packages", n_packages);
423 show_message(req::MessageType::Info, msg, &connection.sender);
424 }
425 world_state.check_watcher.update();
426 }
427
428 if state_changed {
429 update_file_notifications_on_threadpool(
430 pool,
431 world_state.snapshot(),
432 world_state.options.publish_decorations,
433 task_sender.clone(),
434 loop_state.subscriptions.subscriptions(),
435 )
436 }
437
438 let loop_duration = loop_start.elapsed();
439 if loop_duration > Duration::from_millis(100) {
440 log::error!("overly long loop turn: {:?}", loop_duration);
441 if env::var("RA_PROFILE").is_ok() {
442 show_message(
443 req::MessageType::Error,
444 format!("overly long loop turn: {:?}", loop_duration),
445 &connection.sender,
446 );
447 }
448 }
449
450 Ok(())
451}
452
453fn on_task(
454 task: Task,
455 msg_sender: &Sender<Message>,
456 pending_requests: &mut PendingRequests,
457 state: &mut WorldState,
458) {
459 match task {
460 Task::Respond(response) => {
461 if let Some(completed) = pending_requests.finish(&response.id) {
462 log::info!("handled req#{} in {:?}", completed.id, completed.duration);
463 state.complete_request(completed);
464 msg_sender.send(response.into()).unwrap();
465 }
466 }
467 Task::Notify(n) => {
468 msg_sender.send(n.into()).unwrap();
469 }
470 Task::Diagnostic(task) => on_diagnostic_task(task, msg_sender, state),
471 }
472}
473
474fn on_request(
475 world: &mut WorldState,
476 pending_requests: &mut PendingRequests,
477 pool: &ThreadPool,
478 task_sender: &Sender<Task>,
479 msg_sender: &Sender<Message>,
480 request_received: Instant,
481 req: Request,
482) -> Result<()> {
483 let mut pool_dispatcher = PoolDispatcher {
484 req: Some(req),
485 pool,
486 world,
487 task_sender,
488 msg_sender,
489 pending_requests,
490 request_received,
491 };
492 pool_dispatcher
493 .on_sync::<req::CollectGarbage>(|s, ()| Ok(s.collect_garbage()))?
494 .on_sync::<req::JoinLines>(|s, p| handlers::handle_join_lines(s.snapshot(), p))?
495 .on_sync::<req::OnEnter>(|s, p| handlers::handle_on_enter(s.snapshot(), p))?
496 .on_sync::<req::SelectionRangeRequest>(|s, p| {
497 handlers::handle_selection_range(s.snapshot(), p)
498 })?
499 .on_sync::<req::FindMatchingBrace>(|s, p| {
500 handlers::handle_find_matching_brace(s.snapshot(), p)
501 })?
502 .on::<req::AnalyzerStatus>(handlers::handle_analyzer_status)?
503 .on::<req::SyntaxTree>(handlers::handle_syntax_tree)?
504 .on::<req::ExpandMacro>(handlers::handle_expand_macro)?
505 .on::<req::OnTypeFormatting>(handlers::handle_on_type_formatting)?
506 .on::<req::DocumentSymbolRequest>(handlers::handle_document_symbol)?
507 .on::<req::WorkspaceSymbol>(handlers::handle_workspace_symbol)?
508 .on::<req::GotoDefinition>(handlers::handle_goto_definition)?
509 .on::<req::GotoImplementation>(handlers::handle_goto_implementation)?
510 .on::<req::GotoTypeDefinition>(handlers::handle_goto_type_definition)?
511 .on::<req::ParentModule>(handlers::handle_parent_module)?
512 .on::<req::Runnables>(handlers::handle_runnables)?
513 .on::<req::DecorationsRequest>(handlers::handle_decorations)?
514 .on::<req::Completion>(handlers::handle_completion)?
515 .on::<req::CodeActionRequest>(handlers::handle_code_action)?
516 .on::<req::CodeLensRequest>(handlers::handle_code_lens)?
517 .on::<req::CodeLensResolve>(handlers::handle_code_lens_resolve)?
518 .on::<req::FoldingRangeRequest>(handlers::handle_folding_range)?
519 .on::<req::SignatureHelpRequest>(handlers::handle_signature_help)?
520 .on::<req::HoverRequest>(handlers::handle_hover)?
521 .on::<req::PrepareRenameRequest>(handlers::handle_prepare_rename)?
522 .on::<req::Rename>(handlers::handle_rename)?
523 .on::<req::References>(handlers::handle_references)?
524 .on::<req::Formatting>(handlers::handle_formatting)?
525 .on::<req::DocumentHighlightRequest>(handlers::handle_document_highlight)?
526 .on::<req::InlayHints>(handlers::handle_inlay_hints)?
527 .on::<req::CallHierarchyPrepare>(handlers::handle_call_hierarchy_prepare)?
528 .on::<req::CallHierarchyIncomingCalls>(handlers::handle_call_hierarchy_incoming)?
529 .on::<req::CallHierarchyOutgoingCalls>(handlers::handle_call_hierarchy_outgoing)?
530 .on::<req::Ssr>(handlers::handle_ssr)?
531 .finish();
532 Ok(())
533}
534
535fn on_notification(
536 msg_sender: &Sender<Message>,
537 state: &mut WorldState,
538 pending_requests: &mut PendingRequests,
539 subs: &mut Subscriptions,
540 not: Notification,
541) -> Result<()> {
542 let not = match notification_cast::<req::Cancel>(not) {
543 Ok(params) => {
544 let id: RequestId = match params.id {
545 NumberOrString::Number(id) => id.into(),
546 NumberOrString::String(id) => id.into(),
547 };
548 if pending_requests.cancel(&id) {
549 let response = Response::new_err(
550 id,
551 ErrorCode::RequestCanceled as i32,
552 "canceled by client".to_string(),
553 );
554 msg_sender.send(response.into()).unwrap()
555 }
556 return Ok(());
557 }
558 Err(not) => not,
559 };
560 let not = match notification_cast::<req::DidOpenTextDocument>(not) {
561 Ok(params) => {
562 let uri = params.text_document.uri;
563 let path = uri.to_file_path().map_err(|()| format!("invalid uri: {}", uri))?;
564 if let Some(file_id) =
565 state.vfs.write().add_file_overlay(&path, params.text_document.text)
566 {
567 subs.add_sub(FileId(file_id.0));
568 }
569 return Ok(());
570 }
571 Err(not) => not,
572 };
573 let not = match notification_cast::<req::DidChangeTextDocument>(not) {
574 Ok(mut params) => {
575 let uri = params.text_document.uri;
576 let path = uri.to_file_path().map_err(|()| format!("invalid uri: {}", uri))?;
577 let text =
578 params.content_changes.pop().ok_or_else(|| "empty changes".to_string())?.text;
579 state.vfs.write().change_file_overlay(path.as_path(), text);
580 return Ok(());
581 }
582 Err(not) => not,
583 };
584 let not = match notification_cast::<req::DidSaveTextDocument>(not) {
585 Ok(_params) => {
586 state.check_watcher.update();
587 return Ok(());
588 }
589 Err(not) => not,
590 };
591 let not = match notification_cast::<req::DidCloseTextDocument>(not) {
592 Ok(params) => {
593 let uri = params.text_document.uri;
594 let path = uri.to_file_path().map_err(|()| format!("invalid uri: {}", uri))?;
595 if let Some(file_id) = state.vfs.write().remove_file_overlay(path.as_path()) {
596 subs.remove_sub(FileId(file_id.0));
597 }
598 let params =
599 req::PublishDiagnosticsParams { uri, diagnostics: Vec::new(), version: None };
600 let not = notification_new::<req::PublishDiagnostics>(params);
601 msg_sender.send(not.into()).unwrap();
602 return Ok(());
603 }
604 Err(not) => not,
605 };
606 let not = match notification_cast::<req::DidChangeConfiguration>(not) {
607 Ok(_params) => {
608 return Ok(());
609 }
610 Err(not) => not,
611 };
612 let not = match notification_cast::<req::DidChangeWatchedFiles>(not) {
613 Ok(params) => {
614 let mut vfs = state.vfs.write();
615 for change in params.changes {
616 let uri = change.uri;
617 let path = uri.to_file_path().map_err(|()| format!("invalid uri: {}", uri))?;
618 vfs.notify_changed(path)
619 }
620 return Ok(());
621 }
622 Err(not) => not,
623 };
624 log::error!("unhandled notification: {:?}", not);
625 Ok(())
626}
627
628fn on_check_task(
629 task: CheckTask,
630 world_state: &mut WorldState,
631 task_sender: &Sender<Task>,
632) -> Result<()> {
633 match task {
634 CheckTask::ClearDiagnostics => {
635 task_sender.send(Task::Diagnostic(DiagnosticTask::ClearCheck))?;
636 }
637
638 CheckTask::AddDiagnostic { url, diagnostic, fixes } => {
639 let path = url.to_file_path().map_err(|()| format!("invalid uri: {}", url))?;
640 let file_id = match world_state.vfs.read().path2file(&path) {
641 Some(file) => FileId(file.0),
642 None => {
643 log::error!("File with cargo diagnostic not found in VFS: {}", path.display());
644 return Ok(());
645 }
646 };
647
648 task_sender
649 .send(Task::Diagnostic(DiagnosticTask::AddCheck(file_id, diagnostic, fixes)))?;
650 }
651
652 CheckTask::Status(progress) => {
653 let params = req::ProgressParams {
654 token: req::ProgressToken::String("rustAnalyzer/cargoWatcher".to_string()),
655 value: req::ProgressParamsValue::WorkDone(progress),
656 };
657 let not = notification_new::<req::Progress>(params);
658 task_sender.send(Task::Notify(not)).unwrap();
659 }
660 };
661
662 Ok(())
663}
664
665fn on_diagnostic_task(task: DiagnosticTask, msg_sender: &Sender<Message>, state: &mut WorldState) {
666 let subscriptions = state.diagnostics.handle_task(task);
667
668 for file_id in subscriptions {
669 let path = state.vfs.read().file2path(VfsFile(file_id.0));
670 let uri = match url_from_path_with_drive_lowercasing(&path) {
671 Ok(uri) => uri,
672 Err(err) => {
673 log::error!("Couldn't convert path to url ({}): {:?}", err, path.to_string_lossy());
674 continue;
675 }
676 };
677
678 let diagnostics = state.diagnostics.diagnostics_for(file_id).cloned().collect();
679 let params = req::PublishDiagnosticsParams { uri, diagnostics, version: None };
680 let not = notification_new::<req::PublishDiagnostics>(params);
681 msg_sender.send(not.into()).unwrap();
682 }
683}
684
685struct PoolDispatcher<'a> {
686 req: Option<Request>,
687 pool: &'a ThreadPool,
688 world: &'a mut WorldState,
689 pending_requests: &'a mut PendingRequests,
690 msg_sender: &'a Sender<Message>,
691 task_sender: &'a Sender<Task>,
692 request_received: Instant,
693}
694
695impl<'a> PoolDispatcher<'a> {
696 /// Dispatches the request onto the current thread
697 fn on_sync<R>(
698 &mut self,
699 f: fn(&mut WorldState, R::Params) -> Result<R::Result>,
700 ) -> Result<&mut Self>
701 where
702 R: req::Request + 'static,
703 R::Params: DeserializeOwned + panic::UnwindSafe + 'static,
704 R::Result: Serialize + 'static,
705 {
706 let (id, params) = match self.parse::<R>() {
707 Some(it) => it,
708 None => {
709 return Ok(self);
710 }
711 };
712 let world = panic::AssertUnwindSafe(&mut *self.world);
713 let task = panic::catch_unwind(move || {
714 let result = f(world.0, params);
715 result_to_task::<R>(id, result)
716 })
717 .map_err(|_| format!("sync task {:?} panicked", R::METHOD))?;
718 on_task(task, self.msg_sender, self.pending_requests, self.world);
719 Ok(self)
720 }
721
722 /// Dispatches the request onto thread pool
723 fn on<R>(&mut self, f: fn(WorldSnapshot, R::Params) -> Result<R::Result>) -> Result<&mut Self>
724 where
725 R: req::Request + 'static,
726 R::Params: DeserializeOwned + Send + 'static,
727 R::Result: Serialize + 'static,
728 {
729 let (id, params) = match self.parse::<R>() {
730 Some(it) => it,
731 None => {
732 return Ok(self);
733 }
734 };
735
736 self.pool.execute({
737 let world = self.world.snapshot();
738 let sender = self.task_sender.clone();
739 move || {
740 let result = f(world, params);
741 let task = result_to_task::<R>(id, result);
742 sender.send(task).unwrap();
743 }
744 });
745
746 Ok(self)
747 }
748
749 fn parse<R>(&mut self) -> Option<(RequestId, R::Params)>
750 where
751 R: req::Request + 'static,
752 R::Params: DeserializeOwned + 'static,
753 {
754 let req = self.req.take()?;
755 let (id, params) = match req.extract::<R::Params>(R::METHOD) {
756 Ok(it) => it,
757 Err(req) => {
758 self.req = Some(req);
759 return None;
760 }
761 };
762 self.pending_requests.start(PendingRequest {
763 id: id.clone(),
764 method: R::METHOD.to_string(),
765 received: self.request_received,
766 });
767 Some((id, params))
768 }
769
770 fn finish(&mut self) {
771 match self.req.take() {
772 None => (),
773 Some(req) => {
774 log::error!("unknown request: {:?}", req);
775 let resp = Response::new_err(
776 req.id,
777 ErrorCode::MethodNotFound as i32,
778 "unknown request".to_string(),
779 );
780 self.msg_sender.send(resp.into()).unwrap();
781 }
782 }
783 }
784}
785
786fn result_to_task<R>(id: RequestId, result: Result<R::Result>) -> Task
787where
788 R: req::Request + 'static,
789 R::Params: DeserializeOwned + 'static,
790 R::Result: Serialize + 'static,
791{
792 let response = match result {
793 Ok(resp) => Response::new_ok(id, &resp),
794 Err(e) => match e.downcast::<LspError>() {
795 Ok(lsp_error) => Response::new_err(id, lsp_error.code, lsp_error.message),
796 Err(e) => {
797 if is_canceled(&e) {
798 Response::new_err(
799 id,
800 ErrorCode::ContentModified as i32,
801 "content modified".to_string(),
802 )
803 } else {
804 Response::new_err(id, ErrorCode::InternalError as i32, e.to_string())
805 }
806 }
807 },
808 };
809 Task::Respond(response)
810}
811
812fn update_file_notifications_on_threadpool(
813 pool: &ThreadPool,
814 world: WorldSnapshot,
815 publish_decorations: bool,
816 task_sender: Sender<Task>,
817 subscriptions: Vec<FileId>,
818) {
819 log::trace!("updating notifications for {:?}", subscriptions);
820 let publish_diagnostics = world.feature_flags().get("lsp.diagnostics");
821 pool.execute(move || {
822 for file_id in subscriptions {
823 if publish_diagnostics {
824 match handlers::publish_diagnostics(&world, file_id) {
825 Err(e) => {
826 if !is_canceled(&e) {
827 log::error!("failed to compute diagnostics: {:?}", e);
828 }
829 }
830 Ok(task) => {
831 task_sender.send(Task::Diagnostic(task)).unwrap();
832 }
833 }
834 }
835 if publish_decorations {
836 match handlers::publish_decorations(&world, file_id) {
837 Err(e) => {
838 if !is_canceled(&e) {
839 log::error!("failed to compute decorations: {:?}", e);
840 }
841 }
842 Ok(params) => {
843 let not = notification_new::<req::PublishDecorations>(params);
844 task_sender.send(Task::Notify(not)).unwrap();
845 }
846 }
847 }
848 }
849 });
850}
851
852pub fn show_message(typ: req::MessageType, message: impl Into<String>, sender: &Sender<Message>) {
853 let message = message.into();
854 let params = req::ShowMessageParams { typ, message };
855 let not = notification_new::<req::ShowMessage>(params);
856 sender.send(not.into()).unwrap();
857}
858
859fn is_canceled(e: &Box<dyn std::error::Error + Send + Sync>) -> bool {
860 e.downcast_ref::<Canceled>().is_some()
861}
862
863fn notification_is<N: lsp_types::notification::Notification>(notification: &Notification) -> bool {
864 notification.method == N::METHOD
865}
866
867fn notification_cast<N>(notification: Notification) -> std::result::Result<N::Params, Notification>
868where
869 N: lsp_types::notification::Notification,
870 N::Params: DeserializeOwned,
871{
872 notification.extract(N::METHOD)
873}
874
875fn notification_new<N>(params: N::Params) -> Notification
876where
877 N: lsp_types::notification::Notification,
878 N::Params: Serialize,
879{
880 Notification::new(N::METHOD.to_string(), params)
881}
882
883fn request_new<R>(id: RequestId, params: R::Params) -> Request
884where
885 R: lsp_types::request::Request,
886 R::Params: Serialize,
887{
888 Request::new(id, R::METHOD.to_string(), params)
889}
generated by cgit v1.2.3 (git 2.25.1) at 2024-11-02 07:21:36 +0000