Add proof-of-concept QoS implementation

Author: lunacookies

Cargo.lock

@@ -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/Cargo.toml

@@ -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/flycheck/src/lib.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/ide/src/prime_caches.rs

@@ -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/proc-macro-srv/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/Cargo.toml

@@ -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/bin/main.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/src/task_pool.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/rust-analyzer/tests/slow-tests/support.rs

@@ -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]