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

Ensure newer snapshots are always detected in wait_for_snapshot

Browse source
This commit is contained in:
Antonio Scandurra 2022-06-30 18:04:19 +02:00
parent 484af8c7c4
commit 4ee8ee5a06
2 changed files with 42 additions and 30 deletions
Download patch file Download diff file Expand all files Collapse all files

8
crates/project/src/project.rs
View file

@ -1081,7 +1081,7 @@ impl Project {
.ok_or_else(|| anyhow!("missing entry in response"))?; .ok_or_else(|| anyhow!("missing entry in response"))?;
worktree worktree
.update(&mut cx, |worktree, cx| { .update(&mut cx, |worktree, cx| {
worktree.as_remote().unwrap().insert_entry( worktree.as_remote_mut().unwrap().insert_entry(
entry, entry,
response.worktree_scan_id as usize, response.worktree_scan_id as usize,
cx, cx,
@ -1124,7 +1124,7 @@ impl Project {
.ok_or_else(|| anyhow!("missing entry in response"))?; .ok_or_else(|| anyhow!("missing entry in response"))?;
worktree worktree
.update(&mut cx, |worktree, cx| { .update(&mut cx, |worktree, cx| {
worktree.as_remote().unwrap().insert_entry( worktree.as_remote_mut().unwrap().insert_entry(
entry, entry,
response.worktree_scan_id as usize, response.worktree_scan_id as usize,
cx, cx,
@ -1167,7 +1167,7 @@ impl Project {
.ok_or_else(|| anyhow!("missing entry in response"))?; .ok_or_else(|| anyhow!("missing entry in response"))?;
worktree worktree
.update(&mut cx, |worktree, cx| { .update(&mut cx, |worktree, cx| {
worktree.as_remote().unwrap().insert_entry( worktree.as_remote_mut().unwrap().insert_entry(
entry, entry,
response.worktree_scan_id as usize, response.worktree_scan_id as usize,
cx, cx,
@ -1200,7 +1200,7 @@ impl Project {
.await?; .await?;
worktree worktree
.update(&mut cx, move |worktree, cx| { .update(&mut cx, move |worktree, cx| {
worktree.as_remote().unwrap().delete_entry( worktree.as_remote_mut().unwrap().delete_entry(
entry_id, entry_id,
response.worktree_scan_id as usize, response.worktree_scan_id as usize,
cx, cx,

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

@ -9,7 +9,7 @@ use ::ignore::gitignore::{Gitignore, GitignoreBuilder};
use anyhow::{anyhow, Context, Result}; use anyhow::{anyhow, Context, Result};
use client::{proto, Client}; use client::{proto, Client};
use clock::ReplicaId; use clock::ReplicaId;
use collections::HashMap; use collections::{HashMap, VecDeque};
use futures::{ use futures::{
channel::{ channel::{
mpsc::{self, UnboundedSender}, mpsc::{self, UnboundedSender},
@ -82,7 +82,7 @@ pub struct RemoteWorktree {
project_id: u64, project_id: u64,
client: Arc<Client>, client: Arc<Client>,
updates_tx: Option<UnboundedSender<proto::UpdateWorktree>>, updates_tx: Option<UnboundedSender<proto::UpdateWorktree>>,
snapshot_updated_rx: watch::Receiver<()>, snapshot_subscriptions: VecDeque<(usize, oneshot::Sender<()>)>,
replica_id: ReplicaId, replica_id: ReplicaId,
diagnostic_summaries: TreeMap<PathKey, DiagnosticSummary>, diagnostic_summaries: TreeMap<PathKey, DiagnosticSummary>,
visible: bool, visible: bool,
@ -204,7 +204,7 @@ impl Worktree {
snapshot: snapshot.clone(), snapshot: snapshot.clone(),
background_snapshot: background_snapshot.clone(), background_snapshot: background_snapshot.clone(),
updates_tx: Some(updates_tx), updates_tx: Some(updates_tx),
snapshot_updated_rx: snapshot_updated_rx.clone(), snapshot_subscriptions: Default::default(),
client: client.clone(), client: client.clone(),
diagnostic_summaries: Default::default(), diagnostic_summaries: Default::default(),
visible, visible,
@ -227,7 +227,18 @@ impl Worktree {
async move { async move {
while let Some(_) = snapshot_updated_rx.recv().await { while let Some(_) = snapshot_updated_rx.recv().await {
if let Some(this) = this.upgrade(&cx) { if let Some(this) = this.upgrade(&cx) {
this.update(&mut cx, |this, cx| this.poll_snapshot(cx)); this.update(&mut cx, |this, cx| {
this.poll_snapshot(cx);
let this = this.as_remote_mut().unwrap();
while let Some((scan_id, _)) = this.snapshot_subscriptions.front() {
if this.observed_snapshot(*scan_id) {
let (_, tx) = this.snapshot_subscriptions.pop_front().unwrap();
let _ = tx.send(());
} else {
break;
}
}
});
} else { } else {
break; break;
} }
@ -969,25 +980,26 @@ impl RemoteWorktree {
} }
} }
fn wait_for_snapshot( fn observed_snapshot(&self, scan_id: usize) -> bool {
&self, self.scan_id > scan_id || (self.scan_id == scan_id && self.is_complete)
scan_id: usize, }
cx: &mut ModelContext<Worktree>,
) -> Task<Option<()>> { fn wait_for_snapshot(&mut self, scan_id: usize) -> impl Future<Output = ()> {
let mut rx = self.snapshot_updated_rx.clone(); let (tx, rx) = oneshot::channel();
cx.spawn_weak(|worktree, cx| async move { if self.observed_snapshot(scan_id) {
while rx.recv().await.is_some() { let _ = tx.send(());
let snapshot = worktree } else {
.upgrade(&cx)? match self
.read_with(&cx, |worktree, _| worktree.snapshot()); .snapshot_subscriptions
if snapshot.scan_id > scan_id .binary_search_by_key(&scan_id, |probe| probe.0)
|| (snapshot.scan_id == scan_id && snapshot.is_complete) {
{ Ok(ix) | Err(ix) => self.snapshot_subscriptions.insert(ix, (scan_id, tx)),
break;
}
} }
None }
})
async move {
let _ = rx.await;
}
} }
pub fn update_diagnostic_summary( pub fn update_diagnostic_summary(
@ -1009,12 +1021,12 @@ impl RemoteWorktree {
} }
pub fn insert_entry( pub fn insert_entry(
&self, &mut self,
entry: proto::Entry, entry: proto::Entry,
scan_id: usize, scan_id: usize,
cx: &mut ModelContext<Worktree>, cx: &mut ModelContext<Worktree>,
) -> Task<Result<Entry>> { ) -> Task<Result<Entry>> {
let wait_for_snapshot = self.wait_for_snapshot(scan_id, cx); let wait_for_snapshot = self.wait_for_snapshot(scan_id);
cx.spawn(|this, mut cx| async move { cx.spawn(|this, mut cx| async move {
wait_for_snapshot.await; wait_for_snapshot.await;
this.update(&mut cx, |worktree, _| { this.update(&mut cx, |worktree, _| {
@ -1028,12 +1040,12 @@ impl RemoteWorktree {
} }
pub(crate) fn delete_entry( pub(crate) fn delete_entry(
&self, &mut self,
id: ProjectEntryId, id: ProjectEntryId,
scan_id: usize, scan_id: usize,
cx: &mut ModelContext<Worktree>, cx: &mut ModelContext<Worktree>,
) -> Task<Result<()>> { ) -> Task<Result<()>> {
let wait_for_snapshot = self.wait_for_snapshot(scan_id, cx); let wait_for_snapshot = self.wait_for_snapshot(scan_id);
cx.spawn(|this, mut cx| async move { cx.spawn(|this, mut cx| async move {
wait_for_snapshot.await; wait_for_snapshot.await;
this.update(&mut cx, |worktree, _| { this.update(&mut cx, |worktree, _| {