Merge #856

856: Reduce dependnecies of ra_vfs r=pnkfelix a=matklad

In preparation for moving `ra_vfs` to a separate repo with extensive cross-platform CI, remove dependency on `ra_thread_workder` and `ra_arena`.

Co-authored-by: Aleksey Kladov <[email protected]>

3 files changed, 178 insertions, 119 deletions

diff --git a/crates/ra_vfs/src/io.rs b/crates/ra_vfs/src/io.rs
index 0cffc03f3..5969ee0d0 100644
--- a/crates/ra_vfs/src/io.rs
+++ b/crates/ra_vfs/src/io.rs
@@ -3,13 +3,14 @@ use std::{
     path::{Path, PathBuf},
     sync::{mpsc, Arc},
     time::Duration,
+    thread,
 };
-use crossbeam_channel::{Sender, unbounded, RecvError, select};
+use crossbeam_channel::{Sender, Receiver, unbounded, RecvError, select};
 use relative_path::RelativePathBuf;
 use walkdir::WalkDir;
 use notify::{DebouncedEvent, RecommendedWatcher, RecursiveMode, Watcher as _Watcher};
 
-use crate::{Roots, VfsRoot};
+use crate::{Roots, VfsRoot, VfsTask};
 
 pub(crate) enum Task {
     AddRoot { root: VfsRoot },
@@ -18,7 +19,7 @@ pub(crate) enum Task {
 /// `TaskResult` transfers files read on the IO thread to the VFS on the main
 /// thread.
 #[derive(Debug)]
-pub enum TaskResult {
+pub(crate) enum TaskResult {
     /// Emitted when we've recursively scanned a source root during the initial
     /// load.
     BulkLoadRoot { root: VfsRoot, files: Vec<(RelativePathBuf, String)> },
@@ -46,7 +47,46 @@ enum ChangeKind {
 
 const WATCHER_DELAY: Duration = Duration::from_millis(250);
 
-pub(crate) type Worker = thread_worker::Worker<Task, TaskResult>;
+// Like thread::JoinHandle, but joins the thread on drop.
+//
+// This is useful because it guarantees the absence of run-away threads, even if
+// code panics. This is important, because we might see panics in the test and
+// we might be used in an IDE context, where a failed component is just
+// restarted.
+//
+// Because all threads are joined, care must be taken to avoid deadlocks. That
+// typically means ensuring that channels are dropped before the threads.
+struct ScopedThread(Option<thread::JoinHandle<()>>);
+
+impl ScopedThread {
+    fn spawn(name: String, f: impl FnOnce() + Send + 'static) -> ScopedThread {
+        let handle = thread::Builder::new().name(name).spawn(f).unwrap();
+        ScopedThread(Some(handle))
+    }
+}
+
+impl Drop for ScopedThread {
+    fn drop(&mut self) {
+        let res = self.0.take().unwrap().join();
+        if !thread::panicking() {
+            res.unwrap();
+        }
+    }
+}
+
+pub(crate) struct Worker {
+    // XXX: field order is significant here.
+    //
+    // In Rust, fields are dropped in the declaration order, and we rely on this
+    // here. We must close sender first, so that the  `thread` (who holds the
+    // opposite side of the channel) noticed shutdown. Then, we must join the
+    // thread, but we must keep receiver alive so that the thread does not
+    // panic.
+    pub(crate) sender: Sender<Task>,
+    _thread: ScopedThread,
+    pub(crate) receiver: Receiver<VfsTask>,
+}
+
 pub(crate) fn start(roots: Arc<Roots>) -> Worker {
     // This is a pretty elaborate setup of threads & channels! It is
     // explained by the following concerns:
@@ -55,74 +95,75 @@ pub(crate) fn start(roots: Arc<Roots>) -> 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.
-    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| {
-            // 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
-                _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
-                            },
-                            Ok(Task::AddRoot { root }) => {
-                                watch_root(watcher.as_mut(), &output_sender, &*roots, root);
-                            }
-                        },
-                        // 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);
-                            }
+    let _thread;
+    // 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.
+    let (input_sender, input_receiver) = unbounded();
+    let (output_sender, output_receiver) = unbounded();
+
+    _thread = ScopedThread::spawn("vfs".to_string(), move || {
+        // 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 _notify_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
+            _notify_thread = ScopedThread::spawn("notify-convertor".to_string(), 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
                         },
-                    }
+                        Ok(Task::AddRoot { root }) => {
+                            watch_root(watcher.as_mut(), &output_sender, &*roots, root);
+                        }
+                    },
+                    // 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);
+                        }
+                    },
                 }
             }
-            // Drain pending events: we are not interested in them anyways!
-            watcher_receiver.into_iter().for_each(|_| ());
-        },
-    )
+        }
+        // Drain pending events: we are not interested in them anyways!
+        watcher_receiver.into_iter().for_each(|_| ());
+    });
+    Worker { sender: input_sender, _thread, receiver: output_receiver }
 }
 
 fn watch_root(
     watcher: Option<&mut RecommendedWatcher>,
-    sender: &Sender<TaskResult>,
+    sender: &Sender<VfsTask>,
     roots: &Roots,
     root: VfsRoot,
 ) {
@@ -136,7 +177,8 @@ fn watch_root(
             Some((path, text))
         })
         .collect();
-    sender.send(TaskResult::BulkLoadRoot { root, files }).unwrap();
+    let res = TaskResult::BulkLoadRoot { root, files };
+    sender.send(VfsTask(res)).unwrap();
     log::debug!("... loaded {}", root_path.display());
 }
 
@@ -173,7 +215,7 @@ fn convert_notify_event(event: DebouncedEvent, sender: &Sender<(PathBuf, ChangeK
 
 fn handle_change(
     watcher: Option<&mut RecommendedWatcher>,
-    sender: &Sender<TaskResult>,
+    sender: &Sender<VfsTask>,
     roots: &Roots,
     path: PathBuf,
     kind: ChangeKind,
@@ -195,13 +237,15 @@ fn handle_change(
                 .try_for_each(|rel_path| {
                     let abs_path = rel_path.to_path(&roots.path(root));
                     let text = read_to_string(&abs_path);
-                    sender.send(TaskResult::SingleFile { root, path: rel_path, text })
+                    let res = TaskResult::SingleFile { root, path: rel_path, text };
+                    sender.send(VfsTask(res))
                 })
                 .unwrap()
         }
         ChangeKind::Write | ChangeKind::Remove => {
             let text = read_to_string(&path);
-            sender.send(TaskResult::SingleFile { root, path: rel_path, text }).unwrap();
+            let res = TaskResult::SingleFile { root, path: rel_path, text };
+            sender.send(VfsTask(res)).unwrap();
         }
     }
 }
diff --git a/crates/ra_vfs/src/lib.rs b/crates/ra_vfs/src/lib.rs
index 7f555a3c0..808c138df 100644
--- a/crates/ra_vfs/src/lib.rs
+++ b/crates/ra_vfs/src/lib.rs
@@ -25,7 +25,6 @@ use std::{
 };
 
 use crossbeam_channel::Receiver;
-use ra_arena::{impl_arena_id, Arena, RawId, map::ArenaMap};
 use relative_path::{RelativePath, RelativePathBuf};
 use rustc_hash::{FxHashMap, FxHashSet};
 
@@ -34,14 +33,23 @@ use crate::{
     roots::Roots,
 };
 
-pub use crate::{
-    io::TaskResult as VfsTask,
-    roots::VfsRoot,
-};
+pub use crate::roots::VfsRoot;
+
+/// Opaque wrapper around file-system event.
+///
+/// Calling code is expected to just pass `VfsTask` to `handle_task` method. It
+/// is exposed as a public API so that the caller can plug vfs events into the
+/// main event loop and be notified when changes happen.
+pub struct VfsTask(TaskResult);
+
+impl fmt::Debug for VfsTask {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        f.write_str("VfsTask { ... }")
+    }
+}
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
-pub struct VfsFile(pub RawId);
-impl_arena_id!(VfsFile);
+pub struct VfsFile(pub u32);
 
 struct VfsFileData {
     root: VfsRoot,
@@ -52,8 +60,8 @@ struct VfsFileData {
 
 pub struct Vfs {
     roots: Arc<Roots>,
-    files: Arena<VfsFile, VfsFileData>,
-    root2files: ArenaMap<VfsRoot, FxHashSet<VfsFile>>,
+    files: Vec<VfsFileData>,
+    root2files: FxHashMap<VfsRoot, FxHashSet<VfsFile>>,
     pending_changes: Vec<VfsChange>,
     worker: Worker,
 }
@@ -68,18 +76,25 @@ impl fmt::Debug for Vfs {
     }
 }
 
+#[derive(Debug, Clone)]
+pub enum VfsChange {
+    AddRoot { root: VfsRoot, files: Vec<(VfsFile, RelativePathBuf, Arc<String>)> },
+    AddFile { root: VfsRoot, file: VfsFile, path: RelativePathBuf, text: Arc<String> },
+    RemoveFile { root: VfsRoot, file: VfsFile, path: RelativePathBuf },
+    ChangeFile { file: VfsFile, text: Arc<String> },
+}
+
 impl Vfs {
     pub fn new(roots: Vec<PathBuf>) -> (Vfs, Vec<VfsRoot>) {
         let roots = Arc::new(Roots::new(roots));
         let worker = io::start(Arc::clone(&roots));
-        let mut root2files = ArenaMap::default();
+        let mut root2files = FxHashMap::default();
 
         for root in roots.iter() {
             root2files.insert(root, Default::default());
-            worker.sender().send(io::Task::AddRoot { root }).unwrap();
+            worker.sender.send(io::Task::AddRoot { root }).unwrap();
         }
-        let res =
-            Vfs { roots, files: Arena::default(), root2files, worker, pending_changes: Vec::new() };
+        let res = Vfs { roots, files: Vec::new(), root2files, worker, pending_changes: Vec::new() };
         let vfs_roots = res.roots.iter().collect();
         (res, vfs_roots)
     }
@@ -96,8 +111,8 @@ impl Vfs {
     }
 
     pub fn file2path(&self, file: VfsFile) -> PathBuf {
-        let rel_path = &self.files[file].path;
-        let root_path = &self.roots.path(self.files[file].root);
+        let rel_path = &self.file(file).path;
+        let root_path = &self.roots.path(self.file(file).root);
         rel_path.to_path(root_path)
     }
 
@@ -154,23 +169,23 @@ impl Vfs {
         mem::replace(&mut self.pending_changes, Vec::new())
     }
 
-    pub fn task_receiver(&self) -> &Receiver<io::TaskResult> {
-        self.worker.receiver()
+    pub fn task_receiver(&self) -> &Receiver<VfsTask> {
+        &self.worker.receiver
     }
 
-    pub fn handle_task(&mut self, task: io::TaskResult) {
-        match task {
+    pub fn handle_task(&mut self, task: VfsTask) {
+        match task.0 {
             TaskResult::BulkLoadRoot { root, files } => {
                 let mut cur_files = Vec::new();
                 // While we were scanning the root in the background, a file might have
                 // been open in the editor, so we need to account for that.
-                let existing = self.root2files[root]
+                let existing = self.root2files[&root]
                     .iter()
-                    .map(|&file| (self.files[file].path.clone(), file))
+                    .map(|&file| (self.file(file).path.clone(), file))
                     .collect::<FxHashMap<_, _>>();
                 for (path, text) in files {
                     if let Some(&file) = existing.get(&path) {
-                        let text = Arc::clone(&self.files[file].text);
+                        let text = Arc::clone(&self.file(file).text);
                         cur_files.push((file, path, text));
                         continue;
                     }
@@ -184,7 +199,7 @@ impl Vfs {
             }
             TaskResult::SingleFile { root, path, text } => {
                 let existing_file = self.find_file(root, &path);
-                if existing_file.map(|file| self.files[file].is_overlayed) == Some(true) {
+                if existing_file.map(|file| self.file(file).is_overlayed) == Some(true) {
                     return;
                 }
                 match (existing_file, text) {
@@ -240,23 +255,24 @@ impl Vfs {
         is_overlayed: bool,
     ) -> VfsFile {
         let data = VfsFileData { root, path, text, is_overlayed };
-        let file = self.files.alloc(data);
-        self.root2files.get_mut(root).unwrap().insert(file);
+        let file = VfsFile(self.files.len() as u32);
+        self.files.push(data);
+        self.root2files.get_mut(&root).unwrap().insert(file);
         file
     }
 
     fn raw_change_file(&mut self, file: VfsFile, new_text: Arc<String>, is_overlayed: bool) {
-        let mut file_data = &mut self.files[file];
+        let mut file_data = &mut self.file_mut(file);
         file_data.text = new_text;
         file_data.is_overlayed = is_overlayed;
     }
 
     fn raw_remove_file(&mut self, file: VfsFile) {
         // FIXME: use arena with removal
-        self.files[file].text = Default::default();
-        self.files[file].path = Default::default();
-        let root = self.files[file].root;
-        let removed = self.root2files.get_mut(root).unwrap().remove(&file);
+        self.file_mut(file).text = Default::default();
+        self.file_mut(file).path = Default::default();
+        let root = self.file(file).root;
+        let removed = self.root2files.get_mut(&root).unwrap().remove(&file);
         assert!(removed);
     }
 
@@ -267,14 +283,14 @@ impl Vfs {
     }
 
     fn find_file(&self, root: VfsRoot, path: &RelativePath) -> Option<VfsFile> {
-        self.root2files[root].iter().map(|&it| it).find(|&file| self.files[file].path == path)
+        self.root2files[&root].iter().map(|&it| it).find(|&file| self.file(file).path == path)
     }
-}
 
-#[derive(Debug, Clone)]
-pub enum VfsChange {
-    AddRoot { root: VfsRoot, files: Vec<(VfsFile, RelativePathBuf, Arc<String>)> },
-    AddFile { root: VfsRoot, file: VfsFile, path: RelativePathBuf, text: Arc<String> },
-    RemoveFile { root: VfsRoot, file: VfsFile, path: RelativePathBuf },
-    ChangeFile { file: VfsFile, text: Arc<String> },
+    fn file(&self, file: VfsFile) -> &VfsFileData {
+        &self.files[file.0 as usize]
+    }
+
+    fn file_mut(&mut self, file: VfsFile) -> &mut VfsFileData {
+        &mut self.files[file.0 as usize]
+    }
 }
diff --git a/crates/ra_vfs/src/roots.rs b/crates/ra_vfs/src/roots.rs
index 5e2776a35..4503458ee 100644
--- a/crates/ra_vfs/src/roots.rs
+++ b/crates/ra_vfs/src/roots.rs
@@ -1,16 +1,13 @@
 use std::{
     iter,
-    sync::Arc,
     path::{Path, PathBuf},
 };
 
 use relative_path::{ RelativePath, RelativePathBuf};
-use ra_arena::{impl_arena_id, Arena, RawId};
 
 /// VfsRoot identifies a watched directory on the file system.
 #[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
-pub struct VfsRoot(pub RawId);
-impl_arena_id!(VfsRoot);
+pub struct VfsRoot(pub u32);
 
 /// Describes the contents of a single source root.
 ///
@@ -25,12 +22,12 @@ struct RootData {
 }
 
 pub(crate) struct Roots {
-    roots: Arena<VfsRoot, Arc<RootData>>,
+    roots: Vec<RootData>,
 }
 
 impl Roots {
     pub(crate) fn new(mut paths: Vec<PathBuf>) -> Roots {
-        let mut roots = Arena::default();
+        let mut roots = Vec::new();
         // A hack to make nesting work.
         paths.sort_by_key(|it| std::cmp::Reverse(it.as_os_str().len()));
         paths.dedup();
@@ -38,9 +35,7 @@ impl Roots {
             let nested_roots =
                 paths[..i].iter().filter_map(|it| rel_path(path, it)).collect::<Vec<_>>();
 
-            let config = Arc::new(RootData::new(path.clone(), nested_roots));
-
-            roots.alloc(config.clone());
+            roots.push(RootData::new(path.clone(), nested_roots));
         }
         Roots { roots }
     }
@@ -54,20 +49,24 @@ impl Roots {
         self.roots.len()
     }
     pub(crate) fn iter<'a>(&'a self) -> impl Iterator<Item = VfsRoot> + 'a {
-        self.roots.iter().map(|(id, _)| id)
+        (0..self.roots.len()).into_iter().map(|idx| VfsRoot(idx as u32))
     }
     pub(crate) fn path(&self, root: VfsRoot) -> &Path {
-        self.roots[root].path.as_path()
+        self.root(root).path.as_path()
     }
     /// Checks if root contains a path and returns a root-relative path.
     pub(crate) fn contains(&self, root: VfsRoot, path: &Path) -> Option<RelativePathBuf> {
-        let data = &self.roots[root];
+        let data = self.root(root);
         iter::once(&data.path)
             .chain(data.canonical_path.as_ref().into_iter())
             .find_map(|base| rel_path(base, path))
             .filter(|path| !data.excluded_dirs.contains(path))
             .filter(|path| !data.is_excluded(path))
     }
+
+    fn root(&self, root: VfsRoot) -> &RootData {
+        &self.roots[root.0 as usize]
+    }
 }
 
 impl RootData {