diff options
Diffstat (limited to 'src/swarm')
| -rw-r--r-- | src/swarm/blob.rs | 4 | ||||
| -rw-r--r-- | src/swarm/peer/handshake.rs | 4 | ||||
| -rw-r--r-- | src/swarm/peer/keepalive.rs | 20 | ||||
| -rw-r--r-- | src/swarm/relay.rs | 5 |
4 files changed, 20 insertions, 13 deletions
diff --git a/src/swarm/blob.rs b/src/swarm/blob.rs index c153a17..4b69562 100644 --- a/src/swarm/blob.rs +++ b/src/swarm/blob.rs @@ -166,7 +166,7 @@ async fn prepare(state: &AppState, id: BlobId) -> Option<Arc<Prepared>> { } async fn reply(state: &AppState, peer: &str, id: BlobId, body: BlobBody) { - let (epoch, epoch_seq) = state.epoch_stamp().await; + let (epoch, epoch_seq) = state.epoch_stamp(); let message = ParsedMessage::Blob(BlobMessage { epoch, epoch_seq, @@ -220,7 +220,7 @@ impl Drop for Registration<'_> { } async fn request(state: &AppState, peer: &str, id: BlobId, chunks: Vec<u32>) { - let (epoch, epoch_seq) = state.epoch_stamp().await; + let (epoch, epoch_seq) = state.epoch_stamp(); let message = ParsedMessage::Blob(BlobMessage { epoch, epoch_seq, diff --git a/src/swarm/peer/handshake.rs b/src/swarm/peer/handshake.rs index f08929a..7e494c6 100644 --- a/src/swarm/peer/handshake.rs +++ b/src/swarm/peer/handshake.rs @@ -118,7 +118,7 @@ async fn transmit_hello(state: &AppState, peer: &str) -> Result<()> { .as_ref() .map(|key| key.hello_tag(swarm_id.as_deref().unwrap_or(""), state.local_id(), peer)) .unwrap_or_default(); - let (epoch, epoch_seq) = state.epoch_stamp().await; + let (epoch, epoch_seq) = state.epoch_stamp(); // Retries repeat the handshake's number. let handshake = match state.get_peer_state(peer).await { Some(peer_state) => peer_state.lock().await.handshake, @@ -209,7 +209,7 @@ pub(super) async fn send_hello_reply( .map(|key| key.hello_reply_tag(&swarm_id, state.local_id(), peer)) .unwrap_or_default(); let ready = state.is_ready().await; - let (epoch, epoch_seq) = state.epoch_stamp().await; + let (epoch, epoch_seq) = state.epoch_stamp(); let reply = transport::HelloReply { swarm_id, peers, diff --git a/src/swarm/peer/keepalive.rs b/src/swarm/peer/keepalive.rs index 4fcad7e..4ce326e 100644 --- a/src/swarm/peer/keepalive.rs +++ b/src/swarm/peer/keepalive.rs @@ -25,14 +25,14 @@ pub(crate) async fn send_ping(state: &AppState, peer: &str) -> Result<()> { /// Our keepalive for `peer`. pub(crate) async fn local_keepalive(state: &AppState, peer: &str) -> transport::Keepalive { - // Heads and epoch come from the same locked replica, so they always match. The heads - // are cached (`Replica`), so the read lock is enough. + // Heads and epoch are published together (`Replica::stamps`), so they always match, + // and reading them doesn't wait for a sync message being applied. let (heads, epoch, epoch_seq) = { - let replica = state.doc.read().await; + let stamp = state.stamp.borrow(); ( - transport::HeadsSummary::of(replica.heads().to_vec()), - replica.epoch, - replica.epoch_seq, + transport::HeadsSummary::of(stamp.heads.to_vec()), + stamp.epoch, + stamp.epoch_seq, ) }; let syncs_with_you = match state.get_peer_state(peer).await { @@ -62,7 +62,13 @@ pub(super) async fn note_keepalive(state: &AppState, peer: &str, remote: &transp peer_state.syncs_with_us = Some(remote.syncs_with_you); peer_state.reported_heads = remote.heads.heads().map(<[_]>::len); drop(peer_state); - crate::swarm::relay::pull_if_behind(state, peer, remote).await; + // Off the consumer: telling which heads we lack takes the document's lock. + { + let (state, peer, remote) = (state.clone(), peer.to_owned(), remote.clone()); + tokio::spawn(async move { + crate::swarm::relay::pull_if_behind(&state, &peer, &remote).await + }); + } if became_ready { // We may have left the opening to it while it wasn't ready, and nothing else // would make us look again. diff --git a/src/swarm/relay.rs b/src/swarm/relay.rs index 4bfc2b6..34451c9 100644 --- a/src/swarm/relay.rs +++ b/src/swarm/relay.rs @@ -103,8 +103,9 @@ pub(crate) async fn pull_if_behind(state: &AppState, peer: &str, remote: &transp if remote.epoch != state.local_epoch() || !topology::is_neighbour(state, peer) { return; } - // The common case, the same heads, needs only the cached heads (read lock). - if HeadsSummary::of(state.doc.read().await.heads().to_vec()) == remote.heads { + // The common case, the same heads, needs only the published heads. + let ours = HeadsSummary::of(state.stamp.borrow().heads.to_vec()); + if ours == remote.heads { return; } let missing = match &remote.heads { |