Auto merge of #17842 - mo8it:crossbeam-channel, r=<try>

Optimize the usage of channel senders

Used `Sender` directly instead of a boxed closure. There is no need to use the boxed closure. This also allows the caller to decide to do something other than `unwrap` (not a fan of it BTW).

crates/ide-db/Cargo.toml

@@ -14,7 +14,7 @@ doctest = false
 
 [dependencies]
 cov-mark = "2.0.0-pre.1"
-crossbeam-channel = "0.5.5"
+crossbeam-channel.workspace = true
 tracing.workspace = true
 rayon.workspace = true
 fst = { version = "0.4.7", default-features = false }

crates/load-cargo/src/lib.rs

@@ -64,8 +64,7 @@ pub fn load_workspace(
     let (sender, receiver) = unbounded();
     let mut vfs = vfs::Vfs::default();
     let mut loader = {
-        let loader =
-            vfs_notify::NotifyHandle::spawn(Box::new(move |msg| sender.send(msg).unwrap()));
+        let loader = vfs_notify::NotifyHandle::spawn(sender);
         Box::new(loader)
     };
 

crates/rust-analyzer/Cargo.toml

@@ -21,7 +21,7 @@ path = "src/bin/main.rs"
 
 [dependencies]
 anyhow.workspace = true
-crossbeam-channel = "0.5.5"
+crossbeam-channel.workspace = true
 dirs = "5.0.1"
 dissimilar.workspace = true
 itertools.workspace = true
@@ -90,13 +90,13 @@ jemalloc = ["jemallocator", "profile/jemalloc"]
 force-always-assert = ["always-assert/force"]
 sysroot-abi = []
 in-rust-tree = [
-    "sysroot-abi",
-    "syntax/in-rust-tree",
-    "parser/in-rust-tree",
-    "hir/in-rust-tree",
-    "hir-def/in-rust-tree",
-    "hir-ty/in-rust-tree",
-    "load-cargo/in-rust-tree",
+  "sysroot-abi",
+  "syntax/in-rust-tree",
+  "parser/in-rust-tree",
+  "hir/in-rust-tree",
+  "hir-def/in-rust-tree",
+  "hir-ty/in-rust-tree",
+  "load-cargo/in-rust-tree",
 ]
 
 [lints]

crates/rust-analyzer/src/flycheck.rs

@@ -109,7 +109,7 @@ pub(crate) struct FlycheckHandle {
 impl FlycheckHandle {
     pub(crate) fn spawn(
         id: usize,
-        sender: Box<dyn Fn(FlycheckMessage) + Send>,
+        sender: Sender<FlycheckMessage>,
         config: FlycheckConfig,
         sysroot_root: Option<AbsPathBuf>,
         workspace_root: AbsPathBuf,
@@ -199,7 +199,7 @@ enum StateChange {
 struct FlycheckActor {
     /// The workspace id of this flycheck instance.
     id: usize,
-    sender: Box<dyn Fn(FlycheckMessage) + Send>,
+    sender: Sender<FlycheckMessage>,
     config: FlycheckConfig,
     manifest_path: Option<AbsPathBuf>,
     /// Either the workspace root of the workspace we are flychecking,
@@ -235,7 +235,7 @@ pub(crate) const SAVED_FILE_PLACEHOLDER: &str = "$saved_file";
 impl FlycheckActor {
     fn new(
         id: usize,
-        sender: Box<dyn Fn(FlycheckMessage) + Send>,
+        sender: Sender<FlycheckMessage>,
         config: FlycheckConfig,
         sysroot_root: Option<AbsPathBuf>,
         workspace_root: AbsPathBuf,
@@ -256,7 +256,7 @@ impl FlycheckActor {
     }
 
     fn report_progress(&self, progress: Progress) {
-        self.send(FlycheckMessage::Progress { id: self.id, progress });
+        self.sender.send(FlycheckMessage::Progress { id: self.id, progress }).unwrap();
     }
 
     fn next_event(&self, inbox: &Receiver<StateChange>) -> Option<Event> {
@@ -329,7 +329,9 @@ impl FlycheckActor {
                         );
                     }
                     if self.status == FlycheckStatus::Started {
-                        self.send(FlycheckMessage::ClearDiagnostics { id: self.id });
+                        self.sender
+                            .send(FlycheckMessage::ClearDiagnostics { id: self.id })
+                            .unwrap();
                     }
                     self.report_progress(Progress::DidFinish(res));
                     self.status = FlycheckStatus::Finished;
@@ -351,13 +353,17 @@ impl FlycheckActor {
                             "diagnostic received"
                         );
                         if self.status == FlycheckStatus::Started {
-                            self.send(FlycheckMessage::ClearDiagnostics { id: self.id });
+                            self.sender
+                                .send(FlycheckMessage::ClearDiagnostics { id: self.id })
+                                .unwrap();
                         }
-                        self.send(FlycheckMessage::AddDiagnostic {
-                            id: self.id,
-                            workspace_root: self.root.clone(),
-                            diagnostic: msg,
-                        });
+                        self.sender
+                            .send(FlycheckMessage::AddDiagnostic {
+                                id: self.id,
+                                workspace_root: self.root.clone(),
+                                diagnostic: msg,
+                            })
+                            .unwrap();
                         self.status = FlycheckStatus::DiagnosticSent;
                     }
                 },
@@ -477,10 +483,6 @@ impl FlycheckActor {
         cmd.args(args);
         Some(cmd)
     }
-
-    fn send(&self, check_task: FlycheckMessage) {
-        (self.sender)(check_task);
-    }
 }
 
 #[allow(clippy::large_enum_variant)]

crates/rust-analyzer/src/global_state.rs

@@ -185,8 +185,7 @@ impl GlobalState {
     pub(crate) fn new(sender: Sender<lsp_server::Message>, config: Config) -> GlobalState {
         let loader = {
             let (sender, receiver) = unbounded::<vfs::loader::Message>();
-            let handle: vfs_notify::NotifyHandle =
-                vfs::loader::Handle::spawn(Box::new(move |msg| sender.send(msg).unwrap()));
+            let handle: vfs_notify::NotifyHandle = vfs::loader::Handle::spawn(sender);
             let handle = Box::new(handle) as Box<dyn vfs::loader::Handle>;
             Handle { handle, receiver }
         };
@@ -505,7 +504,7 @@ impl GlobalState {
         handler: ReqHandler,
     ) {
         let request = self.req_queue.outgoing.register(R::METHOD.to_owned(), params, handler);
-        self.send(request.into());
+        self.sender.send(request.into()).unwrap();
     }
 
     pub(crate) fn complete_request(&mut self, response: lsp_server::Response) {
@@ -522,7 +521,7 @@ impl GlobalState {
         params: N::Params,
     ) {
         let not = lsp_server::Notification::new(N::METHOD.to_owned(), params);
-        self.send(not.into());
+        self.sender.send(not.into()).unwrap();
     }
 
     pub(crate) fn register_request(
@@ -545,24 +544,20 @@ impl GlobalState {
 
             let duration = start.elapsed();
             tracing::debug!("handled {} - ({}) in {:0.2?}", method, response.id, duration);
-            self.send(response.into());
+            self.sender.send(response.into()).unwrap();
         }
     }
 
     pub(crate) fn cancel(&mut self, request_id: lsp_server::RequestId) {
         if let Some(response) = self.req_queue.incoming.cancel(request_id) {
-            self.send(response.into());
+            self.sender.send(response.into()).unwrap();
         }
     }
 
     pub(crate) fn is_completed(&self, request: &lsp_server::Request) -> bool {
         self.req_queue.incoming.is_completed(&request.id)
     }
 
-    fn send(&self, message: lsp_server::Message) {
-        self.sender.send(message).unwrap()
-    }
-
     pub(crate) fn publish_diagnostics(
         &mut self,
         uri: Url,

crates/rust-analyzer/src/reload.rs

@@ -758,7 +758,7 @@ impl GlobalState {
         self.flycheck = match invocation_strategy {
             crate::flycheck::InvocationStrategy::Once => vec![FlycheckHandle::spawn(
                 0,
-                Box::new(move |msg| sender.send(msg).unwrap()),
+                sender,
                 config,
                 None,
                 self.config.root_path().clone(),
@@ -793,10 +793,9 @@ impl GlobalState {
                         ))
                     })
                     .map(|(id, (root, manifest_path), sysroot_root)| {
-                        let sender = sender.clone();
                         FlycheckHandle::spawn(
                             id,
-                            Box::new(move |msg| sender.send(msg).unwrap()),
+                            sender.clone(),
                             config.clone(),
                             sysroot_root,
                             root.to_path_buf(),

crates/stdx/Cargo.toml

@@ -17,7 +17,7 @@ backtrace = { version = "0.3.67", optional = true }
 always-assert = { version = "0.2.0", features = ["tracing"] }
 jod-thread = "0.1.2"
 libc.workspace = true
-crossbeam-channel = "0.5.5"
+crossbeam-channel.workspace = true
 itertools.workspace = true
 # Think twice before adding anything here
 

crates/vfs-notify/Cargo.toml

@@ -15,7 +15,7 @@ doctest = false
 [dependencies]
 tracing.workspace = true
 walkdir = "2.3.2"
-crossbeam-channel = "0.5.5"
+crossbeam-channel.workspace = true
 notify = "6.1.1"
 rayon = "1.10.0"
 

crates/vfs-notify/src/lib.rs

@@ -119,18 +119,19 @@ impl NotifyActor {
                         self.watched_dir_entries.clear();
                         self.watched_file_entries.clear();
 
-                        let send = |msg| (self.sender)(msg);
-                        send(loader::Message::Progress {
-                            n_total,
-                            n_done: LoadingProgress::Started,
-                            config_version,
-                            dir: None,
-                        });
+                        self.sender
+                            .send(loader::Message::Progress {
+                                n_total,
+                                n_done: LoadingProgress::Started,
+                                config_version,
+                                dir: None,
+                            })
+                            .unwrap();
 
                         let (entry_tx, entry_rx) = unbounded();
                         let (watch_tx, watch_rx) = unbounded();
                         let processed = AtomicUsize::new(0);
-                        config.load.into_par_iter().enumerate().for_each(move |(i, entry)| {
+                        config.load.into_par_iter().enumerate().for_each(|(i, entry)| {
                             let do_watch = config.watch.contains(&i);
                             if do_watch {
                                 _ = entry_tx.send(entry.clone());
@@ -140,25 +141,31 @@ impl NotifyActor {
                                 entry,
                                 do_watch,
                                 |file| {
-                                    send(loader::Message::Progress {
-                                        n_total,
-                                        n_done: LoadingProgress::Progress(
-                                            processed.load(std::sync::atomic::Ordering::Relaxed),
-                                        ),
-                                        dir: Some(file),
-                                        config_version,
-                                    })
+                                    self.sender
+                                        .send(loader::Message::Progress {
+                                            n_total,
+                                            n_done: LoadingProgress::Progress(
+                                                processed
+                                                    .load(std::sync::atomic::Ordering::Relaxed),
+                                            ),
+                                            dir: Some(file),
+                                            config_version,
+                                        })
+                                        .unwrap()
                                 },
                             );
-                            send(loader::Message::Loaded { files });
-                            send(loader::Message::Progress {
-                                n_total,
-                                n_done: LoadingProgress::Progress(
-                                    processed.fetch_add(1, std::sync::atomic::Ordering::AcqRel) + 1,
-                                ),
-                                config_version,
-                                dir: None,
-                            });
+                            self.sender.send(loader::Message::Loaded { files }).unwrap();
+                            self.sender
+                                .send(loader::Message::Progress {
+                                    n_total,
+                                    n_done: LoadingProgress::Progress(
+                                        processed.fetch_add(1, std::sync::atomic::Ordering::AcqRel)
+                                            + 1,
+                                    ),
+                                    config_version,
+                                    dir: None,
+                                })
+                                .unwrap();
                         });
                         for path in watch_rx {
                             self.watch(&path);
@@ -173,17 +180,19 @@ impl NotifyActor {
                                 }
                             }
                         }
-                        self.send(loader::Message::Progress {
-                            n_total,
-                            n_done: LoadingProgress::Finished,
-                            config_version,
-                            dir: None,
-                        });
+                        self.sender
+                            .send(loader::Message::Progress {
+                                n_total,
+                                n_done: LoadingProgress::Finished,
+                                config_version,
+                                dir: None,
+                            })
+                            .unwrap();
                     }
                     Message::Invalidate(path) => {
                         let contents = read(path.as_path());
                         let files = vec![(path, contents)];
-                        self.send(loader::Message::Changed { files });
+                        self.sender.send(loader::Message::Changed { files }).unwrap();
                     }
                 },
                 Event::NotifyEvent(event) => {
@@ -231,7 +240,7 @@ impl NotifyActor {
                                     Some((path, contents))
                                 })
                                 .collect();
-                            self.send(loader::Message::Changed { files });
+                            self.sender.send(loader::Message::Changed { files }).unwrap();
                         }
                     }
                 }
@@ -315,10 +324,6 @@ impl NotifyActor {
             log_notify_error(watcher.watch(path, RecursiveMode::NonRecursive));
         }
     }
-
-    fn send(&self, msg: loader::Message) {
-        (self.sender)(msg);
-    }
 }
 
 fn read(path: &AbsPath) -> Option<Vec<u8>> {

crates/vfs/Cargo.toml

@@ -18,6 +18,7 @@ tracing.workspace = true
 fst = "0.4.7"
 indexmap.workspace = true
 nohash-hasher.workspace = true
+crossbeam-channel.workspace = true
 
 paths.workspace = true
 stdx.workspace = true

crates/vfs/src/loader.rs

@@ -72,7 +72,7 @@ pub enum Message {
 }
 
 /// Type that will receive [`Messages`](Message) from a [`Handle`].
-pub type Sender = Box<dyn Fn(Message) + Send + Sync>;
+pub type Sender = crossbeam_channel::Sender<Message>;
 
 /// Interface for reading and watching files.
 pub trait Handle: fmt::Debug {

lib/lsp-server/Cargo.toml

@@ -10,11 +10,11 @@ edition = "2021"
 log = "0.4.17"
 serde_json = "1.0.108"
 serde = { version = "1.0.192", features = ["derive"] }
-crossbeam-channel = "0.5.8"
+crossbeam-channel.workspace = true
 
 [dev-dependencies]
 lsp-types = "=0.95"
 ctrlc = "3.4.1"
 
 [lints]
-workspace = true
+workspace = true