diff options
| author | tv <tv@krebsco.de> | 2026-10-28 06:02:34 +0100 |
|---|---|---|
| committer | tv <tv@krebsco.de> | 2026-10-28 06:02:34 +0100 |
| commit | a25c3656990934b44ac3d71a80b56125746b2e94 (patch) | |
| tree | 9b4afb896fac383bfa9010c361670cc60f2a4c3f /src/transport | |
| parent | 46e2b985b34831c893500d43261d0b4ee45d8b70 (diff) | |
log slow libp2p sends
Diffstat (limited to 'src/transport')
| -rw-r--r-- | src/transport/libp2p.rs | 26 |
1 files changed, 23 insertions, 3 deletions
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 |
