diff --git a/crates/ipx/CHANGELOG.md b/crates/ipx/CHANGELOG.md index 1ce2a56..9f0cc51 100644 --- a/crates/ipx/CHANGELOG.md +++ b/crates/ipx/CHANGELOG.md @@ -18,6 +18,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - A page of items finds its files by feed as well as item: it read every file's row in the database, about a quarter of a second a page (#344). - A page of items asks for its count beside the items, and the feed list asks its five questions at once, where each waited for the one before (#346). - A Signal's count of items and of unread ones are worked out in one pass, not two (#348). +- Artwork kept on disk is read and written off the threads that answer requests, and two requests for the same missing artwork fetch it once (#345, #349). ### Fixed diff --git a/crates/ipx/src/main.rs b/crates/ipx/src/main.rs index 38ed81e..6c6ec22 100644 --- a/crates/ipx/src/main.rs +++ b/crates/ipx/src/main.rs @@ -1238,8 +1238,9 @@ async fn reap(ctx: &Ctx, dry_run: bool, standalone: bool) -> Result<()> { span.record("entries", r.entries_pruned); if !dry_run { tracing::Instrument::instrument(clean_directory(ctx), db::step("clean_directory")).await?; - let gone = - db::step("trim_artwork").in_scope(|| art::trim(&art::dir(), ctx.cfg().general.art_cache_mb * 1_048_576)); + // Off the async threads: it walks every file kept, on FUSE (#345). + let (span, limit) = (db::step("trim_artwork"), ctx.cfg().general.art_cache_mb * 1_048_576); + let gone = tokio::task::spawn_blocking(move || span.in_scope(|| art::trim(&art::dir(), limit))).await?; if gone > 0 { tracing::debug!(gone, "trimmed the artwork kept on disk"); } diff --git a/crates/ipx/src/web.rs b/crates/ipx/src/web.rs index 1fb4d7f..24854ed 100644 --- a/crates/ipx/src/web.rs +++ b/crates/ipx/src/web.rs @@ -2287,7 +2287,7 @@ async fn art(State(state): State, Query(q): Query) -> Respon // Kept only once it was let through below, so a kept image needs no asking again (#272). let keep = state.ctx.cfg().general.art_cache_mb > 0; let file = crate::art::path(&q.u); - if keep && let Some((kind, body)) = step("art_cached").in_scope(|| crate::art::read(&file)) { + if keep && let Some((kind, body)) = kept_art(&file).await { crate::metrics::STATS.art_hits.fetch_add(1, std::sync::atomic::Ordering::Relaxed); return art_response(kind, body); } @@ -2301,6 +2301,15 @@ async fn art(State(state): State, Query(q): Query) -> Respon if let Some(why) = crate::art::failed(&q.u) { return (StatusCode::BAD_GATEWAY, why).into_response(); } + // One fetch an address at a time: a page asks for fifty images at once, and a cover two of + // them name was fetched twice on a cold cache (#349). The second waits, then finds it kept. + let gate = ART_FETCHES.lock().unwrap().entry(q.u.clone()).or_default().clone(); + let _turn = gate.lock().await; + let _done = ArtFetchDone(&q.u); + if keep && let Some((kind, body)) = kept_art(&file).await { + crate::metrics::STATS.art_hits.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + return art_response(kind, body); + } crate::metrics::STATS.art_misses.fetch_add(1, std::sync::atomic::Ordering::Relaxed); let got = async { let mut r = state.ctx.client.get(&q.u).timeout(crate::feed::FEED_TIMEOUT).send().await?.error_for_status()?; @@ -2316,8 +2325,15 @@ async fn art(State(state): State, Query(q): Query) -> Respon }; match got.instrument(step("art_fetch")).await { Ok((kind, body)) => { - if keep && let Err(e) = step("art_keep").in_scope(|| crate::art::write(&file, &kind, &body)) { - tracing::debug!(error = %e, "could not keep the artwork"); + if keep { + let (span, file, kind, body) = (step("art_keep"), file.clone(), kind.clone(), body.clone()); + match tokio::task::spawn_blocking(move || span.in_scope(|| crate::art::write(&file, &kind, &body))) + .await + { + Ok(Ok(())) => {} + Ok(Err(e)) => tracing::debug!(error = %e, "could not keep the artwork"), + Err(e) => tracing::debug!(error = %e, "could not keep the artwork"), + } } art_response(kind, body) } @@ -2331,6 +2347,31 @@ async fn art(State(state): State, Query(q): Query) -> Respon } } +/// The fetches of artwork under way, by address, each a lock the requests for it take in turn. +static ART_FETCHES: std::sync::LazyLock< + std::sync::Mutex>>>, +> = std::sync::LazyLock::new(Default::default); + +/// Forgets an address's lock once nobody else holds it, however the request ended. +struct ArtFetchDone<'a>(&'a str); + +impl Drop for ArtFetchDone<'_> { + fn drop(&mut self) { + let mut fetches = ART_FETCHES.lock().unwrap(); + // Two: the map's and this request's own. + if fetches.get(self.0).is_some_and(|g| Arc::strong_count(g) <= 2) { + fetches.remove(self.0); + } + } +} + +/// Artwork kept on disk, read off the async threads: the cache is on FUSE, and four calls a +/// hit, fifty images a page, held a worker each, 1 s at p90 for a hit (#345). +async fn kept_art(file: &std::path::Path) -> Option<(String, Vec)> { + let (span, file) = (step("art_cached"), file.to_owned()); + tokio::task::spawn_blocking(move || span.in_scope(|| crate::art::read(&file))).await.ok().flatten() +} + /// A type that says nothing about what the body is, which some publishers send with real artwork: /// Spotify's CloudFront `application/octet-stream`, pippa.io none at all (#279). fn untyped(said: &str) -> bool {