yenru0/zed
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fs: Dispatch watcher events from the reader thread to avoid thrashing (#59537)

Browse Source 
On large worktrees, normal fs activity (like a `git checkout`) could
send the file watcher in to a rescan thrash: we processed each event on
the thread that reads events from the OS, and each event was checked
against every watched directory. On big projects this was too slow to
keep up, so the OS event buffer overflowed and dropped events, and a
dropped event forces a full rescan - creating this unbounded loop.

This PR moves dispatch onto a dedicated `fs-watcher-dispatch` thread.
The reading thread now only drops `Access` events and forwards the rest,
so it should stay fast enough to keep draining the kernel queue and
bursty fs behaviour won't cause an overflow.

In addition, we now index registrations by watched path. Instead of
waking up every watched folder whenever any file changes anywhere, we
now only wake up the folders that actually contain the changed file.

Closes FR-31. Related to #57042.

Release Notes:

- Improved file watcher performance on large worktrees
This commit is contained in:
Neel authored and GitHub committed 2026-06-18 11:17:06 +00:00
1 parent 35e2ef8af5
commit a0e37126e9
1 file changed
+255 -94
Download Patch File Download Diff File Expand all files Collapse all files
crates/fs/src/fs_watcher.rs
+255 -94
View File
@@ -521,19 +521,19 @@ pub struct WatcherRegistrationId(u32);
struct WatcherRegistrationState {
callback: Arc<dyn Fn(&notify::Event) + Send + Sync>,
path: Arc<std::path::Path>,
path: Arc<SanitizedPath>,
mode: WatcherMode,
}
struct PathRegistrationState {
count: u32,
watcher_ids: Vec<WatcherRegistrationId>,
has_os_watcher: bool,
}
struct WatcherState {
watchers: HashMap<WatcherRegistrationId, WatcherRegistrationState>,
native_path_registrations: HashMap<Arc<std::path::Path>, PathRegistrationState>,
poll_path_registrations: HashMap<Arc<std::path::Path>, PathRegistrationState>,
native_path_registrations: HashMap<Arc<SanitizedPath>, PathRegistrationState>,
poll_path_registrations: HashMap<Arc<SanitizedPath>, PathRegistrationState>,
cooldown_until: Option<Instant>,
last_registration: WatcherRegistrationId,
}
@@ -547,7 +547,7 @@ impl WatcherState {
fn path_registrations(
&mut self,
mode: WatcherMode,
) -> &mut HashMap<Arc<std::path::Path>, PathRegistrationState> {
) -> &mut HashMap<Arc<SanitizedPath>, PathRegistrationState> {
match mode {
WatcherMode::Native => &mut self.native_path_registrations,
WatcherMode::Poll => &mut self.poll_path_registrations,
@@ -557,16 +557,16 @@ impl WatcherState {
fn remove_registration(
&mut self,
id: WatcherRegistrationId,
) -> Option<(Arc<std::path::Path>, WatcherMode)> {
) -> Option<(Arc<SanitizedPath>, WatcherMode)> {
let registration_state = self.watchers.remove(&id)?;
let path_registrations = self.path_registrations(registration_state.mode);
let count = path_registrations.get_mut(&registration_state.path)?;
count.count -= 1;
if count.count != 0 {
let path_state = path_registrations.get_mut(&registration_state.path)?;
path_state.watcher_ids.retain(|&existing| existing != id);
if !path_state.watcher_ids.is_empty() {
return None;
}
let was_actually_watched = count.has_os_watcher;
let was_actually_watched = path_state.has_os_watcher;
path_registrations.remove(&registration_state.path);
was_actually_watched.then_some((registration_state.path, registration_state.mode))
@@ -588,6 +588,8 @@ impl<T: notify::Watcher + Send> WatchBackend for T {
}
}
type DispatchEvent = (WatcherMode, Result<notify::Event, notify::Error>);
pub struct GlobalWatcher {
state: Mutex<WatcherState>,
@@ -595,6 +597,7 @@ pub struct GlobalWatcher {
// two mutexes because calling watcher.add triggers watcher.event, which needs watchers.
native_watcher: Mutex<Option<Box<dyn WatchBackend>>>,
poll_watcher: Mutex<Option<Box<dyn WatchBackend>>>,
event_tx: async_channel::Sender<DispatchEvent>,
}
impl GlobalWatcher {
@@ -605,6 +608,7 @@ impl GlobalWatcher {
mode: WatcherMode,
cb: impl Fn(&notify::Event) + Send + Sync + 'static,
) -> anyhow::Result<Option<WatcherRegistrationId>> {
let path = SanitizedPath::from_arc(path);
let mut state = self.state.lock();
let (path_already_covered, path_already_registered) = {
let registrations_for_mode = state.path_registrations(mode);
@@ -620,10 +624,10 @@ impl GlobalWatcher {
}
drop(state);
match self.watch(&path, mode) {
match self.watch(path.as_path(), mode) {
Ok(()) => {}
Err(error) if mode == WatcherMode::Native && is_max_files_watch_error(&error) => {
self.start_native_watch_limit_cooldown(&path);
self.start_native_watch_limit_cooldown(path.as_path());
return Ok(None);
}
Err(error) => return Err(error),
@@ -643,15 +647,85 @@ impl GlobalWatcher {
state
.path_registrations(mode)
.entry(path)
.and_modify(|registration| registration.count += 1)
.or_insert(PathRegistrationState {
count: 1,
.and_modify(|registration| registration.watcher_ids.push(id))
.or_insert_with(|| PathRegistrationState {
watcher_ids: vec![id],
has_os_watcher: !path_already_covered,
});
Ok(Some(id))
}
fn enqueue(&self, mode: WatcherMode, event: Result<notify::Event, notify::Error>) {
if matches!(
event,
Ok(Event {
kind: EventKind::Access(_),
..
})
) {
return;
}
// A failed send only happens once the dispatch thread has shut down, at
// which point there's nothing left to dispatch to.
self.event_tx.try_send((mode, event)).ok();
}
fn dispatch(&self, mode: WatcherMode, event: Result<notify::Event, notify::Error>) {
let event = match event {
Ok(event) => event,
Err(error) => {
log::warn!("watcher error for {mode:?}: {error}");
return;
}
};
log::trace!("global handle event for {mode:?}: {event:?}");
let callbacks = {
let state = self.state.lock();
if event.need_rescan() {
let callbacks = state
.watchers
.values()
.filter(|registration| registration.mode == mode)
.map(|registration| registration.callback.clone())
.collect::<Vec<_>>();
log::warn!(
"filesystem watcher lost sync for {mode:?}; scheduling rescans for {} registrations",
callbacks.len()
);
callbacks
} else {
let path_registrations = match mode {
WatcherMode::Native => &state.native_path_registrations,
WatcherMode::Poll => &state.poll_path_registrations,
};
let mut ids = Vec::new();
for path in &event.paths {
let sanitized = SanitizedPath::new(path);
for ancestor in sanitized.as_path().ancestors() {
let ancestor = SanitizedPath::unchecked_new(ancestor);
if let Some(registration) = path_registrations.get(ancestor) {
ids.extend_from_slice(&registration.watcher_ids);
}
}
}
ids.sort_unstable_by_key(|id| id.0);
ids.dedup();
ids.into_iter()
.filter_map(|id| state.watchers.get(&id))
.map(|registration| registration.callback.clone())
.collect::<Vec<_>>()
}
};
for callback in callbacks {
callback(&event);
}
}
fn start_native_watch_limit_cooldown(&self, path: &Path) {
let mut state = self.state.lock();
let now = Instant::now();
@@ -671,7 +745,7 @@ impl GlobalWatcher {
return;
};
drop(state);
self.unwatch(&path, mode).log_err();
self.unwatch(path.as_path(), mode).log_err();
}
fn watch(&self, path: &Path, mode: WatcherMode) -> anyhow::Result<()> {
@@ -705,20 +779,28 @@ impl GlobalWatcher {
}
fn unwatch(&self, path: &Path, mode: WatcherMode) -> anyhow::Result<()> {
match mode {
WatcherMode::Native => {
if let Some(watcher) = self.native_watcher.lock().as_mut() {
watcher.unwatch(path)?;
}
}
WatcherMode::Poll => {
if let Some(watcher) = self.poll_watcher.lock().as_mut() {
watcher.unwatch(path)?;
}
}
}
let watcher = match mode {
WatcherMode::Native => self
.native_watcher
.lock()
.as_mut()
.map(|watcher| watcher.unwatch(path)),
WatcherMode::Poll => self
.poll_watcher
.lock()
.as_mut()
.map(|watcher| watcher.unwatch(path)),
};
Ok(())
match watcher {
// inotify auto-removes a watch when its directory is deleted, so a
// later unwatch races that and fails with a benign error. Either way
// the path is no longer watched, which is all we wanted.
Some(Err(error)) if !matches!(error.kind, notify::ErrorKind::WatchNotFound) => {
Err(error.into())
}
_ => Ok(()),
}
}
fn ensure_native_watcher(&self) -> anyhow::Result<()> {
@@ -732,8 +814,10 @@ impl GlobalWatcher {
// risk of queue overflows (and thus full rescans) under read-heavy
// workloads like grep or language server indexing.
let config = notify::Config::default().with_event_kinds(notify::EventKindMask::CORE);
let watcher =
<notify::RecommendedWatcher as notify::Watcher>::new(handle_native_event, config)?;
let watcher = <notify::RecommendedWatcher as notify::Watcher>::new(
|event| global_watcher().enqueue(WatcherMode::Native, event),
config,
)?;
*self.native_watcher.lock() = Some(Box::new(watcher));
Ok(())
}
@@ -744,22 +828,26 @@ impl GlobalWatcher {
}
let config = notify::Config::default().with_poll_interval(*POLL_INTERVAL);
let watcher = notify::PollWatcher::new(handle_poll_event, config)?;
let watcher = notify::PollWatcher::new(
|event| global_watcher().enqueue(WatcherMode::Poll, event),
config,
)?;
*self.poll_watcher.lock() = Some(Box::new(watcher));
Ok(())
}
}
fn path_already_covered(
path: &Path,
path_registrations: &HashMap<Arc<std::path::Path>, PathRegistrationState>,
path: &SanitizedPath,
path_registrations: &HashMap<Arc<SanitizedPath>, PathRegistrationState>,
mode: WatcherMode,
) -> bool {
(mode == WatcherMode::Poll || cfg!(any(target_os = "windows", target_os = "macos")))
&& path
.as_path()
.ancestors()
.skip(1)
.any(|ancestor| path_registrations.contains_key(ancestor))
.any(|ancestor| path_registrations.contains_key(SanitizedPath::unchecked_new(ancestor)))
}
fn is_max_files_watch_error(error: &anyhow::Error) -> bool {
@@ -793,68 +881,31 @@ pub fn poll_interval() -> Duration {
static FS_WATCHER_INSTANCE: OnceLock<GlobalWatcher> = OnceLock::new();
fn global_watcher() -> &'static GlobalWatcher {
FS_WATCHER_INSTANCE.get_or_init(|| GlobalWatcher {
state: Mutex::new(WatcherState {
watchers: Default::default(),
native_path_registrations: Default::default(),
poll_path_registrations: Default::default(),
cooldown_until: None,
last_registration: Default::default(),
}),
native_watcher: Mutex::new(None),
poll_watcher: Mutex::new(None),
FS_WATCHER_INSTANCE.get_or_init(|| {
let (event_tx, event_rx) = async_channel::unbounded::<DispatchEvent>();
std::thread::Builder::new()
.name("fs-watcher-dispatch".to_owned())
.spawn(move || {
while let Ok((mode, event)) = event_rx.recv_blocking() {
global_watcher().dispatch(mode, event);
}
})
.expect("failed to spawn fs watcher dispatch thread");
GlobalWatcher {
state: Mutex::new(WatcherState {
watchers: Default::default(),
native_path_registrations: Default::default(),
poll_path_registrations: Default::default(),
cooldown_until: None,
last_registration: Default::default(),
}),
native_watcher: Mutex::new(None),
poll_watcher: Mutex::new(None),
event_tx,
}
})
}
fn handle_native_event(event: Result<notify::Event, notify::Error>) {
handle_event(WatcherMode::Native, event);
}
fn handle_poll_event(event: Result<notify::Event, notify::Error>) {
handle_event(WatcherMode::Poll, event);
}
fn handle_event(mode: WatcherMode, event: Result<notify::Event, notify::Error>) {
if matches!(
event,
Ok(Event {
kind: EventKind::Access(_),
..
})
) {
return;
}
log::trace!("global handle event for {mode:?}: {event:?}");
let callbacks = {
let state = global_watcher().state.lock();
state
.watchers
.values()
.filter(|registration| registration.mode == mode)
.map(|registration| registration.callback.clone())
.collect::<Vec<_>>()
};
match event {
Ok(event) => {
if event.need_rescan() {
log::warn!(
"filesystem watcher lost sync for {mode:?}; scheduling rescans for {} registrations",
callbacks.len()
);
}
for callback in callbacks {
callback(&event);
}
}
Err(error) => {
log::warn!("watcher error for {mode:?}: {error}");
}
}
}
#[cfg(test)]
mod tests {
use super::*;
@@ -916,6 +967,10 @@ mod tests {
native_watcher: Option<Arc<Mutex<FakeWatchBackend>>>,
poll_watcher: Option<Arc<Mutex<FakeWatchBackend>>>,
) -> GlobalWatcher {
// Tests call `handle_event` directly to exercise dispatch synchronously,
// rather than going through the OS watcher callbacks, so nothing is ever
// sent on this channel; the receiver can just be dropped.
let (event_tx, _event_rx) = async_channel::unbounded();
GlobalWatcher {
state: Mutex::new(WatcherState {
watchers: Default::default(),
@@ -934,6 +989,7 @@ mod tests {
Box::new(SharedFakeWatchBackend(watcher)) as Box<dyn WatchBackend>
}),
),
event_tx,
}
}
@@ -994,6 +1050,111 @@ mod tests {
assert_eq!(native_backend.watch_calls, &[first_path.to_path_buf()]);
}
fn modify_event(path: &str) -> notify::Event {
notify::Event {
paths: vec![PathBuf::from(path)],
..notify::Event::new(EventKind::Modify(notify::event::ModifyKind::Any))
}
}
fn recording_watcher() -> (GlobalWatcher, Arc<Mutex<Vec<String>>>) {
let backend = Arc::new(Mutex::new(FakeWatchBackend::default()));
let watcher = test_watcher_with_backends(Some(backend), None);
let fired = Arc::new(Mutex::new(Vec::new()));
for dir in ["/repo/a", "/repo/a/nested", "/repo/b"] {
let fired = fired.clone();
let label = dir.to_owned();
watcher
.add(
Arc::<Path>::from(Path::new(dir)),
WatcherMode::Native,
move |_| {
fired.lock().push(label.clone());
},
)
.expect("add watch")
.expect("watch registered");
}
(watcher, fired)
}
#[test]
fn event_dispatches_only_to_registrations_covering_its_path() {
let (watcher, fired) = recording_watcher();
watcher.dispatch(WatcherMode::Native, Ok(modify_event("/repo/a/file.txt")));
// Only the directory containing the file resolves; siblings stay untouched.
assert_eq!(*fired.lock(), vec!["/repo/a".to_owned()]);
}
#[test]
fn event_dispatches_to_every_ancestor_registration() {
let (watcher, fired) = recording_watcher();
watcher.dispatch(
WatcherMode::Native,
Ok(modify_event("/repo/a/nested/file.txt")),
);
// Both the directory containing the file and the ancestor watching it
// resync, each exactly once, matching the previous broadcast-and-filter
// behavior.
let mut got = fired.lock().clone();
got.sort();
assert_eq!(got, vec!["/repo/a".to_owned(), "/repo/a/nested".to_owned()]);
}
#[test]
fn rescan_event_broadcasts_to_all_registrations_of_the_same_mode() {
let (watcher, fired) = recording_watcher();
let rescan = notify::Event::new(EventKind::Other).set_flag(notify::event::Flag::Rescan);
watcher.dispatch(WatcherMode::Native, Ok(rescan));
// A pathless rescan may have missed events anywhere, so every registration
// of that mode resyncs, regardless of which directory it watches.
let mut got = fired.lock().clone();
got.sort();
assert_eq!(
got,
vec![
"/repo/a".to_owned(),
"/repo/a/nested".to_owned(),
"/repo/b".to_owned(),
]
);
}
#[test]
#[cfg(target_os = "windows")]
fn event_dispatches_when_reported_path_has_verbatim_prefix() {
// `notify` (and Windows APIs generally) can report a changed path using the
// verbatim `\\?\` long-path form even when the directory was registered
// without it.
let backend = Arc::new(Mutex::new(FakeWatchBackend::default()));
let watcher = test_watcher_with_backends(Some(backend), None);
let fired = Arc::new(Mutex::new(Vec::new()));
{
let fired = fired.clone();
watcher
.add(
Arc::<Path>::from(Path::new("C:\\repo\\src")),
WatcherMode::Native,
move |_| fired.lock().push(()),
)
.expect("add watch")
.expect("watch registered");
}
watcher.dispatch(
WatcherMode::Native,
Ok(modify_event("\\\\?\\C:\\repo\\src\\main.rs")),
);
assert_eq!(fired.lock().len(), 1);
}
#[test]
fn test_coalesce_pending_rescans() {
let test_cases = [