diff --git a/src/engine.rs b/src/engine.rs index 8c678d3..961725d 100644 --- a/src/engine.rs +++ b/src/engine.rs @@ -869,15 +869,27 @@ pub struct NetSource { cur: Vec, pos: usize, prebuffer: usize, + /// the most `prebuffer` may grow to after the sound runs dry (see `net_feed_adaptive`); + /// always at least `prebuffer` + grow_to: usize, buffering: bool, } /// A network feed: push blocks (with `try_send`, so a slow consumer never holds the network up) /// into the sender; the `NetSource` plays them. +#[allow(dead_code)] // RadioMaster-Server (the mic) pub fn net_feed(prebuffer: usize) -> (Sender>, NetSource) { + net_feed_adaptive(prebuffer, prebuffer) +} + +/// Like `net_feed`, but each time the sound runs dry after playing, the prebuffer doubles (up to +/// `grow_to` blocks). Over a LAN it stays short; behind a reverse proxy or CDN that passes the sound +/// on in bursts, it grows until a burst's gap fits, instead of cutting in and out for good. +pub fn net_feed_adaptive(prebuffer: usize, grow_to: usize) -> (Sender>, NetSource) { let prebuffer = prebuffer.max(1); - let (tx, rx) = crossbeam_channel::bounded(prebuffer * 4); - (tx, NetSource { rx, cur: Vec::new(), pos: 0, prebuffer, buffering: true }) + let grow_to = grow_to.max(prebuffer); + let (tx, rx) = crossbeam_channel::bounded(grow_to * 4); + (tx, NetSource { rx, cur: Vec::new(), pos: 0, prebuffer, grow_to, buffering: true }) } impl NetSource { @@ -898,6 +910,8 @@ impl NetSource { self.pos = 0; } Err(_) => { + // ran dry while playing: wait for more next time + self.prebuffer = (self.prebuffer * 2).min(self.grow_to); self.buffering = true; self.cur.clear(); self.pos = 0; @@ -1017,6 +1031,39 @@ mod tests { std::fs::write(path, h).unwrap(); } + #[test] + fn network_feed_waits_longer_after_running_dry() { + let (tx, mut src) = net_feed_adaptive(2, 8); + let play = |src: &mut NetSource, frames: usize| (0..frames).filter(|_| src.frame() != (0.0, 0.0)).count(); + // two blocks of one frame each start it + tx.send(vec![1.0, 1.0]).unwrap(); + assert_eq!(play(&mut src, 1), 0, "still buffering"); + tx.send(vec![1.0, 1.0]).unwrap(); + assert_eq!(play(&mut src, 2), 2); + // it ran dry: now it waits for four blocks + assert_eq!(play(&mut src, 1), 0); + for _ in 0..3 { + tx.send(vec![1.0, 1.0]).unwrap(); + } + assert_eq!(play(&mut src, 1), 0, "three aren't enough any more"); + tx.send(vec![1.0, 1.0]).unwrap(); + assert_eq!(play(&mut src, 4), 4); + // never past grow_to; a burst that big still fits + for _ in 0..3 { + assert_eq!(play(&mut src, 1), 0); + } + assert_eq!(src.prebuffer, 8); + for _ in 0..24 { + tx.try_send(vec![1.0, 1.0]).unwrap(); + } + assert_eq!(play(&mut src, 24), 24); + // a fixed feed (the mic) keeps its prebuffer + let (_tx, mut fixed) = net_feed(3); + fixed.frame(); + fixed.frame(); + assert_eq!(fixed.prebuffer, 3); + } + #[test] fn speech_with_pauses_plays_to_the_end_even_with_silence_trimming() { if std::process::Command::new("espeak-ng").arg("--version").output().is_err() { diff --git a/src/link.rs b/src/link.rs index d674c20..0d03740 100644 --- a/src/link.rs +++ b/src/link.rs @@ -455,6 +455,9 @@ impl Fs { /// How much of the server's mix is buffered before it plays (decoded MP3 frames of about 26 ms). const LISTEN_PREBUFFER: usize = 12; +/// How far that may grow when the sound keeps running dry (about 6 s): a reverse proxy or CDN in +/// front of the server (Cloudflare, nginx, IIS) can pass the stream on in bursts a few seconds apart. +const LISTEN_PREBUFFER_MAX: usize = 230; /// MP3 bitrate of the mic sent to the server. const MIC_KBPS: u32 = 128; /// The mic goes up in slices this long. @@ -496,7 +499,7 @@ impl Listener { token: Arc>, say: Box, ) -> Listener { - let (feed, src) = crate::engine::net_feed(LISTEN_PREBUFFER); + let (feed, src) = crate::engine::net_feed_adaptive(LISTEN_PREBUFFER, LISTEN_PREBUFFER_MAX); engine.set_feed(Some(src)); let stop = Arc::new(AtomicBool::new(false)); let state = Arc::new(Mutex::new(Hearing::Connecting)); diff --git a/src/server.rs b/src/server.rs index 0f1be7f..63f9f22 100644 --- a/src/server.rs +++ b/src/server.rs @@ -339,6 +339,9 @@ const TOKEN_IDLE: Duration = Duration::from_secs(24 * 3600); /// MP3 bitrate of the mix sent to the apps. const LISTEN_KBPS: u32 = 192; +/// How many mix blocks (about 23 ms each) may pile up while an app (or a proxy in front of the +/// server) reads slowly before the oldest are skipped: about 3 s. The app trims its own delay. +const LISTEN_BACKLOG: usize = 130; /// An app's mic goes off when nothing arrives for this long (the app quit or lost the network). const MIC_IDLE: Duration = Duration::from_secs(3); /// Mic audio waits for this many decoded MP3 frames (about 26 ms each) before it plays. @@ -1065,11 +1068,12 @@ impl Api { }; let engine = api.core.lock().engine.clone(); let mut w = req.into_writer(); - let head = "HTTP/1.1 200 OK\r\nContent-Type: audio/mpeg\r\nCache-Control: no-cache, no-store\r\nX-Accel-Buffering: no\r\nConnection: close\r\n\r\n"; + let head = "HTTP/1.1 200 OK\r\nContent-Type: audio/mpeg\r\nCache-Control: no-cache, no-store, no-transform\r\nX-Accel-Buffering: no\r\nConnection: close\r\n\r\n"; if w.write_all(head.as_bytes()).and_then(|_| w.flush()).is_err() { return; } - let (tx, rx) = crossbeam_channel::bounded::>>(64); + // room for a few seconds: a proxy in front of the server can stop reading for a moment + let (tx, rx) = crossbeam_channel::bounded::>>(LISTEN_BACKLOG * 2); engine.add_tap(tx); let mut out = Vec::new(); let mut checked = Instant::now(); @@ -1077,7 +1081,7 @@ impl Api { match rx.recv_timeout(Duration::from_secs(1)) { Ok(pcm) => { // keep the delay short: skip what piled up while the network was slow - while rx.len() > 12 { + while rx.len() > LISTEN_BACKLOG { let _ = rx.try_recv(); } out.clear();