diff --git a/crates/qbz-audio/src/alsa_direct.rs b/crates/qbz-audio/src/alsa_direct.rs index c2d05d357..56fe428f4 100644 --- a/crates/qbz-audio/src/alsa_direct.rs +++ b/crates/qbz-audio/src/alsa_direct.rs @@ -12,6 +12,24 @@ use std::sync::atomic::AtomicBool; #[cfg(target_os = "linux")] use std::sync::{Arc, Mutex}; +/// Log a PCM recovery and record it as a network-throttle underrun signal. +/// +/// Each call to ALSA's `pcm.recover()` that returns successfully indicates +/// that the writer thread fell behind the kernel's playback buffer — i.e. +/// an audio underrun. The network throttle treats this as the strongest +/// possible "slow down" signal and immediately drops the prefetch cap to +/// zero for `PANIC_WINDOW_SECS`, so the live stream gets the full pipe to +/// recover. +#[cfg(target_os = "linux")] +fn log_pcm_recovery(suffix: &str) { + if suffix.is_empty() { + log::warn!("[ALSA Direct] Recovered from PCM error"); + } else { + log::warn!("[ALSA Direct] Recovered from PCM error {}", suffix); + } + crate::network_throttle::state().record_underrun(); +} + /// Direct ALSA PCM stream for hw: devices /// /// Field order is significant: Rust drops struct fields top-to-bottom, so the @@ -223,7 +241,7 @@ impl AlsaDirectStream { if let Err(recover_err) = pcm.recover(e.errno() as i32, false) { Err(format!("Failed to recover from error: {}", recover_err)) } else { - log::warn!("[ALSA Direct] Recovered from PCM error"); + log_pcm_recovery(""); Ok(()) } } @@ -252,7 +270,7 @@ impl AlsaDirectStream { if let Err(recover_err) = pcm.recover(e.errno() as i32, false) { Err(format!("Failed to recover from error: {}", recover_err)) } else { - log::warn!("[ALSA Direct] Recovered from PCM error"); + log_pcm_recovery(""); Ok(()) } } @@ -279,7 +297,7 @@ impl AlsaDirectStream { if let Err(recover_err) = pcm.recover(e.errno() as i32, false) { Err(format!("Failed to recover from error: {}", recover_err)) } else { - log::warn!("[ALSA Direct] Recovered from PCM error"); + log_pcm_recovery(""); Ok(()) } } @@ -318,7 +336,7 @@ impl AlsaDirectStream { if let Err(recover_err) = pcm.recover(e.errno() as i32, false) { Err(format!("Failed to recover from error: {}", recover_err)) } else { - log::warn!("[ALSA Direct] Recovered from PCM error (S24_3LE)"); + log_pcm_recovery("(S24_3LE)"); Ok(()) } } @@ -348,7 +366,7 @@ impl AlsaDirectStream { if let Err(recover_err) = pcm.recover(e.errno() as i32, false) { Err(format!("Failed to recover from error: {}", recover_err)) } else { - log::warn!("[ALSA Direct] Recovered from PCM error"); + log_pcm_recovery(""); Ok(()) } } @@ -388,7 +406,7 @@ impl AlsaDirectStream { if let Err(recover_err) = pcm.recover(e.errno() as i32, false) { Err(format!("Failed to recover from error: {}", recover_err)) } else { - log::warn!("[ALSA Direct] Recovered from PCM error"); + log_pcm_recovery(""); Ok(()) } } @@ -420,7 +438,7 @@ impl AlsaDirectStream { if let Err(recover_err) = pcm.recover(e.errno() as i32, false) { Err(format!("Failed to recover from error: {}", recover_err)) } else { - log::warn!("[ALSA Direct] Recovered from PCM error"); + log_pcm_recovery(""); Ok(()) } } @@ -456,7 +474,7 @@ impl AlsaDirectStream { if let Err(recover_err) = pcm.recover(e.errno() as i32, false) { Err(format!("Failed to recover from error: {}", recover_err)) } else { - log::warn!("[ALSA Direct] Recovered from PCM error"); + log_pcm_recovery(""); Ok(()) } } @@ -493,7 +511,7 @@ impl AlsaDirectStream { if let Err(recover_err) = pcm.recover(e.errno() as i32, false) { Err(format!("Failed to recover from error: {}", recover_err)) } else { - log::warn!("[ALSA Direct] Recovered from PCM error (S24_3LE)"); + log_pcm_recovery("(S24_3LE)"); Ok(()) } } @@ -523,7 +541,7 @@ impl AlsaDirectStream { if let Err(recover_err) = pcm.recover(e.errno() as i32, false) { Err(format!("Failed to recover from error: {}", recover_err)) } else { - log::warn!("[ALSA Direct] Recovered from PCM error"); + log_pcm_recovery(""); Ok(()) } } diff --git a/crates/qbz-audio/src/lib.rs b/crates/qbz-audio/src/lib.rs index 91ca71ec6..895df4d8b 100644 --- a/crates/qbz-audio/src/lib.rs +++ b/crates/qbz-audio/src/lib.rs @@ -45,6 +45,7 @@ pub mod dynamic_amplify; pub mod loudness; pub mod loudness_analyzer; pub mod loudness_cache; +pub mod network_throttle; pub mod output_sinks; pub mod settings; pub mod visualizer; diff --git a/crates/qbz-audio/src/network_throttle.rs b/crates/qbz-audio/src/network_throttle.rs new file mode 100644 index 000000000..c119a2758 --- /dev/null +++ b/crates/qbz-audio/src/network_throttle.rs @@ -0,0 +1,244 @@ +//! Adaptive prefetch throttling based on observed network conditions. +//! +//! Default QBZ prefetches up to 5 tracks concurrently (≈30 simultaneous +//! HTTP/2 streams once the CMAF init + segment fan-out is counted). That +//! aggressive concurrency is exactly what makes Hi-Res playback smooth on +//! a fast connection, and we don't want to dial it down for everyone just +//! because some users have a slow link. +//! +//! Instead, this module watches two real-time signals — measured per-segment +//! bandwidth and audio underruns — and computes a *dynamic* cap that the +//! prefetch dispatcher consults. When conditions are good the cap is the +//! full memory-profile default (5 on Normal-class hosts, 1 on LowMemory). +//! When the live stream starts losing the race against the consumer, the +//! cap collapses to 0 and the prefetch fan-out gets out of the way. +//! +//! Recovery follows TCP-style slow-start logic: after `PANIC_WINDOW_SECS` +//! of no fresh underruns the cap walks back up one level at a time, +//! re-validated by each new bandwidth sample. + +use std::sync::OnceLock; +use std::sync::RwLock; +use std::time::Instant; + +/// EMA smoothing factor for observed bandwidth. Higher = more reactive to +/// the latest sample, lower = smoother but laggier. 0.4 is a compromise: +/// a single bad sample dents the estimate but doesn't dominate it. +const BANDWIDTH_EMA_ALPHA: f64 = 0.4; + +/// How long we stay in panic mode after an audio underrun. The user just +/// experienced a glitch; we want the stream to recover with the entire +/// pipe to itself for a meaningful window before letting prefetch back in. +const PANIC_WINDOW_SECS: u64 = 30; + +/// Bandwidth-to-playback ratio thresholds for the throttle levels. +/// +/// - At or below `SURVIVING_RATIO`: no prefetch at all. The live stream +/// barely has bandwidth to keep itself fed. +/// - At or below `CAUTIOUS_RATIO`: only one prefetch track in flight. +/// - At or below `RELAXED_RATIO`: two prefetch tracks. +/// - Above `RELAXED_RATIO`: full memory-profile default. +const SURVIVING_RATIO: f64 = 1.5; +const CAUTIOUS_RATIO: f64 = 2.5; +const RELAXED_RATIO: f64 = 4.0; + +/// Approximate sustained bandwidth required for live playback by quality +/// tier, in MB/s. These numbers are for CMAF/FLAC compressed streams; +/// they're inputs to the ratio comparison above, not hard limits. +pub fn playback_mbps_for_quality(quality_tag: PlaybackQualityTag) -> f64 { + match quality_tag { + PlaybackQualityTag::UltraHiRes => 2.5, // 24-bit / 192 kHz FLAC + PlaybackQualityTag::HiRes => 1.4, // 24-bit / 96 kHz FLAC + PlaybackQualityTag::Lossless => 0.5, // 16-bit / 44.1 kHz FLAC + PlaybackQualityTag::Lossy => 0.04, // 320 kbps MP3 + } +} + +/// Small enum that mirrors `qbz-models::Quality` without taking a runtime +/// dependency on it. The callsite translates its own quality enum into one +/// of these four buckets before consulting the throttle. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum PlaybackQualityTag { + UltraHiRes, + HiRes, + Lossless, + Lossy, +} + +#[derive(Debug, Default)] +struct ThrottleInner { + /// Exponential moving average of observed segment-download bandwidth + /// in MB/s. `None` until the first sample arrives. + bandwidth_ema_mbps: Option, + /// Timestamp of the most recent audio underrun. `None` means we have + /// not observed one this session. + last_underrun: Option, +} + +pub struct ThrottleState { + inner: RwLock, +} + +static GLOBAL: OnceLock = OnceLock::new(); + +/// Singleton accessor. Lazily initialized on first call. +pub fn state() -> &'static ThrottleState { + GLOBAL.get_or_init(|| ThrottleState { + inner: RwLock::new(ThrottleInner::default()), + }) +} + +impl ThrottleState { + /// Feed a fresh per-segment bandwidth measurement (MB/s). Called from + /// the CMAF streaming loop every few segments. + pub fn record_segment_bandwidth(&self, mbps: f64) { + if !mbps.is_finite() || mbps <= 0.0 { + return; + } + if let Ok(mut inner) = self.inner.write() { + inner.bandwidth_ema_mbps = Some(match inner.bandwidth_ema_mbps { + Some(prev) => prev * (1.0 - BANDWIDTH_EMA_ALPHA) + mbps * BANDWIDTH_EMA_ALPHA, + None => mbps, + }); + } + } + + /// Signal that an audio buffer underrun just happened. Forces the + /// throttle into panic mode for `PANIC_WINDOW_SECS`. + pub fn record_underrun(&self) { + if let Ok(mut inner) = self.inner.write() { + inner.last_underrun = Some(Instant::now()); + } + } + + /// Current EMA bandwidth in MB/s, or `None` if no samples yet. + pub fn current_bandwidth_mbps(&self) -> Option { + self.inner.read().ok().and_then(|i| i.bandwidth_ema_mbps) + } + + /// True when an underrun was recorded within `PANIC_WINDOW_SECS`. + pub fn in_panic_mode(&self) -> bool { + let inner = match self.inner.read() { + Ok(g) => g, + Err(_) => return false, + }; + match inner.last_underrun { + Some(t) => t.elapsed().as_secs() < PANIC_WINDOW_SECS, + None => false, + } + } + + /// Decide the prefetch cap given the current track's playback rate and + /// the memory-profile default. The cap is always clamped to `[0, default_cap]` + /// — we never *raise* prefetch above the memory profile, only restrict. + pub fn current_prefetch_cap(&self, playback_mbps: f64, default_cap: usize) -> usize { + if self.in_panic_mode() { + return 0; + } + let bw = match self.current_bandwidth_mbps() { + Some(v) => v, + // No samples yet — trust the memory profile default. The first + // segment of the first track will land within a few seconds and + // give us real numbers. + None => return default_cap, + }; + let ratio = if playback_mbps > 0.0 { + bw / playback_mbps + } else { + f64::INFINITY + }; + if ratio <= SURVIVING_RATIO { + 0 + } else if ratio <= CAUTIOUS_RATIO { + 1.min(default_cap) + } else if ratio <= RELAXED_RATIO { + 2.min(default_cap) + } else { + default_cap + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn fresh_state_returns_default_cap() { + let s = ThrottleState { + inner: RwLock::new(ThrottleInner::default()), + }; + assert_eq!(s.current_prefetch_cap(2.5, 5), 5); + } + + #[test] + fn panic_mode_zeros_cap() { + let s = ThrottleState { + inner: RwLock::new(ThrottleInner::default()), + }; + s.record_underrun(); + assert_eq!(s.current_prefetch_cap(2.5, 5), 0); + } + + #[test] + fn surviving_ratio_zeros_prefetch() { + let s = ThrottleState { + inner: RwLock::new(ThrottleInner::default()), + }; + // bw = 3.0, playback = 2.5 → ratio = 1.2 (< 1.5) + s.record_segment_bandwidth(3.0); + assert_eq!(s.current_prefetch_cap(2.5, 5), 0); + } + + #[test] + fn cautious_ratio_allows_one() { + let s = ThrottleState { + inner: RwLock::new(ThrottleInner::default()), + }; + // bw = 5.0, playback = 2.5 → ratio = 2.0 (between 1.5 and 2.5) + s.record_segment_bandwidth(5.0); + assert_eq!(s.current_prefetch_cap(2.5, 5), 1); + } + + #[test] + fn relaxed_ratio_allows_two() { + let s = ThrottleState { + inner: RwLock::new(ThrottleInner::default()), + }; + // bw = 8.0, playback = 2.5 → ratio = 3.2 (between 2.5 and 4.0) + s.record_segment_bandwidth(8.0); + assert_eq!(s.current_prefetch_cap(2.5, 5), 2); + } + + #[test] + fn abundant_bandwidth_unlocks_default() { + let s = ThrottleState { + inner: RwLock::new(ThrottleInner::default()), + }; + // bw = 20.0, playback = 2.5 → ratio = 8.0 (well above 4.0) + s.record_segment_bandwidth(20.0); + assert_eq!(s.current_prefetch_cap(2.5, 5), 5); + } + + #[test] + fn cap_never_exceeds_default() { + let s = ThrottleState { + inner: RwLock::new(ThrottleInner::default()), + }; + s.record_segment_bandwidth(20.0); + // Memory profile says 1 — never raise above. + assert_eq!(s.current_prefetch_cap(2.5, 1), 1); + } + + #[test] + fn ema_smooths_spikes() { + let s = ThrottleState { + inner: RwLock::new(ThrottleInner::default()), + }; + s.record_segment_bandwidth(10.0); + s.record_segment_bandwidth(1.0); + // EMA: 10 * 0.6 + 1 * 0.4 = 6.4 + let bw = s.current_bandwidth_mbps().unwrap(); + assert!((bw - 6.4).abs() < 0.01); + } +} diff --git a/crates/qbz-player/src/player/streaming_source.rs b/crates/qbz-player/src/player/streaming_source.rs index 2a52e6ac7..b17dc1b95 100644 --- a/crates/qbz-player/src/player/streaming_source.rs +++ b/crates/qbz-player/src/player/streaming_source.rs @@ -282,6 +282,15 @@ impl BufferedMediaSource { /// /// Used to store in cache after streaming playback completes. /// Returns None if download is not complete or failed. + /// + /// IMPORTANT: This clones the buffer rather than moving it. Earlier + /// attempts to use `mem::take` here regressed playback — the + /// `Source` impl is still actively reading from `state.data` when + /// the promotion path calls this, and zeroing the buffer out from + /// under the reader caused immediate EOF (tracks ending 10s into a + /// 104s file). Cloning is the safe choice; the audible hiccup at + /// promotion that the move attempted to fix needs a different + /// approach (shared `Arc>` ownership, or off-thread copy). pub fn take_complete_data(&self) -> Option> { let (lock, _) = &*self.state; if let Ok(state) = lock.lock() { diff --git a/src-tauri/src/commands_v2/helpers.rs b/src-tauri/src/commands_v2/helpers.rs index b4523e433..ebf260a1c 100644 --- a/src-tauri/src/commands_v2/helpers.rs +++ b/src-tauri/src/commands_v2/helpers.rs @@ -960,17 +960,22 @@ pub async fn v2_cmaf_stream( // Progress logging every 5 segments or on last segment if seg_idx % 5 == 0 || seg_idx == n_segments { let elapsed = start.elapsed().as_secs_f64(); + let mbps = if elapsed > 0.0 { + total_written as f64 / (1024.0 * 1024.0) / elapsed + } else { + 0.0 + }; log::info!( "[V2/CMAF-STREAM] Segment {}/{} ({:.1} MB, {:.1} MB/s)", seg_idx, n_segments - 1, total_written as f64 / (1024.0 * 1024.0), - if elapsed > 0.0 { - total_written as f64 / (1024.0 * 1024.0) / elapsed - } else { - 0.0 - } + mbps ); + // Feed the network-throttle state with the latest observed + // bandwidth so the prefetch scheduler can scale down on slow + // connections and stay out of the live stream's way. + qbz_audio::network_throttle::state().record_segment_bandwidth(mbps); } } diff --git a/src-tauri/src/commands_v2/playback.rs b/src-tauri/src/commands_v2/playback.rs index 71d7b3b6a..3c4e83c3b 100644 --- a/src-tauri/src/commands_v2/playback.rs +++ b/src-tauri/src/commands_v2/playback.rs @@ -26,14 +26,53 @@ use super::cached_audio_incompatible_with_hw; // ==================== Prefetch (V2) ==================== -/// Number of Qobuz tracks to prefetch — resolved per-host from the -/// detected memory profile. Normal hosts get 5 tracks (~2 minutes -/// of HiRes cache ahead); LowMemory hosts (issue #331) get 1 to keep -/// the in-memory footprint manageable on devices like the Pi 3B. -fn v2_prefetch_count() -> usize { +/// Memory-profile default number of Qobuz tracks to prefetch. Normal +/// hosts get 5 (~2 minutes of HiRes cache ahead); LowMemory hosts +/// (issue #331) get 1 to keep the in-memory footprint manageable on +/// devices like the Pi 3B. This is the upper bound the adaptive +/// throttle is allowed to walk back up to — it never raises above this. +fn v2_prefetch_count_default() -> usize { qbz_core::system_capabilities::memory_profile().prefetch_count } +/// Translate the active playback `Quality` into the throttle's coarse +/// quality bucket, used to estimate the live stream's sustained +/// bandwidth requirement (MB/s). Inputs to `current_prefetch_cap()`. +fn quality_to_throttle_tag(q: qbz_models::Quality) -> qbz_audio::network_throttle::PlaybackQualityTag { + use qbz_audio::network_throttle::PlaybackQualityTag; + use qbz_models::Quality; + match q { + Quality::UltraHiRes => PlaybackQualityTag::UltraHiRes, + Quality::HiRes => PlaybackQualityTag::HiRes, + Quality::Lossless => PlaybackQualityTag::Lossless, + Quality::Mp3 => PlaybackQualityTag::Lossy, + } +} + +/// Adaptive prefetch cap. Starts at the memory-profile default and gets +/// scaled down by the network-throttle state based on observed +/// bandwidth and recent audio underruns. The throttle never raises the +/// cap above the memory-profile default, so a user with a fat +/// connection still gets the full 5-track fan-out the firmware intends. +fn v2_prefetch_count(quality: qbz_models::Quality) -> usize { + let default_cap = v2_prefetch_count_default(); + let tag = quality_to_throttle_tag(quality); + let playback_mbps = qbz_audio::network_throttle::playback_mbps_for_quality(tag); + let throttled_cap = qbz_audio::network_throttle::state() + .current_prefetch_cap(playback_mbps, default_cap); + if throttled_cap < default_cap { + log::debug!( + "[V2/PREFETCH] Adaptive cap: {} (down from default {}). bw={:?} MB/s, panic={}, playback_need={:.2} MB/s", + throttled_cap, + default_cap, + qbz_audio::network_throttle::state().current_bandwidth_mbps(), + qbz_audio::network_throttle::state().in_panic_mode(), + playback_mbps, + ); + } + throttled_cap +} + /// How far ahead to look for tracks to prefetch (to handle mixed playlists /// with local/offline tracks interspersed with Qobuz tracks) const V2_PREFETCH_LOOKAHEAD: usize = 15; @@ -98,7 +137,7 @@ fn spawn_v2_prefetch_with_hw_check( let mut qobuz_prefetched = 0; - let prefetch_cap = v2_prefetch_count(); + let prefetch_cap = v2_prefetch_count(quality); for track in upcoming_tracks { // Stop once we've prefetched enough Qobuz tracks if qobuz_prefetched >= prefetch_cap {