Don't block when emitting fs events

Antonio Scandurra created 3 years ago

Blocking could happen while processing events, which would cause the
background scanner to never make any further progress.

Change summary

crates/project/src/fs.rs       | 19 ++++++++++---------
crates/project/src/worktree.rs |  2 +-
crates/server/src/rpc.rs       |  4 +++-
3 files changed, 14 insertions(+), 11 deletions(-)

Detailed changes

crates/project/src/fs.rs

@@ -225,7 +225,7 @@ struct FakeFsEntry {
 struct FakeFsState {
     entries: std::collections::BTreeMap<PathBuf, FakeFsEntry>,
     next_inode: u64,
-    events_tx: postage::broadcast::Sender<Vec<fsevent::Event>>,
+    event_txs: Vec<smol::channel::Sender<Vec<fsevent::Event>>>,
 }
 
 #[cfg(any(test, feature = "test-support"))]
@@ -248,8 +248,6 @@ impl FakeFsState {
         I: IntoIterator<Item = T>,
         T: Into<PathBuf>,
     {
-        use postage::prelude::Sink as _;
-
         let events = paths
             .into_iter()
             .map(|path| fsevent::Event {
@@ -257,9 +255,12 @@ impl FakeFsState {
                 flags: fsevent::StreamFlags::empty(),
                 path: path.into(),
             })
-            .collect();
+            .collect::<Vec<_>>();
 
-        let _ = self.events_tx.send(events).await;
+        self.event_txs.retain(|tx| {
+            let _ = tx.try_send(events.clone());
+            !tx.is_closed()
+        });
     }
 }
 
@@ -273,7 +274,6 @@ pub struct FakeFs {
 #[cfg(any(test, feature = "test-support"))]
 impl FakeFs {
     pub fn new(executor: std::sync::Arc<gpui::executor::Background>) -> std::sync::Arc<Self> {
-        let (events_tx, _) = postage::broadcast::channel(2);
         let mut entries = std::collections::BTreeMap::new();
         entries.insert(
             Path::new("/").to_path_buf(),
@@ -292,7 +292,7 @@ impl FakeFs {
             state: futures::lock::Mutex::new(FakeFsState {
                 entries,
                 next_inode: 1,
-                events_tx,
+                event_txs: Default::default(),
             }),
         })
     }
@@ -642,9 +642,10 @@ impl Fs for FakeFs {
         path: &Path,
         _: Duration,
     ) -> Pin<Box<dyn Send + Stream<Item = Vec<fsevent::Event>>>> {
-        let state = self.state.lock().await;
+        let mut state = self.state.lock().await;
         self.simulate_random_delay().await;
-        let rx = state.events_tx.subscribe();
+        let (tx, rx) = smol::channel::unbounded();
+        state.event_txs.push(tx);
         let path = path.to_path_buf();
         Box::pin(futures::StreamExt::filter(rx, move |events| {
             let result = events.iter().any(|event| event.path.starts_with(&path));

crates/project/src/worktree.rs

@@ -386,7 +386,7 @@ impl Worktree {
                     if worktree.poll_task.is_none() {
                         worktree.poll_task = Some(cx.spawn(|this, mut cx| async move {
                             if is_fake_fs {
-                                smol::future::yield_now().await;
+                                cx.background().simulate_random_delay().await;
                             } else {
                                 smol::Timer::after(Duration::from_millis(100)).await;
                             }

crates/server/src/rpc.rs

@@ -4185,7 +4185,9 @@ mod tests {
 
         while operations.get() < max_operations {
             cx.background().simulate_random_delay().await;
-            if clients.len() < max_peers && rng.lock().gen_bool(0.05) {
+            if clients.len() >= max_peers {
+                break;
+            } else if rng.lock().gen_bool(0.05) {
                 operations.set(operations.get() + 1);
 
                 let guest_id = clients.len();