diff options
Diffstat (limited to 'src/tools/rust-analyzer/crates/stdx')
| -rw-r--r-- | src/tools/rust-analyzer/crates/stdx/Cargo.toml | 2 | ||||
| -rw-r--r-- | src/tools/rust-analyzer/crates/stdx/src/hash.rs | 80 | ||||
| -rw-r--r-- | src/tools/rust-analyzer/crates/stdx/src/lib.rs | 2 | ||||
| -rw-r--r-- | src/tools/rust-analyzer/crates/stdx/src/thread.rs | 102 | ||||
| -rw-r--r-- | src/tools/rust-analyzer/crates/stdx/src/thread/intent.rs | 287 | ||||
| -rw-r--r-- | src/tools/rust-analyzer/crates/stdx/src/thread/pool.rs | 92 |
6 files changed, 484 insertions, 81 deletions
diff --git a/src/tools/rust-analyzer/crates/stdx/Cargo.toml b/src/tools/rust-analyzer/crates/stdx/Cargo.toml index c881f2fd3f4..a67f36ae900 100644 --- a/src/tools/rust-analyzer/crates/stdx/Cargo.toml +++ b/src/tools/rust-analyzer/crates/stdx/Cargo.toml @@ -15,6 +15,8 @@ doctest = false libc = "0.2.135" backtrace = { version = "0.3.65", optional = true } always-assert = { version = "0.1.2", features = ["log"] } +jod-thread = "0.1.2" +crossbeam-channel = "0.5.5" # Think twice before adding anything here [target.'cfg(windows)'.dependencies] diff --git a/src/tools/rust-analyzer/crates/stdx/src/hash.rs b/src/tools/rust-analyzer/crates/stdx/src/hash.rs deleted file mode 100644 index 0c21d2674b1..00000000000 --- a/src/tools/rust-analyzer/crates/stdx/src/hash.rs +++ /dev/null @@ -1,80 +0,0 @@ -//! A none hashing [`Hasher`] implementation. -use std::{ - hash::{BuildHasher, Hasher}, - marker::PhantomData, -}; - -pub type NoHashHashMap<K, V> = std::collections::HashMap<K, V, NoHashHasherBuilder<K>>; -pub type NoHashHashSet<K> = std::collections::HashSet<K, NoHashHasherBuilder<K>>; - -#[derive(Copy, Clone, Debug, PartialEq, Eq)] -pub struct NoHashHasherBuilder<T>(PhantomData<T>); - -impl<T> Default for NoHashHasherBuilder<T> { - fn default() -> Self { - Self(Default::default()) - } -} - -pub trait NoHashHashable {} -impl NoHashHashable for usize {} -impl NoHashHashable for u32 {} - -pub struct NoHashHasher(u64); - -impl<T: NoHashHashable> BuildHasher for NoHashHasherBuilder<T> { - type Hasher = NoHashHasher; - fn build_hasher(&self) -> Self::Hasher { - NoHashHasher(0) - } -} - -impl Hasher for NoHashHasher { - fn finish(&self) -> u64 { - self.0 - } - - fn write(&mut self, _: &[u8]) { - unimplemented!("NoHashHasher should only be used for hashing primitive integers") - } - - fn write_u8(&mut self, i: u8) { - self.0 = i as u64; - } - - fn write_u16(&mut self, i: u16) { - self.0 = i as u64; - } - - fn write_u32(&mut self, i: u32) { - self.0 = i as u64; - } - - fn write_u64(&mut self, i: u64) { - self.0 = i; - } - - fn write_usize(&mut self, i: usize) { - self.0 = i as u64; - } - - fn write_i8(&mut self, i: i8) { - self.0 = i as u64; - } - - fn write_i16(&mut self, i: i16) { - self.0 = i as u64; - } - - fn write_i32(&mut self, i: i32) { - self.0 = i as u64; - } - - fn write_i64(&mut self, i: i64) { - self.0 = i as u64; - } - - fn write_isize(&mut self, i: isize) { - self.0 = i as u64; - } -} diff --git a/src/tools/rust-analyzer/crates/stdx/src/lib.rs b/src/tools/rust-analyzer/crates/stdx/src/lib.rs index 5639aaf57cd..24990d6a0e7 100644 --- a/src/tools/rust-analyzer/crates/stdx/src/lib.rs +++ b/src/tools/rust-analyzer/crates/stdx/src/lib.rs @@ -7,11 +7,11 @@ use std::process::Command; use std::{cmp::Ordering, ops, time::Instant}; mod macros; -pub mod hash; pub mod process; pub mod panic_context; pub mod non_empty_vec; pub mod rand; +pub mod thread; pub use always_assert::{always, never}; diff --git a/src/tools/rust-analyzer/crates/stdx/src/thread.rs b/src/tools/rust-analyzer/crates/stdx/src/thread.rs new file mode 100644 index 00000000000..e577eb43137 --- /dev/null +++ b/src/tools/rust-analyzer/crates/stdx/src/thread.rs @@ -0,0 +1,102 @@ +//! A utility module for working with threads that automatically joins threads upon drop +//! and abstracts over operating system quality of service (QoS) APIs +//! through the concept of a “thread intent”. +//! +//! The intent of a thread is frozen at thread creation time, +//! i.e. there is no API to change the intent of a thread once it has been spawned. +//! +//! As a system, rust-analyzer should have the property that +//! old manual scheduling APIs are replaced entirely by QoS. +//! To maintain this invariant, we panic when it is clear that +//! old scheduling APIs have been used. +//! +//! Moreover, we also want to ensure that every thread has an intent set explicitly +//! to force a decision about its importance to the system. +//! Thus, [`ThreadIntent`] has no default value +//! and every entry point to creating a thread requires a [`ThreadIntent`] upfront. + +use std::fmt; + +mod intent; +mod pool; + +pub use intent::ThreadIntent; +pub use pool::Pool; + +pub fn spawn<F, T>(intent: ThreadIntent, f: F) -> JoinHandle<T> +where + F: FnOnce() -> T, + F: Send + 'static, + T: Send + 'static, +{ + Builder::new(intent).spawn(f).expect("failed to spawn thread") +} + +pub struct Builder { + intent: ThreadIntent, + inner: jod_thread::Builder, + allow_leak: bool, +} + +impl Builder { + pub fn new(intent: ThreadIntent) -> Builder { + Builder { intent, inner: jod_thread::Builder::new(), allow_leak: false } + } + + pub fn name(self, name: String) -> Builder { + Builder { inner: self.inner.name(name), ..self } + } + + pub fn stack_size(self, size: usize) -> Builder { + Builder { inner: self.inner.stack_size(size), ..self } + } + + pub fn allow_leak(self, b: bool) -> Builder { + Builder { allow_leak: b, ..self } + } + + pub fn spawn<F, T>(self, f: F) -> std::io::Result<JoinHandle<T>> + where + F: FnOnce() -> T, + F: Send + 'static, + T: Send + 'static, + { + let inner_handle = self.inner.spawn(move || { + self.intent.apply_to_current_thread(); + f() + })?; + + Ok(JoinHandle { inner: Some(inner_handle), allow_leak: self.allow_leak }) + } +} + +pub struct JoinHandle<T = ()> { + // `inner` is an `Option` so that we can + // take ownership of the contained `JoinHandle`. + inner: Option<jod_thread::JoinHandle<T>>, + allow_leak: bool, +} + +impl<T> JoinHandle<T> { + pub fn join(mut self) -> T { + self.inner.take().unwrap().join() + } +} + +impl<T> Drop for JoinHandle<T> { + fn drop(&mut self) { + if !self.allow_leak { + return; + } + + if let Some(join_handle) = self.inner.take() { + join_handle.detach(); + } + } +} + +impl<T> fmt::Debug for JoinHandle<T> { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.pad("JoinHandle { .. }") + } +} diff --git a/src/tools/rust-analyzer/crates/stdx/src/thread/intent.rs b/src/tools/rust-analyzer/crates/stdx/src/thread/intent.rs new file mode 100644 index 00000000000..7b65db30cc5 --- /dev/null +++ b/src/tools/rust-analyzer/crates/stdx/src/thread/intent.rs @@ -0,0 +1,287 @@ +//! An opaque façade around platform-specific QoS APIs. + +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)] +// Please maintain order from least to most priority for the derived `Ord` impl. +pub enum ThreadIntent { + /// Any thread which does work that isn’t in the critical path of the user typing + /// (e.g. processing Go To Definition). + Worker, + + /// Any thread which does work caused by the user typing + /// (e.g. processing syntax highlighting). + LatencySensitive, +} + +impl ThreadIntent { + // These APIs must remain private; + // we only want consumers to set thread intent + // either during thread creation or using our pool impl. + + pub(super) fn apply_to_current_thread(self) { + let class = thread_intent_to_qos_class(self); + set_current_thread_qos_class(class); + } + + pub(super) fn assert_is_used_on_current_thread(self) { + if IS_QOS_AVAILABLE { + let class = thread_intent_to_qos_class(self); + assert_eq!(get_current_thread_qos_class(), Some(class)); + } + } +} + +use imp::QoSClass; + +const IS_QOS_AVAILABLE: bool = imp::IS_QOS_AVAILABLE; + +fn set_current_thread_qos_class(class: QoSClass) { + imp::set_current_thread_qos_class(class) +} + +fn get_current_thread_qos_class() -> Option<QoSClass> { + imp::get_current_thread_qos_class() +} + +fn thread_intent_to_qos_class(intent: ThreadIntent) -> QoSClass { + imp::thread_intent_to_qos_class(intent) +} + +// All Apple platforms use XNU as their kernel +// and thus have the concept of QoS. +#[cfg(target_vendor = "apple")] +mod imp { + use super::ThreadIntent; + + #[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)] + // Please maintain order from least to most priority for the derived `Ord` impl. + pub(super) enum QoSClass { + // Documentation adapted from https://github.com/apple-oss-distributions/libpthread/blob/67e155c94093be9a204b69637d198eceff2c7c46/include/sys/qos.h#L55 + // + /// TLDR: invisible maintenance tasks + /// + /// Contract: + /// + /// * **You do not care about how long it takes for work to finish.** + /// * **You do not care about work being deferred temporarily.** + /// (e.g. if the device’s battery is in a critical state) + /// + /// Examples: + /// + /// * in a video editor: + /// creating periodic backups of project files + /// * in a browser: + /// cleaning up cached sites which have not been accessed in a long time + /// * in a collaborative word processor: + /// creating a searchable index of all documents + /// + /// Use this QoS class for background tasks + /// which the user did not initiate themselves + /// and which are invisible to the user. + /// It is expected that this work will take significant time to complete: + /// minutes or even hours. + /// + /// This QoS class provides the most energy and thermally-efficient execution possible. + /// All other work is prioritized over background tasks. + Background, + + /// TLDR: tasks that don’t block using your app + /// + /// Contract: + /// + /// * **Your app remains useful even as the task is executing.** + /// + /// Examples: + /// + /// * in a video editor: + /// exporting a video to disk – + /// the user can still work on the timeline + /// * in a browser: + /// automatically extracting a downloaded zip file – + /// the user can still switch tabs + /// * in a collaborative word processor: + /// downloading images embedded in a document – + /// the user can still make edits + /// + /// Use this QoS class for tasks which + /// may or may not be initiated by the user, + /// but whose result is visible. + /// It is expected that this work will take a few seconds to a few minutes. + /// Typically your app will include a progress bar + /// for tasks using this class. + /// + /// This QoS class provides a balance between + /// performance, responsiveness and efficiency. + Utility, + + /// TLDR: tasks that block using your app + /// + /// Contract: + /// + /// * **You need this work to complete + /// before the user can keep interacting with your app.** + /// * **Your work will not take more than a few seconds to complete.** + /// + /// Examples: + /// + /// * in a video editor: + /// opening a saved project + /// * in a browser: + /// loading a list of the user’s bookmarks and top sites + /// when a new tab is created + /// * in a collaborative word processor: + /// running a search on the document’s content + /// + /// Use this QoS class for tasks which were initiated by the user + /// and block the usage of your app while they are in progress. + /// It is expected that this work will take a few seconds or less to complete; + /// not long enough to cause the user to switch to something else. + /// Your app will likely indicate progress on these tasks + /// through the display of placeholder content or modals. + /// + /// This QoS class is not energy-efficient. + /// Rather, it provides responsiveness + /// by prioritizing work above other tasks on the system + /// except for critical user-interactive work. + UserInitiated, + + /// TLDR: render loops and nothing else + /// + /// Contract: + /// + /// * **You absolutely need this work to complete immediately + /// or your app will appear to freeze.** + /// * **Your work will always complete virtually instantaneously.** + /// + /// Examples: + /// + /// * the main thread in a GUI application + /// * the update & render loop in a game + /// * a secondary thread which progresses an animation + /// + /// Use this QoS class for any work which, if delayed, + /// will make your user interface unresponsive. + /// It is expected that this work will be virtually instantaneous. + /// + /// This QoS class is not energy-efficient. + /// Specifying this class is a request to run with + /// nearly all available system CPU and I/O bandwidth even under contention. + UserInteractive, + } + + pub(super) const IS_QOS_AVAILABLE: bool = true; + + pub(super) fn set_current_thread_qos_class(class: QoSClass) { + let c = match class { + QoSClass::UserInteractive => libc::qos_class_t::QOS_CLASS_USER_INTERACTIVE, + QoSClass::UserInitiated => libc::qos_class_t::QOS_CLASS_USER_INITIATED, + QoSClass::Utility => libc::qos_class_t::QOS_CLASS_UTILITY, + QoSClass::Background => libc::qos_class_t::QOS_CLASS_BACKGROUND, + }; + + let code = unsafe { libc::pthread_set_qos_class_self_np(c, 0) }; + + if code == 0 { + return; + } + + let errno = unsafe { *libc::__error() }; + + match errno { + libc::EPERM => { + // This thread has been excluded from the QoS system + // due to a previous call to a function such as `pthread_setschedparam` + // which is incompatible with QoS. + // + // Panic instead of returning an error + // to maintain the invariant that we only use QoS APIs. + panic!("tried to set QoS of thread which has opted out of QoS (os error {errno})") + } + + libc::EINVAL => { + // This is returned if we pass something other than a qos_class_t + // to `pthread_set_qos_class_self_np`. + // + // This is impossible, so again panic. + unreachable!( + "invalid qos_class_t value was passed to pthread_set_qos_class_self_np" + ) + } + + _ => { + // `pthread_set_qos_class_self_np`’s documentation + // does not mention any other errors. + unreachable!("`pthread_set_qos_class_self_np` returned unexpected error {errno}") + } + } + } + + pub(super) fn get_current_thread_qos_class() -> Option<QoSClass> { + let current_thread = unsafe { libc::pthread_self() }; + let mut qos_class_raw = libc::qos_class_t::QOS_CLASS_UNSPECIFIED; + let code = unsafe { + libc::pthread_get_qos_class_np(current_thread, &mut qos_class_raw, std::ptr::null_mut()) + }; + + if code != 0 { + // `pthread_get_qos_class_np`’s documentation states that + // an error value is placed into errno if the return code is not zero. + // However, it never states what errors are possible. + // Inspecting the source[0] shows that, as of this writing, it always returns zero. + // + // Whatever errors the function could report in future are likely to be + // ones which we cannot handle anyway + // + // 0: https://github.com/apple-oss-distributions/libpthread/blob/67e155c94093be9a204b69637d198eceff2c7c46/src/qos.c#L171-L177 + let errno = unsafe { *libc::__error() }; + unreachable!("`pthread_get_qos_class_np` failed unexpectedly (os error {errno})"); + } + + match qos_class_raw { + libc::qos_class_t::QOS_CLASS_USER_INTERACTIVE => Some(QoSClass::UserInteractive), + libc::qos_class_t::QOS_CLASS_USER_INITIATED => Some(QoSClass::UserInitiated), + libc::qos_class_t::QOS_CLASS_DEFAULT => None, // QoS has never been set + libc::qos_class_t::QOS_CLASS_UTILITY => Some(QoSClass::Utility), + libc::qos_class_t::QOS_CLASS_BACKGROUND => Some(QoSClass::Background), + + libc::qos_class_t::QOS_CLASS_UNSPECIFIED => { + // Using manual scheduling APIs causes threads to “opt out” of QoS. + // At this point they become incompatible with QoS, + // and as such have the “unspecified” QoS class. + // + // Panic instead of returning an error + // to maintain the invariant that we only use QoS APIs. + panic!("tried to get QoS of thread which has opted out of QoS") + } + } + } + + pub(super) fn thread_intent_to_qos_class(intent: ThreadIntent) -> QoSClass { + match intent { + ThreadIntent::Worker => QoSClass::Utility, + ThreadIntent::LatencySensitive => QoSClass::UserInitiated, + } + } +} + +// FIXME: Windows has QoS APIs, we should use them! +#[cfg(not(target_vendor = "apple"))] +mod imp { + use super::ThreadIntent; + + #[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)] + pub(super) enum QoSClass { + Default, + } + + pub(super) const IS_QOS_AVAILABLE: bool = false; + + pub(super) fn set_current_thread_qos_class(_: QoSClass) {} + + pub(super) fn get_current_thread_qos_class() -> Option<QoSClass> { + None + } + + pub(super) fn thread_intent_to_qos_class(_: ThreadIntent) -> QoSClass { + QoSClass::Default + } +} diff --git a/src/tools/rust-analyzer/crates/stdx/src/thread/pool.rs b/src/tools/rust-analyzer/crates/stdx/src/thread/pool.rs new file mode 100644 index 00000000000..2ddd7da74c2 --- /dev/null +++ b/src/tools/rust-analyzer/crates/stdx/src/thread/pool.rs @@ -0,0 +1,92 @@ +//! [`Pool`] implements a basic custom thread pool +//! inspired by the [`threadpool` crate](http://docs.rs/threadpool). +//! When you spawn a task you specify a thread intent +//! so the pool can schedule it to run on a thread with that intent. +//! rust-analyzer uses this to prioritize work based on latency requirements. +//! +//! The thread pool is implemented entirely using +//! the threading utilities in [`crate::thread`]. + +use std::sync::{ + atomic::{AtomicUsize, Ordering}, + Arc, +}; + +use crossbeam_channel::{Receiver, Sender}; + +use super::{Builder, JoinHandle, ThreadIntent}; + +pub struct Pool { + // `_handles` is never read: the field is present + // only for its `Drop` impl. + + // The worker threads exit once the channel closes; + // make sure to keep `job_sender` above `handles` + // so that the channel is actually closed + // before we join the worker threads! + job_sender: Sender<Job>, + _handles: Vec<JoinHandle>, + extant_tasks: Arc<AtomicUsize>, +} + +struct Job { + requested_intent: ThreadIntent, + f: Box<dyn FnOnce() + Send + 'static>, +} + +impl Pool { + pub fn new(threads: usize) -> Pool { + const STACK_SIZE: usize = 8 * 1024 * 1024; + const INITIAL_INTENT: ThreadIntent = ThreadIntent::Worker; + + let (job_sender, job_receiver) = crossbeam_channel::unbounded(); + let extant_tasks = Arc::new(AtomicUsize::new(0)); + + let mut handles = Vec::with_capacity(threads); + for _ in 0..threads { + let handle = Builder::new(INITIAL_INTENT) + .stack_size(STACK_SIZE) + .name("Worker".into()) + .spawn({ + let extant_tasks = Arc::clone(&extant_tasks); + let job_receiver: Receiver<Job> = job_receiver.clone(); + move || { + let mut current_intent = INITIAL_INTENT; + for job in job_receiver { + if job.requested_intent != current_intent { + job.requested_intent.apply_to_current_thread(); + current_intent = job.requested_intent; + } + extant_tasks.fetch_add(1, Ordering::SeqCst); + (job.f)(); + extant_tasks.fetch_sub(1, Ordering::SeqCst); + } + } + }) + .expect("failed to spawn thread"); + + handles.push(handle); + } + + Pool { _handles: handles, extant_tasks, job_sender } + } + + pub fn spawn<F>(&self, intent: ThreadIntent, f: F) + where + F: FnOnce() + Send + 'static, + { + let f = Box::new(move || { + if cfg!(debug_assertions) { + intent.assert_is_used_on_current_thread(); + } + f() + }); + + let job = Job { requested_intent: intent, f }; + self.job_sender.send(job).unwrap(); + } + + pub fn len(&self) -> usize { + self.extant_tasks.load(Ordering::SeqCst) + } +} |