vfs crate scaffold

Cargo.lock

@@ -749,8 +749,10 @@ name = "ra_vfs"
 version = "0.1.0"
 dependencies = [
  "crossbeam-channel 0.2.6 (registry+https://github.com/rust-lang/crates.io-index)",
+ "log 0.4.6 (registry+https://github.com/rust-lang/crates.io-index)",
  "relative-path 0.4.0 (registry+https://github.com/rust-lang/crates.io-index)",
  "rustc-hash 1.0.1 (registry+https://github.com/rust-lang/crates.io-index)",
+ "thread_worker 0.1.0",
  "walkdir 2.2.7 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
 

crates/ra_lsp_server/src/main_loop.rs

@@ -88,8 +88,8 @@ pub fn main_loop(
     drop(pool);
     log::info!("...threadpool has finished");
 
-    let fs_res = fs_watcher.stop();
-    let ws_res = ws_watcher.stop();
+    let fs_res = fs_watcher.shutdown();
+    let ws_res = ws_watcher.shutdown();
 
     main_res?;
     fs_res.map_err(|_| format_err!("fs watcher died"))?;

crates/ra_vfs/Cargo.toml

@@ -9,3 +9,6 @@ walkdir = "2.2.7"
 relative-path = "0.4.0"
 rustc-hash = "1.0"
 crossbeam-channel = "0.2.4"
+log = "0.4.6"
+
+thread_worker = { path = "../thread_worker" }

crates/ra_vfs/src/io.rs

@@ -6,67 +6,64 @@ use std::{
 
 use walkdir::WalkDir;
 use crossbeam_channel::{Sender, Receiver};
+use thread_worker::{WorkerHandle, Worker};
 
-pub(crate) fn start_io() -> (JoinHandle<(), Sender<()>, Receiver()>) {}
+#[derive(Debug)]
+pub struct FileEvent {
+    pub path: PathBuf,
+    pub kind: FileEventKind,
+}
 
-// use crate::thread_watcher::{ThreadWatcher, Worker};
+#[derive(Debug)]
+pub enum FileEventKind {
+    Add(String),
+}
 
-// #[derive(Debug)]
-// pub struct FileEvent {
-//     pub path: PathBuf,
-//     pub kind: FileEventKind,
-// }
+pub fn start() -> (Worker<PathBuf, (PathBuf, Vec<FileEvent>)>, WorkerHandle) {
+    thread_worker::spawn::<PathBuf, (PathBuf, Vec<FileEvent>), _>(
+        "vfs",
+        128,
+        |input_receiver, output_sender| {
+            input_receiver
+                .map(|path| {
+                    log::debug!("loading {} ...", path.as_path().display());
+                    let events = load_root(path.as_path());
+                    log::debug!("... loaded {}", path.as_path().display());
+                    (path, events)
+                })
+                .for_each(|it| output_sender.send(it))
+        },
+    )
+}
 
-// #[derive(Debug)]
-// pub enum FileEventKind {
-//     Add(String),
-// }
-
-// pub fn roots_loader() -> (Worker<PathBuf, (PathBuf, Vec<FileEvent>)>, ThreadWatcher) {
-//     Worker::<PathBuf, (PathBuf, Vec<FileEvent>)>::spawn(
-//         "roots loader",
-//         128,
-//         |input_receiver, output_sender| {
-//             input_receiver
-//                 .map(|path| {
-//                     log::debug!("loading {} ...", path.as_path().display());
-//                     let events = load_root(path.as_path());
-//                     log::debug!("... loaded {}", path.as_path().display());
-//                     (path, events)
-//                 })
-//                 .for_each(|it| output_sender.send(it))
-//         },
-//     )
-// }
-
-// fn load_root(path: &Path) -> Vec<FileEvent> {
-//     let mut res = Vec::new();
-//     for entry in WalkDir::new(path) {
-//         let entry = match entry {
-//             Ok(entry) => entry,
-//             Err(e) => {
-//                 log::warn!("watcher error: {}", e);
-//                 continue;
-//             }
-//         };
-//         if !entry.file_type().is_file() {
-//             continue;
-//         }
-//         let path = entry.path();
-//         if path.extension().and_then(|os| os.to_str()) != Some("rs") {
-//             continue;
-//         }
-//         let text = match fs::read_to_string(path) {
-//             Ok(text) => text,
-//             Err(e) => {
-//                 log::warn!("watcher error: {}", e);
-//                 continue;
-//             }
-//         };
-//         res.push(FileEvent {
-//             path: path.to_owned(),
-//             kind: FileEventKind::Add(text),
-//         })
-//     }
-//     res
-// }
+fn load_root(path: &Path) -> Vec<FileEvent> {
+    let mut res = Vec::new();
+    for entry in WalkDir::new(path) {
+        let entry = match entry {
+            Ok(entry) => entry,
+            Err(e) => {
+                log::warn!("watcher error: {}", e);
+                continue;
+            }
+        };
+        if !entry.file_type().is_file() {
+            continue;
+        }
+        let path = entry.path();
+        if path.extension().and_then(|os| os.to_str()) != Some("rs") {
+            continue;
+        }
+        let text = match fs::read_to_string(path) {
+            Ok(text) => text,
+            Err(e) => {
+                log::warn!("watcher error: {}", e);
+                continue;
+            }
+        };
+        res.push(FileEvent {
+            path: path.to_owned(),
+            kind: FileEventKind::Add(text),
+        })
+    }
+    res
+}

crates/ra_vfs/src/lib.rs

@@ -15,6 +15,7 @@ mod arena;
 mod io;
 
 use std::{
+    thread,
     cmp::Reverse,
     path::{Path, PathBuf},
     ffi::OsStr,
@@ -22,7 +23,12 @@ use std::{
 };
 
 use relative_path::RelativePathBuf;
-use crate::arena::{ArenaId, Arena};
+use thread_worker::{WorkerHandle, Worker};
+
+use crate::{
+    arena::{ArenaId, Arena},
+    io::FileEvent,
+};
 
 /// `RootFilter` is a predicate that checks if a file can belong to a root
 struct RootFilter {
@@ -76,16 +82,24 @@ struct VfsFileData {
     text: Arc<String>,
 }
 
-#[derive(Default)]
 struct Vfs {
     roots: Arena<VfsRoot, RootFilter>,
     files: Arena<VfsFile, VfsFileData>,
     // pending_changes: Vec<PendingChange>,
+    worker: Worker<PathBuf, (PathBuf, Vec<FileEvent>)>,
+    worker_handle: WorkerHandle,
 }
 
 impl Vfs {
     pub fn new(mut roots: Vec<PathBuf>) -> Vfs {
-        let mut res = Vfs::default();
+        let (worker, worker_handle) = io::start();
+
+        let mut res = Vfs {
+            roots: Arena::default(),
+            files: Arena::default(),
+            worker,
+            worker_handle,
+        };
 
         roots.sort_by_key(|it| Reverse(it.as_os_str().len()));
 
@@ -104,6 +118,11 @@ impl Vfs {
     pub fn commit_changes(&mut self) -> Vec<VfsChange> {
         unimplemented!()
     }
+
+    pub fn shutdown(self) -> thread::Result<()> {
+        let _ = self.worker.shutdown();
+        self.worker_handle.shutdown()
+    }
 }
 
 #[derive(Debug, Clone)]

crates/thread_worker/src/lib.rs

@@ -30,7 +30,7 @@ where
 impl<I, O> Worker<I, O> {
     /// Stops the worker. Returns the message receiver to fetch results which
     /// have become ready before the worker is stopped.
-    pub fn stop(self) -> Receiver<O> {
+    pub fn shutdown(self) -> Receiver<O> {
         self.out
     }
 
@@ -45,11 +45,11 @@ impl WorkerHandle {
         WorkerHandle {
             name,
             thread,
-            bomb: DropBomb::new(format!("WorkerHandle {} was not stopped", name)),
+            bomb: DropBomb::new(format!("WorkerHandle {} was not shutdown", name)),
         }
     }
 
-    pub fn stop(mut self) -> thread::Result<()> {
+    pub fn shutdown(mut self) -> thread::Result<()> {
         log::info!("waiting for {} to finish ...", self.name);
         let name = self.name;
         self.bomb.defuse();