From 4791f55ad81fc589ee62fe374b8852e173cf7c3e Mon Sep 17 00:00:00 2001 From: zarzet <42882290+zarzet@users.noreply.github.com> Date: Mon, 19 Oct 2026 01:22:17 +0700 Subject: [PATCH] refactor(providers): share one Coalescer for single-flight request coalescing The lyrics, MusicBrainz, platform resolver and availability services each carried their own copy of the same single-flight machinery: a Flight with an atomic waiter count, a result slot and condvar, a Waiter drop guard, a worker check that fails once the last waiter leaves, a 25 ms wait_timeout poll loop, a MAX_FLIGHTS (64) bound and a finish step that publishes the result, removes the current flight, decrements the active count and wakes shutdown. Move that into crate::coalescer (Flight, Waiter, Coalescer). The table stays inside each service's own state mutex, so cache lookups, joins and finish remain atomic with that service's cache and generation. Per-site rules stay at the call sites: cache keys, what finish caches and when, error values, thread names, worker post-checks, panic messages, MusicBrainz's atomic try_join, its spawn under the state lock and its uncached spawn-failure path. Existing tests only change their access path to the waiter and active counts. --- .../crates/providers/src/coalescer.rs | 455 ++++++++++++++++++ rust_backend/crates/providers/src/lib.rs | 1 + .../crates/providers/src/lyrics/service.rs | 91 +--- .../providers/src/lyrics/service/tests.rs | 6 +- .../crates/providers/src/musicbrainz.rs | 122 ++--- .../crates/providers/src/musicbrainz/tests.rs | 15 +- .../src/resolver/availability/cache.rs | 92 +--- .../crates/providers/src/resolver/service.rs | 87 +--- 8 files changed, 575 insertions(+), 294 deletions(-) create mode 100644 rust_backend/crates/providers/src/coalescer.rs diff --git a/rust_backend/crates/providers/src/coalescer.rs b/rust_backend/crates/providers/src/coalescer.rs new file mode 100644 index 00000000..62969302 --- /dev/null +++ b/rust_backend/crates/providers/src/coalescer.rs @@ -0,0 +1,455 @@ +//! Single-flight request coalescing shared by the lyrics, MusicBrainz, resolver and +//! availability services. +//! +//! The flight table lives inside each service's own state mutex, so cache lookups, +//! flight joins and [`Coalescer::finish`] stay atomic with that service's cache and +//! generation. Services keep their own key, caching and worker rules; this module +//! owns waiter accounting, cancellation once the last waiter leaves, the poll loop, +//! the `MAX_FLIGHTS` bound and idle notification. + +use std::borrow::Borrow; +use std::collections::BTreeMap; +use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; +use std::sync::{Arc, Condvar, Mutex}; +use std::time::Duration; + +/// Caller checks are opaque callbacks, so waiters poll them while a flight runs. +const POLL: Duration = Duration::from_millis(25); +/// Upper bound on concurrently running workers per service. +pub(crate) const MAX_FLIGHTS: usize = 64; + +/// One shared in-flight request. `T` carries per-service data fixed at creation. +pub(crate) struct Flight { + tag: T, + waiters: AtomicUsize, + result: Mutex>, + ready: Condvar, +} + +impl Flight { + pub(crate) fn new(tag: T) -> Arc { + Arc::new(Self { + tag, + waiters: AtomicUsize::new(0), + result: Mutex::new(None), + ready: Condvar::new(), + }) + } + + pub(crate) fn tag(&self) -> &T { + &self.tag + } + + pub(crate) fn waiters(&self) -> usize { + self.waiters.load(Ordering::Acquire) + } + + /// Registers one more waiter under the service lock. Create the matching + /// [`Waiter`] guard once that lock is released. + pub(crate) fn join(&self) { + self.waiters.fetch_add(1, Ordering::AcqRel); + } + + /// Worker-side liveness: fails once the owner closes or every waiter has left. + pub(crate) fn check( + &self, + closed: &AtomicBool, + closed_message: &str, + cancelled_message: &str, + ) -> Result<(), String> { + if closed.load(Ordering::Acquire) { + Err(closed_message.into()) + } else if self.waiters() == 0 { + Err(cancelled_message.into()) + } else { + Ok(()) + } + } + + /// Blocks until the flight finishes, running the caller's `check` before every + /// poll and once more after the shared result is cloned. + pub(crate) fn wait(&self, check: impl Fn() -> Result<(), E>) -> Result + where + V: Clone, + { + loop { + check()?; + let result = self.result.lock().expect("flight lock"); + if let Some(value) = result.as_ref() { + let value = value.clone(); + drop(result); + check()?; + return Ok(value); + } + drop(self.ready.wait_timeout(result, POLL).expect("flight lock")); + } + } +} + +/// Drop guard for one registered waiter. Dropping the last guard makes +/// [`Flight::check`] fail, which cancels the detached worker. +pub(crate) struct Waiter(Arc>); + +impl Waiter { + pub(crate) fn new(flight: Arc>) -> Self { + Self(flight) + } +} + +impl Drop for Waiter { + fn drop(&mut self) { + self.0.waiters.fetch_sub(1, Ordering::AcqRel); + } +} + +/// Flight table and running-worker count, kept inside a service's state mutex. +pub(crate) struct Coalescer { + flights: BTreeMap>>, + active: usize, +} + +impl Default for Coalescer { + fn default() -> Self { + Self { + flights: BTreeMap::new(), + active: 0, + } + } +} + +impl Coalescer { + /// Workers started and not yet finished; shutdown waits on `idle` until zero. + pub(crate) fn active(&self) -> usize { + self.active + } + + pub(crate) fn is_full(&self) -> bool { + self.active >= MAX_FLIGHTS + } + + /// A flight for `key` that still has waiters. The caller registers itself + /// with [`Flight::join`] while still holding the service lock. + pub(crate) fn existing(&self, key: &Q) -> Option>> + where + K: Borrow, + Q: Ord + ?Sized, + { + self.flights + .get(key) + .filter(|flight| flight.waiters() > 0) + .cloned() + } + + /// Joins a flight for `key` only while its waiter count is still nonzero, + /// so an abandoned flight is never revived between the check and the join. + pub(crate) fn try_join(&self, key: &Q) -> Option>> + where + K: Borrow, + Q: Ord + ?Sized, + { + self.flights + .get(key) + .filter(|flight| { + flight + .waiters + .try_update(Ordering::AcqRel, Ordering::Acquire, |count| { + (count > 0).then_some(count + 1) + }) + .is_ok() + }) + .cloned() + } + + /// Records a new flight and counts its worker as active. Check + /// [`Coalescer::is_full`] first. + pub(crate) fn insert(&mut self, key: K, flight: Arc>) { + self.flights.insert(key, flight); + self.active += 1; + } + + /// Whether `flight` is still the table entry for `key`. + pub(crate) fn is_current(&self, key: &Q, flight: &Arc>) -> bool + where + K: Borrow, + Q: Ord + ?Sized, + { + self.flights + .get(key) + .is_some_and(|current| Arc::ptr_eq(current, flight)) + } + + /// Publishes `result` to the flight's waiters, removes the flight if it is + /// still current, and wakes shutdown. Call under the service lock after any + /// service-specific caching. + pub(crate) fn finish( + &mut self, + key: &Q, + flight: &Arc>, + result: V, + idle: &Condvar, + ) where + K: Borrow, + Q: Ord + ?Sized, + { + *flight.result.lock().expect("flight lock") = Some(result); + flight.ready.notify_all(); + if self.is_current(key, flight) { + self.flights.remove(key); + } + self.active -= 1; + idle.notify_all(); + } + + #[cfg(test)] + pub(crate) fn get(&self, key: &Q) -> Option<&Arc>> + where + K: Borrow, + Q: Ord + ?Sized, + { + self.flights.get(key) + } + + #[cfg(test)] + pub(crate) fn values(&self) -> impl Iterator>> { + self.flights.values() + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::time::Instant; + + type Outcome = Result; + + #[derive(Default)] + struct State { + flights: Coalescer, + cache: BTreeMap, + } + + /// Minimal service following the same flow as the four production owners. + #[derive(Default)] + struct Service { + state: Mutex, + idle: Condvar, + closed: AtomicBool, + calls: AtomicUsize, + release: AtomicBool, + worker_errors: Mutex>, + } + + impl Service { + fn lookup( + self: &Arc, + key: &str, + check: &(dyn Fn() -> Result<(), String> + Sync), + ) -> Outcome { + let (flight, start) = { + let mut state = self.state.lock().unwrap(); + if let Some(value) = state.cache.get(key) { + return Ok(*value); + } + let (flight, start) = if let Some(flight) = state.flights.existing(key) { + (flight, false) + } else { + if state.flights.is_full() { + return Err("busy".into()); + } + let flight = Flight::new(()); + state.flights.insert(key.into(), Arc::clone(&flight)); + (flight, true) + }; + flight.join(); + (flight, start) + }; + let _waiter = Waiter::new(Arc::clone(&flight)); + if start { + let service = Arc::clone(self); + let worker_flight = Arc::clone(&flight); + let worker_key = key.to_owned(); + std::thread::spawn(move || { + let check = || worker_flight.check(&service.closed, "closed", "cancelled"); + service.calls.fetch_add(1, Ordering::AcqRel); + let result = loop { + if let Err(error) = check() { + service.worker_errors.lock().unwrap().push(error.clone()); + break Err(error); + } + if service.release.load(Ordering::Acquire) { + break Ok(7); + } + std::thread::sleep(Duration::from_millis(2)); + }; + let mut state = service.state.lock().unwrap(); + if let Ok(value) = result { + state.cache.insert(worker_key.clone(), value); + } + state.flights.finish( + worker_key.as_str(), + &worker_flight, + result, + &service.idle, + ); + }); + } + flight.wait(check)? + } + } + + fn until(condition: impl Fn() -> bool) { + let deadline = Instant::now() + Duration::from_secs(3); + while !condition() { + assert!(Instant::now() < deadline, "coalescer test timed out"); + std::thread::sleep(Duration::from_millis(2)); + } + } + + fn waiters(service: &Service) -> usize { + let state = service.state.lock().unwrap(); + state.flights.values().map(|flight| flight.waiters()).sum() + } + + #[test] + fn one_flight_serves_every_waiter_and_finish_caches() { + let service = Arc::new(Service::default()); + let threads: Vec<_> = (0..8) + .map(|_| { + let service = Arc::clone(&service); + std::thread::spawn(move || service.lookup("key", &|| Ok(()))) + }) + .collect(); + until(|| waiters(&service) == 8 && service.calls.load(Ordering::Acquire) == 1); + std::thread::sleep(Duration::from_millis(10)); + assert_eq!(service.calls.load(Ordering::Acquire), 1); + service.release.store(true, Ordering::Release); + for thread in threads { + assert_eq!(thread.join().unwrap(), Ok(7)); + } + let state = service.state.lock().unwrap(); + assert_eq!(state.flights.active(), 0); + assert!(state.flights.get("key").is_none()); + assert_eq!(state.cache.get("key"), Some(&7)); + drop(state); + assert_eq!(service.lookup("key", &|| Ok(())), Ok(7)); + assert_eq!(service.calls.load(Ordering::Acquire), 1); + } + + #[test] + fn last_waiter_leaving_cancels_the_worker() { + let service = Arc::new(Service::default()); + let cancelled = Arc::new(AtomicBool::new(false)); + let caller = { + let service = Arc::clone(&service); + let cancelled = Arc::clone(&cancelled); + std::thread::spawn(move || { + service.lookup("key", &|| { + if cancelled.load(Ordering::Acquire) { + Err("caller cancelled".into()) + } else { + Ok(()) + } + }) + }) + }; + until(|| waiters(&service) == 1); + cancelled.store(true, Ordering::Release); + assert_eq!(caller.join().unwrap(), Err("caller cancelled".into())); + until(|| service.state.lock().unwrap().flights.active() == 0); + assert_eq!(*service.worker_errors.lock().unwrap(), ["cancelled"]); + let state = service.state.lock().unwrap(); + assert!(state.cache.is_empty()); + assert!(state.flights.get("key").is_none()); + } + + #[test] + fn closed_owner_is_reported_before_cancellation() { + let flight: Arc> = Flight::new(()); + let closed = AtomicBool::new(true); + assert_eq!( + flight.check(&closed, "closed", "cancelled"), + Err("closed".into()) + ); + closed.store(false, Ordering::Release); + assert_eq!( + flight.check(&closed, "closed", "cancelled"), + Err("cancelled".into()) + ); + flight.join(); + assert_eq!(flight.check(&closed, "closed", "cancelled"), Ok(())); + } + + #[test] + fn running_flights_are_bounded() { + let service = Arc::new(Service::default()); + let threads: Vec<_> = (0..MAX_FLIGHTS) + .map(|index| { + let service = Arc::clone(&service); + std::thread::spawn(move || service.lookup(&index.to_string(), &|| Ok(()))) + }) + .collect(); + until(|| service.state.lock().unwrap().flights.active() == MAX_FLIGHTS); + assert!(service.state.lock().unwrap().flights.is_full()); + assert_eq!(service.lookup("over", &|| Ok(())), Err("busy".into())); + service.release.store(true, Ordering::Release); + for thread in threads { + assert_eq!(thread.join().unwrap(), Ok(7)); + } + assert!(!service.state.lock().unwrap().flights.is_full()); + assert_eq!(service.lookup("over", &|| Ok(())), Ok(7)); + } + + #[test] + fn abandoned_flights_are_not_joined_and_stale_finish_keeps_replacement() { + let idle = Condvar::new(); + let mut flights: Coalescer = Coalescer::default(); + let stale = Flight::new(()); + flights.insert("key".into(), Arc::clone(&stale)); + assert!(flights.existing("key").is_none()); + assert!(flights.try_join("key").is_none()); + assert_eq!(stale.waiters(), 0); + let current = Flight::new(()); + current.join(); + flights.insert("key".into(), Arc::clone(¤t)); + assert!(flights.try_join("key").is_some()); + assert_eq!(current.waiters(), 2); + assert!(!flights.is_current("key", &stale)); + flights.finish("key", &stale, Ok(1), &idle); + assert_eq!(stale.wait(|| Ok::<_, String>(())), Ok(Ok(1))); + assert!(flights.is_current("key", ¤t)); + assert_eq!(flights.active(), 1); + flights.finish("key", ¤t, Ok(2), &idle); + assert!(flights.get("key").is_none()); + assert_eq!(flights.active(), 0); + } + + #[test] + fn finish_wakes_shutdown_waiting_for_idle() { + let service = Arc::new(Service::default()); + let flight = Flight::new(()); + service + .state + .lock() + .unwrap() + .flights + .insert("key".into(), Arc::clone(&flight)); + let shutdown = { + let service = Arc::clone(&service); + std::thread::spawn(move || { + let mut state = service.state.lock().unwrap(); + while state.flights.active() != 0 { + state = service.idle.wait(state).unwrap(); + } + }) + }; + std::thread::sleep(Duration::from_millis(20)); + assert!(!shutdown.is_finished()); + service + .state + .lock() + .unwrap() + .flights + .finish("key", &flight, Ok(1), &service.idle); + shutdown.join().unwrap(); + } +} diff --git a/rust_backend/crates/providers/src/lib.rs b/rust_backend/crates/providers/src/lib.rs index 0a7acbb8..ee267b6f 100644 --- a/rust_backend/crates/providers/src/lib.rs +++ b/rust_backend/crates/providers/src/lib.rs @@ -4,6 +4,7 @@ //! behavior (wire formats, parsing quirks, file layouts) remains a compatibility //! contract. +mod coalescer; pub mod deezer; pub mod lyrics; pub mod musicbrainz; diff --git a/rust_backend/crates/providers/src/lyrics/service.rs b/rust_backend/crates/providers/src/lyrics/service.rs index 49c8f7d2..d7b4eff2 100644 --- a/rust_backend/crates/providers/src/lyrics/service.rs +++ b/rust_backend/crates/providers/src/lyrics/service.rs @@ -2,6 +2,7 @@ //! in-flight requests, and caches negative results. use super::{CallGraph, CallNode, Check, LyricsError, cache::LyricsCache}; +use crate::coalescer::{Coalescer, Flight, Waiter}; use spotiflac_core::lyrics::{LyricsResponse, config, config::FetchOptions, matching}; use spotiflac_core::matching::lowercase; use std::collections::BTreeMap; @@ -16,7 +17,6 @@ const NEGATIVE_TTL: Duration = Duration::from_secs(5 * 60); const UNAVAILABLE_COOLDOWN: Duration = Duration::from_secs(10 * 60); const PRIORITY_GRACE: Duration = Duration::from_secs(5); const MAX_NEGATIVE: usize = 500; -const MAX_FLIGHTS: usize = 64; #[derive(Clone, Debug, Default)] pub struct SearchRequest { @@ -56,20 +56,8 @@ pub trait LyricsFetcher: Send + Sync + 'static { } type FetchResult = Result; - -struct Flight { - node: CallNode, - waiters: AtomicUsize, - result: Mutex>, - ready: Condvar, -} - -struct Waiter(Arc); -impl Drop for Waiter { - fn drop(&mut self) { - self.0.waiters.fetch_sub(1, Ordering::AcqRel); - } -} +/// Each flight carries the call-graph node its worker runs under. +type LyricsFlight = Flight; #[derive(Default)] struct State { @@ -79,8 +67,7 @@ struct State { generation: u64, negative: BTreeMap, health: BTreeMap, - flights: BTreeMap<(u64, String), Arc>, - active: usize, + flights: Coalescer<(u64, String), FetchResult, CallNode>, } struct Inner { @@ -209,44 +196,33 @@ impl LyricsService { } state.negative.remove(&key); let flight_key = (state.generation, key); - let existing = state - .flights - .get(&flight_key) - .filter(|flight| flight.waiters.load(Ordering::Acquire) > 0) - .cloned(); + let existing = state.flights.existing(&flight_key); let (flight, start) = if let Some(flight) = existing { (flight, false) } else { - if state.active >= MAX_FLIGHTS { + if state.flights.is_full() { return Err(LyricsError::Other( "too many concurrent lyrics searches".into(), )); } - let flight = Arc::new(Flight { - node: self.inner.calls.node(), - waiters: AtomicUsize::new(0), - result: Mutex::new(None), - ready: Condvar::new(), - }); - (flight, true) + (Flight::new(self.inner.calls.node()), true) }; let dependency = request .caller .as_ref() - .map(|caller| caller.wait_for(&flight.node)) + .map(|caller| caller.wait_for(flight.tag())) .transpose() .map_err(LyricsError::Recursive)?; if start { state .flights .insert(flight_key.clone(), Arc::clone(&flight)); - state.active += 1; } - request.caller = Some(flight.node.clone()); - flight.waiters.fetch_add(1, Ordering::AcqRel); + request.caller = Some(flight.tag().clone()); + flight.join(); (flight, start, flight_key, providers, dependency) }; - let _waiter = Waiter(Arc::clone(&flight)); + let _waiter = Waiter::new(Arc::clone(&flight)); if start { let inner = Arc::clone(&self.inner); let worker_flight = Arc::clone(&flight); @@ -255,13 +231,11 @@ impl LyricsService { .name("lyrics-search".into()) .spawn(move || { let check = || { - if inner.closed.load(Ordering::Acquire) { - Err("lyrics service is closed".into()) - } else if worker_flight.waiters.load(Ordering::Acquire) == 0 { - Err("lyrics request cancelled".into()) - } else { - Ok(()) - } + worker_flight.check( + &inner.closed, + "lyrics service is closed", + "lyrics request cancelled", + ) }; let result = catch_unwind(AssertUnwindSafe(|| { inner.fetch_uncached( @@ -283,22 +257,7 @@ impl LyricsService { ); } } - loop { - self.inner.check(check)?; - let result = flight.result.lock().expect("lyrics flight lock"); - if let Some(value) = result.as_ref() { - let value = value.clone(); - drop(result); - self.inner.check(check)?; - return value; - } - drop( - flight - .ready - .wait_timeout(result, POLL) - .expect("lyrics flight lock"), - ); - } + flight.wait(|| self.inner.check(check))? } /// Cancel workers, wait for their HTTP/extension calls to unwind, then flush @@ -306,7 +265,7 @@ impl LyricsService { pub fn shutdown(&self) -> Result<(), String> { self.inner.closed.store(true, Ordering::Release); let mut state = self.inner.state.lock().expect("lyrics service lock"); - while state.active > 0 { + while state.flights.active() > 0 { state = self.inner.idle.wait(state).expect("lyrics service lock"); } drop(state); @@ -329,7 +288,7 @@ impl Inner { self.fetcher.check() } - fn finish(&self, key: &(u64, String), flight: &Arc, result: FetchResult) { + fn finish(&self, key: &(u64, String), flight: &Arc, result: FetchResult) { let mut state = self.state.lock().expect("lyrics service lock"); if state.generation == key.0 && !self.closed.load(Ordering::Acquire) { if result.as_ref().is_err_and(|error| { @@ -345,17 +304,7 @@ impl Inner { state.negative.remove(&key.1); } } - *flight.result.lock().expect("lyrics flight lock") = Some(result); - flight.ready.notify_all(); - if state - .flights - .get(key) - .is_some_and(|current| Arc::ptr_eq(current, flight)) - { - state.flights.remove(key); - } - state.active -= 1; - self.idle.notify_all(); + state.flights.finish(key, flight, result, &self.idle); } fn fetch_uncached( diff --git a/rust_backend/crates/providers/src/lyrics/service/tests.rs b/rust_backend/crates/providers/src/lyrics/service/tests.rs index 4ba5df36..298af339 100644 --- a/rust_backend/crates/providers/src/lyrics/service/tests.rs +++ b/rust_backend/crates/providers/src/lyrics/service/tests.rs @@ -52,7 +52,7 @@ fn waiters(service: &LyricsService) -> usize { .unwrap() .flights .values() - .map(|flight| flight.waiters.load(Ordering::Acquire)) + .map(|flight| flight.waiters()) .sum() } fn wait_gate(gate: &AtomicBool, check: &Check<'_>) -> Result<(), LyricsError> { @@ -143,7 +143,7 @@ fn abandoning_all_waiters_cancels_work_without_negative_caching() { worker.join().unwrap(), Err(LyricsError::Cancelled(_)) )); - until(|| service.inner.state.lock().unwrap().active == 0); + until(|| service.inner.state.lock().unwrap().flights.active() == 0); assert_eq!( service.fetch(request(), &|| Ok(())).unwrap().source, "LRCLIB" @@ -389,7 +389,7 @@ fn instrumental_heuristic_skips_network_and_provider_panics_release_workers() { assert_eq!(response.source, "Heuristic: Instrumental"); assert!(service.fetch(request(), &|| Ok(())).is_err()); service.shutdown().unwrap(); - assert_eq!(service.inner.state.lock().unwrap().active, 0); + assert_eq!(service.inner.state.lock().unwrap().flights.active(), 0); } #[test] diff --git a/rust_backend/crates/providers/src/musicbrainz.rs b/rust_backend/crates/providers/src/musicbrainz.rs index 289b5546..203e5ac4 100644 --- a/rust_backend/crates/providers/src/musicbrainz.rs +++ b/rust_backend/crates/providers/src/musicbrainz.rs @@ -4,20 +4,20 @@ mod http; #[cfg(test)] mod tests; +use crate::coalescer::{Coalescer, Flight, Waiter}; use crate::resolver::{Check, ResolverError}; use spotiflac_core::matching::uppercase; use spotiflac_core::metadata::musicbrainz::Response; use spotiflac_network::NetworkService; use std::collections::BTreeMap; use std::panic::{AssertUnwindSafe, catch_unwind}; -use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; +use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Condvar, Mutex}; use std::thread::JoinHandle; use std::time::{Duration, Instant}; const POLL: Duration = Duration::from_millis(25); const MAX_CACHE: usize = 256; -const MAX_FLIGHTS: usize = 64; type Outcome = Result, ResolverError>; /// Trusted native/test configuration. These options are not exposed to JS/FFI. @@ -84,24 +84,9 @@ impl Cache { } } -#[derive(Default)] -struct Flight { - waiters: AtomicUsize, - result: Mutex>, - ready: Condvar, -} - -struct Waiter(Arc); -impl Drop for Waiter { - fn drop(&mut self) { - self.0.waiters.fetch_sub(1, Ordering::AcqRel); - } -} - #[derive(Default)] struct State { - active: usize, - flights: BTreeMap>, + flights: Coalescer, workers: Vec>, cache: Cache, } @@ -192,30 +177,17 @@ impl MusicBrainzClient { self.inner.check(check)?; return value.map(|value| (key, value)); } - if let Some(flight) = state - .flights - .get(&key) - .filter(|flight| { - flight - .waiters - .try_update(Ordering::AcqRel, Ordering::Acquire, |count| { - (count > 0).then_some(count + 1) - }) - .is_ok() - }) - .cloned() - { + if let Some(flight) = state.flights.try_join(&key) { flight } else { - if state.active >= MAX_FLIGHTS { + if state.flights.is_full() { return Err(ResolverError::Failed( "MusicBrainz lookup limit reached".into(), )); } - let flight = Arc::new(Flight::default()); - flight.waiters.store(1, Ordering::Release); + let flight = Flight::new(()); + flight.join(); state.flights.insert(key.clone(), flight.clone()); - state.active += 1; let inner = self.inner.clone(); let worker_key = key.clone(); let worker_flight = flight.clone(); @@ -224,13 +196,11 @@ impl MusicBrainzClient { .name("musicbrainz".into()) .spawn(move || { let check = || { - if inner.closed.load(Ordering::Acquire) { - Err("MusicBrainz client closed".into()) - } else if worker_flight.waiters.load(Ordering::Acquire) == 0 { - Err("MusicBrainz request cancelled".into()) - } else { - Ok(()) - } + worker_flight.check( + &inner.closed, + "MusicBrainz client closed", + "MusicBrainz request cancelled", + ) }; let result = catch_unwind(AssertUnwindSafe(|| { if worker_key.is_empty() { @@ -249,36 +219,28 @@ impl MusicBrainzClient { ); }) { Ok(worker) => state.workers.push(worker), - Err(error) => { - state.flights.remove(&key); - state.active -= 1; - *flight.result.lock().unwrap() = - Some(Err(ResolverError::Failed(error.to_string()))); - self.inner.idle.notify_all(); - } + // Spawn failures are published without caching. + Err(error) => state.flights.finish( + &key, + &flight, + Err(ResolverError::Failed(error.to_string())), + &self.inner.idle, + ), } flight } }; - let _waiter = Waiter(flight.clone()); - loop { - self.inner.check(check)?; - let result_guard = flight.result.lock().unwrap(); - if let Some(result) = result_guard.as_ref() { - let result = result.clone(); - drop(result_guard); - self.inner.check(check)?; - return result.map(|value| (key, value)); - } - drop(flight.ready.wait_timeout(result_guard, POLL).unwrap()); - } + let _waiter = Waiter::new(flight.clone()); + flight + .wait(|| self.inner.check(check))? + .map(|value| (key, value)) } pub fn shutdown(&self) { let _shutdown = self.inner.shutdown.lock().unwrap(); self.inner.closed.store(true, Ordering::Release); let mut state = self.inner.state.lock().unwrap(); - while state.active != 0 { + while state.flights.active() != 0 { state = self.inner.idle.wait(state).unwrap(); } state.cache.0.clear(); @@ -305,29 +267,21 @@ impl Inner { } } - fn finish(&self, key: &str, flight: &Arc, result: Outcome) { + fn finish(&self, key: &str, flight: &Arc>, result: Outcome) { let mut state = self.state.lock().unwrap(); - let current = state - .flights - .get(key) - .is_some_and(|current| Arc::ptr_eq(current, flight)); - if current { - state.flights.remove(key); - if !self.closed.load(Ordering::Acquire) - && flight.waiters.load(Ordering::Acquire) > 0 - && !matches!( - result, - Err(ResolverError::Cancelled(_) | ResolverError::Closed | ResolverError::Busy) - ) - { - state - .cache - .put(key, result.clone(), Instant::now(), &self.options); - } + // Only the current flight caches, and only while a caller still wants it. + if state.flights.is_current(key, flight) + && !self.closed.load(Ordering::Acquire) + && flight.waiters() > 0 + && !matches!( + result, + Err(ResolverError::Cancelled(_) | ResolverError::Closed | ResolverError::Busy) + ) + { + state + .cache + .put(key, result.clone(), Instant::now(), &self.options); } - *flight.result.lock().unwrap() = Some(result); - flight.ready.notify_all(); - state.active -= 1; - self.idle.notify_all(); + state.flights.finish(key, flight, result, &self.idle); } } diff --git a/rust_backend/crates/providers/src/musicbrainz/tests.rs b/rust_backend/crates/providers/src/musicbrainz/tests.rs index e0c84f9e..c9c3dea7 100644 --- a/rust_backend/crates/providers/src/musicbrainz/tests.rs +++ b/rust_backend/crates/providers/src/musicbrainz/tests.rs @@ -1,5 +1,6 @@ use super::*; use spotiflac_core::metadata::musicbrainz::{ArtistCredit, Recording, Release, Tag}; +use std::sync::atomic::AtomicUsize; use std::thread; fn response() -> Arc { @@ -174,9 +175,15 @@ fn genre_and_album_artist_share_a_flight_without_cancelling_other_callers() { }) .collect(); until(|| { - client.inner.state.lock().unwrap().flights["USAA00000101"] - .waiters - .load(Ordering::Acquire) + client + .inner + .state + .lock() + .unwrap() + .flights + .get("USAA00000101") + .unwrap() + .waiters() == 9 }); cancel.store(true, Ordering::Release); @@ -236,7 +243,7 @@ fn abandoned_flight_cannot_publish_over_its_replacement() { probe.release.store(true, Ordering::Release); assert_eq!(second.join().unwrap().unwrap(), "Rock"); probe.hold_cancelled.store(false, Ordering::Release); - until(|| client.inner.state.lock().unwrap().active == 0); + until(|| client.inner.state.lock().unwrap().flights.active() == 0); assert_eq!(client.genre("same", &|| Ok(())).unwrap(), "Rock"); assert_eq!(probe.calls.load(Ordering::Acquire), 2); } diff --git a/rust_backend/crates/providers/src/resolver/availability/cache.rs b/rust_backend/crates/providers/src/resolver/availability/cache.rs index 281f176c..9631ca18 100644 --- a/rust_backend/crates/providers/src/resolver/availability/cache.rs +++ b/rust_backend/crates/providers/src/resolver/availability/cache.rs @@ -4,12 +4,13 @@ use super::{ AlbumAvailability, TrackAvailability, album, deezer_id_from_metadata, from_deezer, from_links, resolved_links, }; +use crate::coalescer::{Coalescer, Flight, Waiter}; use crate::deezer::MetadataLookup; use crate::lyrics::{LyricsError, builtin::TrackResolver}; use crate::resolver::{Check, PlatformResolverService, Resolver, ResolverError}; use std::collections::BTreeMap; use std::panic::{AssertUnwindSafe, catch_unwind}; -use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; +use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Condvar, Mutex}; use std::time::{Duration, Instant}; @@ -38,25 +39,12 @@ struct Entry { result: Outcome, expires: Instant, } -#[derive(Default)] -struct Flight { - waiters: AtomicUsize, - result: Mutex>, - ready: Condvar, -} -struct Waiter(Arc); -impl Drop for Waiter { - fn drop(&mut self) { - self.0.waiters.fetch_sub(1, Ordering::AcqRel); - } -} struct State { generation: u64, region: String, cache: BTreeMap, - flights: BTreeMap<(u64, Key), Arc>, - active: usize, + flights: Coalescer<(u64, Key), Outcome>, } struct Inner { @@ -82,8 +70,7 @@ impl AvailabilityService { generation: 0, region: "US".into(), cache: BTreeMap::new(), - flights: BTreeMap::new(), - active: 0, + flights: Coalescer::default(), }), closed: AtomicBool::new(false), idle: Condvar::new(), @@ -225,26 +212,21 @@ impl AvailabilityService { state.cache.remove(&key); let generation = state.generation; let flight_key = (generation, key.clone()); - let existing = state - .flights - .get(&flight_key) - .filter(|flight| flight.waiters.load(Ordering::Acquire) > 0) - .cloned(); + let existing = state.flights.existing(&flight_key); let (flight, start) = if let Some(flight) = existing { (flight, false) } else { - if state.active >= 64 { + if state.flights.is_full() { return Err(ResolverError::Busy); } - let flight = Arc::new(Flight::default()); + let flight = Flight::new(()); state.flights.insert(flight_key, flight.clone()); - state.active += 1; (flight, true) }; - flight.waiters.fetch_add(1, Ordering::AcqRel); + flight.join(); (key, generation, flight, start) }; - let _waiter = Waiter(flight.clone()); + let _waiter = Waiter::new(flight.clone()); if start { let inner = self.inner.clone(); let worker_key = key.clone(); @@ -253,13 +235,11 @@ impl AvailabilityService { .name("platform-availability".into()) .spawn(move || { let check = || { - if inner.closed.load(Ordering::Acquire) { - Err("availability service closed".into()) - } else if worker_flight.waiters.load(Ordering::Acquire) == 0 { - Err("availability request cancelled".into()) - } else { - Ok(()) - } + worker_flight.check( + &inner.closed, + "availability service closed", + "availability request cancelled", + ) }; let result = catch_unwind(AssertUnwindSafe(|| inner.fetch(&worker_key, &check))) @@ -278,28 +258,13 @@ impl AvailabilityService { ); } } - loop { - self.inner.check(check)?; - let result = flight.result.lock().unwrap(); - if let Some(value) = result.as_ref() { - let value = value.clone(); - drop(result); - self.inner.check(check)?; - return value; - } - drop( - flight - .ready - .wait_timeout(result, Duration::from_millis(25)) - .unwrap(), - ); - } + flight.wait(|| self.inner.check(check))? } pub fn shutdown(&self) { self.inner.closed.store(true, Ordering::Release); let mut state = self.inner.state.lock().unwrap(); - while state.active != 0 { + while state.flights.active() != 0 { state = self.inner.idle.wait(state).unwrap(); } state.cache.clear(); @@ -391,7 +356,7 @@ impl Inner { } } - fn finish(&self, generation: u64, key: &Key, flight: &Arc, result: Outcome) { + fn finish(&self, generation: u64, key: &Key, flight: &Arc>, result: Outcome) { let mut state = self.state.lock().unwrap(); if generation == state.generation && !self.closed.load(Ordering::Acquire) @@ -430,18 +395,9 @@ impl Inner { }, ); } - *flight.result.lock().unwrap() = Some(result); - flight.ready.notify_all(); - let flight_key = (generation, key.clone()); - if state + state .flights - .get(&flight_key) - .is_some_and(|value| Arc::ptr_eq(value, flight)) - { - state.flights.remove(&flight_key); - } - state.active -= 1; - self.idle.notify_all(); + .finish(&(generation, key.clone()), flight, result, &self.idle); } } @@ -450,6 +406,7 @@ mod tests { use super::*; use crate::resolver::{Metadata, Resolution}; use spotiflac_core::metadata::TrackMetadata; + use std::sync::atomic::AtomicUsize; #[derive(Default)] struct Probe { @@ -547,8 +504,7 @@ mod tests { .values() .next() .unwrap() - .waiters - .load(Ordering::Acquire) + .waiters() == 2 }); cancelled.store(true, Ordering::Release); @@ -627,7 +583,7 @@ mod tests { assert_eq!(service.set_region("ID"), Err(ResolverError::Closed)); assert_eq!(service.region(), Err(ResolverError::Closed)); assert_eq!(service.clear_cache(), Err(ResolverError::Closed)); - assert_eq!(service.inner.state.lock().unwrap().active, 0); + assert_eq!(service.inner.state.lock().unwrap().flights.active(), 0); } #[test] @@ -694,7 +650,7 @@ mod tests { ), Err(ResolverError::Cancelled(_)) )); - until(|| service.inner.state.lock().unwrap().active == 0); + until(|| service.inner.state.lock().unwrap().flights.active() == 0); assert!( !service .inner @@ -718,7 +674,7 @@ mod tests { service.check_track(&index.to_string(), "", &|| Ok(())) })); } - until(|| service.inner.state.lock().unwrap().active == 64); + until(|| service.inner.state.lock().unwrap().flights.active() == 64); assert_eq!( service.check_track("over-limit", "", &|| Ok(())), Err(ResolverError::Busy) diff --git a/rust_backend/crates/providers/src/resolver/service.rs b/rust_backend/crates/providers/src/resolver/service.rs index f5335934..1a9397d6 100644 --- a/rust_backend/crates/providers/src/resolver/service.rs +++ b/rust_backend/crates/providers/src/resolver/service.rs @@ -1,18 +1,17 @@ //! Platform resolver service that resolves URLs and Spotify tracks to other platforms, with caching. use super::{Check, Metadata, Resolution, Resolver, ResolverError, check_active}; +use crate::coalescer::{Coalescer, Flight, Waiter}; use crate::lyrics::{LyricsError, builtin::TrackResolver}; use std::collections::BTreeMap; use std::panic::{AssertUnwindSafe, catch_unwind}; -use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; +use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Condvar, Mutex}; use std::time::{Duration, Instant}; const POSITIVE_TTL: Duration = Duration::from_secs(30 * 60); const NEGATIVE_TTL: Duration = Duration::from_secs(5 * 60); const MAX_CACHE: usize = 500; -const MAX_FLIGHTS: usize = 64; -const POLL: Duration = Duration::from_millis(25); type Outcome = Result; #[derive(Clone, PartialEq, Eq, PartialOrd, Ord)] @@ -25,20 +24,6 @@ struct Key { cached: bool, } -#[derive(Default)] -struct Flight { - waiters: AtomicUsize, - result: Mutex>, - ready: Condvar, -} - -struct Waiter(Arc); -impl Drop for Waiter { - fn drop(&mut self) { - self.0.waiters.fetch_sub(1, Ordering::AcqRel); - } -} - struct CacheEntry { value: Outcome, expires: Instant, @@ -47,8 +32,7 @@ struct CacheEntry { struct State { region: String, generation: u64, - active: usize, - flights: BTreeMap>, + flights: Coalescer, cache: BTreeMap<(String, String), CacheEntry>, } @@ -74,8 +58,7 @@ impl PlatformResolverService { state: Mutex::new(State { region: "US".into(), generation: 0, - active: 0, - flights: BTreeMap::new(), + flights: Coalescer::default(), cache: BTreeMap::new(), }), idle: Condvar::new(), @@ -150,26 +133,21 @@ impl PlatformResolverService { artist: hint.artist.clone(), cached, }; - let existing = state - .flights - .get(&key) - .filter(|flight| flight.waiters.load(Ordering::Acquire) > 0) - .cloned(); + let existing = state.flights.existing(&key); let (flight, start) = if let Some(flight) = existing { (flight, false) } else { - if state.active >= MAX_FLIGHTS { + if state.flights.is_full() { return Err(ResolverError::Busy); } - let flight = Arc::new(Flight::default()); + let flight = Flight::new(()); state.flights.insert(key.clone(), Arc::clone(&flight)); - state.active += 1; (flight, true) }; - flight.waiters.fetch_add(1, Ordering::AcqRel); + flight.join(); (key, flight, start) }; - let _waiter = Waiter(Arc::clone(&flight)); + let _waiter = Waiter::new(Arc::clone(&flight)); if start { let inner = Arc::clone(&self.inner); let worker_key = key.clone(); @@ -178,13 +156,11 @@ impl PlatformResolverService { .name("platform-resolver".into()) .spawn(move || { let check = || { - if inner.closed.load(Ordering::Acquire) { - Err("platform resolver closed".into()) - } else if worker_flight.waiters.load(Ordering::Acquire) == 0 { - Err("platform resolution cancelled".into()) - } else { - Ok(()) - } + worker_flight.check( + &inner.closed, + "platform resolver closed", + "platform resolution cancelled", + ) }; let result = catch_unwind(AssertUnwindSafe(|| { inner.resolver.resolve( @@ -209,21 +185,13 @@ impl PlatformResolverService { .finish(&key, &flight, Err(ResolverError::Failed(error.to_string()))); } } - loop { - self.inner.check(check)?; - let result = flight.result.lock().unwrap(); - if let Some(result) = result.as_ref() { - self.inner.check(check)?; - return result.clone(); - } - drop(flight.ready.wait_timeout(result, POLL).unwrap()); - } + flight.wait(|| self.inner.check(check))? } pub fn shutdown(&self) { self.inner.closed.store(true, Ordering::Release); let mut state = self.inner.state.lock().unwrap(); - while state.active != 0 { + while state.flights.active() != 0 { state = self.inner.idle.wait(state).unwrap(); } state.cache.clear(); @@ -269,7 +237,7 @@ impl Inner { } } - fn finish(&self, key: &Key, flight: &Arc, result: Outcome) { + fn finish(&self, key: &Key, flight: &Arc>, result: Outcome) { let mut state = self.state.lock().unwrap(); if key.cached && key.generation == state.generation @@ -303,23 +271,15 @@ impl Inner { }, ); } - *flight.result.lock().unwrap() = Some(result); - flight.ready.notify_all(); - if state - .flights - .get(key) - .is_some_and(|current| Arc::ptr_eq(current, flight)) - { - state.flights.remove(key); - } - state.active -= 1; - self.idle.notify_all(); + state.flights.finish(key, flight, result, &self.idle); } } #[cfg(test)] mod tests { use super::*; + use crate::coalescer::MAX_FLIGHTS; + use std::sync::atomic::AtomicUsize; #[derive(Default)] struct Probe { @@ -392,8 +352,7 @@ mod tests { .values() .next() .unwrap() - .waiters - .load(Ordering::Acquire) + .waiters() == 2 }); cancelled.store(true, Ordering::Release); @@ -434,7 +393,7 @@ mod tests { ), Err(ResolverError::Cancelled(_)) )); - until(|| service.inner.state.lock().unwrap().active == 0); + until(|| service.inner.state.lock().unwrap().flights.active() == 0); assert!(service.inner.state.lock().unwrap().cache.is_empty()); let waiting = { let service = service.clone(); @@ -449,7 +408,7 @@ mod tests { ); assert_eq!(service.set_region("ID"), Err(ResolverError::Closed)); assert_eq!(service.clear_cache(), Err(ResolverError::Closed)); - assert_eq!(service.inner.state.lock().unwrap().active, 0); + assert_eq!(service.inner.state.lock().unwrap().flights.active(), 0); } #[test]