feat(neshub): iroh 1.3, and send each frame as one datagram batch

send_many_datagrams queues a frame's fragments under one lock hold and
one driver wake instead of one per fragment. It also rejects the whole
frame on TooLarge, so a path that shrinks mid-frame no longer leaves half
a frame on the wire for the client to time out. Eviction behaviour is
unchanged. A refused PyroWave frame now keeps its sequence number, since
nothing went out under it. pyro_bench mirrors both ends.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
DatCaptainHorse
2026-10-05 15:57:53 +03:00
co-authored by Claude Opus 5.5
parent 030dcc5a8c
commit 10fdeddc70
6 changed files with 79 additions and 43 deletions
Generated
+6 -6
View File
@@ -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",
+1 -1
View File
@@ -11,7 +11,7 @@ name = "neshub"
path = "src/main.rs"
[dependencies]
iroh = "1.2"
iroh = "1.3"
anyhow.workspace = true
clap.workspace = true
+24 -11
View File
@@ -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<Bytes> = 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) {
+23 -15
View File
@@ -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<Bytes> = (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(())
}
+17 -10
View File
@@ -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<Bytes> = 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);
}
+8
View File
@@ -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,