6 files changed, 179 insertions, 175 deletions

diff --git a/crates/flycheck/src/lib.rs b/crates/flycheck/src/lib.rs
index 3e73cf6ff..9e8205ae7 100644
--- a/crates/flycheck/src/lib.rs
+++ b/crates/flycheck/src/lib.rs
@@ -10,7 +10,7 @@ use std::{
     time::Instant,
 };
 
-use crossbeam_channel::{never, select, unbounded, Receiver, RecvError, Sender};
+use crossbeam_channel::{never, select, unbounded, Receiver, Sender};
 
 pub use cargo_metadata::diagnostic::{
     Applicability, Diagnostic, DiagnosticLevel, DiagnosticSpan, DiagnosticSpanMacroExpansion,
@@ -61,7 +61,7 @@ impl FlycheckHandle {
     ) -> FlycheckHandle {
         let (cmd_send, cmd_recv) = unbounded::<Restart>();
         let handle = jod_thread::spawn(move || {
-            FlycheckActor::new(sender, config, workspace_root).run(&cmd_recv);
+            FlycheckActor::new(sender, config, workspace_root).run(cmd_recv);
         });
         FlycheckHandle { cmd_send, handle }
     }
@@ -98,14 +98,18 @@ struct FlycheckActor {
     config: FlycheckConfig,
     workspace_root: PathBuf,
     last_update_req: Option<Instant>,
-    // XXX: drop order is significant
-    message_recv: Receiver<CheckEvent>,
     /// WatchThread exists to wrap around the communication needed to be able to
     /// run `cargo check` without blocking. Currently the Rust standard library
     /// doesn't provide a way to read sub-process output without blocking, so we
     /// have to wrap sub-processes output handling in a thread and pass messages
     /// back over a channel.
-    check_process: Option<jod_thread::JoinHandle>,
+    // XXX: drop order is significant
+    check_process: Option<(Receiver<CheckEvent>, jod_thread::JoinHandle)>,
+}
+
+enum Event {
+    Restart(Restart),
+    CheckEvent(Option<CheckEvent>),
 }
 
 impl FlycheckActor {
@@ -114,40 +118,48 @@ impl FlycheckActor {
         config: FlycheckConfig,
         workspace_root: PathBuf,
     ) -> FlycheckActor {
-        FlycheckActor {
-            sender,
-            config,
-            workspace_root,
-            last_update_req: None,
-            message_recv: never(),
-            check_process: None,
-        }
+        FlycheckActor { sender, config, workspace_root, last_update_req: None, check_process: None }
     }
 
-    fn run(&mut self, cmd_recv: &Receiver<Restart>) {
+    fn run(&mut self, inbox: Receiver<Restart>) {
         // If we rerun the thread, we need to discard the previous check results first
-        self.clean_previous_results();
-
-        loop {
-            select! {
-                recv(&cmd_recv) -> cmd => match cmd {
-                    Ok(Restart) => self.last_update_req = Some(Instant::now()),
-                    Err(RecvError) => {
-                        // Command channel has closed, so shut down
-                        break;
-                    },
-                },
-                recv(self.message_recv) -> msg => match msg {
-                    Ok(msg) => self.handle_message(msg),
-                    Err(RecvError) => {
-                        // Watcher finished, replace it with a never channel to
-                        // avoid busy-waiting.
-                        self.message_recv = never();
-                        self.check_process = None;
-                    },
+        self.send(Message::ClearDiagnostics);
+        self.send(Message::Progress(Progress::End));
+
+        while let Some(event) = self.next_event(&inbox) {
+            match event {
+                Event::Restart(Restart) => self.last_update_req = Some(Instant::now()),
+                Event::CheckEvent(None) => {
+                    // Watcher finished, replace it with a never channel to
+                    // avoid busy-waiting.
+                    self.check_process = None;
                 }
-            };
+                Event::CheckEvent(Some(event)) => match event {
+                    CheckEvent::Begin => {
+                        self.send(Message::Progress(Progress::Being));
+                    }
+
+                    CheckEvent::End => {
+                        self.send(Message::Progress(Progress::End));
+                    }
+
+                    CheckEvent::Msg(cargo_metadata::Message::CompilerArtifact(msg)) => {
+                        self.send(Message::Progress(Progress::DidCheckCrate(msg.target.name)));
+                    }
+
+                    CheckEvent::Msg(cargo_metadata::Message::CompilerMessage(msg)) => {
+                        self.send(Message::AddDiagnostic {
+                            workspace_root: self.workspace_root.clone(),
+                            diagnostic: msg.message,
+                        });
+                    }
 
+                    CheckEvent::Msg(cargo_metadata::Message::BuildScriptExecuted(_))
+                    | CheckEvent::Msg(cargo_metadata::Message::BuildFinished(_))
+                    | CheckEvent::Msg(cargo_metadata::Message::TextLine(_))
+                    | CheckEvent::Msg(cargo_metadata::Message::Unknown) => {}
+                },
+            }
             if self.should_recheck() {
                 self.last_update_req = None;
                 self.send(Message::ClearDiagnostics);
@@ -156,9 +168,12 @@ impl FlycheckActor {
         }
     }
 
-    fn clean_previous_results(&self) {
-        self.send(Message::ClearDiagnostics);
-        self.send(Message::Progress(Progress::End));
+    fn next_event(&self, inbox: &Receiver<Restart>) -> Option<Event> {
+        let check_chan = self.check_process.as_ref().map(|(chan, _thread)| chan);
+        select! {
+            recv(inbox) -> msg => msg.ok().map(Event::Restart),
+            recv(check_chan.unwrap_or(&never())) -> msg => Some(Event::CheckEvent(msg.ok())),
+        }
     }
 
     fn should_recheck(&mut self) -> bool {
@@ -171,37 +186,8 @@ impl FlycheckActor {
         false
     }
 
-    fn handle_message(&self, msg: CheckEvent) {
-        match msg {
-            CheckEvent::Begin => {
-                self.send(Message::Progress(Progress::Being));
-            }
-
-            CheckEvent::End => {
-                self.send(Message::Progress(Progress::End));
-            }
-
-            CheckEvent::Msg(cargo_metadata::Message::CompilerArtifact(msg)) => {
-                self.send(Message::Progress(Progress::DidCheckCrate(msg.target.name)));
-            }
-
-            CheckEvent::Msg(cargo_metadata::Message::CompilerMessage(msg)) => {
-                self.send(Message::AddDiagnostic {
-                    workspace_root: self.workspace_root.clone(),
-                    diagnostic: msg.message,
-                });
-            }
-
-            CheckEvent::Msg(cargo_metadata::Message::BuildScriptExecuted(_))
-            | CheckEvent::Msg(cargo_metadata::Message::BuildFinished(_))
-            | CheckEvent::Msg(cargo_metadata::Message::TextLine(_))
-            | CheckEvent::Msg(cargo_metadata::Message::Unknown) => {}
-        }
-    }
-
     fn restart_check_process(&mut self) {
         // First, clear and cancel the old thread
-        self.message_recv = never();
         self.check_process = None;
 
         let mut cmd = match &self.config {
@@ -237,8 +223,7 @@ impl FlycheckActor {
         cmd.current_dir(&self.workspace_root);
 
         let (message_send, message_recv) = unbounded();
-        self.message_recv = message_recv;
-        self.check_process = Some(jod_thread::spawn(move || {
+        let thread = jod_thread::spawn(move || {
             // If we trigger an error here, we will do so in the loop instead,
             // which will break out of the loop, and continue the shutdown
             let _ = message_send.send(CheckEvent::Begin);
@@ -267,7 +252,8 @@ impl FlycheckActor {
             // We can ignore any error here, as we are already in the progress
             // of shutting down.
             let _ = message_send.send(CheckEvent::End);
-        }))
+        });
+        self.check_process = Some((message_recv, thread))
     }
 
     fn send(&self, check_task: Message) {
diff --git a/crates/rust-analyzer/src/global_state.rs b/crates/rust-analyzer/src/global_state.rs
index 446207e9e..de6b95686 100644
--- a/crates/rust-analyzer/src/global_state.rs
+++ b/crates/rust-analyzer/src/global_state.rs
@@ -20,8 +20,9 @@ use crate::{
     diagnostics::{CheckFixes, DiagnosticCollection},
     from_proto,
     line_endings::LineEndings,
-    main_loop::ReqQueue,
+    main_loop::{ReqQueue, Task},
     request_metrics::{LatestRequests, RequestMetrics},
+    thread_pool::TaskPool,
     to_proto::url_from_abs_path,
     Result,
 };
@@ -66,6 +67,7 @@ impl Default for Status {
 /// incremental salsa database.
 pub(crate) struct GlobalState {
     pub(crate) config: Config,
+    pub(crate) task_pool: (TaskPool<Task>, Receiver<Task>),
     pub(crate) analysis_host: AnalysisHost,
     pub(crate) loader: Box<dyn vfs::loader::Handle>,
     pub(crate) task_receiver: Receiver<vfs::loader::Message>,
@@ -153,8 +155,15 @@ impl GlobalState {
 
         let mut analysis_host = AnalysisHost::new(lru_capacity);
         analysis_host.apply_change(change);
+
+        let task_pool = {
+            let (sender, receiver) = unbounded();
+            (TaskPool::new(sender), receiver)
+        };
+
         let mut res = GlobalState {
             config,
+            task_pool,
             analysis_host,
             loader,
             task_receiver,
diff --git a/crates/rust-analyzer/src/lib.rs b/crates/rust-analyzer/src/lib.rs
index 794286672..ca788dd3c 100644
--- a/crates/rust-analyzer/src/lib.rs
+++ b/crates/rust-analyzer/src/lib.rs
@@ -30,6 +30,7 @@ mod diagnostics;
 mod line_endings;
 mod request_metrics;
 mod lsp_utils;
+mod thread_pool;
 pub mod lsp_ext;
 pub mod config;
 
diff --git a/crates/rust-analyzer/src/main_loop.rs b/crates/rust-analyzer/src/main_loop.rs
index 3b3b83209..1a9c5ee2c 100644
--- a/crates/rust-analyzer/src/main_loop.rs
+++ b/crates/rust-analyzer/src/main_loop.rs
@@ -2,19 +2,17 @@
 //! requests/replies and notifications back to the client.
 use std::{
     env, fmt, panic,
-    sync::Arc,
     time::{Duration, Instant},
 };
 
-use crossbeam_channel::{never, select, unbounded, RecvError, Sender};
-use lsp_server::{Connection, ErrorCode, Message, Notification, Request, RequestId, Response};
+use crossbeam_channel::{never, select, RecvError, Sender};
+use lsp_server::{Connection, ErrorCode, Notification, Request, RequestId, Response};
 use lsp_types::{request::Request as _, NumberOrString};
 use ra_db::VfsPath;
 use ra_ide::{Canceled, FileId};
 use ra_prof::profile;
 use ra_project_model::{PackageRoot, ProjectWorkspace};
 use serde::{de::DeserializeOwned, Serialize};
-use threadpool::ThreadPool;
 
 use crate::{
     config::{Config, FilesWatcher, LinkedProject},
@@ -118,65 +116,51 @@ pub fn main_loop(config: Config, connection: Connection) -> Result<()> {
         GlobalState::new(workspaces, config.lru_capacity, config, req_queue)
     };
 
-    let pool = ThreadPool::default();
-    let (task_sender, task_receiver) = unbounded::<Task>();
-
     log::info!("server initialized, serving requests");
     {
-        let task_sender = task_sender;
         loop {
             log::trace!("selecting");
             let event = select! {
                 recv(&connection.receiver) -> msg => match msg {
-                    Ok(msg) => Event::Msg(msg),
+                    Ok(msg) => Event::Lsp(msg),
                     Err(RecvError) => return Err("client exited without shutdown".into()),
                 },
-                recv(task_receiver) -> task => Event::Task(task.unwrap()),
+                recv(&global_state.task_pool.1) -> task => Event::Task(task.unwrap()),
                 recv(global_state.task_receiver) -> task => match task {
                     Ok(task) => Event::Vfs(task),
                     Err(RecvError) => return Err("vfs died".into()),
                 },
                 recv(global_state.flycheck.as_ref().map_or(&never(), |it| &it.1)) -> task => match task {
-                    Ok(task) => Event::CheckWatcher(task),
+                    Ok(task) => Event::Flycheck(task),
                     Err(RecvError) => return Err("check watcher died".into()),
                 },
             };
-            if let Event::Msg(Message::Request(req)) = &event {
+            if let Event::Lsp(lsp_server::Message::Request(req)) = &event {
                 if connection.handle_shutdown(&req)? {
                     break;
                 };
             }
             assert!(!global_state.vfs.read().0.has_changes());
-            loop_turn(&pool, &task_sender, &connection, &mut global_state, event)?;
+            loop_turn(&connection, &mut global_state, event)?;
             assert!(!global_state.vfs.read().0.has_changes());
         }
     }
     global_state.analysis_host.request_cancellation();
-    log::info!("waiting for tasks to finish...");
-    task_receiver.into_iter().for_each(|task| on_task(task, &connection.sender, &mut global_state));
-    log::info!("...tasks have finished");
-    log::info!("joining threadpool...");
-    pool.join();
-    drop(pool);
-    log::info!("...threadpool has finished");
-
-    let vfs = Arc::try_unwrap(global_state.vfs).expect("all snapshots should be dead");
-    drop(vfs);
-
     Ok(())
 }
 
 #[derive(Debug)]
-enum Task {
+pub(crate) enum Task {
     Respond(Response),
-    Diagnostic(DiagnosticTask),
+    Diagnostics(Vec<DiagnosticTask>),
+    Unit,
 }
 
 enum Event {
-    Msg(Message),
+    Lsp(lsp_server::Message),
     Task(Task),
     Vfs(vfs::loader::Message),
-    CheckWatcher(flycheck::Message),
+    Flycheck(flycheck::Message),
 }
 
 impl fmt::Debug for Event {
@@ -186,7 +170,7 @@ impl fmt::Debug for Event {
         };
 
         match self {
-            Event::Msg(Message::Notification(not)) => {
+            Event::Lsp(lsp_server::Message::Notification(not)) => {
                 if notification_is::<lsp_types::notification::DidOpenTextDocument>(not)
                     || notification_is::<lsp_types::notification::DidChangeTextDocument>(not)
                 {
@@ -203,10 +187,10 @@ impl fmt::Debug for Event {
             _ => (),
         }
         match self {
-            Event::Msg(it) => fmt::Debug::fmt(it, f),
+            Event::Lsp(it) => fmt::Debug::fmt(it, f),
             Event::Task(it) => fmt::Debug::fmt(it, f),
             Event::Vfs(it) => fmt::Debug::fmt(it, f),
-            Event::CheckWatcher(it) => fmt::Debug::fmt(it, f),
+            Event::Flycheck(it) => fmt::Debug::fmt(it, f),
         }
     }
 }
@@ -215,19 +199,13 @@ pub(crate) type ReqHandler = fn(&mut GlobalState, Response);
 pub(crate) type ReqQueue = lsp_server::ReqQueue<(&'static str, Instant), ReqHandler>;
 const DO_NOTHING: ReqHandler = |_, _| ();
 
-fn loop_turn(
-    pool: &ThreadPool,
-    task_sender: &Sender<Task>,
-    connection: &Connection,
-    global_state: &mut GlobalState,
-    event: Event,
-) -> Result<()> {
+fn loop_turn(connection: &Connection, global_state: &mut GlobalState, event: Event) -> Result<()> {
     let loop_start = Instant::now();
 
     // NOTE: don't count blocking select! call as a loop-turn time
     let _p = profile("main_loop_inner/loop-turn");
     log::info!("loop turn = {:?}", event);
-    let queue_count = pool.queued_count();
+    let queue_count = global_state.task_pool.0.len();
     if queue_count > 0 {
         log::info!("queued count = {}", queue_count);
     }
@@ -269,17 +247,15 @@ fn loop_turn(
                 )
             }
         },
-        Event::CheckWatcher(task) => {
-            on_check_task(task, global_state, task_sender, &connection.sender)?
-        }
-        Event::Msg(msg) => match msg {
-            Message::Request(req) => {
-                on_request(global_state, pool, task_sender, &connection.sender, loop_start, req)?
+        Event::Flycheck(task) => on_check_task(task, global_state, &connection.sender)?,
+        Event::Lsp(msg) => match msg {
+            lsp_server::Message::Request(req) => {
+                on_request(global_state, &connection.sender, loop_start, req)?
             }
-            Message::Notification(not) => {
+            lsp_server::Message::Notification(not) => {
                 on_notification(&connection.sender, global_state, not)?;
             }
-            Message::Response(resp) => {
+            lsp_server::Message::Response(resp) => {
                 let handler = global_state.req_queue.outgoing.complete(resp.id.clone());
                 handler(global_state, resp)
             }
@@ -301,16 +277,14 @@ fn loop_turn(
             .map(|path| global_state.vfs.read().0.file_id(&path).unwrap())
             .collect::<Vec<_>>();
 
-        update_file_notifications_on_threadpool(
-            pool,
-            global_state.snapshot(),
-            task_sender.clone(),
-            subscriptions.clone(),
-        );
-        pool.execute({
+        update_file_notifications_on_threadpool(global_state, subscriptions.clone());
+        global_state.task_pool.0.spawn({
             let subs = subscriptions;
             let snap = global_state.snapshot();
-            move || snap.analysis.prime_caches(subs).unwrap_or_else(|_: Canceled| ())
+            move || {
+                snap.analysis.prime_caches(subs).unwrap_or_else(|_: Canceled| ());
+                Task::Unit
+            }
         });
     }
 
@@ -329,7 +303,7 @@ fn loop_turn(
     Ok(())
 }
 
-fn on_task(task: Task, msg_sender: &Sender<Message>, global_state: &mut GlobalState) {
+fn on_task(task: Task, msg_sender: &Sender<lsp_server::Message>, global_state: &mut GlobalState) {
     match task {
         Task::Respond(response) => {
             if let Some((method, start)) =
@@ -345,26 +319,21 @@ fn on_task(task: Task, msg_sender: &Sender<Message>, global_state: &mut GlobalSt
                 msg_sender.send(response.into()).unwrap();
             }
         }
-        Task::Diagnostic(task) => on_diagnostic_task(task, msg_sender, global_state),
+        Task::Diagnostics(tasks) => {
+            tasks.into_iter().for_each(|task| on_diagnostic_task(task, msg_sender, global_state))
+        }
+        Task::Unit => (),
     }
 }
 
 fn on_request(
     global_state: &mut GlobalState,
-    pool: &ThreadPool,
-    task_sender: &Sender<Task>,
-    msg_sender: &Sender<Message>,
+    msg_sender: &Sender<lsp_server::Message>,
     request_received: Instant,
     req: Request,
 ) -> Result<()> {
-    let mut pool_dispatcher = PoolDispatcher {
-        req: Some(req),
-        pool,
-        global_state,
-        task_sender,
-        msg_sender,
-        request_received,
-    };
+    let mut pool_dispatcher =
+        PoolDispatcher { req: Some(req), global_state, msg_sender, request_received };
     pool_dispatcher
         .on_sync::<lsp_ext::CollectGarbage>(|s, ()| Ok(s.collect_garbage()))?
         .on_sync::<lsp_ext::JoinLines>(|s, p| handlers::handle_join_lines(s.snapshot(), p))?
@@ -415,7 +384,7 @@ fn on_request(
 }
 
 fn on_notification(
-    msg_sender: &Sender<Message>,
+    msg_sender: &Sender<lsp_server::Message>,
     global_state: &mut GlobalState,
     not: Notification,
 ) -> Result<()> {
@@ -552,12 +521,11 @@ fn on_notification(
 fn on_check_task(
     task: flycheck::Message,
     global_state: &mut GlobalState,
-    task_sender: &Sender<Task>,
-    msg_sender: &Sender<Message>,
+    msg_sender: &Sender<lsp_server::Message>,
 ) -> Result<()> {
     match task {
         flycheck::Message::ClearDiagnostics => {
-            task_sender.send(Task::Diagnostic(DiagnosticTask::ClearCheck))?;
+            on_diagnostic_task(DiagnosticTask::ClearCheck, msg_sender, global_state)
         }
 
         flycheck::Message::AddDiagnostic { workspace_root, diagnostic } => {
@@ -576,11 +544,15 @@ fn on_check_task(
                     }
                 };
 
-                task_sender.send(Task::Diagnostic(DiagnosticTask::AddCheck(
-                    file_id,
-                    diag.diagnostic,
-                    diag.fixes.into_iter().map(|it| it.into()).collect(),
-                )))?;
+                on_diagnostic_task(
+                    DiagnosticTask::AddCheck(
+                        file_id,
+                        diag.diagnostic,
+                        diag.fixes.into_iter().map(|it| it.into()).collect(),
+                    ),
+                    msg_sender,
+                    global_state,
+                )
             }
         }
 
@@ -598,7 +570,11 @@ fn on_check_task(
     Ok(())
 }
 
-fn on_diagnostic_task(task: DiagnosticTask, msg_sender: &Sender<Message>, state: &mut GlobalState) {
+fn on_diagnostic_task(
+    task: DiagnosticTask,
+    msg_sender: &Sender<lsp_server::Message>,
+    state: &mut GlobalState,
+) {
     let subscriptions = state.diagnostics.handle_task(task);
 
     for file_id in subscriptions {
@@ -623,7 +599,7 @@ fn percentage(done: usize, total: usize) -> f64 {
 
 fn report_progress(
     global_state: &mut GlobalState,
-    sender: &Sender<Message>,
+    sender: &Sender<lsp_server::Message>,
     title: &str,
     state: Progress,
     message: Option<String>,
@@ -670,10 +646,8 @@ fn report_progress(
 
 struct PoolDispatcher<'a> {
     req: Option<Request>,
-    pool: &'a ThreadPool,
     global_state: &'a mut GlobalState,
-    msg_sender: &'a Sender<Message>,
-    task_sender: &'a Sender<Task>,
+    msg_sender: &'a Sender<lsp_server::Message>,
     request_received: Instant,
 }
 
@@ -721,13 +695,11 @@ impl<'a> PoolDispatcher<'a> {
             }
         };
 
-        self.pool.execute({
+        self.global_state.task_pool.0.spawn({
             let world = self.global_state.snapshot();
-            let sender = self.task_sender.clone();
             move || {
                 let result = f(world, params);
-                let task = result_to_task::<R>(id, result);
-                sender.send(task).unwrap();
+                result_to_task::<R>(id, result)
             }
         });
 
@@ -797,26 +769,27 @@ where
 }
 
 fn update_file_notifications_on_threadpool(
-    pool: &ThreadPool,
-    world: GlobalStateSnapshot,
-    task_sender: Sender<Task>,
+    global_state: &mut GlobalState,
     subscriptions: Vec<FileId>,
 ) {
     log::trace!("updating notifications for {:?}", subscriptions);
-    if world.config.publish_diagnostics {
-        pool.execute(move || {
-            for file_id in subscriptions {
-                match handlers::publish_diagnostics(&world, file_id) {
-                    Err(e) => {
-                        if !is_canceled(&*e) {
-                            log::error!("failed to compute diagnostics: {:?}", e);
-                        }
-                    }
-                    Ok(task) => {
-                        task_sender.send(Task::Diagnostic(task)).unwrap();
-                    }
-                }
-            }
+    if global_state.config.publish_diagnostics {
+        let snapshot = global_state.snapshot();
+        global_state.task_pool.0.spawn(move || {
+            let diagnostics = subscriptions
+                .into_iter()
+                .filter_map(|file_id| {
+                    handlers::publish_diagnostics(&snapshot, file_id)
+                        .map_err(|err| {
+                            if !is_canceled(&*err) {
+                                log::error!("failed to compute diagnostics: {:?}", err);
+                            }
+                            ()
+                        })
+                        .ok()
+                })
+                .collect::<Vec<_>>();
+            Task::Diagnostics(diagnostics)
         })
     }
 }
diff --git a/crates/rust-analyzer/src/thread_pool.rs b/crates/rust-analyzer/src/thread_pool.rs
new file mode 100644
index 000000000..4fa502925
--- /dev/null
+++ b/crates/rust-analyzer/src/thread_pool.rs
@@ -0,0 +1,35 @@
+//! A thin wrapper around `ThreadPool` to make sure that we join all things
+//! properly.
+use crossbeam_channel::Sender;
+
+pub(crate) struct TaskPool<T> {
+    sender: Sender<T>,
+    inner: threadpool::ThreadPool,
+}
+
+impl<T> TaskPool<T> {
+    pub(crate) fn new(sender: Sender<T>) -> TaskPool<T> {
+        TaskPool { sender, inner: threadpool::ThreadPool::default() }
+    }
+
+    pub(crate) fn spawn<F>(&mut self, task: F)
+    where
+        F: FnOnce() -> T + Send + 'static,
+        T: Send + 'static,
+    {
+        self.inner.execute({
+            let sender = self.sender.clone();
+            move || sender.send(task()).unwrap()
+        })
+    }
+
+    pub(crate) fn len(&self) -> usize {
+        self.inner.queued_count()
+    }
+}
+
+impl<T> Drop for TaskPool<T> {
+    fn drop(&mut self) {
+        self.inner.join()
+    }
+}
diff --git a/crates/vfs-notify/src/lib.rs b/crates/vfs-notify/src/lib.rs
index 282cf0358..68fdb8cb0 100644
--- a/crates/vfs-notify/src/lib.rs
+++ b/crates/vfs-notify/src/lib.rs
@@ -83,8 +83,8 @@ impl NotifyActor {
         }
     }
 
-    fn run(mut self, receiver: Receiver<Message>) {
-        while let Some(event) = self.next_event(&receiver) {
+    fn run(mut self, inbox: Receiver<Message>) {
+        while let Some(event) = self.next_event(&inbox) {
             log::debug!("vfs-notify event: {:?}", event);
             match event {
                 Event::Message(msg) => match msg {