diff --git a/crates/tinymemory-integrations/src/cortex/testing/mod.rs b/crates/tinymemory-integrations/src/cortex/testing/mod.rs index 2efba3ee..223796fd 100644 --- a/crates/tinymemory-integrations/src/cortex/testing/mod.rs +++ b/crates/tinymemory-integrations/src/cortex/testing/mod.rs @@ -174,9 +174,13 @@ pub(crate) type Shared = Arc; /// Decrements `counter` if positive; whether a unit was taken. pub(crate) fn take_one(counter: &AtomicUsize) -> bool { - counter + // `fetch_update` is deprecated for `try_update` on newer toolchains, which + // is not stable on this crate's MSRV. + #[allow(deprecated)] + let taken = counter .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |n| n.checked_sub(1)) - .is_ok() + .is_ok(); + taken } impl Double { diff --git a/crates/tinymemory-tools/src/deletion/mod_tests.rs b/crates/tinymemory-tools/src/deletion/mod_tests.rs index 55a9cdc8..946db09b 100644 --- a/crates/tinymemory-tools/src/deletion/mod_tests.rs +++ b/crates/tinymemory-tools/src/deletion/mod_tests.rs @@ -142,14 +142,16 @@ impl ListStore for &Flaky { } fn save(&self, all: &[PendingDeletion]) -> Result<(), crate::list_store::StoreError> { let failing = &self.failing_saves; - if failing + // `fetch_update` is deprecated for `try_update` on newer toolchains, which is not stable on this crate's MSRV. + #[allow(deprecated)] + let taken = failing .fetch_update( std::sync::atomic::Ordering::SeqCst, std::sync::atomic::Ordering::SeqCst, |left| left.checked_sub(1), ) - .is_ok() - { + .is_ok(); + if taken { return Err(crate::list_store::StoreError("full".into())); } self.saves.lock().unwrap().push(all.to_vec()); diff --git a/crates/tinymemory-tools/src/prefetch/mod.rs b/crates/tinymemory-tools/src/prefetch/mod.rs index 33541fa5..167c8baa 100644 --- a/crates/tinymemory-tools/src/prefetch/mod.rs +++ b/crates/tinymemory-tools/src/prefetch/mod.rs @@ -74,14 +74,24 @@ impl WorkerPool { } fn try_acquire(&self) -> Option { - self.free - .fetch_update(Ordering::AcqRel, Ordering::Acquire, |free| { - free.checked_sub(1) - }) - .ok() - .map(|_| Permit { - free: self.free.clone(), - }) + // A compare_exchange loop rather than `fetch_update`: that method is + // deprecated in favour of `try_update` on newer toolchains, but + // `try_update` is not stable on this crate's MSRV. + let mut free = self.free.load(Ordering::Acquire); + loop { + let next = free.checked_sub(1)?; + match self + .free + .compare_exchange_weak(free, next, Ordering::AcqRel, Ordering::Acquire) + { + Ok(_) => { + return Some(Permit { + free: self.free.clone(), + }); + } + Err(actual) => free = actual, + } + } } /// Workers not running a refresh. diff --git a/crates/tinymemory-tools/src/prefetch/mod_tests.rs b/crates/tinymemory-tools/src/prefetch/mod_tests.rs index c0e1a8a9..ae553cce 100644 --- a/crates/tinymemory-tools/src/prefetch/mod_tests.rs +++ b/crates/tinymemory-tools/src/prefetch/mod_tests.rs @@ -203,3 +203,22 @@ fn runtime_shutdown_releases_worker_and_pending_work() { assert_eq!(pool.available(), 1, "the permit is released"); assert!(cache.completed(&"k").is_none()); } + +#[test] +fn concurrent_acquires_never_exceed_the_pool_size() { + let pool = WorkerPool::new(3); + let granted = std::sync::Mutex::new(Vec::new()); + std::thread::scope(|scope| { + for _ in 0..16 { + scope.spawn(|| { + if let Some(permit) = pool.try_acquire() { + granted.lock().unwrap().push(permit); + } + }); + } + }); + assert_eq!(granted.lock().unwrap().len(), 3); + assert_eq!(pool.available(), 0); + granted.lock().unwrap().clear(); + assert_eq!(pool.available(), 3); +}