Mirrors/rust-analyzer
2
0
Fork 0
mirror of https://github.com/rust-lang/rust-analyzer synced 2024-11-10 07:04:22 +00:00
Code Issues Projects Releases Packages Wiki Activity

Refactor main_loop

Browse source
This commit is contained in:
Aleksey Kladov 2020-06-25 17:14:11 +02:00
parent dd20c2ec5b
commit 379a096de9
4 changed files with 406 additions and 418 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

17
crates/flycheck/src/lib.rs
View file

@ -120,7 +120,13 @@ impl FlycheckActor {
) -> FlycheckActor {
FlycheckActor { sender, config, workspace_root, last_update_req: None, check_process: None }
}
fn next_event(&self, inbox: &Receiver<Restart>) -> Option<Event> {
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<Restart>) {
// If we rerun the thread, we need to discard the previous check results first
self.send(Message::ClearDiagnostics);
@ -167,15 +173,6 @@ impl FlycheckActor {
}
}
}
fn next_event(&self, inbox: &Receiver<Restart>) -> Option<Event> {
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

19
crates/rust-analyzer/src/global_state.rs
View file

@ -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 @@ use crate::{
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 @@ impl Default for Status {
/// snapshot of the file systems, and `analysis_host`, which stores our
/// incremental salsa database.
pub(crate) struct GlobalState {
sender: Sender<lsp_server::Message>,
pub(crate) config: Config,
pub(crate) task_pool: (TaskPool<Task>, Receiver<Task>),
pub(crate) analysis_host: AnalysisHost,
@ -95,6 +97,7 @@ pub(crate) struct GlobalStateSnapshot {
impl GlobalState {
pub(crate) fn new(
sender: Sender<lsp_server::Message>,
workspaces: Vec<ProjectWorkspace>,
lru_capacity: Option<usize>,
config: Config,
@ -162,6 +165,7 @@ impl GlobalState {
};
let mut res = GlobalState {
sender,
config,
task_pool,
analysis_host,
@ -252,6 +256,19 @@ impl GlobalState {
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 {

775
crates/rust-analyzer/src/main_loop.rs
View file

@ -5,9 +5,9 @@ use std::{
time::{Duration, Instant},
};
use crossbeam_channel::{never, select, RecvError, Sender};
use crossbeam_channel::{never, select, Receiver};
use lsp_server::{Connection, ErrorCode, Notification, Request, RequestId, Response};
use lsp_types::{request::Request as _, NumberOrString};
use lsp_types::{notification::Notification as _, request::Request as _, NumberOrString};
use ra_db::VfsPath;
use ra_ide::{Canceled, FileId};
use ra_prof::profile;
@ -50,7 +50,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 +113,371 @@ 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<lsp_server::Message>) -> Option<Event> {
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<lsp_server::Message>) -> 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::<Vec<_>>();
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<()> {
let mut pool_dispatcher =
PoolDispatcher { req: Some(req), global_state: self, request_received };
pool_dispatcher
.on_sync::<lsp_ext::CollectGarbage>(|s, ()| Ok(s.collect_garbage()))?
.on_sync::<lsp_ext::JoinLines>(|s, p| handlers::handle_join_lines(s.snapshot(), p))?
.on_sync::<lsp_ext::OnEnter>(|s, p| handlers::handle_on_enter(s.snapshot(), p))?
.on_sync::<lsp_types::request::Shutdown>(|_, ()| Ok(()))?
.on_sync::<lsp_types::request::SelectionRangeRequest>(|s, p| {
handlers::handle_selection_range(s.snapshot(), p)
})?
.on_sync::<lsp_ext::MatchingBrace>(|s, p| {
handlers::handle_matching_brace(s.snapshot(), p)
})?
.on::<lsp_ext::AnalyzerStatus>(handlers::handle_analyzer_status)?
.on::<lsp_ext::SyntaxTree>(handlers::handle_syntax_tree)?
.on::<lsp_ext::ExpandMacro>(handlers::handle_expand_macro)?
.on::<lsp_ext::ParentModule>(handlers::handle_parent_module)?
.on::<lsp_ext::Runnables>(handlers::handle_runnables)?
.on::<lsp_ext::InlayHints>(handlers::handle_inlay_hints)?
.on::<lsp_ext::CodeActionRequest>(handlers::handle_code_action)?
.on::<lsp_ext::ResolveCodeActionRequest>(handlers::handle_resolve_code_action)?
.on::<lsp_ext::HoverRequest>(handlers::handle_hover)?
.on::<lsp_types::request::OnTypeFormatting>(handlers::handle_on_type_formatting)?
.on::<lsp_types::request::DocumentSymbolRequest>(handlers::handle_document_symbol)?
.on::<lsp_types::request::WorkspaceSymbol>(handlers::handle_workspace_symbol)?
.on::<lsp_types::request::GotoDefinition>(handlers::handle_goto_definition)?
.on::<lsp_types::request::GotoImplementation>(handlers::handle_goto_implementation)?
.on::<lsp_types::request::GotoTypeDefinition>(handlers::handle_goto_type_definition)?
.on::<lsp_types::request::Completion>(handlers::handle_completion)?
.on::<lsp_types::request::CodeLensRequest>(handlers::handle_code_lens)?
.on::<lsp_types::request::CodeLensResolve>(handlers::handle_code_lens_resolve)?
.on::<lsp_types::request::FoldingRangeRequest>(handlers::handle_folding_range)?
.on::<lsp_types::request::SignatureHelpRequest>(handlers::handle_signature_help)?
.on::<lsp_types::request::PrepareRenameRequest>(handlers::handle_prepare_rename)?
.on::<lsp_types::request::Rename>(handlers::handle_rename)?
.on::<lsp_types::request::References>(handlers::handle_references)?
.on::<lsp_types::request::Formatting>(handlers::handle_formatting)?
.on::<lsp_types::request::DocumentHighlightRequest>(
handlers::handle_document_highlight,
)?
.on::<lsp_types::request::CallHierarchyPrepare>(
handlers::handle_call_hierarchy_prepare,
)?
.on::<lsp_types::request::CallHierarchyIncomingCalls>(
handlers::handle_call_hierarchy_incoming,
)?
.on::<lsp_types::request::CallHierarchyOutgoingCalls>(
handlers::handle_call_hierarchy_outgoing,
)?
.on::<lsp_types::request::SemanticTokensRequest>(handlers::handle_semantic_tokens)?
.on::<lsp_types::request::SemanticTokensRangeRequest>(
handlers::handle_semantic_tokens_range,
)?
.on::<lsp_ext::Ssr>(handlers::handle_ssr)?
.finish();
Ok(())
}
fn on_notification(&mut self, not: Notification) -> Result<()> {
let not = match notification_cast::<lsp_types::notification::Cancel>(not) {
Ok(params) => {
let id: RequestId = match params.id {
NumberOrString::Number(id) => id.into(),
NumberOrString::String(id) => id.into(),
};
if let Some(response) = self.req_queue.incoming.cancel(id) {
self.send(response.into())
}
return Ok(());
}
Err(not) => not,
};
let not = match notification_cast::<lsp_types::notification::DidOpenTextDocument>(not) {
Ok(params) => {
if let Ok(path) = from_proto::vfs_path(&params.text_document.uri) {
if !self.mem_docs.insert(path.clone()) {
log::error!("duplicate DidOpenTextDocument: {}", path)
}
self.vfs
.write()
.0
.set_file_contents(path, Some(params.text_document.text.into_bytes()));
}
return Ok(());
}
Err(not) => not,
};
let not = match notification_cast::<lsp_types::notification::DidChangeTextDocument>(not) {
Ok(params) => {
if let Ok(path) = from_proto::vfs_path(&params.text_document.uri) {
assert!(self.mem_docs.contains(&path));
let vfs = &mut self.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::<lsp_types::notification::DidCloseTextDocument>(not) {
Ok(params) => {
if let Ok(path) = from_proto::vfs_path(&params.text_document.uri) {
if !self.mem_docs.remove(&path) {
log::error!("orphan DidCloseTextDocument: {}", path)
}
if let Some(path) = path.as_path() {
self.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::<lsp_types::notification::PublishDiagnostics>(params);
self.send(not.into());
return Ok(());
}
Err(not) => not,
};
let not = match notification_cast::<lsp_types::notification::DidSaveTextDocument>(not) {
Ok(_params) => {
if let Some(flycheck) = &self.flycheck {
flycheck.0.update();
}
return Ok(());
}
Err(not) => not,
};
let not = match notification_cast::<lsp_types::notification::DidChangeConfiguration>(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 = self.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"
),
}
},
);
self.send(request.into());
return Ok(());
}
Err(not) => not,
};
let not = match notification_cast::<lsp_types::notification::DidChangeWatchedFiles>(not) {
Ok(params) => {
for change in params.changes {
if let Ok(path) = from_proto::abs_path(&change.uri) {
self.loader.invalidate(path)
}
}
return Ok(());
}
Err(not) => not,
};
if not.method.starts_with("$/") {
return Ok(());
}
log::error!("unhandled notification: {:?}", not);
Ok(())
}
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<FileId>) {
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::<Vec<_>>();
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 +530,10 @@ pub(crate) type ReqHandler = fn(&mut GlobalState, Response);
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::<Vec<_>>();
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<lsp_server::Message>, 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<lsp_server::Message>,
request_received: Instant,
req: Request,
) -> Result<()> {
let mut pool_dispatcher =
PoolDispatcher { req: Some(req), global_state, msg_sender, request_received };
pool_dispatcher
.on_sync::<lsp_ext::CollectGarbage>(|s, ()| Ok(s.collect_garbage()))?
.on_sync::<lsp_ext::JoinLines>(|s, p| handlers::handle_join_lines(s.snapshot(), p))?
.on_sync::<lsp_ext::OnEnter>(|s, p| handlers::handle_on_enter(s.snapshot(), p))?
.on_sync::<lsp_types::request::SelectionRangeRequest>(|s, p| {
handlers::handle_selection_range(s.snapshot(), p)
})?
.on_sync::<lsp_ext::MatchingBrace>(|s, p| handlers::handle_matching_brace(s.snapshot(), p))?
.on::<lsp_ext::AnalyzerStatus>(handlers::handle_analyzer_status)?
.on::<lsp_ext::SyntaxTree>(handlers::handle_syntax_tree)?
.on::<lsp_ext::ExpandMacro>(handlers::handle_expand_macro)?
.on::<lsp_ext::ParentModule>(handlers::handle_parent_module)?
.on::<lsp_ext::Runnables>(handlers::handle_runnables)?
.on::<lsp_ext::InlayHints>(handlers::handle_inlay_hints)?
.on::<lsp_ext::CodeActionRequest>(handlers::handle_code_action)?
.on::<lsp_ext::ResolveCodeActionRequest>(handlers::handle_resolve_code_action)?
.on::<lsp_ext::HoverRequest>(handlers::handle_hover)?
.on::<lsp_types::request::OnTypeFormatting>(handlers::handle_on_type_formatting)?
.on::<lsp_types::request::DocumentSymbolRequest>(handlers::handle_document_symbol)?
.on::<lsp_types::request::WorkspaceSymbol>(handlers::handle_workspace_symbol)?
.on::<lsp_types::request::GotoDefinition>(handlers::handle_goto_definition)?
.on::<lsp_types::request::GotoImplementation>(handlers::handle_goto_implementation)?
.on::<lsp_types::request::GotoTypeDefinition>(handlers::handle_goto_type_definition)?
.on::<lsp_types::request::Completion>(handlers::handle_completion)?
.on::<lsp_types::request::CodeLensRequest>(handlers::handle_code_lens)?
.on::<lsp_types::request::CodeLensResolve>(handlers::handle_code_lens_resolve)?
.on::<lsp_types::request::FoldingRangeRequest>(handlers::handle_folding_range)?
.on::<lsp_types::request::SignatureHelpRequest>(handlers::handle_signature_help)?
.on::<lsp_types::request::PrepareRenameRequest>(handlers::handle_prepare_rename)?
.on::<lsp_types::request::Rename>(handlers::handle_rename)?
.on::<lsp_types::request::References>(handlers::handle_references)?
.on::<lsp_types::request::Formatting>(handlers::handle_formatting)?
.on::<lsp_types::request::DocumentHighlightRequest>(handlers::handle_document_highlight)?
.on::<lsp_types::request::CallHierarchyPrepare>(handlers::handle_call_hierarchy_prepare)?
.on::<lsp_types::request::CallHierarchyIncomingCalls>(
handlers::handle_call_hierarchy_incoming,
)?
.on::<lsp_types::request::CallHierarchyOutgoingCalls>(
handlers::handle_call_hierarchy_outgoing,
)?
.on::<lsp_types::request::SemanticTokensRequest>(handlers::handle_semantic_tokens)?
.on::<lsp_types::request::SemanticTokensRangeRequest>(
handlers::handle_semantic_tokens_range,
)?
.on::<lsp_ext::Ssr>(handlers::handle_ssr)?
.finish();
Ok(())
}
fn on_notification(
msg_sender: &Sender<lsp_server::Message>,
global_state: &mut GlobalState,
not: Notification,
) -> Result<()> {
let not = match notification_cast::<lsp_types::notification::Cancel>(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::<lsp_types::notification::DidOpenTextDocument>(not) {
Ok(params) => {
if let Ok(path) = from_proto::vfs_path(&params.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::<lsp_types::notification::DidChangeTextDocument>(not) {
Ok(params) => {
if let Ok(path) = from_proto::vfs_path(&params.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::<lsp_types::notification::DidCloseTextDocument>(not) {
Ok(params) => {
if let Ok(path) = from_proto::vfs_path(&params.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::<lsp_types::notification::PublishDiagnostics>(params);
msg_sender.send(not.into()).unwrap();
return Ok(());
}
Err(not) => not,
};
let not = match notification_cast::<lsp_types::notification::DidSaveTextDocument>(not) {
Ok(_params) => {
if let Some(flycheck) = &global_state.flycheck {
flycheck.0.update();
}
return Ok(());
}
Err(not) => not,
};
let not = match notification_cast::<lsp_types::notification::DidChangeConfiguration>(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::<lsp_types::notification::DidChangeWatchedFiles>(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<lsp_server::Message>,
) -> 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 +558,6 @@ fn on_check_task(
diag.diagnostic,
diag.fixes.into_iter().map(|it| it.into()).collect(),
),
msg_sender,
global_state,
)
}
@ -563,26 +570,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<lsp_server::Message>,
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::<lsp_types::notification::PublishDiagnostics>(params);
msg_sender.send(not.into()).unwrap();
global_state.send(not.into());
}
}
@ -599,7 +602,6 @@ fn percentage(done: usize, total: usize) -> f64 {
fn report_progress(
global_state: &mut GlobalState,
sender: &Sender<lsp_server::Message>,
title: &str,
state: Progress,
message: Option<String>,
@ -616,7 +618,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,13 +643,12 @@ fn report_progress(
token,
value: lsp_types::ProgressParamsValue::WorkDone(work_done_progress),
});
sender.send(notification.into()).unwrap();
global_state.send(notification.into());
}
struct PoolDispatcher<'a> {
req: Option<Request>,
global_state: &'a mut GlobalState,
msg_sender: &'a Sender<lsp_server::Message>,
request_received: Instant,
}
@ -674,7 +675,7 @@ impl<'a> PoolDispatcher<'a> {
result_to_task::<R>(id, result)
})
.map_err(|_| format!("sync task {:?} panicked", R::METHOD))?;
on_task(task, self.msg_sender, self.global_state);
self.global_state.on_task(task);
Ok(self)
}
@ -736,7 +737,7 @@ impl<'a> PoolDispatcher<'a> {
ErrorCode::MethodNotFound as i32,
"unknown request".to_string(),
);
self.msg_sender.send(resp.into()).unwrap();
self.global_state.send(resp.into());
}
}
}
@ -767,29 +768,3 @@ where
};
Task::Respond(response)
}
fn update_file_notifications_on_threadpool(
global_state: &mut GlobalState,
subscriptions: Vec<FileId>,
) {
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::<Vec<_>>();
Task::Diagnostics(diagnostics)
})
}
}

13
crates/vfs-notify/src/lib.rs
View file

@ -82,7 +82,12 @@ impl NotifyActor {
watcher_receiver,
}
}
fn next_event(&self, receiver: &Receiver<Message>) -> Option<Event> {
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<Message>) {
while let Some(event) = self.next_event(&inbox) {
log::debug!("vfs-notify event: {:?}", event);
@ -154,12 +159,6 @@ impl NotifyActor {
}
}
}
fn next_event(&self, receiver: &Receiver<Message>) -> Option<Event> {
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,