zed-industries/zed
1
1
Fork 0
mirror of https://github.com/zed-industries/zed.git synced 2024-09-19 02:17:35 +03:00
Code Issues Packages Projects Releases Wiki Activity

Change project registration RPC APIs to smooth out UI updates

Browse Source 
* Make `UnregisterProject` a request. This way the client-side project can wait
  to clear out its remote id until the request has completed, so that the
  contacts panel can avoid showing duplicate private/public projects in the
  brief time after unregistering a project, before the next UpdateCollaborators
  message is received.
* Remove the `RegisterWorktree` and `UnregisterWorktree` methods and replace
  them with a single `UpdateProject` method that idempotently updates the
  Project's list of worktrees.
This commit is contained in:
Max Brunsfeld 2022-06-01 13:41:48 -07:00
parent 4d4ec793e2
commit d11beb3c02
6 changed files with 162 additions and 261 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

71
crates/collab/src/rpc.rs
View File

@ -141,12 +141,11 @@ impl Server {
server
.add_request_handler(Server::ping)
.add_request_handler(Server::register_project)
.add_message_handler(Server::unregister_project)
.add_request_handler(Server::unregister_project)
.add_request_handler(Server::join_project)
.add_message_handler(Server::leave_project)
.add_message_handler(Server::respond_to_join_project_request)
.add_request_handler(Server::register_worktree)
.add_message_handler(Server::unregister_worktree)
.add_message_handler(Server::update_project)
.add_request_handler(Server::update_worktree)
.add_message_handler(Server::start_language_server)
.add_message_handler(Server::update_language_server)
@ -484,14 +483,15 @@ impl Server {
user_id = state.user_id_for_connection(request.sender_id)?;
project_id = state.register_project(request.sender_id, user_id);
};
self.update_user_contacts(user_id).await?;
response.send(proto::RegisterProjectResponse { project_id })?;
self.update_user_contacts(user_id).await?;
Ok(())
}
async fn unregister_project(
self: Arc<Server>,
request: TypedEnvelope<proto::UnregisterProject>,
response: Response<proto::UnregisterProject>,
) -> Result<()> {
let (user_id, project) = {
let mut state = self.store_mut().await;
@ -529,6 +529,7 @@ impl Server {
}
self.update_user_contacts(user_id).await?;
response.send(proto::Ack {})?;
Ok(())
}
@ -568,6 +569,7 @@ impl Server {
response: Response<proto::JoinProject>,
) -> Result<()> {
let project_id = request.payload.project_id;
let host_user_id;
let guest_user_id;
let host_connection_id;
@ -768,63 +770,28 @@ impl Server {
Ok(())
}
async fn register_worktree(
async fn update_project(
self: Arc<Server>,
request: TypedEnvelope<proto::RegisterWorktree>,
response: Response<proto::RegisterWorktree>,
request: TypedEnvelope<proto::UpdateProject>,
) -> Result<()> {
let host_user_id;
let user_id;
{
let mut state = self.store_mut().await;
host_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
.read_project(request.payload.project_id, request.sender_id)?
.guest_connection_ids();
state.register_worktree(
state.update_project(
request.payload.project_id,
request.payload.worktree_id,
&request.payload.worktrees,
request.sender_id,
Worktree {
root_name: request.payload.root_name.clone(),
visible: request.payload.visible,
..Default::default()
},
)?;
broadcast(request.sender_id, guest_connection_ids, |connection_id| {
self.peer
.forward_send(request.sender_id, connection_id, request.payload.clone())
});
}
self.update_user_contacts(host_user_id).await?;
response.send(proto::Ack {})?;
Ok(())
}
async fn unregister_worktree(
self: Arc<Server>,
request: TypedEnvelope<proto::UnregisterWorktree>,
) -> Result<()> {
let host_user_id;
let project_id = request.payload.project_id;
let worktree_id = request.payload.worktree_id;
{
let mut state = self.store_mut().await;
let (_, guest_connection_ids) =
state.unregister_worktree(project_id, worktree_id, request.sender_id)?;
host_user_id = state.user_id_for_connection(request.sender_id)?;
broadcast(request.sender_id, guest_connection_ids, |conn_id| {
self.peer.send(
conn_id,
proto::UnregisterWorktree {
project_id,
worktree_id,
},
)
});
}
self.update_user_contacts(host_user_id).await?;
};
self.update_user_contacts(user_id).await?;
Ok(())
}
@ -833,10 +800,11 @@ impl Server {
request: TypedEnvelope<proto::UpdateWorktree>,
response: Response<proto::UpdateWorktree>,
) -> Result<()> {
let connection_ids = self.store_mut().await.update_worktree(
let (connection_ids, metadata_changed) = self.store_mut().await.update_worktree(
request.sender_id,
request.payload.project_id,
request.payload.worktree_id,
&request.payload.root_name,
&request.payload.removed_entries,
&request.payload.updated_entries,
request.payload.scan_id,
@ -846,6 +814,13 @@ impl Server {
self.peer
.forward_send(request.sender_id, connection_id, request.payload.clone())
});
if metadata_changed {
let user_id = self
.store()
.await
.user_id_for_connection(request.sender_id)?;
self.update_user_contacts(user_id).await?;
}
response.send(proto::Ack {})?;
Ok(())
}

54
crates/collab/src/rpc/store.rs
View File

@ -312,19 +312,32 @@ impl Store {
project_id
}
pub fn register_worktree(
pub fn update_project(
&mut self,
project_id: u64,
worktree_id: u64,
worktrees: &[proto::WorktreeMetadata],
connection_id: ConnectionId,
worktree: Worktree,
) -> Result<()> {
let project = self
.projects
.get_mut(&project_id)
.ok_or_else(|| anyhow!("no such project"))?;
if project.host_connection_id == connection_id {
project.worktrees.insert(worktree_id, worktree);
let mut old_worktrees = mem::take(&mut project.worktrees);
for worktree in worktrees {
if let Some(old_worktree) = old_worktrees.remove(&worktree.id) {
project.worktrees.insert(worktree.id, old_worktree);
} else {
project.worktrees.insert(
worktree.id,
Worktree {
root_name: worktree.root_name.clone(),
visible: worktree.visible,
..Default::default()
},
);
}
}
Ok(())
} else {
Err(anyhow!("no such project"))?
@ -374,27 +387,6 @@ impl Store {
}
}
pub fn unregister_worktree(
&mut self,
project_id: u64,
worktree_id: u64,
acting_connection_id: ConnectionId,
) -> Result<(Worktree, Vec<ConnectionId>)> {
let project = self
.projects
.get_mut(&project_id)
.ok_or_else(|| anyhow!("no such project"))?;
if project.host_connection_id != acting_connection_id {
Err(anyhow!("not your worktree"))?;
}
let worktree = project
.worktrees
.remove(&worktree_id)
.ok_or_else(|| anyhow!("no such worktree"))?;
Ok((worktree, project.guest_connection_ids()))
}
pub fn update_diagnostic_summary(
&mut self,
project_id: u64,
@ -573,15 +565,15 @@ impl Store {
connection_id: ConnectionId,
project_id: u64,
worktree_id: u64,
worktree_root_name: &str,
removed_entries: &[u64],
updated_entries: &[proto::Entry],
scan_id: u64,
) -> Result<Vec<ConnectionId>> {
) -> Result<(Vec<ConnectionId>, bool)> {
let project = self.write_project(project_id, connection_id)?;
let worktree = project
.worktrees
.get_mut(&worktree_id)
.ok_or_else(|| anyhow!("no such worktree"))?;
let mut worktree = project.worktrees.entry(worktree_id).or_default();
let metadata_changed = worktree_root_name != worktree.root_name;
worktree.root_name = worktree_root_name.to_string();
for entry_id in removed_entries {
worktree.entries.remove(&entry_id);
}
@ -590,7 +582,7 @@ impl Store {
}
worktree.scan_id = scan_id;
let connection_ids = project.connection_ids();
Ok(connection_ids)
Ok((connection_ids, metadata_changed))
}
pub fn project_connection_ids(

187
crates/project/src/project.rs
View File

@ -282,8 +282,7 @@ impl Project {
client.add_model_message_handler(Self::handle_update_language_server);
client.add_model_message_handler(Self::handle_remove_collaborator);
client.add_model_message_handler(Self::handle_join_project_request_cancelled);
client.add_model_message_handler(Self::handle_register_worktree);
client.add_model_message_handler(Self::handle_unregister_worktree);
client.add_model_message_handler(Self::handle_update_project);
client.add_model_message_handler(Self::handle_unregister_project);
client.add_model_message_handler(Self::handle_project_unshared);
client.add_model_message_handler(Self::handle_update_buffer_file);
@ -338,7 +337,9 @@ impl Project {
.await
.log_err()?;
} else {
this.update(&mut cx, |this, cx| this.unregister(cx));
this.update(&mut cx, |this, cx| this.unregister(cx))
.await
.log_err();
}
}
None
@ -638,30 +639,29 @@ impl Project {
}
}
fn unregister(&mut self, cx: &mut ModelContext<Self>) {
fn unregister(&mut self, cx: &mut ModelContext<Self>) -> Task<Result<()>> {
self.unshared(cx);
for worktree in &self.worktrees {
if let Some(worktree) = worktree.upgrade(cx) {
worktree.update(cx, |worktree, _| {
worktree.as_local_mut().unwrap().unregister();
if let ProjectClientState::Local { remote_id_rx, .. } = &mut self.client_state {
if let Some(remote_id) = *remote_id_rx.borrow() {
let request = self.client.request(proto::UnregisterProject {
project_id: remote_id,
});
return cx.spawn(|this, mut cx| async move {
let response = request.await;
this.update(&mut cx, |this, cx| {
if let ProjectClientState::Local { remote_id_tx, .. } =
&mut this.client_state
{
*remote_id_tx.borrow_mut() = None;
}
this.subscriptions.clear();
this.metadata_changed(cx);
});
response.map(drop)
});
}
}
if let ProjectClientState::Local { remote_id_tx, .. } = &mut self.client_state {
let mut remote_id = remote_id_tx.borrow_mut();
if let Some(remote_id) = *remote_id {
self.client
.send(proto::UnregisterProject {
project_id: remote_id,
})
.log_err();
}
*remote_id = None;
}
self.subscriptions.clear();
self.metadata_changed(cx);
Task::ready(Ok(()))
}
fn register(&mut self, cx: &mut ModelContext<Self>) -> Task<Result<()>> {
@ -674,8 +674,6 @@ impl Project {
let response = self.client.request(proto::RegisterProject {});
cx.spawn(|this, mut cx| async move {
let remote_id = response.await?.project_id;
let mut registrations = Vec::new();
this.update(&mut cx, |this, cx| {
if let ProjectClientState::Local { remote_id_tx, .. } = &mut this.client_state {
*remote_id_tx.borrow_mut() = Some(remote_id);
@ -683,22 +681,10 @@ impl Project {
this.metadata_changed(cx);
cx.emit(Event::RemoteIdChanged(Some(remote_id)));
this.subscriptions
.push(this.client.add_model_for_remote_entity(remote_id, cx));
for worktree in &this.worktrees {
if let Some(worktree) = worktree.upgrade(cx) {
registrations.push(worktree.update(cx, |worktree, cx| {
let worktree = worktree.as_local_mut().unwrap();
worktree.register(remote_id, cx)
}));
}
}
});
futures::future::try_join_all(registrations).await?;
Ok(())
Ok(())
})
})
}
@ -757,7 +743,27 @@ impl Project {
}
fn metadata_changed(&mut self, cx: &mut ModelContext<Self>) {
cx.notify();
self.project_store.update(cx, |_, cx| cx.notify());
if let ProjectClientState::Local { remote_id_rx, .. } = &self.client_state {
if let Some(project_id) = *remote_id_rx.borrow() {
self.client
.send(proto::UpdateProject {
project_id,
worktrees: self
.worktrees
.iter()
.filter_map(|worktree| {
worktree.upgrade(&cx).map(|worktree| {
worktree.read(cx).as_local().unwrap().metadata_proto()
})
})
.collect(),
})
.log_err();
}
}
}
pub fn collaborators(&self) -> &HashMap<PeerId, Collaborator> {
@ -3696,37 +3702,19 @@ impl Project {
});
let worktree = worktree?;
let remote_project_id = project.update(&mut cx, |project, cx| {
let project_id = project.update(&mut cx, |project, cx| {
project.add_worktree(&worktree, cx);
project.remote_id()
project.shared_remote_id()
});
if let Some(project_id) = remote_project_id {
// Because sharing is async, we may have *unshared* the project by the time it completes,
// in which case we need to register the worktree instead.
loop {
if project.read_with(&cx, |project, _| project.is_shared()) {
if worktree
.update(&mut cx, |worktree, cx| {
worktree.as_local_mut().unwrap().share(project_id, cx)
})
.await
.is_ok()
{
break;
}
} else {
worktree
.update(&mut cx, |worktree, cx| {
worktree
.as_local_mut()
.unwrap()
.register(project_id, cx)
})
.await?;
break;
}
}
// Because sharing is async, we may have *unshared* the project by the time it completes.
if let Some(project_id) = project_id {
worktree
.update(&mut cx, |worktree, cx| {
worktree.as_local_mut().unwrap().share(project_id, cx)
})
.await
.log_err();
}
Ok(worktree)
@ -4071,40 +4059,51 @@ impl Project {
Ok(())
}
async fn handle_register_worktree(
async fn handle_update_project(
this: ModelHandle<Self>,
envelope: TypedEnvelope<proto::RegisterWorktree>,
envelope: TypedEnvelope<proto::UpdateProject>,
client: Arc<Client>,
mut cx: AsyncAppContext,
) -> Result<()> {
this.update(&mut cx, |this, cx| {
let remote_id = this.remote_id().ok_or_else(|| anyhow!("invalid project"))?;
let replica_id = this.replica_id();
let worktree = proto::Worktree {
id: envelope.payload.worktree_id,
root_name: envelope.payload.root_name,
entries: Default::default(),
diagnostic_summaries: Default::default(),
visible: envelope.payload.visible,
scan_id: 0,
};
let (worktree, load_task) =
Worktree::remote(remote_id, replica_id, worktree, client, cx);
this.add_worktree(&worktree, cx);
load_task.detach();
Ok(())
})
}
let remote_id = this.remote_id().ok_or_else(|| anyhow!("invalid project"))?;
let mut old_worktrees_by_id = this
.worktrees
.drain(..)
.filter_map(|worktree| {
let worktree = worktree.upgrade(cx)?;
Some((worktree.read(cx).id(), worktree))
})
.collect::<HashMap<_, _>>();
for worktree in envelope.payload.worktrees {
if let Some(old_worktree) =
old_worktrees_by_id.remove(&WorktreeId::from_proto(worktree.id))
{
this.worktrees.push(WorktreeHandle::Strong(old_worktree));
} else {
let worktree = proto::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);
this.add_worktree(&worktree, cx);
load_task.detach();
}
}
this.metadata_changed(cx);
for (id, _) in old_worktrees_by_id {
cx.emit(Event::WorktreeRemoved(id));
}
async fn handle_unregister_worktree(
this: ModelHandle<Self>,
envelope: TypedEnvelope<proto::UnregisterWorktree>,
_: Arc<Client>,
mut cx: AsyncAppContext,
) -> Result<()> {
this.update(&mut cx, |this, cx| {
let worktree_id = WorktreeId::from_proto(envelope.payload.worktree_id);
this.remove_worktree(worktree_id, cx);
Ok(())
})
}

77
crates/project/src/worktree.rs
View File

@ -68,7 +68,6 @@ pub struct LocalWorktree {
last_scan_state_rx: watch::Receiver<ScanState>,
_background_scanner_task: Option<Task<()>>,
poll_task: Option<Task<()>>,
registration: Registration,
share: Option<ShareState>,
diagnostics: HashMap<Arc<Path>, Vec<DiagnosticEntry<PointUtf16>>>,
diagnostic_summaries: TreeMap<PathKey, DiagnosticSummary>,
@ -129,13 +128,6 @@ enum ScanState {
Err(Arc<anyhow::Error>),
}
#[derive(Debug, Eq, PartialEq)]
enum Registration {
None,
Pending,
Done { project_id: u64 },
}
struct ShareState {
project_id: u64,
snapshots_tx: Sender<LocalSnapshot>,
@ -148,12 +140,6 @@ pub enum Event {
impl Entity for Worktree {
type Event = Event;
fn release(&mut self, _: &mut MutableAppContext) {
if let Some(worktree) = self.as_local_mut() {
worktree.unregister();
}
}
}
impl Worktree {
@ -479,7 +465,6 @@ impl LocalWorktree {
background_snapshot: Arc::new(Mutex::new(snapshot)),
last_scan_state_rx,
_background_scanner_task: None,
registration: Registration::None,
share: None,
poll_task: None,
diagnostics: Default::default(),
@ -601,6 +586,14 @@ impl LocalWorktree {
self.snapshot.clone()
}
pub fn metadata_proto(&self) -> proto::WorktreeMetadata {
proto::WorktreeMetadata {
id: self.id().to_proto(),
root_name: self.root_name().to_string(),
visible: self.visible,
}
}
fn load(&self, path: &Path, cx: &mut ModelContext<Worktree>) -> Task<Result<(File, String)>> {
let handle = cx.handle();
let path = Arc::from(path);
@ -897,46 +890,7 @@ impl LocalWorktree {
})
}
pub fn register(
&mut self,
project_id: u64,
cx: &mut ModelContext<Worktree>,
) -> Task<anyhow::Result<()>> {
if self.registration != Registration::None {
return Task::ready(Ok(()));
}
self.registration = Registration::Pending;
let client = self.client.clone();
let register_message = proto::RegisterWorktree {
project_id,
worktree_id: self.id().to_proto(),
root_name: self.root_name().to_string(),
visible: self.visible,
};
let request = client.request(register_message);
cx.spawn(|this, mut cx| async move {
let response = request.await;
this.update(&mut cx, |this, _| {
let worktree = this.as_local_mut().unwrap();
match response {
Ok(_) => {
if worktree.registration == Registration::Pending {
worktree.registration = Registration::Done { project_id };
}
Ok(())
}
Err(error) => {
worktree.registration = Registration::None;
Err(error)
}
}
})
})
}
pub fn share(&mut self, project_id: u64, cx: &mut ModelContext<Worktree>) -> Task<Result<()>> {
let register = self.register(project_id, cx);
let (share_tx, share_rx) = oneshot::channel();
let (snapshots_to_send_tx, snapshots_to_send_rx) =
smol::channel::unbounded::<LocalSnapshot>();
@ -1041,7 +995,6 @@ impl LocalWorktree {
}
cx.spawn_weak(|this, cx| async move {
register.await?;
if let Some(this) = this.upgrade(&cx) {
this.read_with(&cx, |this, _| {
let this = this.as_local().unwrap();
@ -1054,20 +1007,6 @@ impl LocalWorktree {
})
}
pub fn unregister(&mut self) {
self.unshare();
if let Registration::Done { project_id } = self.registration {
self.client
.clone()
.send(proto::UnregisterWorktree {
project_id,
worktree_id: self.id().to_proto(),
})
.log_err();
}
self.registration = Registration::None;
}
pub fn unshare(&mut self) {
self.share.take();
}

26
crates/rpc/proto/zed.proto
View File

@ -35,8 +35,7 @@ message Envelope {
OpenBufferForSymbol open_buffer_for_symbol = 28;
OpenBufferForSymbolResponse open_buffer_for_symbol_response = 29;
RegisterWorktree register_worktree = 30;
UnregisterWorktree unregister_worktree = 31;
UpdateProject update_project = 30;
UpdateWorktree update_worktree = 32;
CreateProjectEntry create_project_entry = 33;
@ -129,6 +128,11 @@ message UnregisterProject {
uint64 project_id = 1;
}
message UpdateProject {
uint64 project_id = 1;
repeated WorktreeMetadata worktrees = 2;
}
message RequestJoinProject {
uint64 requester_id = 1;
uint64 project_id = 2;
@ -177,18 +181,6 @@ message LeaveProject {
uint64 project_id = 1;
}
message RegisterWorktree {
uint64 project_id = 1;
uint64 worktree_id = 2;
string root_name = 3;
bool visible = 4;
}
message UnregisterWorktree {
uint64 project_id = 1;
uint64 worktree_id = 2;
}
message UpdateWorktree {
uint64 project_id = 1;
uint64 worktree_id = 2;
@ -934,3 +926,9 @@ message ProjectMetadata {
repeated string worktree_root_names = 3;
repeated uint64 guests = 4;
}
message WorktreeMetadata {
uint64 id = 1;
string root_name = 2;
bool visible = 3;
}

8
crates/rpc/src/proto.rs
View File

@ -132,7 +132,6 @@ messages!(
(Ping, Foreground),
(ProjectUnshared, Foreground),
(RegisterProject, Foreground),
(RegisterWorktree, Foreground),
(ReloadBuffers, Foreground),
(ReloadBuffersResponse, Foreground),
(RemoveProjectCollaborator, Foreground),
@ -151,7 +150,6 @@ messages!(
(Test, Foreground),
(Unfollow, Foreground),
(UnregisterProject, Foreground),
(UnregisterWorktree, Foreground),
(UpdateBuffer, Foreground),
(UpdateBufferFile, Foreground),
(UpdateContacts, Foreground),
@ -159,6 +157,7 @@ messages!(
(UpdateFollowers, Foreground),
(UpdateInviteInfo, Foreground),
(UpdateLanguageServer, Foreground),
(UpdateProject, Foreground),
(UpdateWorktree, Foreground),
);
@ -192,7 +191,6 @@ request_messages!(
(PerformRename, PerformRenameResponse),
(PrepareRename, PrepareRenameResponse),
(RegisterProject, RegisterProjectResponse),
(RegisterWorktree, Ack),
(ReloadBuffers, ReloadBuffersResponse),
(RequestContact, Ack),
(RemoveContact, Ack),
@ -202,6 +200,7 @@ request_messages!(
(SearchProject, SearchProjectResponse),
(SendChannelMessage, SendChannelMessageResponse),
(Test, Test),
(UnregisterProject, Ack),
(UpdateBuffer, Ack),
(UpdateWorktree, Ack),
);
@ -242,13 +241,12 @@ entity_messages!(
StartLanguageServer,
Unfollow,
UnregisterProject,
UnregisterWorktree,
UpdateBuffer,
UpdateBufferFile,
UpdateDiagnosticSummary,
UpdateFollowers,
UpdateLanguageServer,
RegisterWorktree,
UpdateProject,
UpdateWorktree,
);