1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
|
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};
use crate::{RootConfig, Roots, VfsRoot};
pub(crate) enum Task {
AddRoot { root: VfsRoot, config: Arc<RootConfig> },
}
/// `TaskResult` transfers files read on the IO thread to the VFS on the main
/// thread.
#[derive(Debug)]
pub enum TaskResult {
/// Emitted when we've recursively scanned a source root during the initial
/// load.
BulkLoadRoot { root: VfsRoot, files: Vec<(RelativePathBuf, String)> },
/// Emitted when we've noticed that a single file has changed.
///
/// Note that this by design does not distinguish between
/// create/delete/write events, and instead specifies the *current* state of
/// the file. The idea is to guarantee that in the quiescent state the sum
/// of all results equals to the current state of the file system, while
/// allowing to skip intermediate events in non-quiescent states.
SingleFile { root: VfsRoot, path: RelativePathBuf, text: Option<String> },
}
/// The kind of raw notification we've received from the notify library.
///
/// Note that these are not necessary 100% precise (for example we might receive
/// `Create` instead of `Write`, see #734), but we try do distinguish `Create`s
/// to implement recursive watching of directories.
#[derive(Debug)]
enum ChangeKind {
Create,
Write,
Remove,
}
const WATCHER_DELAY: Duration = Duration::from_millis(250);
pub(crate) struct Worker {
worker: thread_worker::Worker<Task, TaskResult>,
worker_handle: WorkerHandle,
}
impl Worker {
pub(crate) fn start(roots: Arc<Roots>) -> Worker {
// This is a pretty elaborate setup of threads & channels! It is
// explained by the following concerns:
// * we need to burn a thread translating from notify's mpsc to
// crossbeam_channel.
// * 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(
"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();
// These are the corresponding crossbeam channels
let (watcher_sender, watcher_receiver) = unbounded();
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))
});
// 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);
}
},
}
}
// 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 }
}
pub(crate) fn sender(&self) -> &Sender<Task> {
&self.worker.inp
}
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()
}
}
fn watch_root(
watcher: Option<&mut RecommendedWatcher>,
sender: &Sender<TaskResult>,
root: VfsRoot,
config: Arc<RootConfig>,
) {
log::debug!("loading {} ...", config.root.as_path().display());
let files = watch_recursive(watcher, config.root.as_path(), &*config)
.into_iter()
.filter_map(|path| {
let abs_path = path.to_path(&config.root);
let text = read_to_string(&abs_path)?;
Some((path, text))
})
.collect();
sender.send(TaskResult::BulkLoadRoot { root, files }).unwrap();
log::debug!("... loaded {}", config.root.as_path().display());
}
fn convert_notify_event(event: DebouncedEvent, sender: &Sender<(PathBuf, ChangeKind)>) {
// forward relevant events only
match event {
DebouncedEvent::NoticeWrite(_)
| DebouncedEvent::NoticeRemove(_)
| DebouncedEvent::Chmod(_) => {
// ignore
}
DebouncedEvent::Rescan => {
// TODO rescan all roots
}
DebouncedEvent::Create(path) => {
sender.send((path, ChangeKind::Create)).unwrap();
}
DebouncedEvent::Write(path) => {
sender.send((path, ChangeKind::Write)).unwrap();
}
DebouncedEvent::Remove(path) => {
sender.send((path, ChangeKind::Remove)).unwrap();
}
DebouncedEvent::Rename(src, dst) => {
sender.send((src, ChangeKind::Remove)).unwrap();
sender.send((dst, ChangeKind::Create)).unwrap();
}
DebouncedEvent::Error(err, path) => {
// TODO should we reload the file contents?
log::warn!("watcher error \"{}\", {:?}", err, path);
}
}
}
fn handle_change(
watcher: Option<&mut RecommendedWatcher>,
sender: &Sender<TaskResult>,
roots: &Roots,
path: PathBuf,
kind: ChangeKind,
) {
let (root, rel_path) = match roots.find(&path) {
None => return,
Some(it) => it,
};
let config = &roots[root];
match kind {
ChangeKind::Create => {
let mut paths = Vec::new();
if path.is_dir() {
paths.extend(watch_recursive(watcher, &path, &config));
} else {
paths.push(rel_path);
}
paths
.into_iter()
.try_for_each(|rel_path| {
let abs_path = rel_path.to_path(&config.root);
let text = read_to_string(&abs_path);
sender.send(TaskResult::SingleFile { root, path: rel_path, text })
})
.unwrap()
}
ChangeKind::Write | ChangeKind::Remove => {
let text = read_to_string(&path);
sender.send(TaskResult::SingleFile { root, path: rel_path, text }).unwrap();
}
}
}
fn watch_recursive(
mut watcher: Option<&mut RecommendedWatcher>,
dir: &Path,
config: &RootConfig,
) -> Vec<RelativePathBuf> {
let mut files = Vec::new();
for entry in WalkDir::new(dir)
.into_iter()
.filter_entry(|it| config.contains(it.path()).is_some())
.filter_map(|it| it.map_err(|e| log::warn!("watcher error: {}", e)).ok())
{
if entry.file_type().is_dir() {
if let Some(watcher) = &mut watcher {
watch_one(watcher, entry.path());
}
} else {
let path = config.contains(entry.path()).unwrap();
files.push(path.to_owned());
}
}
files
}
fn watch_one(watcher: &mut RecommendedWatcher, dir: &Path) {
match watcher.watch(dir, RecursiveMode::NonRecursive) {
Ok(()) => log::debug!("watching \"{}\"", dir.display()),
Err(e) => log::warn!("could not watch \"{}\": {}", dir.display(), e),
}
}
fn read_to_string(path: &Path) -> Option<String> {
fs::read_to_string(&path).map_err(|e| log::warn!("failed to read file {}", e)).ok()
}
|