From 8d795c29e6c132b457dd5e6ec153fb220f039931 Mon Sep 17 00:00:00 2001 From: Nils Pukropp Date: Tue, 13 Oct 2026 16:40:18 +0200 Subject: [PATCH] perf(stream,player,app): start a resumed episode at its place, with that block already coming Resuming loaded the file from the start and sought on FileLoaded. By then mpv's demuxer had begun reading ahead from the beginning, so the blocks the seek wanted queued behind ones fetched to be thrown away. Player::play_from sets mpv's start option instead, and puts it back once the file has loaded. VideoStream::warm_resume fetches, beside the two edge blocks, the block the resume position falls in, estimated from the running time saved with the watch state. A video's bitrate is uneven, so it can land a block off; then it is one block for nothing, against a round trip saved when it lands. --- crates/pstr-app/src/engine.rs | 22 +++++++++++-- crates/pstr-app/src/playback.rs | 14 +++------ crates/pstr-player/src/player.rs | 50 ++++++++++++++++++++++++++++-- crates/pstr-stream/src/stream.rs | 53 +++++++++++++++++++++++++++++--- 4 files changed, 121 insertions(+), 18 deletions(-) diff --git a/crates/pstr-app/src/engine.rs b/crates/pstr-app/src/engine.rs index 63641c0..0dc798b 100644 --- a/crates/pstr-app/src/engine.rs +++ b/crates/pstr-app/src/engine.rs @@ -1702,8 +1702,12 @@ impl Engine { Ok(stream) => { // Before mpv is even built: building it takes the UI thread // a moment, and the network can spend that moment on the - // two reads every file opens with. - stream.warm_edges(); + // two reads every file opens with — and, resuming, on the + // one it is about to seek to. + match engine.resume_fraction(&target) { + Some(fraction) => stream.warm_resume(fraction), + None => stream.warm_edges(), + } engine.emit(Event::PlaybackReady { target: Box::new(target), stream, @@ -1714,6 +1718,20 @@ impl Engine { }); } + /// How far into the file `target` resumes, as a fraction of its running + /// time — known from the last time it played. + fn resume_fraction(&self, target: &PlaybackTarget) -> Option { + let at = target.resume_at?; + let state = self + .catalog + .lock() + .watch_state(&target.share_id, &target.link_id) + .ok() + .flatten()?; + let duration = state.duration_secs.filter(|duration| *duration > 0.0)?; + Some(at / duration) + } + /// Fetch the blocks a file opens with into the cache, without playing it. /// /// For the episode after the one playing, so that autoplay — or a click on diff --git a/crates/pstr-app/src/playback.rs b/crates/pstr-app/src/playback.rs index e16985c..101e19b 100644 --- a/crates/pstr-app/src/playback.rs +++ b/crates/pstr-app/src/playback.rs @@ -516,7 +516,11 @@ fn run( orders: std::sync::mpsc::Receiver, finished: Arc, ) { - let handle = match player.play(stream) { + let opened = match target.resume_at { + Some(at) => player.play_from(stream, at), + None => player.play(stream), + }; + let handle = match opened { Ok(handle) => handle, Err(error) => { engine.emit(Event::Error(format!("load {}: {error}", target.name))); @@ -529,7 +533,6 @@ fn run( let mut position = target.resume_at.unwrap_or(0.0); let mut duration: Option = None; let mut watched = false; - let mut resumed = target.resume_at.is_none(); let mut last_save = std::time::Instant::now(); 'playback: loop { @@ -563,13 +566,6 @@ fn run( restore_tracks(&player, show); } emit_tracks(&engine, id, &player); - // Only now is there a timeline to seek within. - if !resumed && let Some(at) = target.resume_at { - resumed = true; - if let Err(error) = player.seek_to(at) { - tracing::warn!("resume at {at}: {error}"); - } - } } PlayerEvent::Duration(value) => duration = Some(*value), PlayerEvent::Position(value) => { diff --git a/crates/pstr-player/src/player.rs b/crates/pstr-player/src/player.rs index bbd8fe5..5c7792b 100644 --- a/crates/pstr-player/src/player.rs +++ b/crates/pstr-player/src/player.rs @@ -217,6 +217,9 @@ pub struct Player { /// no stream callback can be running. mpv: Mpv, registry: Arc, + /// [`Self::play_from`] set mpv's `start`, which holds for every file + /// after until it is put back. + start_set: std::sync::atomic::AtomicBool, } impl Player { @@ -298,7 +301,11 @@ impl Player { protocol::register(&mpv, ®istry)?; - let player = Self { mpv, registry }; + let player = Self { + mpv, + registry, + start_set: std::sync::atomic::AtomicBool::new(false), + }; player.observe_playback_properties()?; Ok(player) } @@ -325,11 +332,42 @@ impl Player { /// The returned handle keeps the URL resolvable; drop it once the file is /// done with. Dropping it does not interrupt playback already underway. pub fn play(&self, stream: VideoStream) -> Result { + self.clear_start(); + self.load(stream) + } + + fn load(&self, stream: VideoStream) -> Result { let handle = self.registry.publish(stream); self.mpv.command("loadfile", &[&handle.url()])?; Ok(handle) } + /// Start playing a stream `seconds` in. + /// + /// Through mpv's `start` option rather than a seek once the file has + /// loaded: by then the demuxer has begun reading ahead from the + /// beginning, which on a stream is blocks fetched only to be thrown away, + /// queued in front of the ones the seek wants. + pub fn play_from(&self, stream: VideoStream, seconds: f64) -> Result { + self.mpv.set_property("start", format!("{seconds}"))?; + self.start_set + .store(true, std::sync::atomic::Ordering::Release); + self.load(stream) + } + + /// Put `start` back once the file it was for has loaded, so the next + /// one starts at its beginning. [`Self::play`] puts it back too, for a + /// file that never loaded. + fn clear_start(&self) { + if self + .start_set + .swap(false, std::sync::atomic::Ordering::AcqRel) + && let Err(error) = self.mpv.set_property("start", "none") + { + tracing::warn!(%error, "could not clear mpv's start"); + } + } + /// Play anything else mpv can open: a local path, or one of its own /// pseudo-protocols. /// @@ -351,7 +389,15 @@ impl Player { /// property that briefly has no value is normal, not fatal. pub fn poll_event(&self, timeout: f64) -> Option { match self.mpv.wait_event(timeout)? { - Ok(event) => Some(translate(event)), + Ok(event) => { + let event = translate(event); + // Not on an end of file: loading one file ends the one it + // replaces, before the new one has read `start`. + if matches!(event, PlayerEvent::FileLoaded) { + self.clear_start(); + } + Some(event) + } Err(error) => { tracing::debug!(%error, "mpv event carried an error"); Some(PlayerEvent::Other) diff --git a/crates/pstr-stream/src/stream.rs b/crates/pstr-stream/src/stream.rs index 4868966..5aa61c3 100644 --- a/crates/pstr-stream/src/stream.rs +++ b/crates/pstr-stream/src/stream.rs @@ -223,20 +223,48 @@ impl VideoStream { /// seek, and a seek cancels read-ahead — which here would cancel the very /// fetch it was meant to have started. pub fn warm_edges(&self) { - let Ok(runtime) = tokio::runtime::Handle::try_current() else { - return; - }; let count = self.inner.map.len(); let edges: &[usize] = match count { 0 => &[], 1 => &[0], _ => &[0, count - 1], }; - for &index in edges { + self.warm(edges); + } + + /// [`Self::warm_edges`], and the block `fraction` of the way through the + /// file: where a resumed episode seeks once its index is read. + /// + /// An estimate, from the bytes being spread evenly over the running time. + /// A video's bitrate is not even, so it can land a block off the one the + /// seek wants. Then it is one block fetched for nothing, against a round + /// trip saved, after the two the edges cost, when it lands. + pub fn warm_resume(&self, fraction: f64) { + self.warm_edges(); + if !(0.0..1.0).contains(&fraction) { + return; + } + let offset = (self.inner.map.size() as f64 * fraction) as u64; + let count = self.inner.map.len(); + if let Some(index) = self.inner.map.block_at(offset) + && index != 0 + && index + 1 != count + { + self.warm(&[index]); + } + } + + /// Fetch these blocks in the background. Detached: see + /// [`Self::warm_edges`] for why not read-ahead. + fn warm(&self, indices: &[usize]) { + let Ok(runtime) = tokio::runtime::Handle::try_current() else { + return; + }; + for &index in indices { let inner = Arc::clone(&self.inner); runtime.spawn(async move { if let Err(error) = block_at(inner, index).await { - tracing::debug!(index, %error, "warming an edge block failed"); + tracing::debug!(index, %error, "warming a block failed"); } }); } @@ -1037,6 +1065,21 @@ mod tests { assert_eq!(fake.fetches(), 2); } + #[tokio::test] + async fn warming_a_resume_fetches_the_block_that_far_in_besides_the_edges() { + let fake = Fake::new(&[1000; 10]); + let stream = stream_over(Arc::clone(&fake), 0); + + stream.warm_resume(0.45); + assert!( + eventually(|| { + fake.fetches_of(0) == 1 && fake.fetches_of(4) == 1 && fake.fetches_of(9) == 1 + }) + .await + ); + assert_eq!(fake.fetches(), 3); + } + #[tokio::test] async fn a_fetched_block_reaches_the_disk_cache_without_the_read_waiting_for_it() { let disk = disk_in("fill").await;