automatically wait for worker threads

closes #817

2 files changed, 50 insertions, 62 deletions

diff --git a/crates/ra_vfs/src/io.rs b/crates/ra_vfs/src/io.rs
index 3952b200b..1b70cd8df 100644
--- a/crates/ra_vfs/src/io.rs
+++ b/crates/ra_vfs/src/io.rs
@@ -1,13 +1,11 @@
 use std::{
     fs,
-    thread,
     path::{Path, PathBuf},
     sync::{mpsc, Arc},
     time::Duration,
 };
 use crossbeam_channel::{Receiver, Sender, unbounded, RecvError, select};
 use relative_path::RelativePathBuf;
-use thread_worker::WorkerHandle;
 use walkdir::WalkDir;
 use notify::{DebouncedEvent, RecommendedWatcher, RecursiveMode, Watcher as _Watcher};
 
@@ -49,8 +47,7 @@ enum ChangeKind {
 const WATCHER_DELAY: Duration = Duration::from_millis(250);
 
 pub(crate) struct Worker {
-    worker: thread_worker::Worker<Task, TaskResult>,
-    worker_handle: WorkerHandle,
+    thread_worker: thread_worker::Worker<Task, TaskResult>,
 }
 
 impl Worker {
@@ -62,82 +59,79 @@ impl Worker {
         //    * we want to read all files from a single thread, to guarantee that
         //      we always get fresher versions and never go back in time.
         //    * we want to tear down everything neatly during shutdown.
-        let (worker, worker_handle) = thread_worker::spawn(
+        let thread_worker = thread_worker::Worker::spawn(
             "vfs",
             128,
             // This are the channels we use to communicate with outside world.
             // If `input_receiver` is closed we need to tear ourselves down.
             // `output_sender` should not be closed unless the parent died.
             move |input_receiver, output_sender| {
-                // These are `std` channels notify will send events to
-                let (notify_sender, notify_receiver) = mpsc::channel();
+                // Make sure that the destruction order is
+                //
+                // * notify_sender
+                // * _thread
+                // * watcher_sender
+                //
+                // this is required to avoid deadlocks.
+
                 // These are the corresponding crossbeam channels
                 let (watcher_sender, watcher_receiver) = unbounded();
+                let _thread;
+                {
+                    // These are `std` channels notify will send events to
+                    let (notify_sender, notify_receiver) = mpsc::channel();
 
-                let mut watcher = notify::watcher(notify_sender, WATCHER_DELAY)
-                    .map_err(|e| log::error!("failed to spawn notify {}", e))
-                    .ok();
-                // Start a silly thread to transform between two channels
-                let thread = thread::spawn(move || {
-                    notify_receiver
-                        .into_iter()
-                        .for_each(|event| convert_notify_event(event, &watcher_sender))
-                });
+                    let mut watcher = notify::watcher(notify_sender, WATCHER_DELAY)
+                        .map_err(|e| log::error!("failed to spawn notify {}", e))
+                        .ok();
+                    // Start a silly thread to transform between two channels
+                    _thread = thread_worker::ScopedThread::spawn("notify-convertor", move || {
+                        notify_receiver
+                            .into_iter()
+                            .for_each(|event| convert_notify_event(event, &watcher_sender))
+                    });
 
-                // Process requests from the called or notifications from
-                // watcher until the caller says stop.
-                loop {
-                    select! {
-                        // Received request from the caller. If this channel is
-                        // closed, we should shutdown everything.
-                        recv(input_receiver) -> t => match t {
-                            Err(RecvError) => {
-                                drop(input_receiver);
-                                break
+                    // Process requests from the called or notifications from
+                    // watcher until the caller says stop.
+                    loop {
+                        select! {
+                            // Received request from the caller. If this channel is
+                            // closed, we should shutdown everything.
+                            recv(input_receiver) -> t => match t {
+                                Err(RecvError) => {
+                                    drop(input_receiver);
+                                    break
+                                },
+                                Ok(Task::AddRoot { root, config }) => {
+                                    watch_root(watcher.as_mut(), &output_sender, root, Arc::clone(&config));
+                                }
+                            },
+                            // Watcher send us changes. If **this** channel is
+                            // closed, the watcher has died, which indicates a bug
+                            // -- escalate!
+                            recv(watcher_receiver) -> event => match event {
+                                Err(RecvError) => panic!("watcher is dead"),
+                                Ok((path, change)) => {
+                                    handle_change(watcher.as_mut(), &output_sender, &*roots, path, change);
+                                }
                             },
-                            Ok(Task::AddRoot { root, config }) => {
-                                watch_root(watcher.as_mut(), &output_sender, root, Arc::clone(&config));
-                            }
-                        },
-                        // Watcher send us changes. If **this** channel is
-                        // closed, the watcher has died, which indicates a bug
-                        // -- escalate!
-                        recv(watcher_receiver) -> event => match event {
-                            Err(RecvError) => panic!("watcher is dead"),
-                            Ok((path, change)) => {
-                                handle_change(watcher.as_mut(), &output_sender, &*roots, path, change);
-                            }
-                        },
+                        }
                     }
                 }
-                // Stopped the watcher
-                drop(watcher.take());
                 // Drain pending events: we are not interested in them anyways!
                 watcher_receiver.into_iter().for_each(|_| ());
-
-                let res = thread.join();
-                match &res {
-                    Ok(()) => log::info!("... Watcher terminated with ok"),
-                    Err(_) => log::error!("... Watcher terminated with err"),
-                }
-                res.unwrap();
             },
         );
 
-        Worker { worker, worker_handle }
+        Worker { thread_worker }
     }
 
     pub(crate) fn sender(&self) -> &Sender<Task> {
-        &self.worker.inp
+        &self.thread_worker.sender()
     }
 
     pub(crate) fn receiver(&self) -> &Receiver<TaskResult> {
-        &self.worker.out
-    }
-
-    pub(crate) fn shutdown(self) -> thread::Result<()> {
-        let _ = self.worker.shutdown();
-        self.worker_handle.shutdown()
+        &self.thread_worker.receiver()
     }
 }
 
diff --git a/crates/ra_vfs/src/lib.rs b/crates/ra_vfs/src/lib.rs
index f07657db6..1fb255365 100644
--- a/crates/ra_vfs/src/lib.rs
+++ b/crates/ra_vfs/src/lib.rs
@@ -22,7 +22,6 @@ use std::{
     fmt, fs, mem,
     path::{Path, PathBuf},
     sync::Arc,
-    thread,
 };
 
 use crossbeam_channel::Receiver;
@@ -337,11 +336,6 @@ impl Vfs {
         mem::replace(&mut self.pending_changes, Vec::new())
     }
 
-    /// Shutdown the VFS and terminate the background watching thread.
-    pub fn shutdown(self) -> thread::Result<()> {
-        self.worker.shutdown()
-    }
-
     fn add_file(
         &mut self,
         root: VfsRoot,