Merge #5048

5048: Unify code style for worker threads r=matklad a=matklad

bors r+
🤖

Co-authored-by: Aleksey Kladov <aleksey.kladov@gmail.com>

crates/ra_flycheck/src/lib.rs

@@ -48,21 +48,23 @@ impl fmt::Display for FlycheckConfig {
 /// diagnostics based on the output.
 /// The spawned thread is shut down when this struct is dropped.
 #[derive(Debug)]
-pub struct Flycheck {
+pub struct FlycheckHandle {
     // XXX: drop order is significant
     cmd_send: Sender<CheckCommand>,
     handle: jod_thread::JoinHandle<()>,
-    pub task_recv: Receiver<CheckTask>,
 }
 
-impl Flycheck {
-    pub fn new(config: FlycheckConfig, workspace_root: PathBuf) -> Flycheck {
-        let (task_send, task_recv) = unbounded::<CheckTask>();
+impl FlycheckHandle {
+    pub fn spawn(
+        sender: Box<dyn Fn(CheckTask) + Send>,
+        config: FlycheckConfig,
+        workspace_root: PathBuf,
+    ) -> FlycheckHandle {
         let (cmd_send, cmd_recv) = unbounded::<CheckCommand>();
         let handle = jod_thread::spawn(move || {
-            FlycheckThread::new(config, workspace_root).run(&task_send, &cmd_recv);
+            FlycheckActor::new(sender, config, workspace_root).run(&cmd_recv);
         });
-        Flycheck { task_recv, cmd_send, handle }
+        FlycheckHandle { cmd_send, handle }
     }
 
     /// Schedule a re-start of the cargo check worker.
@@ -95,7 +97,8 @@ pub enum CheckCommand {
     Update,
 }
 
-struct FlycheckThread {
+struct FlycheckActor {
+    sender: Box<dyn Fn(CheckTask) + Send>,
     config: FlycheckConfig,
     workspace_root: PathBuf,
     last_update_req: Option<Instant>,
@@ -109,9 +112,14 @@ struct FlycheckThread {
     check_process: Option<jod_thread::JoinHandle<()>>,
 }
 
-impl FlycheckThread {
-    fn new(config: FlycheckConfig, workspace_root: PathBuf) -> FlycheckThread {
-        FlycheckThread {
+impl FlycheckActor {
+    fn new(
+        sender: Box<dyn Fn(CheckTask) + Send>,
+        config: FlycheckConfig,
+        workspace_root: PathBuf,
+    ) -> FlycheckActor {
+        FlycheckActor {
+            sender,
             config,
             workspace_root,
             last_update_req: None,
@@ -120,9 +128,9 @@ impl FlycheckThread {
         }
     }
 
-    fn run(&mut self, task_send: &Sender<CheckTask>, cmd_recv: &Receiver<CheckCommand>) {
+    fn run(&mut self, cmd_recv: &Receiver<CheckCommand>) {
         // If we rerun the thread, we need to discard the previous check results first
-        self.clean_previous_results(task_send);
+        self.clean_previous_results();
 
         loop {
             select! {
@@ -134,7 +142,7 @@ impl FlycheckThread {
                     },
                 },
                 recv(self.message_recv) -> msg => match msg {
-                    Ok(msg) => self.handle_message(msg, task_send),
+                    Ok(msg) => self.handle_message(msg),
                     Err(RecvError) => {
                         // Watcher finished, replace it with a never channel to
                         // avoid busy-waiting.
@@ -146,15 +154,15 @@ impl FlycheckThread {
 
             if self.should_recheck() {
                 self.last_update_req = None;
-                task_send.send(CheckTask::ClearDiagnostics).unwrap();
+                self.send(CheckTask::ClearDiagnostics);
                 self.restart_check_process();
             }
         }
     }
 
-    fn clean_previous_results(&self, task_send: &Sender<CheckTask>) {
-        task_send.send(CheckTask::ClearDiagnostics).unwrap();
-        task_send.send(CheckTask::Status(Status::End)).unwrap();
+    fn clean_previous_results(&self) {
+        self.send(CheckTask::ClearDiagnostics);
+        self.send(CheckTask::Status(Status::End));
     }
 
     fn should_recheck(&mut self) -> bool {
@@ -173,27 +181,25 @@ impl FlycheckThread {
         }
     }
 
-    fn handle_message(&self, msg: CheckEvent, task_send: &Sender<CheckTask>) {
+    fn handle_message(&self, msg: CheckEvent) {
         match msg {
             CheckEvent::Begin => {
-                task_send.send(CheckTask::Status(Status::Being)).unwrap();
+                self.send(CheckTask::Status(Status::Being));
             }
 
             CheckEvent::End => {
-                task_send.send(CheckTask::Status(Status::End)).unwrap();
+                self.send(CheckTask::Status(Status::End));
             }
 
             CheckEvent::Msg(Message::CompilerArtifact(msg)) => {
-                task_send.send(CheckTask::Status(Status::Progress(msg.target.name))).unwrap();
+                self.send(CheckTask::Status(Status::Progress(msg.target.name)));
             }
 
             CheckEvent::Msg(Message::CompilerMessage(msg)) => {
-                task_send
-                    .send(CheckTask::AddDiagnostic {
-                        workspace_root: self.workspace_root.clone(),
-                        diagnostic: msg.message,
-                    })
-                    .unwrap();
+                self.send(CheckTask::AddDiagnostic {
+                    workspace_root: self.workspace_root.clone(),
+                    diagnostic: msg.message,
+                });
             }
 
             CheckEvent::Msg(Message::BuildScriptExecuted(_msg)) => {}
@@ -271,6 +277,10 @@ impl FlycheckThread {
             let _ = message_send.send(CheckEvent::End);
         }))
     }
+
+    fn send(&self, check_task: CheckTask) {
+        (self.sender)(check_task)
+    }
 }
 
 enum CheckEvent {

crates/rust-analyzer/src/cli/load_cargo.rs

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

crates/rust-analyzer/src/global_state.rs

@@ -9,7 +9,7 @@ use crossbeam_channel::{unbounded, Receiver};
 use lsp_types::Url;
 use parking_lot::RwLock;
 use ra_db::{CrateId, SourceRoot, VfsPath};
-use ra_flycheck::{Flycheck, FlycheckConfig};
+use ra_flycheck::{CheckTask, FlycheckConfig, FlycheckHandle};
 use ra_ide::{Analysis, AnalysisChange, AnalysisHost, CrateGraph, FileId};
 use ra_project_model::{CargoWorkspace, ProcMacroClient, ProjectWorkspace, Target};
 use stdx::format_to;
@@ -27,12 +27,18 @@ use crate::{
 };
 use rustc_hash::{FxHashMap, FxHashSet};
 
-fn create_flycheck(workspaces: &[ProjectWorkspace], config: &FlycheckConfig) -> Option<Flycheck> {
+fn create_flycheck(
+    workspaces: &[ProjectWorkspace],
+    config: &FlycheckConfig,
+) -> Option<(FlycheckHandle, Receiver<CheckTask>)> {
     // FIXME: Figure out the multi-workspace situation
     workspaces.iter().find_map(move |w| match w {
         ProjectWorkspace::Cargo { cargo, .. } => {
+            let (sender, receiver) = unbounded();
+            let sender = Box::new(move |msg| sender.send(msg).unwrap());
             let cargo_project_root = cargo.workspace_root().to_path_buf();
-            Some(Flycheck::new(config.clone(), cargo_project_root.into()))
+            let flycheck = FlycheckHandle::spawn(sender, config.clone(), cargo_project_root.into());
+            Some((flycheck, receiver))
         }
         ProjectWorkspace::Json { .. } => {
             log::warn!("Cargo check watching only supported for cargo workspaces, disabling");
@@ -63,7 +69,7 @@ pub(crate) struct GlobalState {
     pub(crate) analysis_host: AnalysisHost,
     pub(crate) loader: Box<dyn vfs::loader::Handle>,
     pub(crate) task_receiver: Receiver<vfs::loader::Message>,
-    pub(crate) flycheck: Option<Flycheck>,
+    pub(crate) flycheck: Option<(FlycheckHandle, Receiver<CheckTask>)>,
     pub(crate) diagnostics: DiagnosticCollection,
     pub(crate) mem_docs: FxHashSet<VfsPath>,
     pub(crate) vfs: Arc<RwLock<(vfs::Vfs, FxHashMap<FileId, LineEndings>)>>,
@@ -115,7 +121,7 @@ impl GlobalState {
         };
 
         let mut loader = {
-            let loader = vfs_notify::LoaderHandle::spawn(Box::new(move |msg| {
+            let loader = vfs_notify::NotifyHandle::spawn(Box::new(move |msg| {
                 task_sender.send(msg).unwrap()
             }));
             Box::new(loader)

crates/rust-analyzer/src/main_loop.rs

@@ -136,7 +136,7 @@ pub fn main_loop(config: Config, connection: Connection) -> Result<()> {
                     Ok(task) => Event::Vfs(task),
                     Err(RecvError) => return Err("vfs died".into()),
                 },
-                recv(global_state.flycheck.as_ref().map_or(&never(), |it| &it.task_recv)) -> task => match task {
+                recv(global_state.flycheck.as_ref().map_or(&never(), |it| &it.1)) -> task => match task {
                     Ok(task) => Event::CheckWatcher(task),
                     Err(RecvError) => return Err("check watcher died".into()),
                 },
@@ -290,7 +290,7 @@ fn loop_turn(
 
     if became_ready {
         if let Some(flycheck) = &global_state.flycheck {
-            flycheck.update();
+            flycheck.0.update();
         }
     }
 
@@ -486,7 +486,7 @@ fn on_notification(
     let not = match notification_cast::<lsp_types::notification::DidSaveTextDocument>(not) {
         Ok(_params) => {
             if let Some(flycheck) = &global_state.flycheck {
-                flycheck.update();
+                flycheck.0.update();
             }
             return Ok(());
         }

crates/vfs-notify/src/lib.rs

@@ -20,7 +20,7 @@ use walkdir::WalkDir;
 use crate::include::Include;
 
 #[derive(Debug)]
-pub struct LoaderHandle {
+pub struct NotifyHandle {
     // Relative order of fields below is significant.
     sender: crossbeam_channel::Sender<Message>,
     _thread: jod_thread::JoinHandle,
@@ -32,12 +32,12 @@ enum Message {
     Invalidate(AbsPathBuf),
 }
 
-impl loader::Handle for LoaderHandle {
-    fn spawn(sender: loader::Sender) -> LoaderHandle {
-        let actor = LoaderActor::new(sender);
+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::spawn(move || actor.run(receiver));
-        LoaderHandle { sender, _thread: thread }
+        NotifyHandle { sender, _thread: thread }
     }
     fn set_config(&mut self, config: loader::Config) {
         self.sender.send(Message::Config(config)).unwrap()
@@ -52,10 +52,10 @@ impl loader::Handle for LoaderHandle {
 
 type NotifyEvent = notify::Result<notify::Event>;
 
-struct LoaderActor {
+struct NotifyActor {
+    sender: loader::Sender,
     config: Vec<(AbsPathBuf, Include, bool)>,
     watched_paths: FxHashSet<AbsPathBuf>,
-    sender: loader::Sender,
     // Drop order of fields bellow is significant,
     watcher: Option<RecommendedWatcher>,
     watcher_receiver: Receiver<NotifyEvent>,
@@ -67,19 +67,19 @@ enum Event {
     NotifyEvent(NotifyEvent),
 }
 
-impl LoaderActor {
-    fn new(sender: loader::Sender) -> LoaderActor {
+impl NotifyActor {
+    fn new(sender: loader::Sender) -> NotifyActor {
         let (watcher_sender, watcher_receiver) = unbounded();
         let watcher = log_notify_error(Watcher::new_immediate(move |event| {
             watcher_sender.send(event).unwrap()
         }));
 
-        LoaderActor {
-            watcher,
-            watcher_receiver,
-            watched_paths: FxHashSet::default(),
+        NotifyActor {
             sender,
             config: Vec::new(),
+            watched_paths: FxHashSet::default(),
+            watcher,
+            watcher_receiver,
         }
     }