From a0e37126e9456da0f6855cbfb202fa4efead6fec Mon Sep 17 00:00:00 2001 From: Neel Date: Thu, 18 Jun 2026 12:17:06 +0100 Subject: [PATCH] fs: Dispatch watcher events from the reader thread to avoid thrashing (#59537) 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 --- crates/fs/src/fs_watcher.rs | 349 ++++++++++++++++++++++++++---------- 1 file changed, 255 insertions(+), 94 deletions(-) diff --git a/crates/fs/src/fs_watcher.rs b/crates/fs/src/fs_watcher.rs index 82aeaf2a73..98184da38e 100644 --- a/crates/fs/src/fs_watcher.rs +++ b/crates/fs/src/fs_watcher.rs @@ -521,19 +521,19 @@ pub struct WatcherRegistrationId(u32); struct WatcherRegistrationState { callback: Arc, - path: Arc, + path: Arc, mode: WatcherMode, } struct PathRegistrationState { - count: u32, + watcher_ids: Vec, has_os_watcher: bool, } struct WatcherState { watchers: HashMap, - native_path_registrations: HashMap, PathRegistrationState>, - poll_path_registrations: HashMap, PathRegistrationState>, + native_path_registrations: HashMap, PathRegistrationState>, + poll_path_registrations: HashMap, PathRegistrationState>, cooldown_until: Option, last_registration: WatcherRegistrationId, } @@ -547,7 +547,7 @@ impl WatcherState { fn path_registrations( &mut self, mode: WatcherMode, - ) -> &mut HashMap, PathRegistrationState> { + ) -> &mut HashMap, 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, WatcherMode)> { + ) -> Option<(Arc, WatcherMode)> { let registration_state = self.watchers.remove(&id)?; let path_registrations = self.path_registrations(registration_state.mode); - let count = path_registrations.get_mut(®istration_state.path)?; - count.count -= 1; - if count.count != 0 { + let path_state = path_registrations.get_mut(®istration_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(®istration_state.path); was_actually_watched.then_some((registration_state.path, registration_state.mode)) @@ -588,6 +588,8 @@ impl WatchBackend for T { } } +type DispatchEvent = (WatcherMode, Result); + pub struct GlobalWatcher { state: Mutex, @@ -595,6 +597,7 @@ pub struct GlobalWatcher { // two mutexes because calling watcher.add triggers watcher.event, which needs watchers. native_watcher: Mutex>>, poll_watcher: Mutex>>, + event_tx: async_channel::Sender, } impl GlobalWatcher { @@ -605,6 +608,7 @@ impl GlobalWatcher { mode: WatcherMode, cb: impl Fn(¬ify::Event) + Send + Sync + 'static, ) -> anyhow::Result> { + 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) { + 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) { + 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::>(); + 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(®istration.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::>() + } + }; + + 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 = - ::new(handle_native_event, config)?; + let 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, PathRegistrationState>, + path: &SanitizedPath, + path_registrations: &HashMap, 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 = 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::(); + 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) { - handle_event(WatcherMode::Native, event); -} - -fn handle_poll_event(event: Result) { - handle_event(WatcherMode::Poll, event); -} - -fn handle_event(mode: WatcherMode, event: Result) { - 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::>() - }; - - 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>>, poll_watcher: Option>>, ) -> 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 }), ), + 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>>) { + 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::::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::::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 = [