Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Add mandatory panic contexts to all threadpool tasks #14965

Merged
merged 1 commit into from
Jun 4, 2023
Merged

Add mandatory panic contexts to all threadpool tasks #14965

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 4 additions & 10 deletions crates/rust-analyzer/src/dispatch.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -104,13 +104,10 @@ impl<'a> RequestDispatcher<'a> {
None => return self,
};

self.global_state.task_pool.handle.spawn(ThreadIntent::Worker, {
self.global_state.task_pool.handle.spawn(ThreadIntent::Worker, panic_context, {
let world = self.global_state.snapshot();
move || {
let result = panic::catch_unwind(move || {
let _pctx = stdx::panic_context::enter(panic_context);
f(world, params)
});
let result = panic::catch_unwind(move || f(world, params));
match thread_result_to_response::<R>(req.id.clone(), result) {
Ok(response) => Task::Response(response),
Err(_) => Task::Response(lsp_server::Response::new_err(
Expand Down Expand Up @@ -178,13 +175,10 @@ impl<'a> RequestDispatcher<'a> {
None => return self,
};

self.global_state.task_pool.handle.spawn(intent, {
self.global_state.task_pool.handle.spawn(intent, panic_context, {
let world = self.global_state.snapshot();
move || {
let result = panic::catch_unwind(move || {
let _pctx = stdx::panic_context::enter(panic_context);
f(world, params)
});
let result = panic::catch_unwind(move || f(world, params));
match thread_result_to_response::<R>(req.id.clone(), result) {
Ok(response) => Task::Response(response),
Err(_) => Task::Retry(req),
Expand Down
14 changes: 9 additions & 5 deletions crates/rust-analyzer/src/handlers/notification.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -291,11 +291,15 @@ fn run_flycheck(state: &mut GlobalState, vfs_path: VfsPath) -> bool {
}
Ok(())
};
state.task_pool.handle.spawn_with_sender(stdx::thread::ThreadIntent::Worker, move |_| {
if let Err(e) = std::panic::catch_unwind(task) {
tracing::error!("flycheck task panicked: {e:?}")
}
});
state.task_pool.handle.spawn_with_sender(
stdx::thread::ThreadIntent::Worker,
"flycheck",
move |_| {
if let Err(e) = std::panic::catch_unwind(task) {
tracing::error!("flycheck task panicked: {e:?}")
}
},
);
true
} else {
false
Expand Down
140 changes: 76 additions & 64 deletions crates/rust-analyzer/src/main_loop.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -397,19 +397,25 @@ impl GlobalState {
tracing::debug!(%cause, "will prime caches");
let num_worker_threads = self.config.prime_caches_num_threads();

self.task_pool.handle.spawn_with_sender(stdx::thread::ThreadIntent::Worker, {
let analysis = self.snapshot().analysis;
move |sender| {
sender.send(Task::PrimeCaches(PrimeCachesProgress::Begin)).unwrap();
let res = analysis.parallel_prime_caches(num_worker_threads, |progress| {
let report = PrimeCachesProgress::Report(progress);
sender.send(Task::PrimeCaches(report)).unwrap();
});
sender
.send(Task::PrimeCaches(PrimeCachesProgress::End { cancelled: res.is_err() }))
.unwrap();
}
});
self.task_pool.handle.spawn_with_sender(
stdx::thread::ThreadIntent::Worker,
"prime_caches",
{
let analysis = self.snapshot().analysis;
move |sender| {
sender.send(Task::PrimeCaches(PrimeCachesProgress::Begin)).unwrap();
let res = analysis.parallel_prime_caches(num_worker_threads, |progress| {
let report = PrimeCachesProgress::Report(progress);
sender.send(Task::PrimeCaches(report)).unwrap();
});
sender
.send(Task::PrimeCaches(PrimeCachesProgress::End {
cancelled: res.is_err(),
}))
.unwrap();
}
},
);
}

fn update_status_or_notify(&mut self) {
Expand Down Expand Up @@ -796,56 +802,62 @@ impl GlobalState {

// Diagnostics are triggered by the user typing
// so we run them on a latency sensitive thread.
self.task_pool.handle.spawn(stdx::thread::ThreadIntent::LatencySensitive, move || {
let _p = profile::span("publish_diagnostics");
let diagnostics = subscriptions
.into_iter()
.filter_map(|file_id| {
let line_index = snapshot.file_line_index(file_id).ok()?;
Some((
file_id,
line_index,
snapshot
.analysis
.diagnostics(
&snapshot.config.diagnostics(),
ide::AssistResolveStrategy::None,
file_id,
)
.ok()?,
))
})
.map(|(file_id, line_index, it)| {
(
file_id,
it.into_iter()
.map(move |d| lsp_types::Diagnostic {
range: crate::to_proto::range(&line_index, d.range),
severity: Some(crate::to_proto::diagnostic_severity(d.severity)),
code: Some(lsp_types::NumberOrString::String(
d.code.as_str().to_string(),
)),
code_description: Some(lsp_types::CodeDescription {
href: lsp_types::Url::parse(&format!(
"https://rust-analyzer.github.io/manual.html#{}",
d.code.as_str()
))
.unwrap(),
}),
source: Some("rust-analyzer".to_string()),
message: d.message,
related_information: None,
tags: if d.unused {
Some(vec![lsp_types::DiagnosticTag::UNNECESSARY])
} else {
None
},
data: None,
})
.collect::<Vec<_>>(),
)
});
Task::Diagnostics(diagnostics.collect())
});
self.task_pool.handle.spawn(
stdx::thread::ThreadIntent::LatencySensitive,
"publish_diagnostics",
move || {
let _p = profile::span("publish_diagnostics");
let diagnostics = subscriptions
.into_iter()
.filter_map(|file_id| {
let line_index = snapshot.file_line_index(file_id).ok()?;
Some((
file_id,
line_index,
snapshot
.analysis
.diagnostics(
&snapshot.config.diagnostics(),
ide::AssistResolveStrategy::None,
file_id,
)
.ok()?,
))
})
.map(|(file_id, line_index, it)| {
(
file_id,
it.into_iter()
.map(move |d| lsp_types::Diagnostic {
range: crate::to_proto::range(&line_index, d.range),
severity: Some(crate::to_proto::diagnostic_severity(
d.severity,
)),
code: Some(lsp_types::NumberOrString::String(
d.code.as_str().to_string(),
)),
code_description: Some(lsp_types::CodeDescription {
href: lsp_types::Url::parse(&format!(
"https://rust-analyzer.github.io/manual.html#{}",
d.code.as_str()
))
.unwrap(),
}),
source: Some("rust-analyzer".to_string()),
message: d.message,
related_information: None,
tags: if d.unused {
Some(vec![lsp_types::DiagnosticTag::UNNECESSARY])
} else {
None
},
data: None,
})
.collect::<Vec<_>>(),
)
});
Task::Diagnostics(diagnostics.collect())
},
);
}
}
116 changes: 63 additions & 53 deletions crates/rust-analyzer/src/reload.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -185,7 +185,7 @@ impl GlobalState {
pub(crate) fn fetch_workspaces(&mut self, cause: Cause) {
tracing::info!(%cause, "will fetch workspaces");

self.task_pool.handle.spawn_with_sender(ThreadIntent::Worker, {
self.task_pool.handle.spawn_with_sender(ThreadIntent::Worker, "fetch_workspaces", {
let linked_projects = self.config.linked_projects();
let detached_files = self.config.detached_files().to_vec();
let cargo_config = self.config.cargo();
Expand Down Expand Up @@ -260,70 +260,80 @@ impl GlobalState {
tracing::info!(%cause, "will fetch build data");
let workspaces = Arc::clone(&self.workspaces);
let config = self.config.cargo();
self.task_pool.handle.spawn_with_sender(ThreadIntent::Worker, move |sender| {
sender.send(Task::FetchBuildData(BuildDataProgress::Begin)).unwrap();
self.task_pool.handle.spawn_with_sender(
ThreadIntent::Worker,
"fetch_build_data",
move |sender| {
sender.send(Task::FetchBuildData(BuildDataProgress::Begin)).unwrap();

let progress = {
let sender = sender.clone();
move |msg| {
sender.send(Task::FetchBuildData(BuildDataProgress::Report(msg))).unwrap()
}
};
let res = ProjectWorkspace::run_all_build_scripts(&workspaces, &config, &progress);
let progress = {
let sender = sender.clone();
move |msg| {
sender.send(Task::FetchBuildData(BuildDataProgress::Report(msg))).unwrap()
}
};
let res = ProjectWorkspace::run_all_build_scripts(&workspaces, &config, &progress);

sender.send(Task::FetchBuildData(BuildDataProgress::End((workspaces, res)))).unwrap();
});
sender
.send(Task::FetchBuildData(BuildDataProgress::End((workspaces, res))))
.unwrap();
},
);
}

pub(crate) fn fetch_proc_macros(&mut self, cause: Cause, paths: Vec<ProcMacroPaths>) {
tracing::info!(%cause, "will load proc macros");
let dummy_replacements = self.config.dummy_replacements().clone();
let proc_macro_clients = self.proc_macro_clients.clone();

self.task_pool.handle.spawn_with_sender(ThreadIntent::Worker, move |sender| {
sender.send(Task::LoadProcMacros(ProcMacroProgress::Begin)).unwrap();
self.task_pool.handle.spawn_with_sender(
ThreadIntent::Worker,
"fetch_proc_macros",
move |sender| {
sender.send(Task::LoadProcMacros(ProcMacroProgress::Begin)).unwrap();

let dummy_replacements = &dummy_replacements;
let progress = {
let sender = sender.clone();
&move |msg| {
sender.send(Task::LoadProcMacros(ProcMacroProgress::Report(msg))).unwrap()
}
};

let dummy_replacements = &dummy_replacements;
let progress = {
let sender = sender.clone();
&move |msg| {
sender.send(Task::LoadProcMacros(ProcMacroProgress::Report(msg))).unwrap()
let mut res = FxHashMap::default();
let chain = proc_macro_clients
.iter()
.map(|res| res.as_ref().map_err(|e| e.to_string()))
.chain(iter::repeat_with(|| Err("Proc macros servers are not running".into())));
for (client, paths) in chain.zip(paths) {
res.extend(paths.into_iter().map(move |(crate_id, res)| {
(
crate_id,
res.map_or_else(
|_| Err("proc macro crate is missing dylib".to_owned()),
|(crate_name, path)| {
progress(path.display().to_string());
client.as_ref().map_err(Clone::clone).and_then(|client| {
load_proc_macro(
client,
&path,
crate_name
.as_deref()
.and_then(|crate_name| {
dummy_replacements.get(crate_name).map(|v| &**v)
})
.unwrap_or_default(),
)
})
},
),
)
}));
}
};

let mut res = FxHashMap::default();
let chain = proc_macro_clients
.iter()
.map(|res| res.as_ref().map_err(|e| e.to_string()))
.chain(iter::repeat_with(|| Err("Proc macros servers are not running".into())));
for (client, paths) in chain.zip(paths) {
res.extend(paths.into_iter().map(move |(crate_id, res)| {
(
crate_id,
res.map_or_else(
|_| Err("proc macro crate is missing dylib".to_owned()),
|(crate_name, path)| {
progress(path.display().to_string());
client.as_ref().map_err(Clone::clone).and_then(|client| {
load_proc_macro(
client,
&path,
crate_name
.as_deref()
.and_then(|crate_name| {
dummy_replacements.get(crate_name).map(|v| &**v)
})
.unwrap_or_default(),
)
})
},
),
)
}));
}

sender.send(Task::LoadProcMacros(ProcMacroProgress::End(res))).unwrap();
});
sender.send(Task::LoadProcMacros(ProcMacroProgress::End(res))).unwrap();
},
);
}

pub(crate) fn set_proc_macros(&mut self, proc_macros: ProcMacros) {
Expand Down
28 changes: 22 additions & 6 deletions crates/rust-analyzer/src/task_pool.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,25 +14,41 @@ impl<T> TaskPool<T> {
TaskPool { sender, pool: Pool::new(threads) }
}

pub(crate) fn spawn<F>(&mut self, intent: ThreadIntent, task: F)
where
pub(crate) fn spawn<F>(
&mut self,
intent: ThreadIntent,
panic_context: impl Into<String>,
task: F,
) where
F: FnOnce() -> T + Send + 'static,
T: Send + 'static,
{
let panic_context = panic_context.into();
self.pool.spawn(intent, {
let sender = self.sender.clone();
move || sender.send(task()).unwrap()
move || {
let _pctx = stdx::panic_context::enter(panic_context);
sender.send(task()).unwrap()
}
})
}

pub(crate) fn spawn_with_sender<F>(&mut self, intent: ThreadIntent, task: F)
where
pub(crate) fn spawn_with_sender<F>(
&mut self,
intent: ThreadIntent,
panic_context: impl Into<String>,
task: F,
) where
F: FnOnce(Sender<T>) + Send + 'static,
T: Send + 'static,
{
let panic_context = panic_context.into();
self.pool.spawn(intent, {
let sender = self.sender.clone();
move || task(sender)
move || {
let _pctx = stdx::panic_context::enter(panic_context);
task(sender)
}
})
}

Expand Down