From 8f78f5c1b13234ebd882f3bf9f66214e680fe5ce Mon Sep 17 00:00:00 2001 From: Nils Pukropp Date: Fri, 2 Oct 2026 19:27:27 +0200 Subject: [PATCH] perf(stream,core): take the disk cache off the read path, warm file edges, report crawl progress A block the player waits for no longer waits on its disk-cache write and two fsyncs; the put runs in a spawned task and only the block itself is synced before the rename. Opening a stream can fetch the first and last block together, where MKV cues and the MP4 moov live. Downloads open a copying stream that reads the disk cache but never fills it. crawl_reporting reports the running file count after each listing. --- crates/pstr-core/src/shares.rs | 23 +++++ crates/pstr-stream/src/disk.rs | 8 +- crates/pstr-stream/src/source.rs | 18 ++++ crates/pstr-stream/src/stream.rs | 154 ++++++++++++++++++++++++++++++- 4 files changed, 195 insertions(+), 8 deletions(-) diff --git a/crates/pstr-core/src/shares.rs b/crates/pstr-core/src/shares.rs index 3ed8bf3..e14a953 100644 --- a/crates/pstr-core/src/shares.rs +++ b/crates/pstr-core/src/shares.rs @@ -358,6 +358,19 @@ impl SharedLibrary { /// and a catalog that reorders itself between two identical crawls is /// needlessly hard to diff. pub async fn crawl(&self, share_id: &str) -> Result> { + self.crawl_reporting(share_id, |_| {}).await + } + + /// [`Self::crawl`], calling `progress` with how many nodes have been found + /// so far each time a folder's listing comes back. + /// + /// A crawl of a large share takes minutes, and "crawling…" for minutes + /// reads as stuck. A count that keeps rising does not. + pub async fn crawl_reporting( + &self, + share_id: &str, + progress: impl Fn(usize), + ) -> Result> { use futures::stream::{StreamExt as _, TryStreamExt as _}; let client = self @@ -392,6 +405,7 @@ impl SharedLibrary { } found.push(child); } + progress(found.len()); } level = next; } @@ -428,6 +442,15 @@ fn client_configuration() -> ProtonClientConfiguration { /// Parsed here as well as in the SDK because the store needs a stable id for a /// share *before* it has opened it — and the id must not be derived from the /// password, which would put a secret in the config file by the back door. +/// The share token a link names, or why it is not a usable share link. +/// +/// Public so a form can say what is wrong with a link while it is still being +/// pasted, rather than after a round trip through [`ShareStore::add`] — which +/// asks exactly this first. +pub fn share_token(url: &str) -> Result { + token_from_url(url) +} + fn token_from_url(url: &str) -> Result { let (_, tail) = url .split_once("/urls/") diff --git a/crates/pstr-stream/src/disk.rs b/crates/pstr-stream/src/disk.rs index 10f0caf..a39c656 100644 --- a/crates/pstr-stream/src/disk.rs +++ b/crates/pstr-stream/src/disk.rs @@ -321,11 +321,9 @@ async fn write_entry(path: &Path, block: &[u8]) -> Result<()> { } tokio::fs::rename(&temp, path).await?; - let mut sidecar = tokio::fs::File::create(meta_path(path)).await?; - sidecar - .write_all(block.len().to_string().as_bytes()) - .await?; - sidecar.sync_all().await?; + // Not synced. The block's sync is the one the ordering depends on; a + // sidecar lost to a crash only turns a valid entry into a miss. + tokio::fs::write(meta_path(path), block.len().to_string()).await?; Ok(()) } diff --git a/crates/pstr-stream/src/source.rs b/crates/pstr-stream/src/source.rs index 534994c..1dc592d 100644 --- a/crates/pstr-stream/src/source.rs +++ b/crates/pstr-stream/src/source.rs @@ -155,6 +155,24 @@ impl StreamSource { .await } + /// Open a file to copy it out whole, rather than to play it. + /// + /// Not the playback stream, and not shared with it: its blocks are read + /// once and written to a file, so caching them in memory only evicts the + /// blocks a viewer is about to need, and caching them on disk stores every + /// byte twice while pushing out what was just watched. It does still read + /// the disk cache — an episode watched and then downloaded is not fetched + /// again. + pub async fn open_for_copy(&self, share_id: &str, uid: &NodeUid) -> Result { + let blocks = self.inner.opener.open(share_id, uid).await?; + Ok(VideoStream::copying( + uid.clone(), + blocks, + Arc::new(BlockRing::new(0)), + self.inner.disk.clone(), + )) + } + /// Drop a stream and everything cached for it in memory. /// /// The disk cache is deliberately untouched: it exists precisely so the next diff --git a/crates/pstr-stream/src/stream.rs b/crates/pstr-stream/src/stream.rs index 56e6824..30d29e4 100644 --- a/crates/pstr-stream/src/stream.rs +++ b/crates/pstr-stream/src/stream.rs @@ -103,6 +103,11 @@ struct Inner { map: BlockMap, ring: Arc, disk: Option>, + /// Whether blocks fetched from the network are written to `disk`. Off for + /// a stream being copied out to a file, which reads what the cache already + /// holds but should not push what a viewer just watched out of it to make + /// room for bytes that are about to exist on disk anyway. + fill_disk: bool, /// One fetch per block, however many readers want it. flight: Arc>, readahead: usize, @@ -131,6 +136,28 @@ impl VideoStream { ring: Arc, disk: Option>, readahead: usize, + ) -> Self { + Self::build(uid, blocks, ring, disk, readahead, true) + } + + /// A stream that reads through the disk cache but never adds to it. See + /// [`crate::StreamSource::open_for_copy`]. + pub(crate) fn copying( + uid: NodeUid, + blocks: SharedBlocks, + ring: Arc, + disk: Option>, + ) -> Self { + Self::build(uid, blocks, ring, disk, 0, false) + } + + fn build( + uid: NodeUid, + blocks: SharedBlocks, + ring: Arc, + disk: Option>, + readahead: usize, + fill_disk: bool, ) -> Self { let map = BlockMap::new(blocks.block_sizes()); let readahead = clamp_readahead(readahead, ring.budget(), blocks.block_sizes()); @@ -141,6 +168,7 @@ impl VideoStream { map, ring, disk, + fill_disk, flight: Arc::new(SingleFlight::new()), readahead, readahead_slots: Arc::new(Semaphore::new(readahead.max(1))), @@ -176,6 +204,38 @@ impl VideoStream { self.inner.blocks.block_sizes() } + /// Start fetching the first and last blocks, without waiting for them. + /// + /// Every demuxer reads the head of a file first, and the containers this + /// app plays keep their index at the tail — Matroska's cues, an MP4's + /// `moov` when it was not written for streaming. Opening a file is + /// therefore two round trips, one after the other. Asked for together, + /// they are one. The player's own reads join these through the + /// single-flight, so nothing is fetched twice. + /// + /// Detached rather than read-ahead: the jump to the tail looks like a + /// 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 { + 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"); + } + }); + } + } + /// Fill `buf` from `offset`, returning how many bytes were written. /// /// A read at or past EOF returns 0 rather than failing — that is how a @@ -413,10 +473,17 @@ async fn block_at(inner: Arc, index: usize) -> Result { .fetched_bytes .fetch_add(bytes.len() as u64, Ordering::Relaxed); - if let Some(disk) = inner.disk.as_ref() { - disk.put(&fetch_key, &bytes).await; - } let block: CachedBlock = bytes.into(); + // Written behind the read, not before it: the cache is disposable, + // and a block the player is waiting for should not also wait for a + // 4 MiB write and its fsync. That wait was on every cold block, + // which is to say on the first frame and on every seek. + if let Some(disk) = inner.disk.as_ref().filter(|_| inner.fill_disk) { + let disk = Arc::clone(disk); + let key = fetch_key.clone(); + let bytes = Arc::clone(&block); + tokio::spawn(async move { disk.put(&key, &bytes).await }); + } inner.ring.insert(fetch_key, Arc::clone(&block)); Ok(block) }) @@ -891,4 +958,85 @@ mod tests { assert_eq!(stream.read_at(80, &mut buf).await.unwrap(), 20); assert_eq!(&buf[..20], &expected(80, 20)[..]); } + + async fn disk_in(tag: &str) -> Arc { + let root = std::env::temp_dir().join(format!("pstr-stream-{tag}-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&root); + Arc::new( + DiskCache::open(crate::disk::DiskCacheConfig::new(root)) + .await + .unwrap(), + ) + } + + /// Waits on a detached task without sleeping for a guessed duration. + async fn eventually(mut done: impl FnMut() -> bool) -> bool { + for _ in 0..200 { + if done() { + return true; + } + tokio::time::sleep(std::time::Duration::from_millis(5)).await; + } + done() + } + + #[tokio::test] + async fn warming_fetches_the_first_and_last_blocks_and_nothing_between() { + let fake = Fake::new(&[1000; 10]); + let stream = stream_over(Arc::clone(&fake), 0); + + stream.warm_edges(); + assert!(eventually(|| fake.fetches_of(0) == 1 && fake.fetches_of(9) == 1).await); + assert_eq!(fake.fetches(), 2); + + // And the player's own reads are then served without a second fetch. + stream.read_range(0, 10).await.unwrap(); + stream.read_range(9_990, 10).await.unwrap(); + assert_eq!(fake.fetches(), 2); + } + + #[tokio::test] + async fn a_fetched_block_reaches_the_disk_cache_without_the_read_waiting_for_it() { + let disk = disk_in("fill").await; + let fake = Fake::new(&[1000, 1000]); + let stream = VideoStream::new( + uid(), + fake, + Arc::new(BlockRing::new(DEFAULT_RING_BYTES)), + Some(Arc::clone(&disk)), + 0, + ); + + assert_eq!(stream.read_range(0, 10).await.unwrap(), expected(0, 10)); + assert!(eventually(|| disk.stats().entries == 1).await); + } + + #[tokio::test] + async fn a_copying_stream_reads_the_disk_cache_but_never_adds_to_it() { + let disk = disk_in("copy").await; + let fake = Fake::new(&[1000, 1000]); + let key = BlockKey::new(&uid(), "rev1", 0); + disk.put(&key, &expected(0, 1000)).await; + + let stream = VideoStream::copying( + uid(), + Arc::clone(&fake) as SharedBlocks, + Arc::new(BlockRing::new(0)), + Some(Arc::clone(&disk)), + ); + assert_eq!(stream.read_range(0, 2000).await.unwrap(), expected(0, 2000)); + + assert_eq!( + fake.fetches_of(0), + 0, + "block 0 should have come from the disk" + ); + assert_eq!(fake.fetches_of(1), 1); + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + assert_eq!( + disk.stats().entries, + 1, + "the copy wrote into the playback cache" + ); + } }