oak-team/oak-gpui
3
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Add Server::{state,state_mut} to catch most deadlocks statically

Browse Source
This commit is contained in:
Antonio Scandurra
2021-09-21 12:19:52 +02:00
parent 0b11192fe3
commit d381020a60
1 changed files with 71 additions and 79 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/src/rpc.rs
+71 -79
View File
@@ -112,12 +112,11 @@ impl Server {
addr: String, addr: String,
user_id: UserId, user_id: UserId,
) -> impl Future<Output = ()> { ) -> impl Future<Output = ()> {
let this = self.clone(); let mut this = self.clone();
async move { async move {
let (connection_id, handle_io, mut incoming_rx) = let (connection_id, handle_io, mut incoming_rx) =
this.peer.add_connection(connection).await; this.peer.add_connection(connection).await;
this.store this.state_mut()
.write()
.await .await
.add_connection(connection_id, user_id); .add_connection(connection_id, user_id);
if let Err(err) = this.update_collaborators_for_users(&[user_id]).await { if let Err(err) = this.update_collaborators_for_users(&[user_id]).await {
@@ -167,9 +166,9 @@ impl Server {
} }
} }
async fn sign_out(self: &Arc<Self>, connection_id: ConnectionId) -> tide::Result<()> { async fn sign_out(self: &mut Arc<Self>, connection_id: ConnectionId) -> tide::Result<()> {
self.peer.disconnect(connection_id).await; self.peer.disconnect(connection_id).await;
let removed_connection = self.store.write().await.remove_connection(connection_id)?; let removed_connection = self.state_mut().await.remove_connection(connection_id)?;
for (worktree_id, worktree) in removed_connection.hosted_worktrees { for (worktree_id, worktree) in removed_connection.hosted_worktrees {
if let Some(share) = worktree.share { if let Some(share) = worktree.share {
@@ -210,13 +209,12 @@ impl Server {
} }
async fn open_worktree( async fn open_worktree(
self: Arc<Server>, mut self: Arc<Server>,
request: TypedEnvelope<proto::OpenWorktree>, request: TypedEnvelope<proto::OpenWorktree>,
) -> tide::Result<()> { ) -> tide::Result<()> {
let receipt = request.receipt(); let receipt = request.receipt();
let host_user_id = self let host_user_id = self
.store .state()
.read()
.await .await
.user_id_for_connection(request.sender_id)?; .user_id_for_connection(request.sender_id)?;
@@ -238,7 +236,7 @@ impl Server {
} }
let collaborator_user_ids = collaborator_user_ids.into_iter().collect::<Vec<_>>(); let collaborator_user_ids = collaborator_user_ids.into_iter().collect::<Vec<_>>();
let worktree_id = self.store.write().await.add_worktree(Worktree { let worktree_id = self.state_mut().await.add_worktree(Worktree {
host_connection_id: request.sender_id, host_connection_id: request.sender_id,
collaborator_user_ids: collaborator_user_ids.clone(), collaborator_user_ids: collaborator_user_ids.clone(),
root_name: request.payload.root_name, root_name: request.payload.root_name,
@@ -255,13 +253,12 @@ impl Server {
} }
async fn close_worktree( async fn close_worktree(
self: Arc<Server>, mut self: Arc<Server>,
request: TypedEnvelope<proto::CloseWorktree>, request: TypedEnvelope<proto::CloseWorktree>,
) -> tide::Result<()> { ) -> tide::Result<()> {
let worktree_id = request.payload.worktree_id; let worktree_id = request.payload.worktree_id;
let worktree = self let worktree = self
.store .state_mut()
.write()
.await .await
.remove_worktree(worktree_id, request.sender_id)?; .remove_worktree(worktree_id, request.sender_id)?;
@@ -282,7 +279,7 @@ impl Server {
} }
async fn share_worktree( async fn share_worktree(
self: Arc<Server>, mut self: Arc<Server>,
mut request: TypedEnvelope<proto::ShareWorktree>, mut request: TypedEnvelope<proto::ShareWorktree>,
) -> tide::Result<()> { ) -> tide::Result<()> {
let worktree = request let worktree = request
@@ -296,8 +293,7 @@ impl Server {
.collect(); .collect();
let collaborator_user_ids = let collaborator_user_ids =
self.store self.state_mut()
.write()
.await .await
.share_worktree(worktree.id, request.sender_id, entries); .share_worktree(worktree.id, request.sender_id, entries);
if let Some(collaborator_user_ids) = collaborator_user_ids { if let Some(collaborator_user_ids) = collaborator_user_ids {
@@ -320,13 +316,12 @@ impl Server {
} }
async fn unshare_worktree( async fn unshare_worktree(
self: Arc<Server>, mut self: Arc<Server>,
request: TypedEnvelope<proto::UnshareWorktree>, request: TypedEnvelope<proto::UnshareWorktree>,
) -> tide::Result<()> { ) -> tide::Result<()> {
let worktree_id = request.payload.worktree_id; let worktree_id = request.payload.worktree_id;
let worktree = self let worktree = self
.store .state_mut()
.write()
.await .await
.unshare_worktree(worktree_id, request.sender_id)?; .unshare_worktree(worktree_id, request.sender_id)?;
@@ -342,20 +337,16 @@ impl Server {
} }
async fn join_worktree( async fn join_worktree(
self: Arc<Server>, mut self: Arc<Server>,
request: TypedEnvelope<proto::JoinWorktree>, request: TypedEnvelope<proto::JoinWorktree>,
) -> tide::Result<()> { ) -> tide::Result<()> {
let worktree_id = request.payload.worktree_id; let worktree_id = request.payload.worktree_id;
let user_id = self let user_id = self
.store .state()
.read()
.await .await
.user_id_for_connection(request.sender_id)?; .user_id_for_connection(request.sender_id)?;
let response; let mut state = self.state_mut().await;
let connection_ids;
let collaborator_user_ids;
let mut state = self.store.write().await;
match state.join_worktree(request.sender_id, user_id, worktree_id) { match state.join_worktree(request.sender_id, user_id, worktree_id) {
Ok(JoinedWorktree { Ok(JoinedWorktree {
replica_id, replica_id,
@@ -376,7 +367,7 @@ impl Server {
}); });
} }
} }
response = proto::JoinWorktreeResponse { let response = proto::JoinWorktreeResponse {
worktree: Some(proto::Worktree { worktree: Some(proto::Worktree {
id: worktree_id, id: worktree_id,
root_name: worktree.root_name.clone(), root_name: worktree.root_name.clone(),
@@ -385,10 +376,29 @@ impl Server {
replica_id: replica_id as u32, replica_id: replica_id as u32,
peers, peers,
}; };
connection_ids = worktree.connection_ids(); let connection_ids = worktree.connection_ids();
collaborator_user_ids = worktree.collaborator_user_ids.clone(); let collaborator_user_ids = worktree.collaborator_user_ids.clone();
drop(state);
broadcast(request.sender_id, connection_ids, |conn_id| {
self.peer.send(
conn_id,
proto::AddPeer {
worktree_id,
peer: Some(proto::Peer {
peer_id: request.sender_id.0,
replica_id: response.replica_id,
}),
},
)
})
.await?;
self.peer.respond(request.receipt(), response).await?;
self.update_collaborators_for_users(&collaborator_user_ids)
.await?;
} }
Err(error) => { Err(error) => {
drop(state);
self.peer self.peer
.respond_with_error( .respond_with_error(
request.receipt(), request.receipt(),
@@ -397,44 +407,23 @@ impl Server {
}, },
) )
.await?; .await?;
return Ok(());
} }
} }
drop(state);
broadcast(request.sender_id, connection_ids, |conn_id| {
self.peer.send(
conn_id,
proto::AddPeer {
worktree_id,
peer: Some(proto::Peer {
peer_id: request.sender_id.0,
replica_id: response.replica_id,
}),
},
)
})
.await?;
self.peer.respond(request.receipt(), response).await?;
self.update_collaborators_for_users(&collaborator_user_ids)
.await?;
Ok(()) Ok(())
} }
async fn leave_worktree( async fn leave_worktree(
self: Arc<Server>, mut self: Arc<Server>,
request: TypedEnvelope<proto::LeaveWorktree>, request: TypedEnvelope<proto::LeaveWorktree>,
) -> tide::Result<()> { ) -> tide::Result<()> {
let sender_id = request.sender_id; let sender_id = request.sender_id;
let worktree_id = request.payload.worktree_id; let worktree_id = request.payload.worktree_id;
let worktree = self
if let Some(worktree) = self .state_mut()
.store
.write()
.await .await
.leave_worktree(sender_id, worktree_id) .leave_worktree(sender_id, worktree_id);
{ if let Some(worktree) = worktree {
broadcast(sender_id, worktree.connection_ids, |conn_id| { broadcast(sender_id, worktree.connection_ids, |conn_id| {
self.peer.send( self.peer.send(
conn_id, conn_id,
@@ -452,10 +441,10 @@ impl Server {
} }
async fn update_worktree( async fn update_worktree(
self: Arc<Server>, mut self: Arc<Server>,
request: TypedEnvelope<proto::UpdateWorktree>, request: TypedEnvelope<proto::UpdateWorktree>,
) -> tide::Result<()> { ) -> tide::Result<()> {
let connection_ids = self.store.write().await.update_worktree( let connection_ids = self.state_mut().await.update_worktree(
request.sender_id, request.sender_id,
request.payload.worktree_id, request.payload.worktree_id,
&request.payload.removed_entries, &request.payload.removed_entries,
@@ -477,8 +466,7 @@ impl Server {
) -> tide::Result<()> { ) -> tide::Result<()> {
let receipt = request.receipt(); let receipt = request.receipt();
let host_connection_id = self let host_connection_id = self
.store .state()
.read()
.await .await
.worktree_host_connection_id(request.sender_id, request.payload.worktree_id)?; .worktree_host_connection_id(request.sender_id, request.payload.worktree_id)?;
let response = self let response = self
@@ -494,8 +482,7 @@ impl Server {
request: TypedEnvelope<proto::CloseBuffer>, request: TypedEnvelope<proto::CloseBuffer>,
) -> tide::Result<()> { ) -> tide::Result<()> {
let host_connection_id = self let host_connection_id = self
.store .state()
.read()
.await .await
.worktree_host_connection_id(request.sender_id, request.payload.worktree_id)?; .worktree_host_connection_id(request.sender_id, request.payload.worktree_id)?;
self.peer self.peer
@@ -511,7 +498,7 @@ impl Server {
let host; let host;
let guests; let guests;
{ {
let state = self.store.read().await; let state = self.state().await;
host = state host = state
.worktree_host_connection_id(request.sender_id, request.payload.worktree_id)?; .worktree_host_connection_id(request.sender_id, request.payload.worktree_id)?;
guests = state guests = state
@@ -547,8 +534,7 @@ impl Server {
) -> tide::Result<()> { ) -> tide::Result<()> {
broadcast( broadcast(
request.sender_id, request.sender_id,
self.store self.state()
.read()
.await .await
.worktree_connection_ids(request.sender_id, request.payload.worktree_id)?, .worktree_connection_ids(request.sender_id, request.payload.worktree_id)?,
|connection_id| { |connection_id| {
@@ -585,8 +571,7 @@ impl Server {
request: TypedEnvelope<proto::GetChannels>, request: TypedEnvelope<proto::GetChannels>,
) -> tide::Result<()> { ) -> tide::Result<()> {
let user_id = self let user_id = self
.store .state()
.read()
.await .await
.user_id_for_connection(request.sender_id)?; .user_id_for_connection(request.sender_id)?;
let channels = self.app_state.db.get_accessible_channels(user_id).await?; let channels = self.app_state.db.get_accessible_channels(user_id).await?;
@@ -637,7 +622,7 @@ impl Server {
) -> tide::Result<()> { ) -> tide::Result<()> {
let mut send_futures = Vec::new(); let mut send_futures = Vec::new();
let state = self.store.read().await; let state = self.state().await;
for user_id in user_ids { for user_id in user_ids {
let collaborators = state.collaborators_for_user(*user_id); let collaborators = state.collaborators_for_user(*user_id);
for connection_id in state.connection_ids_for_user(*user_id) { for connection_id in state.connection_ids_for_user(*user_id) {
@@ -657,12 +642,11 @@ impl Server {
} }
async fn join_channel( async fn join_channel(
self: Arc<Self>, mut self: Arc<Self>,
request: TypedEnvelope<proto::JoinChannel>, request: TypedEnvelope<proto::JoinChannel>,
) -> tide::Result<()> { ) -> tide::Result<()> {
let user_id = self let user_id = self
.store .state()
.read()
.await .await
.user_id_for_connection(request.sender_id)?; .user_id_for_connection(request.sender_id)?;
let channel_id = ChannelId::from_proto(request.payload.channel_id); let channel_id = ChannelId::from_proto(request.payload.channel_id);
@@ -675,8 +659,7 @@ impl Server {
Err(anyhow!("access denied"))?; Err(anyhow!("access denied"))?;
} }
self.store self.state_mut()
.write()
.await .await
.join_channel(request.sender_id, channel_id); .join_channel(request.sender_id, channel_id);
let messages = self let messages = self
@@ -706,12 +689,11 @@ impl Server {
} }
async fn leave_channel( async fn leave_channel(
self: Arc<Self>, mut self: Arc<Self>,
request: TypedEnvelope<proto::LeaveChannel>, request: TypedEnvelope<proto::LeaveChannel>,
) -> tide::Result<()> { ) -> tide::Result<()> {
let user_id = self let user_id = self
.store .state()
.read()
.await .await
.user_id_for_connection(request.sender_id)?; .user_id_for_connection(request.sender_id)?;
let channel_id = ChannelId::from_proto(request.payload.channel_id); let channel_id = ChannelId::from_proto(request.payload.channel_id);
@@ -724,8 +706,7 @@ impl Server {
Err(anyhow!("access denied"))?; Err(anyhow!("access denied"))?;
} }
self.store self.state_mut()
.write()
.await .await
.leave_channel(request.sender_id, channel_id); .leave_channel(request.sender_id, channel_id);
@@ -741,7 +722,7 @@ impl Server {
let user_id; let user_id;
let connection_ids; let connection_ids;
{ {
let state = self.store.read().await; let state = self.state().await;
user_id = state.user_id_for_connection(request.sender_id)?; user_id = state.user_id_for_connection(request.sender_id)?;
if let Some(ids) = state.channel_connection_ids(channel_id) { if let Some(ids) = state.channel_connection_ids(channel_id) {
connection_ids = ids; connection_ids = ids;
@@ -829,8 +810,7 @@ impl Server {
request: TypedEnvelope<proto::GetChannelMessages>, request: TypedEnvelope<proto::GetChannelMessages>,
) -> tide::Result<()> { ) -> tide::Result<()> {
let user_id = self let user_id = self
.store .state()
.read()
.await .await
.user_id_for_connection(request.sender_id)?; .user_id_for_connection(request.sender_id)?;
let channel_id = ChannelId::from_proto(request.payload.channel_id); let channel_id = ChannelId::from_proto(request.payload.channel_id);
@@ -872,6 +852,18 @@ impl Server {
.await?; .await?;
Ok(()) Ok(())
} }
fn state<'a>(
self: &'a Arc<Self>,
) -> impl Future<Output = async_std::sync::RwLockReadGuard<'a, Store>> {
self.store.read()
}
fn state_mut<'a>(
self: &'a mut Arc<Self>,
) -> impl Future<Output = async_std::sync::RwLockWriteGuard<'a, Store>> {
self.store.write()
}
} }
pub async fn broadcast<F, T>( pub async fn broadcast<F, T>(