diff options
| -rw-r--r-- | src/main.rs | 2 | ||||
| -rw-r--r-- | src/transport/libp2p.rs | 26 |
2 files changed, 24 insertions, 4 deletions
diff --git a/src/main.rs b/src/main.rs index 8a61200..f174c4e 100644 --- a/src/main.rs +++ b/src/main.rs @@ -245,7 +245,7 @@ async fn main() -> Result<()> { } other => anyhow::bail!("transports: no {other} in this build"), }; - debug!( + info!( transport = transport.0, address = transport.1.local_id(), "Transport ready" diff --git a/src/transport/libp2p.rs b/src/transport/libp2p.rs index 237fd8a..995d074 100644 --- a/src/transport/libp2p.rs +++ b/src/transport/libp2p.rs @@ -67,6 +67,8 @@ use crate::transport::{ }; const PROTOCOL: StreamProtocol = StreamProtocol::new("/continuum/1"); +/// Sends taking this long are logged with where the time went (at debug). +const SLOW_SEND: Duration = Duration::from_millis(5); /// The largest message read: anyone who can connect can send anything, so this bounds /// what one stream costs. Sync messages above 16 chunks travel as blobs. const MAX_MESSAGE: usize = 128 << 20; @@ -1390,21 +1392,39 @@ impl Transport for Libp2pTransport { async fn send_encoded(&self, message: &EncodedMessage) -> Result<()> { let peer = PeerId::from_str(&message.peer) .with_context(|| format!("{} is no libp2p peer ID", message.peer))?; + let started = std::time::Instant::now(); self.connect(peer).await?; + let connected = started.elapsed(); let mut control = self.control.clone(); let body = message.body.clone(); let write = async move { let mut stream = control.open_stream(peer, PROTOCOL).await?; + let opened = started.elapsed(); stream.write_all(&(body.len() as u32).to_be_bytes()).await?; stream.write_all(&body).await?; stream.close().await?; - Ok::<(), anyhow::Error>(()) + Ok::<_, anyhow::Error>(opened) }; let timeout = stream_timeout(message.body.len()); - tokio::time::timeout(timeout, write) + let opened = tokio::time::timeout(timeout, write) .await .map_err(|_| anyhow!("sending to {peer} took longer than {timeout:?}"))? - .with_context(|| format!("Sending to {peer}")) + .with_context(|| format!("Sending to {peer}"))?; + let took = started.elapsed(); + if took >= SLOW_SEND { + // Where a send's time goes: the swarm task (connecting), the stream's protocol + // negotiation (opening), or the connection (writing). + debug!( + peer = %peer, + kind = ?message.kind, + bytes = message.body.len(), + connect_ms = connected.as_secs_f64() * 1000.0, + open_ms = (opened - connected).as_secs_f64() * 1000.0, + write_ms = (took - opened).as_secs_f64() * 1000.0, + "Slow send" + ); + } + Ok(()) } /// Reads inbound streams (`READS` at once) and handles their messages one at a |