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;