mikros/rust
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge #856

Browse Source 
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 <aleksey.kladov@gmail.com>
This commit is contained in:
bors[bot] 2019-02-18 13:21:01 +00:00
commit a591c3460b
5 changed files with 176 additions and 124 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
Cargo.lock generated
View File

@ -1123,16 +1123,13 @@ name = "ra_vfs"
version = "0.1.0"
dependencies = [
"crossbeam-channel 0.3.8 (registry+https://github.com/rust-lang/crates.io-index)",
"drop_bomb 0.1.4 (registry+https://github.com/rust-lang/crates.io-index)",
"flexi_logger 0.10.5 (registry+https://github.com/rust-lang/crates.io-index)",
"log 0.4.6 (registry+https://github.com/rust-lang/crates.io-index)",
"notify 4.0.9 (registry+https://github.com/rust-lang/crates.io-index)",
"parking_lot 0.7.1 (registry+https://github.com/rust-lang/crates.io-index)",
"ra_arena 0.1.0",
"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)",
"tempfile 3.0.7 (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)",
]

4
crates/ra_vfs/Cargo.toml
View File

@ -11,12 +11,8 @@ rustc-hash = "1.0"
crossbeam-channel = "0.3.5"
log = "0.4.6"
notify = "4.0.9"
drop_bomb = "0.1.0"
parking_lot = "0.7.0"
thread_worker = { path = "../thread_worker" }
ra_arena = { path = "../ra_arena" }
[dev-dependencies]
tempfile = "3"
flexi_logger = "0.10.0"

176
crates/ra_vfs/src/io.rs
View File

@ -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.
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();
// 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();
_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.
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))
});
// 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();
// 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);
}
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
},
// 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 }) => {
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();
}
}
}

94
crates/ra_vfs/src/lib.rs
View File

@ -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)
}
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]
}
}
#[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> },
}

23
crates/ra_vfs/src/roots.rs
View File

@ -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 {