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

Merge pull request #1265 from zed-industries/worktree-performance

Browse source 
Fix problems that arise when large numbers of files change on disk
This commit is contained in:
Antonio Scandurra 2022-07-04 09:19:18 +02:00 • committed by GitHub
commit 0b2452f666
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
9 changed files with 393 additions and 453 deletions
Download patch file Download diff file Expand all files Collapse all files

17
crates/client/src/client.rs
View file

@ -28,10 +28,7 @@ use std::{
convert::TryFrom, convert::TryFrom,
fmt::Write as _, fmt::Write as _,
future::Future, future::Future,
sync::{ sync::{Arc, Weak},
atomic::{AtomicUsize, Ordering},
Arc, Weak,
},
time::{Duration, Instant}, time::{Duration, Instant},
}; };
use thiserror::Error; use thiserror::Error;
@ -232,12 +229,8 @@ impl Drop for Subscription {
impl Client { impl Client {
pub fn new(http: Arc<dyn HttpClient>) -> Arc<Self> { pub fn new(http: Arc<dyn HttpClient>) -> Arc<Self> {
lazy_static! {
static ref NEXT_CLIENT_ID: AtomicUsize = AtomicUsize::default();
}
Arc::new(Self { Arc::new(Self {
id: NEXT_CLIENT_ID.fetch_add(1, Ordering::SeqCst), id: 0,
peer: Peer::new(), peer: Peer::new(),
http, http,
state: Default::default(), state: Default::default(),
@ -257,6 +250,12 @@ impl Client {
self.http.clone() self.http.clone()
} }
#[cfg(any(test, feature = "test-support"))]
pub fn set_id(&mut self, id: usize) -> &Self {
self.id = id;
self
}
#[cfg(any(test, feature = "test-support"))] #[cfg(any(test, feature = "test-support"))]
pub fn tear_down(&self) { pub fn tear_down(&self) {
let mut state = self.state.write(); let mut state = self.state.write();

11
crates/collab/src/db.rs
View file

@ -2117,7 +2117,7 @@ pub mod tests {
Self { Self {
background, background,
users: Default::default(), users: Default::default(),
next_user_id: Mutex::new(1), next_user_id: Mutex::new(0),
projects: Default::default(), projects: Default::default(),
worktree_extensions: Default::default(), worktree_extensions: Default::default(),
next_project_id: Mutex::new(1), next_project_id: Mutex::new(1),
@ -2181,6 +2181,7 @@ pub mod tests {
} }
async fn get_user_by_id(&self, id: UserId) -> Result<Option<User>> { async fn get_user_by_id(&self, id: UserId) -> Result<Option<User>> {
self.background.simulate_random_delay().await;
Ok(self.get_users_by_ids(vec![id]).await?.into_iter().next()) Ok(self.get_users_by_ids(vec![id]).await?.into_iter().next())
} }
@ -2195,6 +2196,7 @@ pub mod tests {
} }
async fn get_user_by_github_login(&self, github_login: &str) -> Result<Option<User>> { async fn get_user_by_github_login(&self, github_login: &str) -> Result<Option<User>> {
self.background.simulate_random_delay().await;
Ok(self Ok(self
.users .users
.lock() .lock()
@ -2228,6 +2230,7 @@ pub mod tests {
} }
async fn get_invite_code_for_user(&self, _id: UserId) -> Result<Option<(String, u32)>> { async fn get_invite_code_for_user(&self, _id: UserId) -> Result<Option<(String, u32)>> {
self.background.simulate_random_delay().await;
Ok(None) Ok(None)
} }
@ -2265,6 +2268,7 @@ pub mod tests {
} }
async fn unregister_project(&self, project_id: ProjectId) -> Result<()> { async fn unregister_project(&self, project_id: ProjectId) -> Result<()> {
self.background.simulate_random_delay().await;
self.projects self.projects
.lock() .lock()
.get_mut(&project_id) .get_mut(&project_id)
@ -2370,6 +2374,7 @@ pub mod tests {
requester_id: UserId, requester_id: UserId,
responder_id: UserId, responder_id: UserId,
) -> Result<()> { ) -> Result<()> {
self.background.simulate_random_delay().await;
let mut contacts = self.contacts.lock(); let mut contacts = self.contacts.lock();
for contact in contacts.iter_mut() { for contact in contacts.iter_mut() {
if contact.requester_id == requester_id && contact.responder_id == responder_id { if contact.requester_id == requester_id && contact.responder_id == responder_id {
@ -2399,6 +2404,7 @@ pub mod tests {
} }
async fn remove_contact(&self, requester_id: UserId, responder_id: UserId) -> Result<()> { async fn remove_contact(&self, requester_id: UserId, responder_id: UserId) -> Result<()> {
self.background.simulate_random_delay().await;
self.contacts.lock().retain(|contact| { self.contacts.lock().retain(|contact| {
!(contact.requester_id == requester_id && contact.responder_id == responder_id) !(contact.requester_id == requester_id && contact.responder_id == responder_id)
}); });
@ -2410,6 +2416,7 @@ pub mod tests {
user_id: UserId, user_id: UserId,
contact_user_id: UserId, contact_user_id: UserId,
) -> Result<()> { ) -> Result<()> {
self.background.simulate_random_delay().await;
let mut contacts = self.contacts.lock(); let mut contacts = self.contacts.lock();
for contact in contacts.iter_mut() { for contact in contacts.iter_mut() {
if contact.requester_id == contact_user_id if contact.requester_id == contact_user_id
@ -2436,6 +2443,7 @@ pub mod tests {
requester_id: UserId, requester_id: UserId,
accept: bool, accept: bool,
) -> Result<()> { ) -> Result<()> {
self.background.simulate_random_delay().await;
let mut contacts = self.contacts.lock(); let mut contacts = self.contacts.lock();
for (ix, contact) in contacts.iter_mut().enumerate() { for (ix, contact) in contacts.iter_mut().enumerate() {
if contact.requester_id == requester_id && contact.responder_id == responder_id { if contact.requester_id == requester_id && contact.responder_id == responder_id {
@ -2631,6 +2639,7 @@ pub mod tests {
count: usize, count: usize,
before_id: Option<MessageId>, before_id: Option<MessageId>,
) -> Result<Vec<ChannelMessage>> { ) -> Result<Vec<ChannelMessage>> {
self.background.simulate_random_delay().await;
let mut messages = self let mut messages = self
.channel_messages .channel_messages
.lock() .lock()

39
crates/collab/src/integration_tests.rs
View file

@ -50,7 +50,6 @@ use std::{
time::Duration, time::Duration,
}; };
use theme::ThemeRegistry; use theme::ThemeRegistry;
use tokio::sync::RwLockReadGuard;
use workspace::{Item, SplitDirection, ToggleFollow, Workspace}; use workspace::{Item, SplitDirection, ToggleFollow, Workspace};
#[ctor::ctor] #[ctor::ctor]
@ -589,7 +588,7 @@ async fn test_offline_projects(
deterministic.run_until_parked(); deterministic.run_until_parked();
assert!(server assert!(server
.store .store
.read() .lock()
.await .await
.project_metadata_for_user(user_a) .project_metadata_for_user(user_a)
.is_empty()); .is_empty());
@ -620,7 +619,7 @@ async fn test_offline_projects(
cx_a.foreground().advance_clock(rpc::RECEIVE_TIMEOUT); cx_a.foreground().advance_clock(rpc::RECEIVE_TIMEOUT);
assert!(server assert!(server
.store .store
.read() .lock()
.await .await
.project_metadata_for_user(user_a) .project_metadata_for_user(user_a)
.is_empty()); .is_empty());
@ -1446,7 +1445,7 @@ async fn test_collaborating_with_diagnostics(
// Wait for server to see the diagnostics update. // Wait for server to see the diagnostics update.
deterministic.run_until_parked(); deterministic.run_until_parked();
{ {
let store = server.store.read().await; let store = server.store.lock().await;
let project = store.project(ProjectId::from_proto(project_id)).unwrap(); let project = store.project(ProjectId::from_proto(project_id)).unwrap();
let worktree = project.worktrees.get(&worktree_id.to_proto()).unwrap(); let worktree = project.worktrees.get(&worktree_id.to_proto()).unwrap();
assert!(!worktree.diagnostic_summaries.is_empty()); assert!(!worktree.diagnostic_summaries.is_empty());
@ -1472,6 +1471,7 @@ async fn test_collaborating_with_diagnostics(
// Join project as client C and observe the diagnostics. // Join project as client C and observe the diagnostics.
let project_c = client_c.build_remote_project(&project_a, cx_a, cx_c).await; let project_c = client_c.build_remote_project(&project_a, cx_a, cx_c).await;
deterministic.run_until_parked();
project_c.read_with(cx_c, |project, cx| { project_c.read_with(cx_c, |project, cx| {
assert_eq!( assert_eq!(
project.diagnostic_summaries(cx).collect::<Vec<_>>(), project.diagnostic_summaries(cx).collect::<Vec<_>>(),
@ -3171,7 +3171,7 @@ async fn test_basic_chat(cx_a: &mut TestAppContext, cx_b: &mut TestAppContext) {
assert_eq!( assert_eq!(
server server
.state() .store()
.await .await
.channel(channel_id) .channel(channel_id)
.unwrap() .unwrap()
@ -4425,8 +4425,16 @@ async fn test_random_collaboration(
let mut server = TestServer::start(cx.foreground(), cx.background()).await; let mut server = TestServer::start(cx.foreground(), cx.background()).await;
let db = server.app_state.db.clone(); let db = server.app_state.db.clone();
let host_user_id = db.create_user("host", None, false).await.unwrap(); let host_user_id = db.create_user("host", None, false).await.unwrap();
for username in ["guest-1", "guest-2", "guest-3", "guest-4"] { let mut available_guests = vec![
"guest-1".to_string(),
"guest-2".to_string(),
"guest-3".to_string(),
"guest-4".to_string(),
];
for username in &available_guests {
let guest_user_id = db.create_user(username, None, false).await.unwrap(); let guest_user_id = db.create_user(username, None, false).await.unwrap();
assert_eq!(*username, format!("guest-{}", guest_user_id));
server server
.app_state .app_state
.db .db
@ -4620,12 +4628,7 @@ async fn test_random_collaboration(
} else { } else {
max_operations max_operations
}; };
let mut available_guests = vec![
"guest-1".to_string(),
"guest-2".to_string(),
"guest-3".to_string(),
"guest-4".to_string(),
];
let mut operations = 0; let mut operations = 0;
while operations < max_operations { while operations < max_operations {
if operations == disconnect_host_at { if operations == disconnect_host_at {
@ -4656,7 +4659,7 @@ async fn test_random_collaboration(
.unwrap(); .unwrap();
let contacts = server let contacts = server
.store .store
.read() .lock()
.await .await
.build_initial_contacts_update(contacts) .build_initial_contacts_update(contacts)
.contacts; .contacts;
@ -4728,6 +4731,7 @@ async fn test_random_collaboration(
server.disconnect_client(removed_guest_id); server.disconnect_client(removed_guest_id);
deterministic.advance_clock(RECEIVE_TIMEOUT); deterministic.advance_clock(RECEIVE_TIMEOUT);
deterministic.start_waiting(); deterministic.start_waiting();
log::info!("Waiting for guest {} to exit...", removed_guest_id);
let (guest, guest_project, mut guest_cx, guest_err) = guest.await; let (guest, guest_project, mut guest_cx, guest_err) = guest.await;
deterministic.finish_waiting(); deterministic.finish_waiting();
server.allow_connections(); server.allow_connections();
@ -4740,7 +4744,7 @@ async fn test_random_collaboration(
let contacts = server.app_state.db.get_contacts(*user_id).await.unwrap(); let contacts = server.app_state.db.get_contacts(*user_id).await.unwrap();
let contacts = server let contacts = server
.store .store
.read() .lock()
.await .await
.build_initial_contacts_update(contacts) .build_initial_contacts_update(contacts)
.contacts; .contacts;
@ -4944,6 +4948,7 @@ impl TestServer {
Arc::get_mut(&mut client) Arc::get_mut(&mut client)
.unwrap() .unwrap()
.set_id(user_id.0 as usize)
.override_authenticate(move |cx| { .override_authenticate(move |cx| {
cx.spawn(|_| async move { cx.spawn(|_| async move {
let access_token = "the-token".to_string(); let access_token = "the-token".to_string();
@ -5071,10 +5076,6 @@ impl TestServer {
}) })
} }
async fn state<'a>(&'a self) -> RwLockReadGuard<'a, Store> {
self.server.store.read().await
}
async fn condition<F>(&mut self, mut predicate: F) async fn condition<F>(&mut self, mut predicate: F)
where where
F: FnMut(&Store) -> bool, F: FnMut(&Store) -> bool,
@ -5083,7 +5084,7 @@ impl TestServer {
self.foreground.parking_forbidden(), self.foreground.parking_forbidden(),
"you must call forbid_parking to use server conditions so we don't block indefinitely" "you must call forbid_parking to use server conditions so we don't block indefinitely"
); );
while !(predicate)(&*self.server.store.read().await) { while !(predicate)(&*self.server.store.lock().await) {
self.foreground.start_waiting(); self.foreground.start_waiting();
self.notifications.next().await; self.notifications.next().await;
self.foreground.finish_waiting(); self.foreground.finish_waiting();

169
crates/collab/src/rpc.rs
View file

@ -51,7 +51,7 @@ use std::{
}; };
use time::OffsetDateTime; use time::OffsetDateTime;
use tokio::{ use tokio::{
sync::{RwLock, RwLockReadGuard, RwLockWriteGuard}, sync::{Mutex, MutexGuard},
time::Sleep, time::Sleep,
}; };
use tower::ServiceBuilder; use tower::ServiceBuilder;
@ -97,7 +97,7 @@ impl<R: RequestMessage> Response<R> {
pub struct Server { pub struct Server {
peer: Arc<Peer>, peer: Arc<Peer>,
pub(crate) store: RwLock<Store>, pub(crate) store: Mutex<Store>,
app_state: Arc<AppState>, app_state: Arc<AppState>,
handlers: HashMap<TypeId, MessageHandler>, handlers: HashMap<TypeId, MessageHandler>,
notifications: Option<mpsc::UnboundedSender<()>>, notifications: Option<mpsc::UnboundedSender<()>>,
@ -115,13 +115,8 @@ pub struct RealExecutor;
const MESSAGE_COUNT_PER_PAGE: usize = 100; const MESSAGE_COUNT_PER_PAGE: usize = 100;
const MAX_MESSAGE_LEN: usize = 1024; const MAX_MESSAGE_LEN: usize = 1024;
struct StoreReadGuard<'a> { pub(crate) struct StoreGuard<'a> {
guard: RwLockReadGuard<'a, Store>, guard: MutexGuard<'a, Store>,
_not_send: PhantomData<Rc<()>>,
}
struct StoreWriteGuard<'a> {
guard: RwLockWriteGuard<'a, Store>,
_not_send: PhantomData<Rc<()>>, _not_send: PhantomData<Rc<()>>,
} }
@ -129,7 +124,7 @@ struct StoreWriteGuard<'a> {
pub struct ServerSnapshot<'a> { pub struct ServerSnapshot<'a> {
peer: &'a Peer, peer: &'a Peer,
#[serde(serialize_with = "serialize_deref")] #[serde(serialize_with = "serialize_deref")]
store: RwLockReadGuard<'a, Store>, store: StoreGuard<'a>,
} }
pub fn serialize_deref<S, T, U>(value: &T, serializer: S) -> Result<S::Ok, S::Error> pub fn serialize_deref<S, T, U>(value: &T, serializer: S) -> Result<S::Ok, S::Error>
@ -384,7 +379,7 @@ impl Server {
).await?; ).await?;
{ {
let mut store = this.store_mut().await; let mut store = this.store().await;
store.add_connection(connection_id, user_id, user.admin); store.add_connection(connection_id, user_id, user.admin);
this.peer.send(connection_id, store.build_initial_contacts_update(contacts))?; this.peer.send(connection_id, store.build_initial_contacts_update(contacts))?;
@ -471,7 +466,7 @@ impl Server {
let mut projects_to_unregister = Vec::new(); let mut projects_to_unregister = Vec::new();
let removed_user_id; let removed_user_id;
{ {
let mut store = self.store_mut().await; let mut store = self.store().await;
let removed_connection = store.remove_connection(connection_id)?; let removed_connection = store.remove_connection(connection_id)?;
for (project_id, project) in removed_connection.hosted_projects { for (project_id, project) in removed_connection.hosted_projects {
@ -605,7 +600,7 @@ impl Server {
.await .await
.user_id_for_connection(request.sender_id)?; .user_id_for_connection(request.sender_id)?;
let project_id = self.app_state.db.register_project(user_id).await?; let project_id = self.app_state.db.register_project(user_id).await?;
self.store_mut() self.store()
.await .await
.register_project(request.sender_id, project_id)?; .register_project(request.sender_id, project_id)?;
@ -623,7 +618,7 @@ impl Server {
) -> Result<()> { ) -> Result<()> {
let project_id = ProjectId::from_proto(request.payload.project_id); let project_id = ProjectId::from_proto(request.payload.project_id);
let (user_id, project) = { let (user_id, project) = {
let mut state = self.store_mut().await; let mut state = self.store().await;
let project = state.unregister_project(project_id, request.sender_id)?; let project = state.unregister_project(project_id, request.sender_id)?;
(state.user_id_for_connection(request.sender_id)?, project) (state.user_id_for_connection(request.sender_id)?, project)
}; };
@ -725,7 +720,7 @@ impl Server {
return Err(anyhow!("no such project"))?; return Err(anyhow!("no such project"))?;
} }
self.store_mut().await.request_join_project( self.store().await.request_join_project(
guest_user_id, guest_user_id,
project_id, project_id,
response.into_receipt(), response.into_receipt(),
@ -747,7 +742,7 @@ impl Server {
let host_user_id; let host_user_id;
{ {
let mut state = self.store_mut().await; let mut state = self.store().await;
let project_id = ProjectId::from_proto(request.payload.project_id); let project_id = ProjectId::from_proto(request.payload.project_id);
let project = state.project(project_id)?; let project = state.project(project_id)?;
if project.host_connection_id != request.sender_id { if project.host_connection_id != request.sender_id {
@ -791,20 +786,10 @@ impl Server {
let worktrees = project let worktrees = project
.worktrees .worktrees
.iter() .iter()
.filter_map(|(id, shared_worktree)| { .map(|(id, worktree)| proto::WorktreeMetadata {
let worktree = project.worktrees.get(&id)?; id: *id,
Some(proto::Worktree { root_name: worktree.root_name.clone(),
id: *id, visible: worktree.visible,
root_name: worktree.root_name.clone(),
entries: shared_worktree.entries.values().cloned().collect(),
diagnostic_summaries: shared_worktree
.diagnostic_summaries
.values()
.cloned()
.collect(),
visible: worktree.visible,
scan_id: shared_worktree.scan_id,
})
}) })
.collect::<Vec<_>>(); .collect::<Vec<_>>();
@ -840,14 +825,15 @@ impl Server {
} }
} }
for (receipt, replica_id) in receipts_with_replica_ids { // First, we send the metadata associated with each worktree.
for (receipt, replica_id) in &receipts_with_replica_ids {
self.peer.respond( self.peer.respond(
receipt, receipt.clone(),
proto::JoinProjectResponse { proto::JoinProjectResponse {
variant: Some(proto::join_project_response::Variant::Accept( variant: Some(proto::join_project_response::Variant::Accept(
proto::join_project_response::Accept { proto::join_project_response::Accept {
worktrees: worktrees.clone(), worktrees: worktrees.clone(),
replica_id: replica_id as u32, replica_id: *replica_id as u32,
collaborators: collaborators.clone(), collaborators: collaborators.clone(),
language_servers: project.language_servers.clone(), language_servers: project.language_servers.clone(),
}, },
@ -855,6 +841,43 @@ impl Server {
}, },
)?; )?;
} }
for (worktree_id, worktree) in &project.worktrees {
#[cfg(any(test, feature = "test-support"))]
const MAX_CHUNK_SIZE: usize = 2;
#[cfg(not(any(test, feature = "test-support")))]
const MAX_CHUNK_SIZE: usize = 256;
// Stream this worktree's entries.
let message = proto::UpdateWorktree {
project_id: project_id.to_proto(),
worktree_id: *worktree_id,
root_name: worktree.root_name.clone(),
updated_entries: worktree.entries.values().cloned().collect(),
removed_entries: Default::default(),
scan_id: worktree.scan_id,
is_last_update: worktree.is_complete,
};
for update in proto::split_worktree_update(message, MAX_CHUNK_SIZE) {
for (receipt, _) in &receipts_with_replica_ids {
self.peer.send(receipt.sender_id, update.clone())?;
}
}
// Stream this worktree's diagnostics.
for summary in worktree.diagnostic_summaries.values() {
for (receipt, _) in &receipts_with_replica_ids {
self.peer.send(
receipt.sender_id,
proto::UpdateDiagnosticSummary {
project_id: project_id.to_proto(),
worktree_id: *worktree_id,
summary: Some(summary.clone()),
},
)?;
}
}
}
} }
self.update_user_contacts(host_user_id).await?; self.update_user_contacts(host_user_id).await?;
@ -869,7 +892,7 @@ impl Server {
let project_id = ProjectId::from_proto(request.payload.project_id); let project_id = ProjectId::from_proto(request.payload.project_id);
let project; let project;
{ {
let mut store = self.store_mut().await; let mut store = self.store().await;
project = store.leave_project(sender_id, project_id)?; project = store.leave_project(sender_id, project_id)?;
tracing::info!( tracing::info!(
%project_id, %project_id,
@ -920,7 +943,7 @@ impl Server {
let project_id = ProjectId::from_proto(request.payload.project_id); let project_id = ProjectId::from_proto(request.payload.project_id);
let user_id; let user_id;
{ {
let mut state = self.store_mut().await; let mut state = self.store().await;
user_id = state.user_id_for_connection(request.sender_id)?; user_id = state.user_id_for_connection(request.sender_id)?;
let guest_connection_ids = state let guest_connection_ids = state
.read_project(project_id, request.sender_id)? .read_project(project_id, request.sender_id)?
@ -939,7 +962,7 @@ impl Server {
self: Arc<Server>, self: Arc<Server>,
request: TypedEnvelope<proto::RegisterProjectActivity>, request: TypedEnvelope<proto::RegisterProjectActivity>,
) -> Result<()> { ) -> Result<()> {
self.store_mut().await.register_project_activity( self.store().await.register_project_activity(
ProjectId::from_proto(request.payload.project_id), ProjectId::from_proto(request.payload.project_id),
request.sender_id, request.sender_id,
)?; )?;
@ -954,7 +977,7 @@ impl Server {
let project_id = ProjectId::from_proto(request.payload.project_id); let project_id = ProjectId::from_proto(request.payload.project_id);
let worktree_id = request.payload.worktree_id; let worktree_id = request.payload.worktree_id;
let (connection_ids, metadata_changed, extension_counts) = { let (connection_ids, metadata_changed, extension_counts) = {
let mut store = self.store_mut().await; let mut store = self.store().await;
let (connection_ids, metadata_changed, extension_counts) = store.update_worktree( let (connection_ids, metadata_changed, extension_counts) = store.update_worktree(
request.sender_id, request.sender_id,
project_id, project_id,
@ -963,6 +986,7 @@ impl Server {
&request.payload.removed_entries, &request.payload.removed_entries,
&request.payload.updated_entries, &request.payload.updated_entries,
request.payload.scan_id, request.payload.scan_id,
request.payload.is_last_update,
)?; )?;
(connection_ids, metadata_changed, extension_counts.clone()) (connection_ids, metadata_changed, extension_counts.clone())
}; };
@ -995,7 +1019,7 @@ impl Server {
.summary .summary
.clone() .clone()
.ok_or_else(|| anyhow!("invalid summary"))?; .ok_or_else(|| anyhow!("invalid summary"))?;
let receiver_ids = self.store_mut().await.update_diagnostic_summary( let receiver_ids = self.store().await.update_diagnostic_summary(
ProjectId::from_proto(request.payload.project_id), ProjectId::from_proto(request.payload.project_id),
request.payload.worktree_id, request.payload.worktree_id,
request.sender_id, request.sender_id,
@ -1013,7 +1037,7 @@ impl Server {
self: Arc<Server>, self: Arc<Server>,
request: TypedEnvelope<proto::StartLanguageServer>, request: TypedEnvelope<proto::StartLanguageServer>,
) -> Result<()> { ) -> Result<()> {
let receiver_ids = self.store_mut().await.start_language_server( let receiver_ids = self.store().await.start_language_server(
ProjectId::from_proto(request.payload.project_id), ProjectId::from_proto(request.payload.project_id),
request.sender_id, request.sender_id,
request request
@ -1052,20 +1076,23 @@ impl Server {
where where
T: EntityMessage + RequestMessage, T: EntityMessage + RequestMessage,
{ {
let project_id = ProjectId::from_proto(request.payload.remote_entity_id());
let host_connection_id = self let host_connection_id = self
.store() .store()
.await .await
.read_project( .read_project(project_id, request.sender_id)?
ProjectId::from_proto(request.payload.remote_entity_id()),
request.sender_id,
)?
.host_connection_id; .host_connection_id;
let payload = self
.peer
.forward_request(request.sender_id, host_connection_id, request.payload)
.await?;
response.send( // Ensure project still exists by the time we get the response from the host.
self.peer self.store()
.forward_request(request.sender_id, host_connection_id, request.payload) .await
.await?, .read_project(project_id, request.sender_id)?;
)?;
response.send(payload)?;
Ok(()) Ok(())
} }
@ -1106,7 +1133,7 @@ impl Server {
) -> Result<()> { ) -> Result<()> {
let project_id = ProjectId::from_proto(request.payload.project_id); let project_id = ProjectId::from_proto(request.payload.project_id);
let receiver_ids = { let receiver_ids = {
let mut store = self.store_mut().await; let mut store = self.store().await;
store.register_project_activity(project_id, request.sender_id)?; store.register_project_activity(project_id, request.sender_id)?;
store.project_connection_ids(project_id, request.sender_id)? store.project_connection_ids(project_id, request.sender_id)?
}; };
@ -1173,7 +1200,7 @@ impl Server {
let leader_id = ConnectionId(request.payload.leader_id); let leader_id = ConnectionId(request.payload.leader_id);
let follower_id = request.sender_id; let follower_id = request.sender_id;
{ {
let mut store = self.store_mut().await; let mut store = self.store().await;
if !store if !store
.project_connection_ids(project_id, follower_id)? .project_connection_ids(project_id, follower_id)?
.contains(&leader_id) .contains(&leader_id)
@ -1198,7 +1225,7 @@ impl Server {
async fn unfollow(self: Arc<Self>, request: TypedEnvelope<proto::Unfollow>) -> Result<()> { async fn unfollow(self: Arc<Self>, request: TypedEnvelope<proto::Unfollow>) -> Result<()> {
let project_id = ProjectId::from_proto(request.payload.project_id); let project_id = ProjectId::from_proto(request.payload.project_id);
let leader_id = ConnectionId(request.payload.leader_id); let leader_id = ConnectionId(request.payload.leader_id);
let mut store = self.store_mut().await; let mut store = self.store().await;
if !store if !store
.project_connection_ids(project_id, request.sender_id)? .project_connection_ids(project_id, request.sender_id)?
.contains(&leader_id) .contains(&leader_id)
@ -1216,7 +1243,7 @@ impl Server {
request: TypedEnvelope<proto::UpdateFollowers>, request: TypedEnvelope<proto::UpdateFollowers>,
) -> Result<()> { ) -> Result<()> {
let project_id = ProjectId::from_proto(request.payload.project_id); let project_id = ProjectId::from_proto(request.payload.project_id);
let mut store = self.store_mut().await; let mut store = self.store().await;
store.register_project_activity(project_id, request.sender_id)?; store.register_project_activity(project_id, request.sender_id)?;
let connection_ids = store.project_connection_ids(project_id, request.sender_id)?; let connection_ids = store.project_connection_ids(project_id, request.sender_id)?;
let leader_id = request let leader_id = request
@ -1474,7 +1501,7 @@ impl Server {
Err(anyhow!("access denied"))?; Err(anyhow!("access denied"))?;
} }
self.store_mut() self.store()
.await .await
.join_channel(request.sender_id, channel_id); .join_channel(request.sender_id, channel_id);
let messages = self let messages = self
@ -1516,7 +1543,7 @@ impl Server {
Err(anyhow!("access denied"))?; Err(anyhow!("access denied"))?;
} }
self.store_mut() self.store()
.await .await
.leave_channel(request.sender_id, channel_id); .leave_channel(request.sender_id, channel_id);
@ -1624,25 +1651,13 @@ impl Server {
Ok(()) Ok(())
} }
async fn store<'a>(self: &'a Arc<Self>) -> StoreReadGuard<'a> { pub(crate) async fn store<'a>(&'a self) -> StoreGuard<'a> {
#[cfg(test)] #[cfg(test)]
tokio::task::yield_now().await; tokio::task::yield_now().await;
let guard = self.store.read().await; let guard = self.store.lock().await;
#[cfg(test)] #[cfg(test)]
tokio::task::yield_now().await; tokio::task::yield_now().await;
StoreReadGuard { StoreGuard {
guard,
_not_send: PhantomData,
}
}
async fn store_mut<'a>(self: &'a Arc<Self>) -> StoreWriteGuard<'a> {
#[cfg(test)]
tokio::task::yield_now().await;
let guard = self.store.write().await;
#[cfg(test)]
tokio::task::yield_now().await;
StoreWriteGuard {
guard, guard,
_not_send: PhantomData, _not_send: PhantomData,
} }
@ -1650,13 +1665,13 @@ impl Server {
pub async fn snapshot<'a>(self: &'a Arc<Self>) -> ServerSnapshot<'a> { pub async fn snapshot<'a>(self: &'a Arc<Self>) -> ServerSnapshot<'a> {
ServerSnapshot { ServerSnapshot {
store: self.store.read().await, store: self.store().await,
peer: &self.peer, peer: &self.peer,
} }
} }
} }
impl<'a> Deref for StoreReadGuard<'a> { impl<'a> Deref for StoreGuard<'a> {
type Target = Store; type Target = Store;
fn deref(&self) -> &Self::Target { fn deref(&self) -> &Self::Target {
@ -1664,21 +1679,13 @@ impl<'a> Deref for StoreReadGuard<'a> {
} }
} }
impl<'a> Deref for StoreWriteGuard<'a> { impl<'a> DerefMut for StoreGuard<'a> {
type Target = Store;
fn deref(&self) -> &Self::Target {
&*self.guard
}
}
impl<'a> DerefMut for StoreWriteGuard<'a> {
fn deref_mut(&mut self) -> &mut Self::Target { fn deref_mut(&mut self) -> &mut Self::Target {
&mut *self.guard &mut *self.guard
} }
} }
impl<'a> Drop for StoreWriteGuard<'a> { impl<'a> Drop for StoreGuard<'a> {
fn drop(&mut self) { fn drop(&mut self) {
#[cfg(test)] #[cfg(test)]
self.check_invariants(); self.check_invariants();

3
crates/collab/src/rpc/store.rs
View file

@ -62,6 +62,7 @@ pub struct Worktree {
#[serde(skip)] #[serde(skip)]
pub diagnostic_summaries: BTreeMap<PathBuf, proto::DiagnosticSummary>, pub diagnostic_summaries: BTreeMap<PathBuf, proto::DiagnosticSummary>,
pub scan_id: u64, pub scan_id: u64,
pub is_complete: bool,
} }
#[derive(Default)] #[derive(Default)]
@ -615,6 +616,7 @@ impl Store {
removed_entries: &[u64], removed_entries: &[u64],
updated_entries: &[proto::Entry], updated_entries: &[proto::Entry],
scan_id: u64, scan_id: u64,
is_last_update: bool,
) -> Result<(Vec<ConnectionId>, bool, HashMap<String, usize>)> { ) -> Result<(Vec<ConnectionId>, bool, HashMap<String, usize>)> {
let project = self.write_project(project_id, connection_id)?; let project = self.write_project(project_id, connection_id)?;
let connection_ids = project.connection_ids(); let connection_ids = project.connection_ids();
@ -657,6 +659,7 @@ impl Store {
} }
worktree.scan_id = scan_id; worktree.scan_id = scan_id;
worktree.is_complete = is_last_update;
Ok(( Ok((
connection_ids, connection_ids,
metadata_changed, metadata_changed,

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

@ -507,10 +507,9 @@ impl Project {
let mut worktrees = Vec::new(); let mut worktrees = Vec::new();
for worktree in response.worktrees { for worktree in response.worktrees {
let (worktree, load_task) = cx let worktree = cx
.update(|cx| Worktree::remote(remote_id, replica_id, worktree, client.clone(), cx)); .update(|cx| Worktree::remote(remote_id, replica_id, worktree, client.clone(), cx));
worktrees.push(worktree); worktrees.push(worktree);
load_task.detach();
} }
let (opened_buffer_tx, opened_buffer_rx) = watch::channel(); let (opened_buffer_tx, opened_buffer_rx) = watch::channel();
@ -1102,7 +1101,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,
@ -1145,7 +1144,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,
@ -1188,7 +1187,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,
@ -1221,7 +1220,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,
@ -1352,12 +1351,13 @@ impl Project {
let client = self.client.clone(); let client = self.client.clone();
cx.foreground() cx.foreground()
.spawn(async move { .spawn(async move {
share.await?;
client.send(proto::RespondToJoinProjectRequest { client.send(proto::RespondToJoinProjectRequest {
requester_id, requester_id,
project_id, project_id,
allow, allow,
}) })?;
share.await?;
anyhow::Ok(())
}) })
.detach_and_log_err(cx); .detach_and_log_err(cx);
} }
@ -4552,18 +4552,9 @@ impl Project {
{ {
this.worktrees.push(WorktreeHandle::Strong(old_worktree)); this.worktrees.push(WorktreeHandle::Strong(old_worktree));
} else { } else {
let worktree = proto::Worktree { let worktree =
id: worktree.id,
root_name: worktree.root_name,
entries: Default::default(),
diagnostic_summaries: Default::default(),
visible: worktree.visible,
scan_id: 0,
};
let (worktree, load_task) =
Worktree::remote(remote_id, replica_id, worktree, client.clone(), cx); Worktree::remote(remote_id, replica_id, worktree, client.clone(), cx);
this.add_worktree(&worktree, cx); this.add_worktree(&worktree, cx);
load_task.detach();
} }
} }
@ -4587,8 +4578,8 @@ impl Project {
if let Some(worktree) = this.worktree_for_id(worktree_id, cx) { if let Some(worktree) = this.worktree_for_id(worktree_id, cx) {
worktree.update(cx, |worktree, _| { worktree.update(cx, |worktree, _| {
let worktree = worktree.as_remote_mut().unwrap(); let worktree = worktree.as_remote_mut().unwrap();
worktree.update_from_remote(envelope) worktree.update_from_remote(envelope.payload);
})?; });
} }
Ok(()) Ok(())
}) })
@ -8125,7 +8116,10 @@ mod tests {
} }
#[gpui::test(retries = 5)] #[gpui::test(retries = 5)]
async fn test_rescan_and_remote_updates(cx: &mut gpui::TestAppContext) { async fn test_rescan_and_remote_updates(
deterministic: Arc<Deterministic>,
cx: &mut gpui::TestAppContext,
) {
let dir = temp_tree(json!({ let dir = temp_tree(json!({
"a": { "a": {
"file1": "", "file1": "",
@ -8169,17 +8163,24 @@ mod tests {
// Create a remote copy of this worktree. // Create a remote copy of this worktree.
let tree = project.read_with(cx, |project, cx| project.worktrees(cx).next().unwrap()); let tree = project.read_with(cx, |project, cx| project.worktrees(cx).next().unwrap());
let initial_snapshot = tree.read_with(cx, |tree, _| tree.as_local().unwrap().snapshot()); let initial_snapshot = tree.read_with(cx, |tree, _| tree.as_local().unwrap().snapshot());
let (remote, load_task) = cx.update(|cx| { let remote = cx.update(|cx| {
Worktree::remote( Worktree::remote(
1, 1,
1, 1,
initial_snapshot.to_proto(&Default::default(), true), proto::WorktreeMetadata {
id: initial_snapshot.id().to_proto(),
root_name: initial_snapshot.root_name().into(),
visible: true,
},
rpc.clone(), rpc.clone(),
cx, cx,
) )
}); });
// tree remote.update(cx, |remote, _| {
load_task.await; let update = initial_snapshot.build_initial_update(1);
remote.as_remote_mut().unwrap().update_from_remote(update);
});
deterministic.run_until_parked();
cx.read(|cx| { cx.read(|cx| {
assert!(!buffer2.read(cx).is_dirty()); assert!(!buffer2.read(cx).is_dirty());
@ -8245,19 +8246,16 @@ mod tests {
// Update the remote worktree. Check that it becomes consistent with the // Update the remote worktree. Check that it becomes consistent with the
// local worktree. // local worktree.
remote.update(cx, |remote, cx| { remote.update(cx, |remote, cx| {
let update_message = tree.read(cx).as_local().unwrap().snapshot().build_update( let update = tree.read(cx).as_local().unwrap().snapshot().build_update(
&initial_snapshot, &initial_snapshot,
1, 1,
1, 1,
true, true,
); );
remote remote.as_remote_mut().unwrap().update_from_remote(update);
.as_remote_mut() });
.unwrap() deterministic.run_until_parked();
.snapshot remote.read_with(cx, |remote, _| {
.apply_remote_update(update_message)
.unwrap();
assert_eq!( assert_eq!(
remote remote
.paths() .paths()

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

@ -7,9 +7,9 @@ use super::{
}; };
use ::ignore::gitignore::{Gitignore, GitignoreBuilder}; use ::ignore::gitignore::{Gitignore, GitignoreBuilder};
use anyhow::{anyhow, Context, Result}; use anyhow::{anyhow, Context, Result};
use client::{proto, Client, TypedEnvelope}; 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},
@ -40,11 +40,11 @@ use std::{
ffi::{OsStr, OsString}, ffi::{OsStr, OsString},
fmt, fmt,
future::Future, future::Future,
mem,
ops::{Deref, DerefMut}, ops::{Deref, DerefMut},
os::unix::prelude::{OsStrExt, OsStringExt}, os::unix::prelude::{OsStrExt, OsStringExt},
path::{Path, PathBuf}, path::{Path, PathBuf},
sync::{atomic::AtomicUsize, Arc}, sync::{atomic::AtomicUsize, Arc},
task::Poll,
time::{Duration, SystemTime}, time::{Duration, SystemTime},
}; };
use sum_tree::{Bias, Edit, SeekTarget, SumTree, TreeMap}; use sum_tree::{Bias, Edit, SeekTarget, SumTree, TreeMap};
@ -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>>,
last_scan_id_rx: watch::Receiver<usize>, 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,
@ -96,6 +96,7 @@ pub struct Snapshot {
entries_by_path: SumTree<Entry>, entries_by_path: SumTree<Entry>,
entries_by_id: SumTree<PathEntry>, entries_by_id: SumTree<PathEntry>,
scan_id: usize, scan_id: usize,
is_complete: bool,
} }
#[derive(Clone)] #[derive(Clone)]
@ -124,13 +125,16 @@ impl DerefMut for LocalSnapshot {
#[derive(Clone, Debug)] #[derive(Clone, Debug)]
enum ScanState { enum ScanState {
Idle, Idle,
Scanning, /// The worktree is performing its initial scan of the filesystem.
Initializing,
/// The worktree is updating in response to filesystem events.
Updating,
Err(Arc<anyhow::Error>), Err(Arc<anyhow::Error>),
} }
struct ShareState { struct ShareState {
project_id: u64, project_id: u64,
snapshots_tx: Sender<LocalSnapshot>, snapshots_tx: watch::Sender<LocalSnapshot>,
_maintain_remote_snapshot: Option<Task<Option<()>>>, _maintain_remote_snapshot: Option<Task<Option<()>>>,
} }
@ -171,10 +175,10 @@ impl Worktree {
pub fn remote( pub fn remote(
project_remote_id: u64, project_remote_id: u64,
replica_id: ReplicaId, replica_id: ReplicaId,
worktree: proto::Worktree, worktree: proto::WorktreeMetadata,
client: Arc<Client>, client: Arc<Client>,
cx: &mut MutableAppContext, cx: &mut MutableAppContext,
) -> (ModelHandle<Self>, Task<()>) { ) -> ModelHandle<Self> {
let remote_id = worktree.id; let remote_id = worktree.id;
let root_char_bag: CharBag = worktree let root_char_bag: CharBag = worktree
.root_name .root_name
@ -189,13 +193,13 @@ impl Worktree {
root_char_bag, root_char_bag,
entries_by_path: Default::default(), entries_by_path: Default::default(),
entries_by_id: Default::default(), entries_by_id: Default::default(),
scan_id: worktree.scan_id as usize, scan_id: 0,
is_complete: false,
}; };
let (updates_tx, mut updates_rx) = mpsc::unbounded(); let (updates_tx, mut updates_rx) = mpsc::unbounded();
let background_snapshot = Arc::new(Mutex::new(snapshot.clone())); let background_snapshot = Arc::new(Mutex::new(snapshot.clone()));
let (mut snapshot_updated_tx, mut snapshot_updated_rx) = watch::channel(); let (mut snapshot_updated_tx, mut snapshot_updated_rx) = watch::channel();
let (mut last_scan_id_tx, last_scan_id_rx) = watch::channel_with(worktree.scan_id as usize);
let worktree_handle = cx.add_model(|_: &mut ModelContext<Worktree>| { let worktree_handle = cx.add_model(|_: &mut ModelContext<Worktree>| {
Worktree::Remote(RemoteWorktree { Worktree::Remote(RemoteWorktree {
project_id: project_remote_id, project_id: project_remote_id,
@ -203,96 +207,50 @@ 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),
last_scan_id_rx, snapshot_subscriptions: Default::default(),
client: client.clone(), client: client.clone(),
diagnostic_summaries: TreeMap::from_ordered_entries( diagnostic_summaries: Default::default(),
worktree.diagnostic_summaries.into_iter().map(|summary| {
(
PathKey(PathBuf::from(summary.path).into()),
DiagnosticSummary {
language_server_id: summary.language_server_id as usize,
error_count: summary.error_count as usize,
warning_count: summary.warning_count as usize,
},
)
}),
),
visible, visible,
}) })
}); });
let deserialize_task = cx.spawn({ cx.background()
let worktree_handle = worktree_handle.clone(); .spawn(async move {
|cx| async move { while let Some(update) = updates_rx.next().await {
let (entries_by_path, entries_by_id) = cx if let Err(error) = background_snapshot.lock().apply_remote_update(update) {
.background() log::error!("error applying worktree update: {}", error);
.spawn(async move { }
let mut entries_by_path_edits = Vec::new();
let mut entries_by_id_edits = Vec::new();
for entry in worktree.entries {
match Entry::try_from((&root_char_bag, entry)) {
Ok(entry) => {
entries_by_id_edits.push(Edit::Insert(PathEntry {
id: entry.id,
path: entry.path.clone(),
is_ignored: entry.is_ignored,
scan_id: 0,
}));
entries_by_path_edits.push(Edit::Insert(entry));
}
Err(err) => log::warn!("error for remote worktree entry {:?}", err),
}
}
let mut entries_by_path = SumTree::new();
let mut entries_by_id = SumTree::new();
entries_by_path.edit(entries_by_path_edits, &());
entries_by_id.edit(entries_by_id_edits, &());
(entries_by_path, entries_by_id)
})
.await;
{
let mut snapshot = background_snapshot.lock();
snapshot.entries_by_path = entries_by_path;
snapshot.entries_by_id = entries_by_id;
snapshot_updated_tx.send(()).await.ok(); snapshot_updated_tx.send(()).await.ok();
} }
})
.detach();
cx.background() cx.spawn(|mut cx| {
.spawn(async move { let this = worktree_handle.downgrade();
while let Some(update) = updates_rx.next().await { async move {
if let Err(error) = while let Some(_) = snapshot_updated_rx.recv().await {
background_snapshot.lock().apply_remote_update(update) if let Some(this) = this.upgrade(&cx) {
{ this.update(&mut cx, |this, cx| {
log::error!("error applying worktree update: {}", error); 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;
}
} }
snapshot_updated_tx.send(()).await.ok(); });
} } else {
}) break;
.detach();
cx.spawn(|mut cx| {
let this = worktree_handle.downgrade();
async move {
while let Some(_) = snapshot_updated_rx.recv().await {
if let Some(this) = this.upgrade(&cx) {
this.update(&mut cx, |this, cx| {
this.poll_snapshot(cx);
let this = this.as_remote_mut().unwrap();
*last_scan_id_tx.borrow_mut() = this.snapshot.scan_id;
});
} else {
break;
}
}
} }
}) }
.detach();
} }
}); })
(worktree_handle, deserialize_task) .detach();
worktree_handle
} }
pub fn as_local(&self) -> Option<&LocalWorktree> { pub fn as_local(&self) -> Option<&LocalWorktree> {
@ -376,38 +334,9 @@ impl Worktree {
fn poll_snapshot(&mut self, cx: &mut ModelContext<Self>) { fn poll_snapshot(&mut self, cx: &mut ModelContext<Self>) {
match self { match self {
Self::Local(worktree) => { Self::Local(worktree) => worktree.poll_snapshot(false, cx),
let is_fake_fs = worktree.fs.is_fake(); Self::Remote(worktree) => worktree.poll_snapshot(cx),
worktree.snapshot = worktree.background_snapshot.lock().clone();
if worktree.is_scanning() {
if worktree.poll_task.is_none() {
worktree.poll_task = Some(cx.spawn_weak(|this, mut cx| async move {
if is_fake_fs {
#[cfg(any(test, feature = "test-support"))]
cx.background().simulate_random_delay().await;
} else {
smol::Timer::after(Duration::from_millis(100)).await;
}
if let Some(this) = this.upgrade(&cx) {
this.update(&mut cx, |this, cx| {
this.as_local_mut().unwrap().poll_task = None;
this.poll_snapshot(cx);
});
}
}));
}
} else {
worktree.poll_task.take();
cx.emit(Event::UpdatedEntries);
}
}
Self::Remote(worktree) => {
worktree.snapshot = worktree.background_snapshot.lock().clone();
cx.emit(Event::UpdatedEntries);
}
}; };
cx.notify();
} }
} }
@ -435,7 +364,8 @@ impl LocalWorktree {
.context("failed to stat worktree path")?; .context("failed to stat worktree path")?;
let (scan_states_tx, mut scan_states_rx) = mpsc::unbounded(); let (scan_states_tx, mut scan_states_rx) = mpsc::unbounded();
let (mut last_scan_state_tx, last_scan_state_rx) = watch::channel_with(ScanState::Scanning); let (mut last_scan_state_tx, last_scan_state_rx) =
watch::channel_with(ScanState::Initializing);
let tree = cx.add_model(move |cx: &mut ModelContext<Worktree>| { let tree = cx.add_model(move |cx: &mut ModelContext<Worktree>| {
let mut snapshot = LocalSnapshot { let mut snapshot = LocalSnapshot {
abs_path, abs_path,
@ -449,6 +379,7 @@ impl LocalWorktree {
entries_by_path: Default::default(), entries_by_path: Default::default(),
entries_by_id: Default::default(), entries_by_id: Default::default(),
scan_id: 0, scan_id: 0,
is_complete: true,
}, },
}; };
if let Some(metadata) = metadata { if let Some(metadata) = metadata {
@ -479,11 +410,7 @@ impl LocalWorktree {
while let Some(scan_state) = scan_states_rx.next().await { while let Some(scan_state) = scan_states_rx.next().await {
if let Some(this) = this.upgrade(&cx) { if let Some(this) = this.upgrade(&cx) {
last_scan_state_tx.blocking_send(scan_state).ok(); last_scan_state_tx.blocking_send(scan_state).ok();
this.update(&mut cx, |this, cx| { this.update(&mut cx, |this, cx| this.poll_snapshot(cx));
this.poll_snapshot(cx);
this.as_local().unwrap().broadcast_snapshot()
})
.await;
} else { } else {
break; break;
} }
@ -567,22 +494,53 @@ impl LocalWorktree {
Ok(updated) Ok(updated)
} }
fn poll_snapshot(&mut self, force: bool, cx: &mut ModelContext<Worktree>) {
self.poll_task.take();
match self.scan_state() {
ScanState::Idle => {
self.snapshot = self.background_snapshot.lock().clone();
if let Some(share) = self.share.as_mut() {
*share.snapshots_tx.borrow_mut() = self.snapshot.clone();
}
cx.emit(Event::UpdatedEntries);
}
ScanState::Initializing => {
let is_fake_fs = self.fs.is_fake();
self.snapshot = self.background_snapshot.lock().clone();
self.poll_task = Some(cx.spawn_weak(|this, mut cx| async move {
if is_fake_fs {
#[cfg(any(test, feature = "test-support"))]
cx.background().simulate_random_delay().await;
} else {
smol::Timer::after(Duration::from_millis(100)).await;
}
if let Some(this) = this.upgrade(&cx) {
this.update(&mut cx, |this, cx| this.poll_snapshot(cx));
}
}));
cx.emit(Event::UpdatedEntries);
}
_ => {
if force {
self.snapshot = self.background_snapshot.lock().clone();
}
}
}
cx.notify();
}
pub fn scan_complete(&self) -> impl Future<Output = ()> { pub fn scan_complete(&self) -> impl Future<Output = ()> {
let mut scan_state_rx = self.last_scan_state_rx.clone(); let mut scan_state_rx = self.last_scan_state_rx.clone();
async move { async move {
let mut scan_state = Some(scan_state_rx.borrow().clone()); let mut scan_state = Some(scan_state_rx.borrow().clone());
while let Some(ScanState::Scanning) = scan_state { while let Some(ScanState::Initializing | ScanState::Updating) = scan_state {
scan_state = scan_state_rx.recv().await; scan_state = scan_state_rx.recv().await;
} }
} }
} }
fn is_scanning(&self) -> bool { fn scan_state(&self) -> ScanState {
if let ScanState::Scanning = *self.last_scan_state_rx.borrow() { self.last_scan_state_rx.borrow().clone()
true
} else {
false
}
} }
pub fn snapshot(&self) -> LocalSnapshot { pub fn snapshot(&self) -> LocalSnapshot {
@ -612,7 +570,6 @@ impl LocalWorktree {
.refresh_entry(path, abs_path, None, cx) .refresh_entry(path, abs_path, None, cx)
}) })
.await?; .await?;
this.update(&mut cx, |this, cx| this.poll_snapshot(cx));
Ok(( Ok((
File { File {
entry_id: Some(entry.id), entry_id: Some(entry.id),
@ -710,16 +667,14 @@ impl LocalWorktree {
Some(cx.spawn(|this, mut cx| async move { Some(cx.spawn(|this, mut cx| async move {
delete.await?; delete.await?;
this.update(&mut cx, |this, _| {
let this = this.as_local_mut().unwrap();
let mut snapshot = this.background_snapshot.lock();
snapshot.delete_entry(entry_id);
});
this.update(&mut cx, |this, cx| { this.update(&mut cx, |this, cx| {
this.poll_snapshot(cx); let this = this.as_local_mut().unwrap();
this.as_local().unwrap().broadcast_snapshot() {
}) let mut snapshot = this.background_snapshot.lock();
.await; snapshot.delete_entry(entry_id);
}
this.poll_snapshot(true, cx);
});
Ok(()) Ok(())
})) }))
} }
@ -755,11 +710,6 @@ impl LocalWorktree {
) )
}) })
.await?; .await?;
this.update(&mut cx, |this, cx| {
this.poll_snapshot(cx);
this.as_local().unwrap().broadcast_snapshot()
})
.await;
Ok(entry) Ok(entry)
})) }))
} }
@ -795,11 +745,6 @@ impl LocalWorktree {
) )
}) })
.await?; .await?;
this.update(&mut cx, |this, cx| {
this.poll_snapshot(cx);
this.as_local().unwrap().broadcast_snapshot()
})
.await;
Ok(entry) Ok(entry)
})) }))
} }
@ -833,11 +778,6 @@ impl LocalWorktree {
.refresh_entry(path, abs_path, None, cx) .refresh_entry(path, abs_path, None, cx)
}) })
.await?; .await?;
this.update(&mut cx, |this, cx| {
this.poll_snapshot(cx);
this.as_local().unwrap().broadcast_snapshot()
})
.await;
Ok(entry) Ok(entry)
}) })
} }
@ -870,61 +810,55 @@ impl LocalWorktree {
let this = this let this = this
.upgrade(&cx) .upgrade(&cx)
.ok_or_else(|| anyhow!("worktree was dropped"))?; .ok_or_else(|| anyhow!("worktree was dropped"))?;
let (entry, snapshot, snapshots_tx) = this.read_with(&cx, |this, _| { this.update(&mut cx, |this, cx| {
let this = this.as_local().unwrap(); let this = this.as_local_mut().unwrap();
let mut snapshot = this.background_snapshot.lock(); let inserted_entry;
entry.is_ignored = snapshot {
.ignore_stack_for_path(&path, entry.is_dir()) let mut snapshot = this.background_snapshot.lock();
.is_path_ignored(&path, entry.is_dir()); entry.is_ignored = snapshot
if let Some(old_path) = old_path { .ignore_stack_for_path(&path, entry.is_dir())
snapshot.remove_path(&old_path); .is_path_ignored(&path, entry.is_dir());
if let Some(old_path) = old_path {
snapshot.remove_path(&old_path);
}
inserted_entry = snapshot.insert_entry(entry, fs.as_ref());
snapshot.scan_id += 1;
} }
let entry = snapshot.insert_entry(entry, fs.as_ref()); this.poll_snapshot(true, cx);
snapshot.scan_id += 1; Ok(inserted_entry)
let snapshots_tx = this.share.as_ref().map(|s| s.snapshots_tx.clone()); })
(entry, snapshot.clone(), snapshots_tx)
});
this.update(&mut cx, |this, cx| this.poll_snapshot(cx));
if let Some(snapshots_tx) = snapshots_tx {
snapshots_tx.send(snapshot).await.ok();
}
Ok(entry)
}) })
} }
pub fn share(&mut self, project_id: u64, cx: &mut ModelContext<Worktree>) -> Task<Result<()>> { pub fn share(&mut self, project_id: u64, cx: &mut ModelContext<Worktree>) -> Task<Result<()>> {
let (share_tx, share_rx) = oneshot::channel(); let (share_tx, share_rx) = oneshot::channel();
let (snapshots_to_send_tx, snapshots_to_send_rx) =
smol::channel::unbounded::<LocalSnapshot>();
if self.share.is_some() { if self.share.is_some() {
let _ = share_tx.send(Ok(())); let _ = share_tx.send(Ok(()));
} else { } else {
let (snapshots_tx, mut snapshots_rx) = watch::channel_with(self.snapshot());
let rpc = self.client.clone(); let rpc = self.client.clone();
let worktree_id = cx.model_id() as u64; let worktree_id = cx.model_id() as u64;
let maintain_remote_snapshot = cx.background().spawn({ let maintain_remote_snapshot = cx.background().spawn({
let rpc = rpc.clone(); let rpc = rpc.clone();
let diagnostic_summaries = self.diagnostic_summaries.clone(); let diagnostic_summaries = self.diagnostic_summaries.clone();
async move { async move {
let mut prev_snapshot = match snapshots_to_send_rx.recv().await { let mut prev_snapshot = match snapshots_rx.recv().await {
Ok(snapshot) => { Some(snapshot) => {
if let Err(error) = rpc let update = proto::UpdateWorktree {
.request(proto::UpdateWorktree { project_id,
project_id, worktree_id,
worktree_id, root_name: snapshot.root_name().to_string(),
root_name: snapshot.root_name().to_string(), updated_entries: snapshot
updated_entries: snapshot .entries_by_path
.entries_by_path .iter()
.iter() .map(Into::into)
.filter(|e| !e.is_ignored) .collect(),
.map(Into::into) removed_entries: Default::default(),
.collect(), scan_id: snapshot.scan_id as u64,
removed_entries: Default::default(), is_last_update: true,
scan_id: snapshot.scan_id as u64, };
}) if let Err(error) = send_worktree_update(&rpc, update).await {
.await
{
let _ = share_tx.send(Err(error)); let _ = share_tx.send(Err(error));
return Err(anyhow!("failed to send initial update worktree")); return Err(anyhow!("failed to send initial update worktree"));
} else { } else {
@ -932,8 +866,10 @@ impl LocalWorktree {
snapshot snapshot
} }
} }
Err(error) => { None => {
let _ = share_tx.send(Err(error.into())); share_tx
.send(Err(anyhow!("worktree dropped before share completed")))
.ok();
return Err(anyhow!("failed to send initial update worktree")); return Err(anyhow!("failed to send initial update worktree"));
} }
}; };
@ -946,44 +882,12 @@ impl LocalWorktree {
})?; })?;
} }
// Stream ignored entries in chunks. while let Some(snapshot) = snapshots_rx.recv().await {
{ send_worktree_update(
let mut ignored_entries = prev_snapshot &rpc,
.entries_by_path snapshot.build_update(&prev_snapshot, project_id, worktree_id, true),
.iter() )
.filter(|e| e.is_ignored); .await?;
let mut ignored_entries_to_send = Vec::new();
loop {
#[cfg(any(test, feature = "test-support"))]
const CHUNK_SIZE: usize = 2;
#[cfg(not(any(test, feature = "test-support")))]
const CHUNK_SIZE: usize = 256;
let entry = ignored_entries.next();
if ignored_entries_to_send.len() >= CHUNK_SIZE || entry.is_none() {
rpc.request(proto::UpdateWorktree {
project_id,
worktree_id,
root_name: prev_snapshot.root_name().to_string(),
updated_entries: mem::take(&mut ignored_entries_to_send),
removed_entries: Default::default(),
scan_id: prev_snapshot.scan_id as u64,
})
.await?;
}
if let Some(entry) = entry {
ignored_entries_to_send.push(entry.into());
} else {
break;
}
}
}
while let Ok(snapshot) = snapshots_to_send_rx.recv().await {
let message =
snapshot.build_update(&prev_snapshot, project_id, worktree_id, true);
rpc.request(message).await?;
prev_snapshot = snapshot; prev_snapshot = snapshot;
} }
@ -993,18 +897,12 @@ impl LocalWorktree {
}); });
self.share = Some(ShareState { self.share = Some(ShareState {
project_id, project_id,
snapshots_tx: snapshots_to_send_tx.clone(), snapshots_tx,
_maintain_remote_snapshot: Some(maintain_remote_snapshot), _maintain_remote_snapshot: Some(maintain_remote_snapshot),
}); });
} }
cx.spawn_weak(|this, cx| async move { cx.foreground().spawn(async move {
if let Some(this) = this.upgrade(&cx) {
this.read_with(&cx, |this, _| {
let this = this.as_local().unwrap();
let _ = snapshots_to_send_tx.try_send(this.snapshot());
});
}
share_rx share_rx
.await .await
.unwrap_or_else(|_| Err(anyhow!("share ended"))) .unwrap_or_else(|_| Err(anyhow!("share ended")))
@ -1018,23 +916,6 @@ impl LocalWorktree {
pub fn is_shared(&self) -> bool { pub fn is_shared(&self) -> bool {
self.share.is_some() self.share.is_some()
} }
fn broadcast_snapshot(&self) -> impl Future<Output = ()> {
let mut to_send = None;
if !self.is_scanning() {
if let Some(share) = self.share.as_ref() {
to_send = Some((self.snapshot(), share.snapshots_tx.clone()));
}
}
async move {
if let Some((snapshot, snapshots_to_send_tx)) = to_send {
if let Err(err) = snapshots_to_send_tx.send(snapshot).await {
log::error!("error submitting snapshot to send {}", err);
}
}
}
}
} }
impl RemoteWorktree { impl RemoteWorktree {
@ -1042,31 +923,45 @@ impl RemoteWorktree {
self.snapshot.clone() self.snapshot.clone()
} }
fn poll_snapshot(&mut self, cx: &mut ModelContext<Worktree>) {
self.snapshot = self.background_snapshot.lock().clone();
cx.emit(Event::UpdatedEntries);
cx.notify();
}
pub fn disconnected_from_host(&mut self) { pub fn disconnected_from_host(&mut self) {
self.updates_tx.take(); self.updates_tx.take();
self.snapshot_subscriptions.clear();
} }
pub fn update_from_remote( pub fn update_from_remote(&mut self, update: proto::UpdateWorktree) {
&mut self,
envelope: TypedEnvelope<proto::UpdateWorktree>,
) -> Result<()> {
if let Some(updates_tx) = &self.updates_tx { if let Some(updates_tx) = &self.updates_tx {
updates_tx updates_tx
.unbounded_send(envelope.payload) .unbounded_send(update)
.expect("consumer runs to completion"); .expect("consumer runs to completion");
} }
Ok(())
} }
fn wait_for_snapshot(&self, scan_id: usize) -> impl Future<Output = ()> { fn observed_snapshot(&self, scan_id: usize) -> bool {
let mut rx = self.last_scan_id_rx.clone(); self.scan_id > scan_id || (self.scan_id == scan_id && self.is_complete)
async move { }
while let Some(applied_scan_id) = rx.next().await {
if applied_scan_id >= scan_id { fn wait_for_snapshot(&mut self, scan_id: usize) -> impl Future<Output = ()> {
return; let (tx, rx) = oneshot::channel();
} if self.observed_snapshot(scan_id) {
let _ = tx.send(());
} else {
match self
.snapshot_subscriptions
.binary_search_by_key(&scan_id, |probe| probe.0)
{
Ok(ix) | Err(ix) => self.snapshot_subscriptions.insert(ix, (scan_id, tx)),
} }
} }
async move {
let _ = rx.await;
}
} }
pub fn update_diagnostic_summary( pub fn update_diagnostic_summary(
@ -1088,7 +983,7 @@ 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>,
@ -1107,7 +1002,7 @@ 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>,
@ -1183,7 +1078,7 @@ impl Snapshot {
for entry_id in update.removed_entries { for entry_id in update.removed_entries {
let entry = self let entry = self
.entry_for_id(ProjectEntryId::from_proto(entry_id)) .entry_for_id(ProjectEntryId::from_proto(entry_id))
.ok_or_else(|| anyhow!("unknown entry"))?; .ok_or_else(|| anyhow!("unknown entry {}", entry_id))?;
entries_by_path_edits.push(Edit::Remove(PathKey(entry.path.clone()))); entries_by_path_edits.push(Edit::Remove(PathKey(entry.path.clone())));
entries_by_id_edits.push(Edit::Remove(entry.id)); entries_by_id_edits.push(Edit::Remove(entry.id));
} }
@ -1205,6 +1100,7 @@ impl Snapshot {
self.entries_by_path.edit(entries_by_path_edits, &()); self.entries_by_path.edit(entries_by_path_edits, &());
self.entries_by_id.edit(entries_by_id_edits, &()); self.entries_by_id.edit(entries_by_id_edits, &());
self.scan_id = update.scan_id as usize; self.scan_id = update.scan_id as usize;
self.is_complete = update.is_last_update;
Ok(()) Ok(())
} }
@ -1326,27 +1222,16 @@ impl LocalSnapshot {
} }
#[cfg(test)] #[cfg(test)]
pub(crate) fn to_proto( pub(crate) fn build_initial_update(&self, project_id: u64) -> proto::UpdateWorktree {
&self,
diagnostic_summaries: &TreeMap<PathKey, DiagnosticSummary>,
visible: bool,
) -> proto::Worktree {
let root_name = self.root_name.clone(); let root_name = self.root_name.clone();
proto::Worktree { proto::UpdateWorktree {
id: self.id.0 as u64, project_id,
worktree_id: self.id().to_proto(),
root_name, root_name,
entries: self updated_entries: self.entries_by_path.iter().map(Into::into).collect(),
.entries_by_path removed_entries: Default::default(),
.iter()
.filter(|e| !e.is_ignored)
.map(Into::into)
.collect(),
diagnostic_summaries: diagnostic_summaries
.iter()
.map(|(path, summary)| summary.to_proto(&path.0))
.collect(),
visible,
scan_id: self.scan_id as u64, scan_id: self.scan_id as u64,
is_last_update: true,
} }
} }
@ -1413,6 +1298,7 @@ impl LocalSnapshot {
updated_entries, updated_entries,
removed_entries, removed_entries,
scan_id: self.scan_id as u64, scan_id: self.scan_id as u64,
is_last_update: true,
} }
} }
@ -2050,7 +1936,7 @@ impl BackgroundScanner {
} }
async fn run(mut self, events_rx: impl Stream<Item = Vec<fsevent::Event>>) { async fn run(mut self, events_rx: impl Stream<Item = Vec<fsevent::Event>>) {
if self.notify.unbounded_send(ScanState::Scanning).is_err() { if self.notify.unbounded_send(ScanState::Initializing).is_err() {
return; return;
} }
@ -2069,8 +1955,13 @@ impl BackgroundScanner {
} }
futures::pin_mut!(events_rx); futures::pin_mut!(events_rx);
while let Some(events) = events_rx.next().await {
if self.notify.unbounded_send(ScanState::Scanning).is_err() { while let Some(mut events) = events_rx.next().await {
while let Poll::Ready(Some(additional_events)) = futures::poll!(events_rx.next()) {
events.extend(additional_events);
}
if self.notify.unbounded_send(ScanState::Updating).is_err() {
break; break;
} }
@ -2722,6 +2613,19 @@ impl<'a> TryFrom<(&'a CharBag, proto::Entry)> for Entry {
} }
} }
async fn send_worktree_update(client: &Arc<Client>, update: proto::UpdateWorktree) -> Result<()> {
#[cfg(any(test, feature = "test-support"))]
const MAX_CHUNK_SIZE: usize = 2;
#[cfg(not(any(test, feature = "test-support")))]
const MAX_CHUNK_SIZE: usize = 256;
for update in proto::split_worktree_update(update, MAX_CHUNK_SIZE) {
client.request(update).await?;
}
Ok(())
}
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use super::*; use super::*;
@ -2931,6 +2835,7 @@ mod tests {
root_name: Default::default(), root_name: Default::default(),
root_char_bag: Default::default(), root_char_bag: Default::default(),
scan_id: 0, scan_id: 0,
is_complete: true,
}, },
}; };
initial_snapshot.insert_entry( initial_snapshot.insert_entry(

12
crates/rpc/proto/zed.proto
View file

@ -168,7 +168,7 @@ message JoinProjectResponse {
message Accept { message Accept {
uint32 replica_id = 1; uint32 replica_id = 1;
repeated Worktree worktrees = 2; repeated WorktreeMetadata worktrees = 2;
repeated Collaborator collaborators = 3; repeated Collaborator collaborators = 3;
repeated LanguageServer language_servers = 4; repeated LanguageServer language_servers = 4;
} }
@ -195,6 +195,7 @@ message UpdateWorktree {
repeated Entry updated_entries = 4; repeated Entry updated_entries = 4;
repeated uint64 removed_entries = 5; repeated uint64 removed_entries = 5;
uint64 scan_id = 6; uint64 scan_id = 6;
bool is_last_update = 7;
} }
message CreateProjectEntry { message CreateProjectEntry {
@ -765,15 +766,6 @@ message User {
string avatar_url = 3; string avatar_url = 3;
} }
message Worktree {
uint64 id = 1;
string root_name = 2;
repeated Entry entries = 3;
repeated DiagnosticSummary diagnostic_summaries = 4;
bool visible = 5;
uint64 scan_id = 6;
}
message File { message File {
uint64 worktree_id = 1; uint64 worktree_id = 1;
optional uint64 entry_id = 2; optional uint64 entry_id = 2;

26
crates/rpc/src/proto.rs
View file

@ -5,6 +5,7 @@ use futures::{SinkExt as _, StreamExt as _};
use prost::Message as _; use prost::Message as _;
use serde::Serialize; use serde::Serialize;
use std::any::{Any, TypeId}; use std::any::{Any, TypeId};
use std::{cmp, iter, mem};
use std::{ use std::{
fmt::Debug, fmt::Debug,
io, io,
@ -390,6 +391,31 @@ impl From<Nonce> for u128 {
} }
} }
pub fn split_worktree_update(
mut message: UpdateWorktree,
max_chunk_size: usize,
) -> impl Iterator<Item = UpdateWorktree> {
let mut done = false;
iter::from_fn(move || {
if done {
return None;
}
let chunk_size = cmp::min(message.updated_entries.len(), max_chunk_size);
let updated_entries = message.updated_entries.drain(..chunk_size).collect();
done = message.updated_entries.is_empty();
Some(UpdateWorktree {
project_id: message.project_id,
worktree_id: message.worktree_id,
root_name: message.root_name.clone(),
updated_entries,
removed_entries: mem::take(&mut message.removed_entries),
scan_id: message.scan_id,
is_last_update: done && message.is_last_update,
})
})
}
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use super::*; use super::*;