Add proof-of-concept QoS implementation

Cargo.lock

@@ -408,7 +408,6 @@ dependencies = [
  "cargo_metadata",
  "command-group",
  "crossbeam-channel",
- "jod-thread",
  "paths",
  "rustc-hash",
  "serde",
@@ -1278,6 +1277,7 @@ dependencies = [
  "paths",
  "proc-macro-api",
  "proc-macro-test",
+ "stdx",
  "tt",
 ]
 
@@ -1493,7 +1493,6 @@ dependencies = [
  "ide-db",
  "ide-ssr",
  "itertools",
- "jod-thread",
  "lsp-server",
  "lsp-types",
  "mbe",
@@ -1712,6 +1711,7 @@ version = "0.0.0"
 dependencies = [
  "always-assert",
  "backtrace",
+ "jod-thread",
  "libc",
  "miow",
  "winapi",
@@ -2123,9 +2123,9 @@ name = "vfs-notify"
 version = "0.0.0"
 dependencies = [
  "crossbeam-channel",
- "jod-thread",
  "notify",
  "paths",
+ "stdx",
  "tracing",
  "vfs",
  "walkdir",

crates/flycheck/Cargo.toml

@@ -18,7 +18,6 @@ cargo_metadata = "0.15.0"
 rustc-hash = "1.1.0"
 serde_json.workspace = true
 serde.workspace = true
-jod-thread = "0.1.2"
 command-group = "2.0.1"
 
 # local deps

crates/flycheck/src/lib.rs

@@ -77,7 +77,7 @@ impl fmt::Display for FlycheckConfig {
 pub struct FlycheckHandle {
     // XXX: drop order is significant
     sender: Sender<StateChange>,
-    _thread: jod_thread::JoinHandle,
+    _thread: stdx::thread::JoinHandle,
     id: usize,
 }
 
@@ -90,7 +90,7 @@ impl FlycheckHandle {
     ) -> FlycheckHandle {
         let actor = FlycheckActor::new(id, sender, config, workspace_root);
         let (sender, receiver) = unbounded::<StateChange>();
-        let thread = jod_thread::Builder::new()
+        let thread = stdx::thread::Builder::new(stdx::thread::QoSClass::Utility)
             .name("Flycheck".to_owned())
             .spawn(move || actor.run(receiver))
             .expect("failed to spawn thread");
@@ -395,7 +395,7 @@ struct CargoHandle {
     /// The handle to the actual cargo process. As we cannot cancel directly from with
     /// a read syscall dropping and therefore terminating the process is our best option.
     child: JodGroupChild,
-    thread: jod_thread::JoinHandle<io::Result<(bool, String)>>,
+    thread: stdx::thread::JoinHandle<io::Result<(bool, String)>>,
     receiver: Receiver<CargoMessage>,
 }
 
@@ -409,7 +409,7 @@ impl CargoHandle {
 
         let (sender, receiver) = unbounded();
         let actor = CargoActor::new(sender, stdout, stderr);
-        let thread = jod_thread::Builder::new()
+        let thread = stdx::thread::Builder::new(stdx::thread::QoSClass::Utility)
             .name("CargoHandle".to_owned())
             .spawn(move || actor.run())
             .expect("failed to spawn thread");

crates/ide/src/prime_caches.rs

@@ -80,7 +80,11 @@ pub(crate) fn parallel_prime_caches(
         for _ in 0..num_worker_threads {
             let worker = prime_caches_worker.clone();
             let db = db.snapshot();
-            std::thread::spawn(move || Cancelled::catch(|| worker(db)));
+
+            stdx::thread::Builder::new(stdx::thread::QoSClass::Utility)
+                .allow_leak(true)
+                .spawn(move || Cancelled::catch(|| worker(db)))
+                .expect("failed to spawn thread");
         }
 
         (work_sender, progress_receiver)

crates/proc-macro-srv/Cargo.toml

@@ -22,6 +22,7 @@ object = { version = "0.30.2", default-features = false, features = [
 libloading = "0.7.3"
 memmap2 = "0.5.4"
 
+stdx.workspace = true
 tt.workspace = true
 mbe.workspace = true
 paths.workspace = true

crates/rust-analyzer/Cargo.toml

@@ -86,7 +86,6 @@ jemallocator = { version = "0.5.0", package = "tikv-jemallocator", optional = tr
 
 [dev-dependencies]
 expect-test = "1.4.0"
-jod-thread = "0.1.2"
 xshell = "0.2.2"
 
 test-utils.workspace = true

crates/rust-analyzer/src/bin/main.rs

@@ -78,7 +78,7 @@ fn try_main(flags: flags::RustAnalyzer) -> Result<()> {
                 println!("rust-analyzer {}", rust_analyzer::version());
                 return Ok(());
             }
-            with_extra_thread("LspServer", run_server)?;
+            with_extra_thread("LspServer", stdx::thread::QoSClass::Utility, run_server)?;
         }
         flags::RustAnalyzerCmd::Parse(cmd) => cmd.run()?,
         flags::RustAnalyzerCmd::Symbols(cmd) => cmd.run()?,
@@ -136,14 +136,17 @@ const STACK_SIZE: usize = 1024 * 1024 * 8;
 /// space.
 fn with_extra_thread(
     thread_name: impl Into<String>,
+    qos_class: stdx::thread::QoSClass,
     f: impl FnOnce() -> Result<()> + Send + 'static,
 ) -> Result<()> {
-    let handle =
-        std::thread::Builder::new().name(thread_name.into()).stack_size(STACK_SIZE).spawn(f)?;
-    match handle.join() {
-        Ok(res) => res,
-        Err(panic) => std::panic::resume_unwind(panic),
-    }
+    let handle = stdx::thread::Builder::new(qos_class)
+        .name(thread_name.into())
+        .stack_size(STACK_SIZE)
+        .spawn(f)?;
+
+    handle.join()?;
+
+    Ok(())
 }
 
 fn run_server() -> Result<()> {

crates/rust-analyzer/src/task_pool.rs

@@ -1,5 +1,7 @@
 //! A thin wrapper around `ThreadPool` to make sure that we join all things
 //! properly.
+use std::sync::{Arc, Barrier};
+
 use crossbeam_channel::Sender;
 
 pub(crate) struct TaskPool<T> {
@@ -16,6 +18,18 @@ impl<T> TaskPool<T> {
             .thread_stack_size(STACK_SIZE)
             .num_threads(threads)
             .build();
+
+        // Set QoS of all threads in threadpool.
+        let barrier = Arc::new(Barrier::new(threads + 1));
+        for _ in 0..threads {
+            let barrier = barrier.clone();
+            inner.execute(move || {
+                stdx::thread::set_current_thread_qos_class(stdx::thread::QoSClass::Utility);
+                barrier.wait();
+            });
+        }
+        barrier.wait();
+
         TaskPool { sender, inner }
     }
 
@@ -26,7 +40,16 @@ impl<T> TaskPool<T> {
     {
         self.inner.execute({
             let sender = self.sender.clone();
-            move || sender.send(task()).unwrap()
+            move || {
+                if stdx::thread::IS_QOS_AVAILABLE {
+                    debug_assert_eq!(
+                        stdx::thread::get_current_thread_qos_class(),
+                        Some(stdx::thread::QoSClass::Utility)
+                    );
+                }
+
+                sender.send(task()).unwrap()
+            }
         })
     }
 

crates/rust-analyzer/tests/slow-tests/support.rs

@@ -155,7 +155,7 @@ pub(crate) fn project(fixture: &str) -> Server {
 pub(crate) struct Server {
     req_id: Cell<i32>,
     messages: RefCell<Vec<Message>>,
-    _thread: jod_thread::JoinHandle<()>,
+    _thread: stdx::thread::JoinHandle,
     client: Connection,
     /// XXX: remove the tempdir last
     dir: TestDir,
@@ -165,7 +165,7 @@ impl Server {
     fn new(dir: TestDir, config: Config) -> Server {
         let (connection, client) = Connection::memory();
 
-        let _thread = jod_thread::Builder::new()
+        let _thread = stdx::thread::Builder::new(stdx::thread::QoSClass::Utility)
             .name("test server".to_string())
             .spawn(move || main_loop(config, connection).unwrap())
             .expect("failed to spawn a thread");

crates/stdx/Cargo.toml

@@ -15,6 +15,7 @@ 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"
 # Think twice before adding anything here
 
 [target.'cfg(windows)'.dependencies]

crates/stdx/src/lib.rs

@@ -11,6 +11,7 @@ pub mod process;
 pub mod panic_context;
 pub mod non_empty_vec;
 pub mod rand;
+pub mod thread;
 
 pub use always_assert::{always, never};
 

crates/stdx/src/thread.rs

@@ -0,0 +1,200 @@
+//! A utility module for working with threads that automatically joins threads upon drop
+//! and provides functionality for interfacing with operating system quality of service (QoS) APIs.
+//!
+//! 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 a QoS set explicitly
+//! to force a decision about its importance to the system.
+//! Thus, [`QoSClass`] has no default value
+//! and every entry point to creating a thread requires a [`QoSClass`] upfront.
+
+use std::fmt;
+
+pub fn spawn<F, T>(qos_class: QoSClass, f: F) -> JoinHandle<T>
+where
+    F: FnOnce() -> T,
+    F: Send + 'static,
+    T: Send + 'static,
+{
+    Builder::new(qos_class).spawn(f).expect("failed to spawn thread")
+}
+
+pub struct Builder {
+    qos_class: QoSClass,
+    inner: jod_thread::Builder,
+    allow_leak: bool,
+}
+
+impl Builder {
+    pub fn new(qos_class: QoSClass) -> Builder {
+        Builder { qos_class, 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 || {
+            set_current_thread_qos_class(self.qos_class);
+            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`
+    // in the `Drop` impl below.
+    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 { .. }")
+    }
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
+pub enum QoSClass {
+    // Maintain order in priority from least to most.
+    Background,
+    Utility,
+    UserInitiated,
+    UserInteractive,
+}
+
+#[cfg(target_vendor = "apple")]
+pub const IS_QOS_AVAILABLE: bool = true;
+
+#[cfg(not(target_vendor = "apple"))]
+pub const IS_QOS_AVAILABLE: bool = false;
+
+// All Apple platforms use XNU as their kernel
+// and thus have the concept of QoS.
+#[cfg(target_vendor = "apple")]
+pub 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.
+            //
+            // Let’s just panic here because rust-analyzer as a system
+            // should have the property that QoS is used consistently
+            // instead of old manual scheduling management 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}")
+        }
+    }
+}
+
+#[cfg(not(target_vendor = "apple"))]
+pub fn set_current_thread_qos_class(class: QoSClass) {
+    // FIXME: Windows has QoS APIs, we should use them!
+}
+
+#[cfg(target_vendor = "apple")]
+pub 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 => {
+            // We panic here because rust-analyzer should never use
+            panic!("tried to get QoS of thread which has opted out of QoS")
+        }
+    }
+}
+
+#[cfg(not(target_vendor = "apple"))]
+pub fn get_current_thread_qos_class() -> Option<QoSClass> {
+    None
+}

crates/vfs-notify/Cargo.toml

@@ -13,10 +13,10 @@ doctest = false
 
 [dependencies]
 tracing = "0.1.35"
-jod-thread = "0.1.2"
 walkdir = "2.3.2"
 crossbeam-channel = "0.5.5"
 notify = "5.0"
 
+stdx.workspace = true
 vfs.workspace = true
 paths.workspace = true

crates/vfs-notify/src/lib.rs

@@ -21,7 +21,7 @@ use walkdir::WalkDir;
 pub struct NotifyHandle {
     // Relative order of fields below is significant.
     sender: Sender<Message>,
-    _thread: jod_thread::JoinHandle,
+    _thread: stdx::thread::JoinHandle,
 }
 
 #[derive(Debug)]
@@ -34,7 +34,7 @@ impl loader::Handle for NotifyHandle {
     fn spawn(sender: loader::Sender) -> NotifyHandle {
         let actor = NotifyActor::new(sender);
         let (sender, receiver) = unbounded::<Message>();
-        let thread = jod_thread::Builder::new()
+        let thread = stdx::thread::Builder::new(stdx::thread::QoSClass::Utility)
             .name("VfsLoader".to_owned())
             .spawn(move || actor.run(receiver))
             .expect("failed to spawn thread");