diff --git a/crates/fs/src/fs.rs b/crates/fs/src/fs.rs index ad9ba139a9..548e4a87cc 100644 --- a/crates/fs/src/fs.rs +++ b/crates/fs/src/fs.rs @@ -446,7 +446,8 @@ pub struct RealFs { this: std::sync::Weak, bundled_git_binary_path: Option, executor: BackgroundExecutor, - global_watcher: Arc, + native_watcher: Arc, + poll_watcher: Arc, next_job_id: Arc, job_event_subscribers: Arc>>, trash: Arc>>, @@ -556,7 +557,14 @@ impl RealFs { Arc::new_cyclic(|this| Self { this: this.clone(), bundled_git_binary_path: git_binary_path, - global_watcher: fs_watcher::GlobalWatcher::new(executor.clone()), + native_watcher: fs_watcher::OsWatcher::new( + fs_watcher::OsWatcherKind::Native, + executor.clone(), + ), + poll_watcher: fs_watcher::OsWatcher::new( + fs_watcher::OsWatcherKind::Poll, + executor.clone(), + ), executor, next_job_id: Arc::new(AtomicUsize::new(0)), job_event_subscribers: Arc::new(Mutex::new(Vec::new())), @@ -1164,7 +1172,7 @@ impl Fs for RealFs { } fn start_native_watcher(&self) -> Result<()> { - self.global_watcher.ensure_native_watcher() + self.native_watcher.ensure_backend() } fn path_exists(&self, path: &Path) -> bool { @@ -1193,7 +1201,8 @@ impl Fs for RealFs { .expect("RealFs is only constructed inside an Arc"); fs_watcher::watch( this, - self.global_watcher.clone(), + self.native_watcher.clone(), + self.poll_watcher.clone(), self.executor.clone(), path, latency, @@ -1419,7 +1428,8 @@ pub struct FakeFs { // Use an unfair lock to ensure tests are deterministic. state: Arc>, executor: gpui::BackgroundExecutor, - global_watcher: Arc, + native_watcher: Arc, + poll_watcher: Arc, } #[cfg(feature = "test-support")] @@ -1443,7 +1453,7 @@ struct FakeFsState { /// The kernel's side of file watching, as far as the real watcher code above /// notify can tell: the paths the backend has registered and the callback that -/// delivers events for them into the `GlobalWatcher`. +/// delivers events for them into the native `OsWatcher`. #[cfg(feature = "test-support")] #[derive(Default)] struct FakeWatches { @@ -1810,19 +1820,23 @@ impl FakeFs { remove_dir_errors: Default::default(), case_sensitive: true, })); - let global_watcher = fs_watcher::GlobalWatcher::with_native_backend( + let native_watcher = fs_watcher::OsWatcher::with_backend( + fs_watcher::OsWatcherKind::Native, executor.clone(), Some(Box::new(FakeWatchBackend { state: state.clone(), })), ); - state.lock().watches.event_sink = Some(Box::new(global_watcher.native_event_sink())); + state.lock().watches.event_sink = Some(Box::new(native_watcher.event_sink())); + let poll_watcher = + fs_watcher::OsWatcher::new(fs_watcher::OsWatcherKind::Poll, executor.clone()); let this = Arc::new_cyclic(|this| Self { this: this.clone(), executor: executor.clone(), state, - global_watcher, + native_watcher, + poll_watcher, }); executor.spawn({ @@ -3331,7 +3345,8 @@ impl Fs for FakeFs { // own, so a real debounce would stall every test until `advance_clock`. let (events, watcher) = fs_watcher::watch( self.this.upgrade().unwrap(), - self.global_watcher.clone(), + self.native_watcher.clone(), + self.poll_watcher.clone(), self.executor.clone(), path, Duration::ZERO, diff --git a/crates/fs/src/fs_watcher.rs b/crates/fs/src/fs_watcher.rs index 431723fca3..8965ea5b58 100644 --- a/crates/fs/src/fs_watcher.rs +++ b/crates/fs/src/fs_watcher.rs @@ -14,16 +14,24 @@ use util::{ResultExt, paths::SanitizedPath}; use crate::{Fs, PathEvent, PathEventKind, Watcher}; -#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Hash)] -pub enum WatcherMode { - #[default] +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum OsWatcherKind { Native, Poll, } +impl OsWatcherKind { + /// Poll watches and native macOS/Windows watches cover their whole subtree; + /// native Linux (inotify) watches cover one directory. + fn is_recursive(self) -> bool { + self == Self::Poll || cfg!(any(target_os = "windows", target_os = "macos")) + } +} + pub(crate) async fn watch( fs: Arc, - global_watcher: Arc, + native_watcher: Arc, + poll_watcher: Arc, executor: BackgroundExecutor, path: &Path, latency: Duration, @@ -35,7 +43,8 @@ pub(crate) async fn watch( let pending_paths: Arc>> = Default::default(); let watcher: Arc = Arc::new(FsWatcher::new( - global_watcher, + native_watcher, + poll_watcher, fs.clone(), executor.clone(), tx, @@ -90,7 +99,8 @@ pub(crate) async fn watch( } pub struct FsWatcher { - global_watcher: Arc, + native_watcher: Arc, + poll_watcher: Arc, fs: Arc, executor: BackgroundExecutor, tx: async_channel::Sender<()>, @@ -99,22 +109,24 @@ pub struct FsWatcher { pending_registrations: Arc, Task<()>>>>, } -#[derive(Clone, Copy)] +#[derive(Clone)] struct FsWatcherRegistration { id: WatcherRegistrationId, - mode: WatcherMode, + os_watcher: Arc, } impl FsWatcher { pub(crate) fn new( - global_watcher: Arc, + native_watcher: Arc, + poll_watcher: Arc, fs: Arc, executor: BackgroundExecutor, tx: async_channel::Sender<()>, pending_path_events: Arc>>, ) -> Self { Self { - global_watcher, + native_watcher, + poll_watcher, fs, executor, tx, @@ -132,7 +144,8 @@ impl FsWatcher { return Ok(()); } match register_existing_path( - &self.global_watcher, + &self.native_watcher, + &self.poll_watcher, self.fs.as_ref(), path.clone(), case_insensitive, @@ -160,7 +173,8 @@ impl FsWatcher { } let task = self.executor.spawn(poll_path_until_created( - self.global_watcher.clone(), + self.native_watcher.clone(), + self.poll_watcher.clone(), self.fs.clone(), self.executor.clone(), path.clone(), @@ -184,7 +198,7 @@ impl Drop for FsWatcher { } for (_, registration) in registrations { - self.global_watcher.remove(registration.id); + registration.os_watcher.remove(registration.id); } } } @@ -231,7 +245,7 @@ impl Watcher for FsWatcher { .or_else(|| registrations.remove(&WatchKey::folded(sanitized))) }; if let Some(registration) = registration { - self.global_watcher.remove(registration.id); + registration.os_watcher.remove(registration.id); } Ok(()) } @@ -249,10 +263,9 @@ fn path_covered_by_recursive_registration( [WatchKey::exact(ancestor), WatchKey::folded(ancestor)] .iter() .any(|key| { - registrations.get(key).is_some_and(|registration| { - registration.mode == WatcherMode::Poll - || cfg!(any(target_os = "windows", target_os = "macos")) - }) + registrations + .get(key) + .is_some_and(|registration| registration.os_watcher.is_recursive()) }) }) } @@ -289,31 +302,29 @@ pub fn requires_poll_watcher(path: &Path) -> bool { } fn register_existing_path( - global_watcher: &GlobalWatcher, + native_watcher: &Arc, + poll_watcher: &Arc, fs: &dyn Fs, path: Arc, case_insensitive: bool, tx: async_channel::Sender<()>, pending_path_events: Arc>>, ) -> anyhow::Result> { - let mode = if fs.requires_poll_watcher(path.as_ref()) { + let os_watcher = if fs.requires_poll_watcher(path.as_ref()) { log::info!( "Using poll watcher ({}ms interval) for {}", poll_interval().as_millis(), path.display() ); telemetry::event!("fs_watcher_poll", path = path.display().to_string()); - WatcherMode::Poll + poll_watcher } else { - WatcherMode::Native + native_watcher }; let root_path = SanitizedPath::new_arc(path.as_ref()); let path_for_callback = path.clone(); - let Some(registration_id) = global_watcher.add( - path, - mode, - case_insensitive, - move |event: ¬ify::Event| { + let Some(registration_id) = + os_watcher.add(path, case_insensitive, move |event: ¬ify::Event| { log::trace!("watcher received event: {event:?}"); push_notify_event( &tx, @@ -323,14 +334,13 @@ fn register_existing_path( path_for_callback.as_ref(), event, ); - }, - )? + })? else { return Ok(None); }; Ok(Some(FsWatcherRegistration { id: registration_id, - mode, + os_watcher: os_watcher.clone(), })) } @@ -560,7 +570,8 @@ impl WatchKey { } async fn poll_path_until_created( - global_watcher: Arc, + native_watcher: Arc, + poll_watcher: Arc, fs: Arc, executor: BackgroundExecutor, path: Arc, @@ -591,7 +602,8 @@ async fn poll_path_until_created( } match register_existing_path( - &global_watcher, + &native_watcher, + &poll_watcher, fs.as_ref(), path.clone(), case_insensitive, @@ -602,7 +614,7 @@ async fn poll_path_until_created( { let mut pending_registrations = pending_registrations.lock(); if pending_registrations.remove(path.as_ref()).is_none() { - global_watcher.remove(registration.id); + registration.os_watcher.remove(registration.id); return; } registrations.lock().insert(key, registration); @@ -779,7 +791,6 @@ struct WatcherRegistrationState { callback: Arc, key: WatchKey, path: Arc, - mode: WatcherMode, } struct PathRegistrationState { @@ -787,7 +798,7 @@ struct PathRegistrationState { has_os_watcher: bool, } -/// The registered watch paths for one watcher mode, keyed by [`WatchKey`] so that +/// The registered watch paths for one backend, keyed by [`WatchKey`] so that /// add (dedup), remove, and dispatch share a single notion of path identity. #[derive(Default)] struct WatchPaths(HashMap); @@ -814,8 +825,8 @@ impl WatchPaths { /// True if a recursive registration on a strict ancestor already covers /// `path`. Only poll watches and native macOS/Windows watches are recursive. - fn covered_by_recursive_ancestor(&self, path: &SanitizedPath, mode: WatcherMode) -> bool { - if mode != WatcherMode::Poll && !cfg!(any(target_os = "windows", target_os = "macos")) { + fn covered_by_recursive_ancestor(&self, path: &SanitizedPath, recursive: bool) -> bool { + if !recursive { return false; } path.as_path().ancestors().skip(1).any(|ancestor| { @@ -843,8 +854,7 @@ impl WatchPaths { struct WatcherState { watchers: HashMap, - native_path_registrations: WatchPaths, - poll_path_registrations: WatchPaths, + paths: WatchPaths, cooldown_until: Option, last_registration: WatcherRegistrationId, } @@ -855,29 +865,18 @@ impl WatcherState { .is_some_and(|cooldown_until| cooldown_until > Instant::now()) } - fn path_registrations(&mut self, mode: WatcherMode) -> &mut WatchPaths { - match mode { - WatcherMode::Native => &mut self.native_path_registrations, - WatcherMode::Poll => &mut self.poll_path_registrations, - } - } - - fn remove_registration( - &mut self, - id: WatcherRegistrationId, - ) -> Option<(Arc, WatcherMode)> { + fn remove_registration(&mut self, id: WatcherRegistrationId) -> Option> { let registration_state = self.watchers.remove(&id)?; - let path_registrations = self.path_registrations(registration_state.mode); - let path_state = path_registrations.get_mut(®istration_state.key)?; + let path_state = self.paths.get_mut(®istration_state.key)?; path_state.watcher_ids.retain(|&existing| existing != id); if !path_state.watcher_ids.is_empty() { return None; } let was_actually_watched = path_state.has_os_watcher; - path_registrations.remove(®istration_state.key); + self.paths.remove(®istration_state.key); - was_actually_watched.then_some((registration_state.path, registration_state.mode)) + was_actually_watched.then_some(registration_state.path) } } @@ -896,37 +895,38 @@ impl WatchBackend for T { } } -type DispatchEvent = (WatcherMode, Result); - -pub struct GlobalWatcher { +pub struct OsWatcher { + kind: OsWatcherKind, + recursive: bool, state: Arc>, - // DANGER: never keep state lock while holding watcher lock - // two mutexes because calling watcher.add triggers watcher.event, which needs watchers. - native_watcher: Mutex>>, - poll_watcher: Mutex>>, - event_tx: async_channel::Sender, + // Never hold the state lock while calling the backend: a backend call can be + // slow (a poll watch scans its whole tree up front, an FSEvents watch rebuilds + // the stream) and dispatch needs the state to route every other registration's + // events. Whoever re-locks the state afterwards must re-check what they assumed. + backend: Mutex>>, + event_tx: async_channel::Sender>, _dispatch_task: Task<()>, } -impl GlobalWatcher { - /// Watches with notify's OS backends, created lazily on first use so that a - /// failure to start them can be reported by `ensure_native_watcher`. - pub fn new(executor: BackgroundExecutor) -> Arc { - Self::with_native_backend(executor, None) +impl OsWatcher { + /// Watches with notify's backend of `kind`, created lazily on first use so + /// that a failure to start it can be reported by `ensure_backend`. + pub fn new(kind: OsWatcherKind, executor: BackgroundExecutor) -> Arc { + Self::with_backend(kind, executor, None) } - /// Watches with the given backend in place of the OS one; the returned sink - /// is how that backend reports events, standing in for notify's `EventHandler`. - pub(crate) fn with_native_backend( + /// Watches with the given backend in place of notify's; `event_sink` is how + /// that backend reports events, standing in for notify's `EventHandler`. + pub(crate) fn with_backend( + kind: OsWatcherKind, executor: BackgroundExecutor, - native_watcher: Option>, + backend: Option>, ) -> Arc { - let (event_tx, event_rx) = async_channel::unbounded::(); + let (event_tx, event_rx) = async_channel::unbounded(); let state = Arc::new(Mutex::new(WatcherState { watchers: Default::default(), - native_path_registrations: Default::default(), - poll_path_registrations: Default::default(), + paths: Default::default(), cooldown_until: None, last_registration: Default::default(), })); @@ -934,56 +934,59 @@ impl GlobalWatcher { let state = state.clone(); async move { while let Ok(first) = event_rx.recv().await { - dispatch_batch(&state, first, &event_rx); + dispatch_batch(kind, &state, first, &event_rx); } } }); Arc::new(Self { + kind, + recursive: kind.is_recursive(), state, - native_watcher: Mutex::new(native_watcher), - poll_watcher: Mutex::new(None), + backend: Mutex::new(backend), event_tx, _dispatch_task: dispatch_task, }) } - /// The callback a substitute native backend uses to report events, standing + /// The callback a substitute backend uses to report events, standing /// in for the `EventHandler` notify would have been constructed with. - pub(crate) fn native_event_sink( + pub(crate) fn event_sink( &self, ) -> impl Fn(notify::Result) + Send + Sync + 'static { let event_tx = self.event_tx.clone(); - move |event| enqueue(&event_tx, WatcherMode::Native, event) + move |event| enqueue(&event_tx, event) + } + + pub(crate) fn is_recursive(&self) -> bool { + self.recursive } #[must_use] fn add( &self, path: Arc, - mode: WatcherMode, case_insensitive: bool, cb: impl Fn(¬ify::Event) + Send + Sync + 'static, ) -> anyhow::Result> { let path = SanitizedPath::from_arc(path); let key = WatchKey::for_registration(&path, case_insensitive); let mut state = self.state.lock(); - let (path_already_covered, path_already_registered) = { - let registrations_for_mode = state.path_registrations(mode); - ( - registrations_for_mode.covered_by_recursive_ancestor(&path, mode), - registrations_for_mode.contains(&key), - ) - }; + let path_already_covered = state + .paths + .covered_by_recursive_ancestor(&path, self.recursive); + let path_already_registered = state.paths.contains(&key); if !path_already_covered && !path_already_registered { - if mode == WatcherMode::Native && state.is_native_watch_limit_cooldown_active() { + if self.kind == OsWatcherKind::Native && state.is_native_watch_limit_cooldown_active() { return Ok(None); } drop(state); - match self.watch(path.as_path(), mode) { + match self.watch(path.as_path()) { Ok(()) => {} - Err(error) if mode == WatcherMode::Native && is_max_files_watch_error(&error) => { + Err(error) + if self.kind == OsWatcherKind::Native && is_max_files_watch_error(&error) => + { self.start_native_watch_limit_cooldown(path.as_path()); return Ok(None); } @@ -999,11 +1002,10 @@ impl GlobalWatcher { callback: Arc::new(cb), key: key.clone(), path, - mode, }; state.watchers.insert(id, registration_state); state - .path_registrations(mode) + .paths .entry(key) .and_modify(|registration| registration.watcher_ids.push(id)) .or_insert_with(|| PathRegistrationState { @@ -1015,17 +1017,17 @@ impl GlobalWatcher { } #[cfg(test)] - fn dispatch(&self, mode: WatcherMode, event: Result) { - dispatch(&self.state, mode, event); + fn dispatch(&self, event: notify::Result) { + dispatch(self.kind, &self.state, event); } #[cfg(test)] fn dispatch_batch( &self, - first: DispatchEvent, - event_rx: &async_channel::Receiver, + first: notify::Result, + event_rx: &async_channel::Receiver>, ) { - dispatch_batch(&self.state, first, event_rx); + dispatch_batch(self.kind, &self.state, first, event_rx); } fn start_native_watch_limit_cooldown(&self, path: &Path) { @@ -1043,56 +1045,36 @@ impl GlobalWatcher { pub fn remove(&self, id: WatcherRegistrationId) { let mut state = self.state.lock(); - let Some((path, mode)) = state.remove_registration(id) else { + let Some(path) = state.remove_registration(id) else { return; }; drop(state); - self.unwatch(path.as_path(), mode).log_err(); + self.unwatch(path.as_path()).log_err(); } - fn watch(&self, path: &Path, mode: WatcherMode) -> anyhow::Result<()> { - match mode { - WatcherMode::Native => { - self.ensure_native_watcher()?; - self.native_watcher - .lock() - .as_mut() - .expect("native watcher initialized") - .watch( - path, - if cfg!(any(target_os = "windows", target_os = "macos")) { - notify::RecursiveMode::Recursive - } else { - notify::RecursiveMode::NonRecursive - }, - )?; - } - WatcherMode::Poll => { - self.ensure_poll_watcher()?; - self.poll_watcher - .lock() - .as_mut() - .expect("poll watcher initialized") - .watch(path, notify::RecursiveMode::Recursive)?; - } - } - + fn watch(&self, path: &Path) -> anyhow::Result<()> { + self.ensure_backend()?; + self.backend + .lock() + .as_mut() + .expect("watcher backend initialized") + .watch( + path, + if self.recursive { + notify::RecursiveMode::Recursive + } else { + notify::RecursiveMode::NonRecursive + }, + )?; Ok(()) } - fn unwatch(&self, path: &Path, mode: WatcherMode) -> anyhow::Result<()> { - 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)), - }; + fn unwatch(&self, path: &Path) -> anyhow::Result<()> { + let watcher = self + .backend + .lock() + .as_mut() + .map(|watcher| watcher.unwatch(path)); match watcher { // inotify auto-removes a watch when its directory is deleted, so a @@ -1105,57 +1087,51 @@ impl GlobalWatcher { } } - pub(crate) fn ensure_native_watcher(&self) -> anyhow::Result<()> { + pub(crate) fn ensure_backend(&self) -> anyhow::Result<()> { // The lock is held across creation: with a check-then-insert under two // separate lock acquisitions, concurrent callers could each create a // watcher and the loser's insert would silently drop the winner's // watcher along with every path registered on it. - let mut native_watcher = self.native_watcher.lock(); - if native_watcher.is_none() { - // CORE excludes Access events, which Zed discards anyway. Without this, - // the default mask subscribes to inotify OPEN/CLOSE_* on Linux, so every - // file read in a watched directory would queue events, increasing the - // 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( - self.native_event_sink(), - config, - )?; - *native_watcher = Some(Box::new(watcher)); - } - Ok(()) - } - - fn ensure_poll_watcher(&self) -> anyhow::Result<()> { - let mut poll_watcher = self.poll_watcher.lock(); - if poll_watcher.is_none() { - let config = notify::Config::default().with_poll_interval(*POLL_INTERVAL); - let event_tx = self.event_tx.clone(); - let watcher = notify::PollWatcher::new( - move |event| enqueue(&event_tx, WatcherMode::Poll, event), - config, - )?; - *poll_watcher = Some(Box::new(watcher)); + let mut backend = self.backend.lock(); + if backend.is_none() { + *backend = Some(match self.kind { + OsWatcherKind::Native => { + // CORE excludes Access events, which Zed discards anyway. Without this, + // the default mask subscribes to inotify OPEN/CLOSE_* on Linux, so every + // file read in a watched directory would queue events, increasing the + // 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); + Box::new(::new( + self.event_sink(), + config, + )?) + } + OsWatcherKind::Poll => { + let config = notify::Config::default().with_poll_interval(*POLL_INTERVAL); + Box::new(notify::PollWatcher::new(self.event_sink(), config)?) + } + }); } Ok(()) } } fn dispatch( + kind: OsWatcherKind, state: &Mutex, - mode: WatcherMode, - event: Result, + event: notify::Result, ) { let event = match event { Ok(event) => event, Err(error) => { - log::warn!("watcher error for {mode:?}: {error}"); + log::warn!("{kind:?} watcher error: {error}"); return; } }; - log::trace!("global handle event for {mode:?}: {event:?}"); + log::trace!("handle {kind:?} watcher event: {event:?}"); let callbacks = { let state = state.lock(); @@ -1163,23 +1139,18 @@ fn dispatch( 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", + "{kind:?} filesystem watcher lost sync; 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); - path_registrations.watcher_ids_covering(sanitized, &mut ids); + state.paths.watcher_ids_covering(sanitized, &mut ids); } ids.sort_unstable_by_key(|id| id.0); ids.dedup(); @@ -1196,42 +1167,35 @@ fn dispatch( } fn dispatch_batch( + kind: OsWatcherKind, state: &Mutex, - first: DispatchEvent, - event_rx: &async_channel::Receiver, + first: notify::Result, + event_rx: &async_channel::Receiver>, ) { // A single backend overflow can enqueue many pathless rescan markers, each - // of which rescans every watched root, so one per mode covers the entire + // of which rescans every watched root, so one covers the entire // drained batch. Rescans that name paths only cover those subtrees and are // all dispatched; ordinary events still run. - let mut native_rescan_dispatched = false; - let mut poll_rescan_dispatched = false; + let mut rescan_dispatched = false; - for (mode, event) in - std::iter::once(first).chain(std::iter::from_fn(|| event_rx.try_recv().ok())) - { - let rescan_dispatched = match mode { - WatcherMode::Native => &mut native_rescan_dispatched, - WatcherMode::Poll => &mut poll_rescan_dispatched, - }; + for event in std::iter::once(first).chain(std::iter::from_fn(|| event_rx.try_recv().ok())) { if event .as_ref() .is_ok_and(|event| event.need_rescan() && event.paths.is_empty()) { - if *rescan_dispatched { + if rescan_dispatched { continue; } - *rescan_dispatched = true; + rescan_dispatched = true; } - dispatch(state, mode, event); + dispatch(kind, state, event); } } fn enqueue( - event_tx: &async_channel::Sender, - mode: WatcherMode, - event: Result, + event_tx: &async_channel::Sender>, + event: notify::Result, ) { if matches!( event, @@ -1245,7 +1209,7 @@ fn enqueue( // A failed send only happens once the dispatch thread has shut down, at // which point there's nothing left to dispatch to. - event_tx.try_send((mode, event)).ok(); + event_tx.try_send(event).ok(); } fn is_max_files_watch_error(error: &anyhow::Error) -> bool { @@ -1329,33 +1293,25 @@ mod tests { } } - fn test_watcher(poll_watcher: Arc>) -> GlobalWatcher { - test_watcher_with_backends(None, Some(poll_watcher)) - } - - fn test_watcher_with_backends( - native_watcher: Option>>, - poll_watcher: Option>>, - ) -> GlobalWatcher { - // Tests call `handle_event` directly to exercise dispatch synchronously, + fn test_os_watcher( + kind: OsWatcherKind, + backend: Option>>, + ) -> OsWatcher { + // Tests call `dispatch` 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 { + OsWatcher { + kind, + recursive: kind.is_recursive(), state: Arc::new(Mutex::new(WatcherState { watchers: Default::default(), - native_path_registrations: Default::default(), - poll_path_registrations: Default::default(), + paths: Default::default(), cooldown_until: None, last_registration: Default::default(), })), - native_watcher: Mutex::new( - native_watcher.map(|watcher| { - Box::new(SharedFakeWatchBackend(watcher)) as Box - }), - ), - poll_watcher: Mutex::new( - poll_watcher.map(|watcher| { + backend: Mutex::new( + backend.map(|watcher| { Box::new(SharedFakeWatchBackend(watcher)) as Box }), ), @@ -1375,16 +1331,16 @@ mod tests { #[test] fn covered_child_registration_is_not_unwatched_after_parent_is_removed() { let backend = Arc::new(Mutex::new(FakeWatchBackend::default())); - let watcher = test_watcher(backend.clone()); + let watcher = test_os_watcher(OsWatcherKind::Poll, Some(backend.clone())); let parent = Arc::::from(Path::new("/repo")); let child = Arc::::from(Path::new("/repo/foo.csproj")); let parent_registration = watcher - .add(parent.as_ref().into(), WatcherMode::Poll, false, |_| {}) + .add(parent.as_ref().into(), false, |_| {}) .expect("add parent watch") .expect("parent watch registered"); let child_registration = watcher - .add(child.as_ref().into(), WatcherMode::Poll, false, |_| {}) + .add(child.as_ref().into(), false, |_| {}) .expect("add covered child watch") .expect("child watch registered"); @@ -1404,7 +1360,8 @@ mod tests { let (tx, rx) = async_channel::unbounded(); let pending_path_events: Arc>> = Default::default(); let watcher = FsWatcher::new( - GlobalWatcher::new(cx.executor()), + OsWatcher::new(OsWatcherKind::Native, cx.executor()), + OsWatcher::new(OsWatcherKind::Poll, cx.executor()), crate::RealFs::new(None, cx.executor()), cx.executor(), tx, @@ -1458,16 +1415,15 @@ mod tests { fail_with_watch_limit: true, ..Default::default() })); - let poll_backend = Arc::new(Mutex::new(FakeWatchBackend::default())); - let watcher = test_watcher_with_backends(Some(native_backend.clone()), Some(poll_backend)); + let watcher = test_os_watcher(OsWatcherKind::Native, Some(native_backend.clone())); let first_path = Arc::::from(Path::new("/repo/first")); let second_path = Arc::::from(Path::new("/repo/second")); let first_registration = watcher - .add(first_path.clone(), WatcherMode::Native, false, |_| {}) + .add(first_path.clone(), false, |_| {}) .expect("native watch limit is handled"); let second_registration = watcher - .add(second_path, WatcherMode::Native, false, |_| {}) + .add(second_path, false, |_| {}) .expect("native watch limit backoff is handled"); assert!(first_registration.is_none()); @@ -1484,22 +1440,17 @@ mod tests { } } - fn recording_watcher() -> (GlobalWatcher, Arc>>) { + fn recording_watcher() -> (OsWatcher, Arc>>) { let backend = Arc::new(Mutex::new(FakeWatchBackend::default())); - let watcher = test_watcher_with_backends(Some(backend), None); + let watcher = test_os_watcher(OsWatcherKind::Native, Some(backend)); 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, - false, - move |_| { - fired.lock().push(label.clone()); - }, - ) + .add(Arc::::from(Path::new(dir)), false, move |_| { + fired.lock().push(label.clone()); + }) .expect("add watch") .expect("watch registered"); } @@ -1510,7 +1461,7 @@ mod tests { 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"))); + watcher.dispatch(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()]); @@ -1520,10 +1471,7 @@ mod tests { fn event_dispatches_to_every_ancestor_registration() { let (watcher, fired) = recording_watcher(); - watcher.dispatch( - WatcherMode::Native, - Ok(modify_event("/repo/a/nested/file.txt")), - ); + watcher.dispatch(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 @@ -1574,84 +1522,56 @@ mod tests { #[test] fn case_insensitive_registration_matches_differently_cased_event() { let (fired, cb) = fired_count(); - let watcher = test_watcher_with_backends(Some(Default::default()), None); + let watcher = test_os_watcher(OsWatcherKind::Native, Some(Default::default())); watcher - .add( - Path::new("/Repo/Project").into(), - WatcherMode::Native, - true, - cb, - ) + .add(Path::new("/Repo/Project").into(), true, cb) .expect("add") .expect("registered"); // Event arrives lowercased (as TSGO/macOS may report it). - watcher.dispatch( - WatcherMode::Native, - Ok(modify_event("/repo/project/file.txt")), - ); + watcher.dispatch(Ok(modify_event("/repo/project/file.txt"))); assert_eq!(*fired.lock(), 1); } #[test] fn case_insensitive_registration_survives_case_only_rename() { let (fired, cb) = fired_count(); - let watcher = test_watcher_with_backends(Some(Default::default()), None); + let watcher = test_os_watcher(OsWatcherKind::Native, Some(Default::default())); watcher - .add( - Path::new("/Repo/Proj").into(), - WatcherMode::Native, - true, - cb, - ) + .add(Path::new("/Repo/Proj").into(), true, cb) .expect("add") .expect("registered"); // The watched directory was renamed to a different casing; events now // arrive under the new spelling. - watcher.dispatch(WatcherMode::Native, Ok(modify_event("/Repo/PROJ/file.txt"))); + watcher.dispatch(Ok(modify_event("/Repo/PROJ/file.txt"))); assert_eq!(*fired.lock(), 1); } #[test] fn case_sensitive_registration_ignores_differently_cased_event() { let (fired, cb) = fired_count(); - let watcher = test_watcher_with_backends(Some(Default::default()), None); + let watcher = test_os_watcher(OsWatcherKind::Native, Some(Default::default())); watcher - .add( - Path::new("/Repo/proj").into(), - WatcherMode::Native, - false, - cb, - ) + .add(Path::new("/Repo/proj").into(), false, cb) .expect("add") .expect("registered"); // On a case-sensitive volume these are genuinely different directories. - watcher.dispatch(WatcherMode::Native, Ok(modify_event("/Repo/PROJ/file.txt"))); + watcher.dispatch(Ok(modify_event("/Repo/PROJ/file.txt"))); assert_eq!(*fired.lock(), 0); } #[test] fn differently_cased_adds_dedupe_on_case_insensitive_volume() { let backend = Arc::new(Mutex::new(FakeWatchBackend::default())); - let watcher = test_watcher_with_backends(Some(backend.clone()), None); + let watcher = test_os_watcher(OsWatcherKind::Native, Some(backend.clone())); watcher - .add( - Path::new("/Repo/Proj").into(), - WatcherMode::Native, - true, - |_| {}, - ) + .add(Path::new("/Repo/Proj").into(), true, |_| {}) .expect("add") .expect("registered"); watcher - .add( - Path::new("/repo/proj").into(), - WatcherMode::Native, - true, - |_| {}, - ) + .add(Path::new("/repo/proj").into(), true, |_| {}) .expect("add") .expect("registered"); @@ -1662,18 +1582,13 @@ mod tests { #[test] fn recursive_parent_covers_differently_cased_child() { let backend = Arc::new(Mutex::new(FakeWatchBackend::default())); - let watcher = test_watcher(backend.clone()); + let watcher = test_os_watcher(OsWatcherKind::Poll, Some(backend.clone())); watcher - .add(Path::new("/Repo").into(), WatcherMode::Poll, true, |_| {}) + .add(Path::new("/Repo").into(), true, |_| {}) .expect("add") .expect("registered"); watcher - .add( - Path::new("/repo/child").into(), - WatcherMode::Poll, - true, - |_| {}, - ) + .add(Path::new("/repo/child").into(), true, |_| {}) .expect("add"); // The child is covered by the recursive parent despite the case mismatch. @@ -1681,14 +1596,14 @@ mod tests { } #[test] - fn rescan_event_broadcasts_to_all_registrations_of_the_same_mode() { + fn rescan_event_broadcasts_to_all_registrations_of_the_same_backend() { let (watcher, fired) = recording_watcher(); let rescan = notify::Event::new(EventKind::Other).set_flag(notify::event::Flag::Rescan); - watcher.dispatch(WatcherMode::Native, Ok(rescan)); + watcher.dispatch(Ok(rescan)); // A pathless rescan may have missed events anywhere, so every registration - // of that mode resyncs, regardless of which directory it watches. + // of that backend resyncs, regardless of which directory it watches. let mut got = fired.lock().clone(); got.sort(); assert_eq!( @@ -1707,13 +1622,11 @@ mod tests { let (event_tx, event_rx) = async_channel::unbounded(); let rescan = || notify::Event::new(EventKind::Other).set_flag(notify::event::Flag::Rescan); + event_tx.try_send(Ok(rescan())).unwrap(); event_tx - .try_send((WatcherMode::Native, Ok(rescan()))) + .try_send(Ok(modify_event("/repo/a/file.txt"))) .unwrap(); - event_tx - .try_send((WatcherMode::Native, Ok(modify_event("/repo/a/file.txt")))) - .unwrap(); - watcher.dispatch_batch((WatcherMode::Native, Ok(rescan())), &event_rx); + watcher.dispatch_batch(Ok(rescan()), &event_rx); let mut got = fired.lock().clone(); got.sort(); @@ -1738,17 +1651,12 @@ mod tests { .add_path(PathBuf::from(path)) }; - event_tx - .try_send((WatcherMode::Native, Ok(rescan_of("/repo/b")))) - .unwrap(); - watcher.dispatch_batch( - (WatcherMode::Native, Ok(rescan_of("/repo/a/nested"))), - &event_rx, - ); + event_tx.try_send(Ok(rescan_of("/repo/b"))).unwrap(); + watcher.dispatch_batch(Ok(rescan_of("/repo/a/nested")), &event_rx); // Each named rescan only covers its own subtree once it reaches the // callbacks, so collapsing the second one would lose /repo/b's resync. - // Both are broadcast to every registration of the mode. + // Both are broadcast to every registration of the backend. let mut got = fired.lock().clone(); got.sort(); assert_eq!( @@ -1771,14 +1679,13 @@ mod tests { // 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 watcher = test_os_watcher(OsWatcherKind::Native, Some(backend)); let fired = Arc::new(Mutex::new(Vec::new())); { let fired = fired.clone(); watcher .add( Arc::::from(Path::new("C:\\repo\\src")), - WatcherMode::Native, false, move |_| fired.lock().push(()), ) @@ -1786,10 +1693,7 @@ mod tests { .expect("watch registered"); } - watcher.dispatch( - WatcherMode::Native, - Ok(modify_event("\\\\?\\C:\\repo\\src\\main.rs")), - ); + watcher.dispatch(Ok(modify_event("\\\\?\\C:\\repo\\src\\main.rs"))); assert_eq!(fired.lock().len(), 1); } diff --git a/crates/recent_projects/src/dev_container_suggest.rs b/crates/recent_projects/src/dev_container_suggest.rs index 2cb1dfde82..c5d429fa1c 100644 --- a/crates/recent_projects/src/dev_container_suggest.rs +++ b/crates/recent_projects/src/dev_container_suggest.rs @@ -77,13 +77,16 @@ pub fn suggest_on_worktree_updated( window: &mut Window, cx: &mut Context, ) { - let cli_auto_open = workspace.open_in_dev_container(); + if workspace.open_in_dev_container() { + open_dev_container_from_cli(workspace, window, cx); + return; + } let devcontainer_updated = updated_entries.iter().any(|(path, _, _)| { path.as_ref() == devcontainer_dir_path() || path.as_ref() == devcontainer_json_path() }); - if !devcontainer_updated && !cli_auto_open { + if !devcontainer_updated { return; } @@ -97,35 +100,7 @@ pub fn suggest_on_worktree_updated( return; } - let has_configs = !find_configs_in_snapshot(worktree).is_empty(); - - if cli_auto_open { - workspace.set_open_in_dev_container(false); - let task = cx.spawn_in(window, async move |workspace, cx| { - let scans_complete = - workspace.update(cx, |workspace, cx| workspace.worktree_scans_complete(cx))?; - scans_complete.await; - - workspace.update_in(cx, |workspace, window, cx| { - let has_configs = workspace - .project() - .read(cx) - .worktrees(cx) - .any(|wt| !find_configs_in_snapshot(wt.read(cx)).is_empty()); - if has_configs { - cx.on_next_frame(window, move |_workspace, window, cx| { - window.dispatch_action(Box::new(zed_actions::OpenDevContainer), cx); - }); - } else { - log::warn!("--dev-container: no devcontainer configuration found in project"); - } - }) - }); - workspace.set_dev_container_task(task); - return; - } - - if !has_configs { + if find_configs_in_snapshot(worktree).is_empty() { return; } @@ -193,3 +168,32 @@ pub fn suggest_on_worktree_updated( }); }); } + +pub fn open_dev_container_from_cli( + workspace: &mut Workspace, + window: &mut Window, + cx: &mut Context, +) { + workspace.set_open_in_dev_container(false); + let task = cx.spawn_in(window, async move |workspace, cx| { + let scans_complete = + workspace.update(cx, |workspace, cx| workspace.worktree_scans_complete(cx))?; + scans_complete.await; + + workspace.update_in(cx, |workspace, window, cx| { + let has_configs = workspace + .project() + .read(cx) + .worktrees(cx) + .any(|worktree| !find_configs_in_snapshot(worktree.read(cx)).is_empty()); + if has_configs { + cx.on_next_frame(window, move |_workspace, window, cx| { + window.dispatch_action(Box::new(zed_actions::OpenDevContainer), cx); + }); + } else { + log::warn!("--dev-container: no devcontainer configuration found in project"); + } + }) + }); + workspace.set_dev_container_task(task); +} diff --git a/crates/recent_projects/src/recent_projects.rs b/crates/recent_projects/src/recent_projects.rs index c26d7537b5..5578ae9382 100644 --- a/crates/recent_projects/src/recent_projects.rs +++ b/crates/recent_projects/src/recent_projects.rs @@ -545,6 +545,12 @@ pub fn init(cx: &mut App) { let Some(window) = window else { return; }; + // A workspace opened with `--dev-container` has its worktrees scanning + // before this observer runs, so their update events can't be relied on + // to trigger the auto-open. + if workspace.open_in_dev_container() { + dev_container_suggest::open_dev_container_from_cli(workspace, window, cx); + } cx.subscribe_in( workspace.project(), window,