diff --git a/Cargo.lock b/Cargo.lock index 5afef537..ecb4533a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1705,9 +1705,9 @@ checksum = "791930b43c0d5973160d90a8f3894509f2b273430f5c5c73b668636d0287c5c0" [[package]] name = "iroh" -version = "1.2.0" +version = "1.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b2f8d1cfffc83efe39a1031aab423ce09cb8048071baba550508931e9a81ce46" +checksum = "885787b892b5e2507c701f132ecbd45d2bad4bbb75157a19427087c16dadb833" dependencies = [ "backon", "blake3", @@ -1755,9 +1755,9 @@ dependencies = [ [[package]] name = "iroh-base" -version = "1.2.0" +version = "1.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ffd1efd1aaecd68d4d1b0727a30c5811f43fb822843e48f65f6e5db3b7256cc9" +checksum = "6ae9092be76c9f5429776ab3394f5a5a78dc19dc2524267e884a5b9781546c3f" dependencies = [ "curve25519-dalek", "data-encoding", @@ -1824,9 +1824,9 @@ dependencies = [ [[package]] name = "iroh-relay" -version = "1.2.0" +version = "1.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "beb2294a9749d6a25fd7cd8bcf0fccd932f20716d4f967d85d3f23c7135ae6a2" +checksum = "ee116a8233f84980574bb8d11e4517503080cd2c3605bafca04ee9c35d708bd9" dependencies = [ "blake3", "bytes", diff --git a/apps/neshub/Cargo.toml b/apps/neshub/Cargo.toml index e15082f8..3b8e66c9 100644 --- a/apps/neshub/Cargo.toml +++ b/apps/neshub/Cargo.toml @@ -11,7 +11,7 @@ name = "neshub" path = "src/main.rs" [dependencies] -iroh = "1.2" +iroh = "1.3" anyhow.workspace = true clap.workspace = true diff --git a/apps/neshub/examples/pyro_bench.rs b/apps/neshub/examples/pyro_bench.rs index 0907e60f..cd5a5f9f 100644 --- a/apps/neshub/examples/pyro_bench.rs +++ b/apps/neshub/examples/pyro_bench.rs @@ -31,6 +31,9 @@ use nesprotocol::pyrowave::{ should_skip, }; +/// Datagrams taken per receive, as the client takes them. +const RECV_BATCH: usize = 64; + struct Opts { mbps: u64, fps: u64, @@ -110,10 +113,16 @@ async fn send(conn: Connection, o: &Opts) -> Result<()> { } else { let buf = Bytes::from(d.buf); let mut at = 0; - for len in d.lens { - conn.send_datagram(buf.slice(at..at + len))?; - at += len; - } + let batch: Vec = d + .lens + .iter() + .map(|&len| { + let datagram = buf.slice(at..at + len); + at += len; + datagram + }) + .collect(); + conn.send_many_datagrams(&batch)?; seq = seq.wrapping_add(1); sent += 1; } @@ -147,17 +156,21 @@ async fn receive(conn: Connection) -> Result<()> { let (mut all_bytes, mut whole, mut partial, mut lost) = (0u64, 0u64, 0u64, 0u64); let mut second = Instant::now(); let mut busy = Duration::ZERO; + let mut batch = vec![Bytes::new(); RECV_BATCH]; loop { - let d = match conn.read_datagram().await { - Ok(d) => d, + let n = match conn.read_many_datagrams(&mut batch).await { + Ok(n) => n, Err(_) => break, }; let t0 = Instant::now(); - datagrams += 1; - bytes += d.len() as u64; - if let Some((h, p)) = decode_pyro_datagram(&d) { - let at = d.len() - p.len(); - c.push(h, d.slice(at..), start.elapsed().as_micros() as u64); + for d in &mut batch[..n] { + let d = std::mem::take(d); + datagrams += 1; + bytes += d.len() as u64; + if let Some((h, p)) = decode_pyro_datagram(&d) { + let at = d.len() - p.len(); + c.push(h, d.slice(at..), start.elapsed().as_micros() as u64); + } } busy += t0.elapsed(); if second.elapsed() >= Duration::from_secs(1) { diff --git a/apps/neshub/src/dgram.rs b/apps/neshub/src/dgram.rs index bd8964fb..3743210f 100644 --- a/apps/neshub/src/dgram.rs +++ b/apps/neshub/src/dgram.rs @@ -102,9 +102,11 @@ impl DatagramSender { /// used, minus the length prefix a datagram does not need. /// /// A failure is reported but never retried. Retrying a real-time frame means - /// delivering it late, which is the behaviour this whole change removes. A - /// half-sent frame is abandoned where it stands; the receiver times its - /// fragments out and asks for a keyframe. + /// delivering it late, which is the behaviour this whole change removes. The + /// frame goes out whole or not at all: every fragment is queued under one + /// hold of the connection lock, so a path that shrinks mid-frame rejects the + /// frame rather than leaving half of it on the wire for the receiver to time + /// out. pub fn send_frame(&self, seq: u16, body: &[u8]) -> Result<(), SendFrameError> { // Ask QUIC what fits right now rather than assuming. The estimate moves // over a connection's life as path MTU discovery runs, and it can shrink @@ -142,18 +144,24 @@ impl DatagramSender { buf.extend_from_slice(&body[start..end]); } - for index in 0..total { - let start = index * payload_size; - let end = (start + payload_size).min(body.len()); - let datagram = buf.split_to(DGRAM_HDR_LEN + (end - start)).freeze(); - // Deliberately not `send_datagram_wait`: that waits for buffer space - // under congestion, which prioritises old datagrams over new ones. - // For live media the opposite is right — drop the backlog, send the - // frame that is actually current. - self.conn - .send_datagram(datagram) - .map_err(SendFrameError::Quic)?; - } + let datagrams: Vec = (0..total) + .map(|index| { + let start = index * payload_size; + let end = (start + payload_size).min(body.len()); + buf.split_to(DGRAM_HDR_LEN + (end - start)).freeze() + }) + .collect(); + + // One call for the whole frame: one lock hold and one driver wake rather + // than one per fragment, and a `TooLarge` rejects every fragment instead + // of the ones after the path shrank. Like `send_datagram`, it evicts the + // oldest queued datagrams to make room -- deliberately not the waiting + // kind, which prioritises old datagrams over new ones. For live media + // the opposite is right: drop the backlog, send the frame that is + // actually current. + self.conn + .send_many_datagrams(&datagrams) + .map_err(SendFrameError::Quic)?; Ok(()) } diff --git a/apps/neshub/src/pyro.rs b/apps/neshub/src/pyro.rs index 5763e289..524c3fc7 100644 --- a/apps/neshub/src/pyro.rs +++ b/apps/neshub/src/pyro.rs @@ -71,16 +71,23 @@ impl PyroSender { // Taking the `Vec` is free, and each datagram is a slice of it. let buf = Bytes::from(datagrams.buf); let mut at = 0; - for len in datagrams.lens { - debug_assert!(len >= PYRO_DGRAM_HDR_LEN); - let datagram = buf.slice(at..at + len); - at += len; - // Not `send_datagram_wait`, for the reason `dgram::send_frame` - // gives; the guard above is what keeps this from evicting. - if let Err(e) = self.conn.send_datagram(datagram) { - debug!("{label}: PyroWave frame {} cut short: {e}", frame.frame); - break; - } + let batch: Vec = datagrams + .lens + .iter() + .map(|&len| { + debug_assert!(len >= PYRO_DGRAM_HDR_LEN); + let datagram = buf.slice(at..at + len); + at += len; + datagram + }) + .collect(); + // Evicting, for the reason `dgram::send_frame` gives; the guard above is + // what keeps it from evicting anything. The batch is queued whole or + // not at all, so a refused frame put nothing on the wire and its + // sequence number goes to the next frame. + if let Err(e) = self.conn.send_many_datagrams(&batch) { + debug!("{label}: PyroWave frame {} not sent: {e}", frame.frame); + return; } self.seq = self.seq.wrapping_add(1); } diff --git a/docs/media-transport.md b/docs/media-transport.md index 2dcbfe7c..058caab8 100644 --- a/docs/media-transport.md +++ b/docs/media-transport.md @@ -90,6 +90,14 @@ or a connection that is already gone, never for one the path could not carry. prioritising old frames over current ones, which for live media is the wrong trade.) +The hub sends each frame with `send_many_datagrams` (iroh 1.3), which is the +same call over a batch: `send_many(datagrams, true)`, evicting exactly as +above and returning the full count. What it adds is that the batch is queued +under one lock hold, and a `TooLarge` on any fragment rejects the whole batch, +so a frame goes out whole or not at all. noq-proto's `send_many` with +`drop: false` would queue until full and return a partial count — a real +backpressure signal — but iroh does not expose that mode. + A sender overrunning the path is therefore told nothing at all. Any counter fed by that return value cannot be non-zero however badly things are going, which is exactly how a field report came to contain "encoder perfectly healthy,