From b2a942baa2bff00c8e77440309f6d257f92fa096 Mon Sep 17 00:00:00 2001 From: defaultdino Date: Thu, 22 Oct 2026 23:38:45 +0200 Subject: [PATCH] move player decoding to separate thread --- src/app/playback.rs | 2 +- src/audio/player.rs | 331 +++++++++++++++++++++++++++----------------- src/audio/source.rs | 2 +- 3 files changed, 206 insertions(+), 129 deletions(-) diff --git a/src/app/playback.rs b/src/app/playback.rs index 5b87c39..eb9f3d2 100644 --- a/src/app/playback.rs +++ b/src/app/playback.rs @@ -99,7 +99,7 @@ impl Playback { self.notice = Some(Message::info(text)); } - /// never blocks the UI: the player thread can be stuck in a stalled read for a while + /// never blocks the UI, even if the player falls behind fn send(&mut self, cmd: PlayerCmd) { match self.player.try_send(cmd) { Ok(()) => {} diff --git a/src/audio/player.rs b/src/audio/player.rs index 7232a63..9d374b9 100644 --- a/src/audio/player.rs +++ b/src/audio/player.rs @@ -22,6 +22,7 @@ use crate::plex::models::Track; const POSITION_EVERY: Duration = Duration::from_millis(250); const IDLE_POLL: Duration = Duration::from_millis(20); const FILL_BUDGET: Duration = Duration::from_millis(50); +const DECODE_AHEAD_PACKETS: usize = 32; pub fn spawn(client: Arc, cache_dir: PathBuf, volume: f32) -> Result { let (cmd_tx, cmd_rx) = sync_channel(32); @@ -39,17 +40,14 @@ pub fn spawn(client: Arc, cache_dir: PathBuf, volume: f32) -> Result } }; let _ = ready_tx.send(Ok(())); - let (opened_tx, opened_rx) = channel(); Player { out, client, cache_dir, events: ev_tx, - opened_tx, - opened_rx, decoding: None, next: None, - next_pending: None, + starved: false, resampler: None, pcm: Vec::new(), ready: Vec::new(), @@ -68,9 +66,94 @@ pub fn spawn(client: Arc, cache_dir: PathBuf, volume: f32) -> Result Ok((cmd_tx, ev_rx)) } -struct Loaded { +enum Chunk { + Pcm(u32, Vec), + Seeked { + generation: u64, + reached_ms: Result, + }, + End, + Failed(String), +} + +struct DecodeThread { track: Arc, - dec: Decoder, + chunks: Receiver, + seeks: Sender<(u64, u64)>, + start_queued: bool, + seek_gen: u64, + awaiting_seek: bool, +} + +impl DecodeThread { + fn spawn( + client: Arc, + cache_dir: PathBuf, + track: Arc, + events: SyncSender, + ) -> Self { + let (chunk_tx, chunks) = sync_channel(DECODE_AHEAD_PACKETS); + let (seeks, seek_rx) = channel(); + let t = track.clone(); + let spawned = thread::Builder::new() + .name("decode".into()) + .spawn(move || decode(&client, &cache_dir, t, &events, &seek_rx, &chunk_tx)); + if let Err(e) = spawned { + warn!("spawning decode: {e}"); + } + Self { + track, + chunks, + seeks, + start_queued: false, + seek_gen: 0, + awaiting_seek: false, + } + } +} + +fn decode( + client: &Arc, + cache_dir: &std::path::Path, + track: Arc, + events: &SyncSender, + seeks: &Receiver<(u64, u64)>, + out: &SyncSender, +) { + let mut dec = match open(client, cache_dir, &track, events) { + Ok(d) => d, + Err(e) => { + let _ = out.send(Chunk::Failed(format!("{}: {e:#}", track.title))); + return; + } + }; + loop { + if let Some((generation, ms)) = seeks.try_iter().last() { + let reached_ms = dec.seek(ms).map_err(|e| format!("seek: {e:#}")); + if out + .send(Chunk::Seeked { + generation, + reached_ms, + }) + .is_err() + { + return; + } + } + let mut pcm = Vec::new(); + let chunk = match dec.next(&mut pcm) { + Ok(true) => Chunk::Pcm(dec.rate(), pcm), + Ok(false) => Chunk::End, + Err(e) => { + warn!("decode error in {}: {e:#}", track.title); + Chunk::Failed(format!("{}: {e:#}", track.title)) + } + }; + let last = !matches!(chunk, Chunk::Pcm(..)); + if out.send(chunk).is_err() || last { + return; + } + } } struct Playing { @@ -84,11 +167,9 @@ struct Player { client: Arc, cache_dir: PathBuf, events: SyncSender, - opened_tx: Sender<(String, Result)>, - opened_rx: Receiver<(String, Result)>, - decoding: Option, - next: Option, - next_pending: Option, + decoding: Option, + next: Option, + starved: bool, resampler: Option, pcm: Vec, ready: Vec, @@ -108,11 +189,16 @@ struct Player { impl Player { fn run(mut self, cmds: Receiver) { loop { - let cmd = if self.wants_data() { + let cmd = if self.wants_data() && !self.starved { cmds.try_recv() .map_err(|e| matches!(e, TryRecvError::Disconnected)) } else if self.active() && !self.paused { - cmds.recv_timeout(self.idle_wait()) + let wait = if self.starved { + IDLE_POLL + } else { + self.idle_wait() + }; + cmds.recv_timeout(wait) .map_err(|e| matches!(e, RecvTimeoutError::Disconnected)) } else { cmds.recv().map_err(|_| true) @@ -125,9 +211,6 @@ impl Player { if self.out.failed() && !self.paused { self.recover_device(); } - while let Ok((rk, res)) = self.opened_rx.try_recv() { - self.on_opened(rk, res); - } self.fill(); self.track_output(); } @@ -202,10 +285,7 @@ impl Player { } fn active(&self) -> bool { - self.playing.is_some() - || self.decoding.is_some() - || !self.starts.is_empty() - || self.next_pending.is_some() + self.playing.is_some() || self.decoding.is_some() || !self.starts.is_empty() } fn handle(&mut self, cmd: PlayerCmd) { @@ -225,17 +305,13 @@ impl Player { PlayerCmd::Stop => { self.reset_pipeline(); self.next = None; - self.next_pending = None; self.out.suspend(); } PlayerCmd::SetVolume(v) => { self.volume = v; self.out.set_volume(v); } - PlayerCmd::ClearNext => { - self.next = None; - self.next_pending = None; - } + PlayerCmd::ClearNext => self.next = None, } } @@ -251,67 +327,41 @@ impl Player { self.written = self.out.played(); } + fn spawn_decoding(&self, track: Arc) -> DecodeThread { + DecodeThread::spawn( + self.client.clone(), + self.cache_dir.clone(), + track, + self.events.clone(), + ) + } + fn load(&mut self, track: Arc) { self.reset_pipeline(); - let loaded = match self.next.take() { - Some(n) if n.track.rating_key == track.rating_key => Ok(n), - _ => open(&self.client, &self.cache_dir, track.clone(), &self.events), - }; - self.next_pending = None; - match loaded { - Ok(l) => { - self.out.resume(); - self.starts.push_back((self.written, l.track.clone())); - self.decoding = Some(l); - self.paused = false; - self.out.set_paused(false); - } - Err(e) => self.error(&track.rating_key, format!("{}: {e:#}", track.title)), - } + let next = self + .next + .take() + .filter(|n| n.track.rating_key == track.rating_key); + self.decoding = Some(next.unwrap_or_else(|| self.spawn_decoding(track))); + self.out.resume(); + self.paused = false; + self.out.set_paused(false); } fn enqueue_next(&mut self, track: Arc) { - let rk = track.rating_key.clone(); - if self.next_pending.as_ref() == Some(&rk) - || self.next.as_ref().is_some_and(|n| n.track.rating_key == rk) + if self + .next + .as_ref() + .is_some_and(|n| n.track.rating_key == track.rating_key) { return; } - self.next = None; - self.next_pending = Some(rk.clone()); - let (client, dir, tx, ev) = ( - self.client.clone(), - self.cache_dir.clone(), - self.opened_tx.clone(), - self.events.clone(), - ); - let spawned = thread::Builder::new() - .name("open-next".into()) - .spawn(move || { - let _ = tx.send((rk, open(&client, &dir, track, &ev))); - }); - if let Err(e) = spawned { - warn!("spawning open-next: {e}"); - self.next_pending = None; - } - } - - fn on_opened(&mut self, rk: String, res: Result) { - if self.next_pending.as_ref() != Some(&rk) { - return; - } - self.next_pending = None; - match res { - Ok(l) => { - self.next = Some(l); - // decoding already ran out (or playback fully ended): chain it in right away - if self.decoding.is_none() { - self.end_at = None; - self.out.resume(); - self.advance(); - } - } - Err(e) => self.error(&rk, format!("{e:#}")), + self.next = Some(self.spawn_decoding(track)); + // decoding already ran out (or playback fully ended): chain it in right away + if self.decoding.is_none() { + self.end_at = None; + self.out.resume(); + self.advance(); } } @@ -324,14 +374,12 @@ impl Player { debug!("seek ignored: decoder already moved past the playing track"); return; } - let reached = match cur.dec.seek(ms) { - Ok(r) => r, - Err(e) => { - let rk = cur.track.rating_key.clone(); - self.error(&rk, format!("seek: {e:#}")); - return; - } - }; + cur.seek_gen += 1; + if cur.seeks.send((cur.seek_gen, ms)).is_err() { + debug!("seek ignored: the track already finished decoding"); + return; + } + cur.awaiting_seek = true; if let Some(r) = self.resampler.as_mut() { r.reset(); } @@ -342,14 +390,36 @@ impl Player { self.written = self.out.played(); if let Some(p) = self.playing.as_mut() { p.start = self.written; - p.offset_ms = reached; + p.offset_ms = ms; } self.last_position = Instant::now() .checked_sub(POSITION_EVERY) .unwrap_or_else(Instant::now); } + fn on_seeked(&mut self, generation: u64, reached_ms: Result) { + let Some(cur) = self.decoding.as_mut() else { + return; + }; + if generation != cur.seek_gen { + return; + } + cur.awaiting_seek = false; + match reached_ms { + Ok(ms) => { + if let Some(p) = self.playing.as_mut() { + p.offset_ms = ms; + } + } + Err(e) => { + let rk = cur.track.rating_key.clone(); + self.error(&rk, e); + } + } + } + fn fill(&mut self) { + self.starved = false; if self.paused { return; } @@ -371,29 +441,54 @@ impl Player { let Some(cur) = self.decoding.as_mut() else { return; }; - self.pcm.clear(); - match cur.dec.next(&mut self.pcm) { - Ok(true) => { - let (rate, rk) = (cur.dec.rate(), cur.track.rating_key.clone()); + let chunk = match cur.chunks.try_recv() { + Ok(c) => c, + Err(TryRecvError::Empty) => { + self.starved = true; + return; + } + Err(TryRecvError::Disconnected) => Chunk::Failed("decoder thread exited".into()), + }; + let (track, awaiting_seek) = (cur.track.clone(), cur.awaiting_seek); + match chunk { + Chunk::Pcm(..) if awaiting_seek => {} + Chunk::Pcm(rate, pcm) => { + if !cur.start_queued { + cur.start_queued = true; + self.queue_start(track.clone(), rate); + } + self.pcm = pcm; if let Err(e) = self.resample(rate) { - self.error(&rk, format!("resampling: {e:#}")); + self.error(&track.rating_key, format!("resampling: {e:#}")); self.advance(); } } - Ok(false) => self.advance(), - Err(e) => { - warn!("decode error in {}: {e:#}", cur.track.title); - let (rk, msg) = ( - cur.track.rating_key.clone(), - format!("{}: {e:#}", cur.track.title), - ); - self.error(&rk, msg); + Chunk::Seeked { + generation, + reached_ms, + } => self.on_seeked(generation, reached_ms), + Chunk::End => self.advance(), + Chunk::Failed(msg) => { + self.error(&track.rating_key, msg); self.advance(); } } } } + fn queue_start(&mut self, track: Arc, rate: u32) { + if self + .resampler + .as_ref() + .is_some_and(|r| r.input_rate() != rate) + { + self.flush_resampler(); + } + let owed = self.resampler.as_ref().map_or(0, Resampler::owed); + let at = self.written + self.queued_frames() + owed; + self.starts.push_back((at, track)); + } + fn resample(&mut self, rate: u32) -> Result<()> { if rate == self.out.rate { self.ready.extend_from_slice(&self.pcm); @@ -420,25 +515,10 @@ impl Player { /// chains the pre-opened next track or marks the end /// if the decoding track is exhausted fn advance(&mut self) { - self.decoding = None; - match self.next.take() { - Some(n) => { - let keeps_rate = self - .resampler - .as_ref() - .is_some_and(|r| r.input_rate() == n.dec.rate()); - if !keeps_rate { - self.flush_resampler(); - } - let owed = self.resampler.as_ref().map_or(0, Resampler::owed); - let at = self.written + self.queued_frames() + owed; - self.starts.push_back((at, n.track.clone())); - self.decoding = Some(n); - } - None => { - self.flush_resampler(); - self.end_at = Some(self.written + self.queued_frames()); - } + self.decoding = self.next.take(); + if self.decoding.is_none() { + self.flush_resampler(); + self.end_at = Some(self.written + self.queued_frames()); } } @@ -468,9 +548,7 @@ impl Player { if self.playing.take().is_some() { self.emit(PlayerEvent::TrackEnded); } - if self.next_pending.is_none() { - self.out.suspend(); - } + self.out.suspend(); } if let Some(p) = &self.playing && !self.paused @@ -487,9 +565,9 @@ impl Player { fn open( client: &Arc, cache_dir: &std::path::Path, - track: Arc, + track: &Track, events: &SyncSender, -) -> Result { +) -> Result { static SEQ: AtomicU64 = AtomicU64::new(0); if track.part_key.is_empty() { anyhow::bail!("{} has no playable media", track.title); @@ -500,7 +578,6 @@ fn open( let _ = ev.try_send(PlayerEvent::BufferingChanged(b)); }); let source = CachedHttpSource::open(client, &track.part_key, path, on_buffering)?; - let dec = Decoder::open(Box::new(source), track.container.as_deref()) - .with_context(|| format!("{} ({:?})", track.title, track.audio_codec))?; - Ok(Loaded { track, dec }) + Decoder::open(Box::new(source), track.container.as_deref()) + .with_context(|| format!("{} ({:?})", track.title, track.audio_codec)) } diff --git a/src/audio/source.rs b/src/audio/source.rs index f8434a0..d52d083 100644 --- a/src/audio/source.rs +++ b/src/audio/source.rs @@ -16,7 +16,7 @@ const BUFFERING_AFTER: Duration = Duration::from_millis(200); const JUMP_AHEAD: u64 = 256 * 1024; /// a blocked read with no download progress for this long restarts the download const STALL_AFTER: Duration = Duration::from_secs(10); -/// a read blocked this long without progress fails, so the player thread is never stuck for good +/// a read blocked this long without progress fails, so a decode thread is never stuck for good const GIVE_UP_AFTER: Duration = Duration::from_secs(30); /// network errors tolerated in a row (2 + 4 + 8 + 8 + 8 s of backoff) before giving up const RETRIES: u32 = 5;