Yehowshua/ZIm
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Handle path changes and progress updates from all worker threads during initial scan

Browse source
This commit is contained in:
Max Brunsfeld 2023-03-24 14:35:18 -07:00
parent 027def6800
commit 455ffb17f1
1 changed files with 72 additions and 63 deletions
Download patch file Download diff file Expand all files Collapse all files

135
crates/project/src/worktree.rs
View file

@ -9,7 +9,7 @@ use collections::{HashMap, VecDeque};
use fs::{repository::GitRepository, Fs, LineEnding}; use fs::{repository::GitRepository, Fs, LineEnding};
use futures::{ use futures::{
channel::{ channel::{
mpsc::{self, UnboundedReceiver, UnboundedSender}, mpsc::{self, UnboundedSender},
oneshot, oneshot,
}, },
select_biased, Stream, StreamExt, select_biased, Stream, StreamExt,
@ -44,7 +44,10 @@ use std::{
mem, mem,
ops::{Deref, DerefMut}, ops::{Deref, DerefMut},
path::{Path, PathBuf}, path::{Path, PathBuf},
sync::{atomic::AtomicUsize, Arc}, sync::{
atomic::{AtomicUsize, Ordering::SeqCst},
Arc,
},
task::Poll, task::Poll,
time::{Duration, SystemTime}, time::{Duration, SystemTime},
}; };
@ -61,7 +64,7 @@ pub enum Worktree {
pub struct LocalWorktree { pub struct LocalWorktree {
snapshot: LocalSnapshot, snapshot: LocalSnapshot,
path_changes_tx: mpsc::UnboundedSender<(Vec<PathBuf>, barrier::Sender)>, path_changes_tx: channel::Sender<(Vec<PathBuf>, barrier::Sender)>,
is_scanning: (watch::Sender<bool>, watch::Receiver<bool>), is_scanning: (watch::Sender<bool>, watch::Receiver<bool>),
_background_scanner_task: Task<()>, _background_scanner_task: Task<()>,
share: Option<ShareState>, share: Option<ShareState>,
@ -238,7 +241,7 @@ impl Worktree {
); );
} }
let (path_changes_tx, path_changes_rx) = mpsc::unbounded(); let (path_changes_tx, path_changes_rx) = channel::unbounded();
let (scan_states_tx, mut scan_states_rx) = mpsc::unbounded(); let (scan_states_tx, mut scan_states_rx) = mpsc::unbounded();
cx.spawn_weak(|this, mut cx| async move { cx.spawn_weak(|this, mut cx| async move {
@ -837,7 +840,7 @@ impl LocalWorktree {
this.as_local_mut() this.as_local_mut()
.unwrap() .unwrap()
.path_changes_tx .path_changes_tx
.unbounded_send((vec![abs_path], tx)) .try_send((vec![abs_path], tx))
.unwrap(); .unwrap();
}); });
rx.recv().await; rx.recv().await;
@ -930,7 +933,7 @@ impl LocalWorktree {
} }
let (tx, mut rx) = barrier::channel(); let (tx, mut rx) = barrier::channel();
path_changes_tx.unbounded_send((paths, tx)).unwrap(); path_changes_tx.try_send((paths, tx)).unwrap();
rx.recv().await; rx.recv().await;
this.upgrade(&cx) this.upgrade(&cx)
.ok_or_else(|| anyhow!("worktree was dropped"))? .ok_or_else(|| anyhow!("worktree was dropped"))?
@ -2165,7 +2168,7 @@ impl BackgroundScanner {
async fn run( async fn run(
self, self,
events_rx: impl Stream<Item = Vec<fsevent::Event>>, events_rx: impl Stream<Item = Vec<fsevent::Event>>,
mut changed_paths: UnboundedReceiver<(Vec<PathBuf>, barrier::Sender)>, mut changed_paths: channel::Receiver<(Vec<PathBuf>, barrier::Sender)>,
) { ) {
use futures::FutureExt as _; use futures::FutureExt as _;
@ -2225,64 +2228,70 @@ impl BackgroundScanner {
.unwrap(); .unwrap();
drop(tx); drop(tx);
let progress_update_count = AtomicUsize::new(0);
self.executor self.executor
.scoped(|scope| { .scoped(|scope| {
// While the scan is running, listen for path update requests from the worktree, for _ in 0..self.executor.num_cpus() {
// and report updates to the worktree based on a timer.
scope.spawn(async {
let reporting_timer = self.pause_between_initializing_updates().fuse();
futures::pin_mut!(reporting_timer);
loop {
select_biased! {
job = changed_paths.next().fuse() => {
let Some((abs_paths, barrier)) = job else { break };
self.update_entries_for_paths(abs_paths, None).await;
if self
.notify
.unbounded_send(ScanState::Initializing {
snapshot: self.snapshot.lock().clone(),
barrier: Some(barrier),
})
.is_err()
{
break;
}
}
_ = reporting_timer => {
if self
.notify
.unbounded_send(ScanState::Initializing {
snapshot: self.snapshot.lock().clone(),
barrier: None,
})
.is_err()
{
break;
}
reporting_timer.set(self.pause_between_initializing_updates().fuse());
}
job = rx.recv().fuse() => {
let Ok(job) = job else { break };
if let Err(err) = self
.scan_dir(root_char_bag, next_entry_id.clone(), &job)
.await
{
log::error!("error scanning {:?}: {}", job.abs_path, err);
}
}
}
}
});
// Spawn worker threads to scan the directory recursively.
for _ in 1..self.executor.num_cpus() {
scope.spawn(async { scope.spawn(async {
while let Ok(job) = rx.recv().await { let mut last_progress_update_count = 0;
if let Err(err) = self let progress_update_timer = self.pause_between_progress_updates().fuse();
.scan_dir(root_char_bag, next_entry_id.clone(), &job) futures::pin_mut!(progress_update_timer);
.await loop {
{ select_biased! {
log::error!("error scanning {:?}: {}", job.abs_path, err); // Send periodic progress updates to the worktree. Use an atomic counter
// to ensure that only one of the workers sends a progress update after
// the update interval elapses.
_ = progress_update_timer => {
match progress_update_count.compare_exchange(
last_progress_update_count,
last_progress_update_count + 1,
SeqCst,
SeqCst
) {
Ok(_) => {
last_progress_update_count += 1;
if self
.notify
.unbounded_send(ScanState::Initializing {
snapshot: self.snapshot.lock().clone(),
barrier: None,
})
.is_err()
{
break;
}
}
Err(current_count) => last_progress_update_count = current_count,
}
progress_update_timer.set(self.pause_between_progress_updates().fuse());
}
// Refresh any paths requested by the main thread.
job = changed_paths.recv().fuse() => {
let Ok((abs_paths, barrier)) = job else { break };
self.update_entries_for_paths(abs_paths, None).await;
if self
.notify
.unbounded_send(ScanState::Initializing {
snapshot: self.snapshot.lock().clone(),
barrier: Some(barrier),
})
.is_err()
{
break;
}
}
// Recursively load directories from the file system.
job = rx.recv().fuse() => {
let Ok(job) = job else { break };
if let Err(err) = self
.scan_dir(root_char_bag, next_entry_id.clone(), &job)
.await
{
log::error!("error scanning {:?}: {}", job.abs_path, err);
}
}
} }
} }
}); });
@ -2370,7 +2379,7 @@ impl BackgroundScanner {
} }
} }
async fn pause_between_initializing_updates(&self) { async fn pause_between_progress_updates(&self) {
#[cfg(any(test, feature = "test-support"))] #[cfg(any(test, feature = "test-support"))]
if self.fs.is_fake() { if self.fs.is_fake() {
return self.executor.simulate_random_delay().await; return self.executor.simulate_random_delay().await;