From 1cf548ebcb4a8d721843e0d00a09ffada1a7620c Mon Sep 17 00:00:00 2001 From: tison Date: Sat, 6 Jun 2026 20:29:19 +0800 Subject: [PATCH 1/2] Add keyed fair semaphore --- README.md | 2 + mea/src/keyed_fair_semaphore/mod.rs | 584 ++++++++++++++++++++++++++ mea/src/keyed_fair_semaphore/tests.rs | 192 +++++++++ mea/src/lib.rs | 12 + 4 files changed, 790 insertions(+) create mode 100644 mea/src/keyed_fair_semaphore/mod.rs create mode 100644 mea/src/keyed_fair_semaphore/tests.rs diff --git a/README.md b/README.md index b1eab3b..f8b092a 100644 --- a/README.md +++ b/README.md @@ -24,6 +24,7 @@ Mea (Make Easy Async) is a runtime-agnostic library providing essential synchron * [**Barrier**](https://docs.rs/mea/*/mea/barrier/struct.Barrier.html): A synchronization primitive that enables tasks to wait until all participants arrive. * [**Condvar**](https://docs.rs/mea/*/mea/condvar/struct.Condvar.html): A condition variable that allows tasks to wait for a notification. +* [**KeyedFairSemaphore**](https://docs.rs/mea/*/mea/keyed_fair_semaphore/struct.KeyedFairSemaphore.html): A semaphore that schedules waiters fairly across keys. * [**Latch**](https://docs.rs/mea/*/mea/latch/struct.Latch.html): A synchronization primitive that allows one or more tasks to wait until a set of operations completes. * [**Mutex**](https://docs.rs/mea/*/mea/mutex/struct.Mutex.html): A mutual exclusion primitive for protecting shared data. * [**Once**](https://docs.rs/mea/*/mea/once/struct.Once.html): A primitive that ensures a one-time asynchronous operation runs at most once, even when called concurrently. @@ -72,6 +73,7 @@ This crate collects runtime-agnostic synchronization primitives from spare parts * **Barrier** is inspired by `std::sync::Barrier` and `tokio::sync::Barrier`, with a different implementation based on the internal `WaitSet` primitive. * **Condvar** is inspired by `std::sync::Condvar` and `async_std::sync::Condvar`, with a different implementation based on the internal `Semaphore` primitive. Different from the async_std implementation, this condvar is fair. +* **KeyedFairSemaphore** is written from scratch to bound global concurrency while balancing released permits across keys by their current in-flight counts. * **Latch** is inspired by [`latches`](https://github.com/mirromutth/latches), with a different implementation based on the internal `CountdownState` primitive. No `wait` or `watch` method is provided, since it can be easily implemented by [composing delay futures](https://docs.rs/fastimer/*/fastimer/fn.timeout.html). No sync variant is provided, since it can be easily implemented with block_on of any runtime. * **Mutex** is derived from `tokio::sync::Mutex`. No blocking method is provided, since it can be easily implemented with block_on of any runtime. * **OnceCell** is derived from `tokio::sync::OnceCell`, but using our own semaphore implementation. diff --git a/mea/src/keyed_fair_semaphore/mod.rs b/mea/src/keyed_fair_semaphore/mod.rs new file mode 100644 index 0000000..f8431b1 --- /dev/null +++ b/mea/src/keyed_fair_semaphore/mod.rs @@ -0,0 +1,584 @@ +// Copyright 2024 tison +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! A keyed semaphore that schedules waiters fairly across keys. +//! +//! A [`KeyedFairSemaphore`] maintains a fixed set of permits. Each acquire +//! operation belongs to a key, and the semaphore prefers pending waiters whose +//! key has the fewest in-flight permits. Ties are resolved by queue order, so +//! waiters with the same in-flight count are served first-in, first-out. +//! +//! This is useful when many tasks compete for the same bounded resource and +//! each task belongs to a key, such as a tenant, statement, partition, or work +//! queue. Compared with a plain FIFO semaphore, it prevents one busy key from +//! monopolizing all released permits while other keys are waiting. +//! +//! # Examples +//! +//! ``` +//! # #[tokio::main] +//! # async fn main() { +//! use std::sync::Arc; +//! +//! use mea::keyed_fair_semaphore::KeyedFairSemaphore; +//! +//! let semaphore = Arc::new(KeyedFairSemaphore::new(2)); +//! let permit_a = semaphore.clone().acquire_owned("a").await; +//! let permit_b = semaphore.clone().acquire_owned("b").await; +//! +//! assert_eq!(semaphore.available_permits(), 0); +//! assert_eq!(permit_a.key(), &"a"); +//! assert_eq!(permit_b.key(), &"b"); +//! # } +//! ``` + +use std::collections::HashMap; +use std::collections::VecDeque; +use std::future::Future; +use std::hash::BuildHasher; +use std::hash::Hash; +use std::hash::RandomState; +use std::pin::Pin; +use std::sync::Arc; +use std::task::Context; +use std::task::Poll; +use std::task::Waker; + +use crate::internal::Mutex; + +#[cfg(test)] +mod tests; + +/// A keyed semaphore that schedules waiters fairly across keys. +/// +/// See the [module level documentation](self) for more. +#[derive(Debug)] +pub struct KeyedFairSemaphore +where + K: Eq + Hash, + S: BuildHasher, +{ + state: Mutex>, +} + +impl KeyedFairSemaphore +where + K: Eq + Hash, +{ + /// Creates a new keyed fair semaphore with the given number of permits. + /// + /// # Panics + /// + /// Panics if `permits` is zero. + /// + /// # Examples + /// + /// ``` + /// use mea::keyed_fair_semaphore::KeyedFairSemaphore; + /// + /// let semaphore = KeyedFairSemaphore::::new(3); + /// assert_eq!(semaphore.total_permits(), 3); + /// assert_eq!(semaphore.available_permits(), 3); + /// ``` + pub fn new(permits: usize) -> Self { + Self::with_hasher(permits, RandomState::new()) + } +} + +impl KeyedFairSemaphore +where + K: Eq + Hash, + S: BuildHasher, +{ + /// Creates a new keyed fair semaphore with the given number of permits and hasher. + /// + /// # Panics + /// + /// Panics if `permits` is zero. + pub fn with_hasher(permits: usize, hasher: S) -> Self { + assert!( + permits > 0, + "a keyed fair semaphore requires at least one permit" + ); + Self { + state: Mutex::new(KeyedFairSemaphoreState::new(permits, hasher)), + } + } + + /// Returns the total number of permits managed by this semaphore. + pub fn total_permits(&self) -> usize { + self.state.lock().total_permits + } + + /// Returns the current number of available permits. + /// + /// Granted waiters that have not yet been polled again already count as + /// in-flight, so they are not included in this number. + pub fn available_permits(&self) -> usize { + self.state.lock().available_permits + } + + /// Returns the number of pending waiters. + /// + /// Waiters that have already been granted a permit but have not yet been + /// polled again are not counted. + pub fn queued_waiters(&self) -> usize { + self.state.lock().pending_waiters() + } + + /// Returns the number of in-flight permits currently held by `key`. + pub fn in_flight_for(&self, key: &K) -> usize { + self.state.lock().in_flight_for(key) + } + + /// Returns `true` when all permits are available and no waiters remain. + pub fn is_idle(&self) -> bool { + let state = self.state.lock(); + state.available_permits == state.total_permits + && state.in_flight_by_key.is_empty() + && state.pending_waiters() == 0 + } + + /// Attempts to acquire one permit for `key` without waiting. + /// + /// This method succeeds only when a permit is available and there are no + /// pending waiters. If another key is already waiting, new callers must + /// join the queue so the keyed fair scheduling rule is preserved. + pub fn try_acquire(&self, key: K) -> Option> { + let key = Arc::new(key); + let acquired = { + let mut state = self.state.lock(); + state.try_acquire_now(key.clone()) + }; + + acquired.then_some(KeyedFairSemaphorePermit { + semaphore: self, + key, + }) + } + + /// Acquires one permit for `key`. + /// + /// # Cancel safety + /// + /// This method uses a queue to fairly distribute permits across keys. + /// Cancelling a call to `acquire` makes you lose your place in the queue. + /// If the waiter has already been granted a permit but the future is + /// cancelled before returning, the permit is released. + pub async fn acquire(&self, key: K) -> KeyedFairSemaphorePermit<'_, K, S> { + let key = Arc::new(key); + Acquire::new(self, key.clone()).await; + KeyedFairSemaphorePermit { + semaphore: self, + key, + } + } + + /// Attempts to acquire one owned permit for `key` without waiting. + /// + /// The semaphore must be wrapped in an [`Arc`] to call this method. + pub fn try_acquire_owned( + self: Arc, + key: K, + ) -> Option> { + let key = Arc::new(key); + let acquired = { + let mut state = self.state.lock(); + state.try_acquire_now(key.clone()) + }; + + acquired.then_some(OwnedKeyedFairSemaphorePermit { + semaphore: self, + key, + }) + } + + /// Acquires one owned permit for `key`. + /// + /// The semaphore must be wrapped in an [`Arc`] to call this method. + /// + /// # Cancel safety + /// + /// This method has the same cancel-safety behavior as [`Self::acquire`]. + pub async fn acquire_owned(self: Arc, key: K) -> OwnedKeyedFairSemaphorePermit { + let key = Arc::new(key); + Acquire::new(&self, key.clone()).await; + OwnedKeyedFairSemaphorePermit { + semaphore: self, + key, + } + } + + fn release(&self, key: &Arc) { + let mut wakers = Vec::new(); + { + let mut state = self.state.lock(); + state.release(key); + state.grant_waiters(&mut wakers); + } + wake_all(wakers); + } +} + +#[derive(Debug)] +struct KeyedFairSemaphoreState +where + K: Eq + Hash, + S: BuildHasher, +{ + total_permits: usize, + available_permits: usize, + next_waiter_id: u64, + in_flight_by_key: HashMap, usize, S>, + waiters: VecDeque>, +} + +impl KeyedFairSemaphoreState +where + K: Eq + Hash, + S: BuildHasher, +{ + fn new(permits: usize, hasher: S) -> Self { + Self { + total_permits: permits, + available_permits: permits, + next_waiter_id: 0, + in_flight_by_key: HashMap::with_hasher(hasher), + waiters: VecDeque::new(), + } + } + + fn pending_waiters(&self) -> usize { + self.waiters.iter().filter(|waiter| !waiter.granted).count() + } + + fn has_pending_waiters(&self) -> bool { + self.waiters.iter().any(|waiter| !waiter.granted) + } + + fn in_flight_for(&self, key: &K) -> usize { + self.in_flight_by_key.get(key).copied().unwrap_or_default() + } + + fn try_acquire_now(&mut self, key: Arc) -> bool { + if self.available_permits == 0 || self.has_pending_waiters() { + return false; + } + + self.available_permits -= 1; + self.increment_in_flight(key); + true + } + + fn push_waiter(&mut self, key: Arc, waker: &Waker) -> u64 { + let waiter_id = self.next_waiter_id; + self.next_waiter_id = self + .next_waiter_id + .checked_add(1) + .expect("number of keyed fair semaphore waiters would overflow u64::MAX"); + + self.waiters.push_back(Waiter { + id: waiter_id, + key, + waker: Some(waker.clone()), + granted: false, + }); + + waiter_id + } + + fn poll_waiter(&mut self, waiter_id: u64, waker: &Waker) -> Poll<()> { + let Some(index) = self.waiter_index(waiter_id) else { + panic!("keyed fair semaphore waiter is missing"); + }; + + let waiter = &mut self.waiters[index]; + if waiter.granted { + self.waiters.remove(index); + Poll::Ready(()) + } else { + let update_waker = waiter + .waker + .as_ref() + .is_none_or(|current| !current.will_wake(waker)); + if update_waker { + waiter.waker = Some(waker.clone()); + } + Poll::Pending + } + } + + fn remove_waiter(&mut self, waiter_id: u64) -> Option> { + let index = self.waiter_index(waiter_id)?; + self.waiters.remove(index) + } + + fn grant_waiters(&mut self, wakers: &mut Vec) { + while self.available_permits > 0 { + let Some(index) = self.next_waiter_to_grant() else { + return; + }; + + self.available_permits -= 1; + let (key, waker) = { + let waiter = &mut self.waiters[index]; + waiter.granted = true; + (waiter.key.clone(), waiter.waker.take()) + }; + self.increment_in_flight(key); + + if let Some(waker) = waker { + wakers.push(waker); + } + } + } + + fn next_waiter_to_grant(&self) -> Option { + let mut best = None; + let mut best_in_flight = usize::MAX; + + for (index, waiter) in self.waiters.iter().enumerate() { + if waiter.granted { + continue; + } + + let in_flight = self + .in_flight_by_key + .get(&waiter.key) + .copied() + .unwrap_or_default(); + if in_flight < best_in_flight { + best = Some(index); + best_in_flight = in_flight; + } + } + + best + } + + fn release(&mut self, key: &Arc) { + let Some(in_flight) = self.in_flight_by_key.get_mut(key) else { + return; + }; + + *in_flight -= 1; + if *in_flight == 0 { + self.in_flight_by_key.remove(key); + } + + self.available_permits += 1; + debug_assert!(self.available_permits <= self.total_permits); + } + + fn increment_in_flight(&mut self, key: Arc) { + *self.in_flight_by_key.entry(key).or_default() += 1; + } + + fn waiter_index(&self, waiter_id: u64) -> Option { + self.waiters + .iter() + .position(|waiter| waiter.id == waiter_id) + } +} + +#[derive(Debug)] +struct Waiter +where + K: Eq + Hash, +{ + id: u64, + key: Arc, + waker: Option, + granted: bool, +} + +#[derive(Debug)] +struct Acquire<'a, K, S = RandomState> +where + K: Eq + Hash, + S: BuildHasher, +{ + semaphore: &'a KeyedFairSemaphore, + key: Arc, + waiter_id: Option, + done: bool, +} + +impl<'a, K, S> Acquire<'a, K, S> +where + K: Eq + Hash, + S: BuildHasher, +{ + fn new(semaphore: &'a KeyedFairSemaphore, key: Arc) -> Self { + Self { + semaphore, + key, + waiter_id: None, + done: false, + } + } +} + +impl Drop for Acquire<'_, K, S> +where + K: Eq + Hash, + S: BuildHasher, +{ + fn drop(&mut self) { + let Some(waiter_id) = self.waiter_id.take() else { + return; + }; + + let mut wakers = Vec::new(); + { + let mut state = self.semaphore.state.lock(); + if let Some(waiter) = state.remove_waiter(waiter_id) + && waiter.granted + { + state.release(&waiter.key); + state.grant_waiters(&mut wakers); + } + } + wake_all(wakers); + } +} + +impl Future for Acquire<'_, K, S> +where + K: Eq + Hash, + S: BuildHasher, +{ + type Output = (); + + fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { + let this = self.get_mut(); + if this.done { + return Poll::Ready(()); + } + + if let Some(waiter_id) = this.waiter_id { + let mut state = this.semaphore.state.lock(); + if state.poll_waiter(waiter_id, cx.waker()).is_ready() { + this.waiter_id = None; + this.done = true; + return Poll::Ready(()); + } + return Poll::Pending; + } + + let mut wakers = Vec::new(); + let mut ready = false; + { + let mut state = this.semaphore.state.lock(); + if state.try_acquire_now(this.key.clone()) { + this.done = true; + return Poll::Ready(()); + } + + let waiter_id = state.push_waiter(this.key.clone(), cx.waker()); + this.waiter_id = Some(waiter_id); + state.grant_waiters(&mut wakers); + if state.poll_waiter(waiter_id, cx.waker()).is_ready() { + this.waiter_id = None; + this.done = true; + ready = true; + } + } + wake_all(wakers); + + if ready { + Poll::Ready(()) + } else { + Poll::Pending + } + } +} + +/// A permit acquired from a [`KeyedFairSemaphore`]. +/// +/// When the permit is dropped, it is released back to the semaphore. +#[must_use = "permits are released immediately when dropped"] +#[derive(Debug)] +pub struct KeyedFairSemaphorePermit<'a, K, S = RandomState> +where + K: Eq + Hash, + S: BuildHasher, +{ + semaphore: &'a KeyedFairSemaphore, + key: Arc, +} + +impl KeyedFairSemaphorePermit<'_, K, S> +where + K: Eq + Hash, + S: BuildHasher, +{ + /// Returns the key that this permit belongs to. + pub fn key(&self) -> &K { + &self.key + } +} + +impl Drop for KeyedFairSemaphorePermit<'_, K, S> +where + K: Eq + Hash, + S: BuildHasher, +{ + fn drop(&mut self) { + self.semaphore.release(&self.key); + } +} + +/// An owned permit acquired from an [`Arc`]. +/// +/// When the permit is dropped, it is released back to the semaphore. +/// +/// [`Arc`]: std::sync::Arc +#[must_use = "permits are released immediately when dropped"] +#[derive(Debug)] +pub struct OwnedKeyedFairSemaphorePermit +where + K: Eq + Hash, + S: BuildHasher, +{ + semaphore: Arc>, + key: Arc, +} + +impl OwnedKeyedFairSemaphorePermit +where + K: Eq + Hash, + S: BuildHasher, +{ + /// Returns the key that this permit belongs to. + pub fn key(&self) -> &K { + &self.key + } +} + +impl Drop for OwnedKeyedFairSemaphorePermit +where + K: Eq + Hash, + S: BuildHasher, +{ + fn drop(&mut self) { + self.semaphore.release(&self.key); + } +} + +fn wake_all(wakers: Vec) { + for waker in wakers { + waker.wake(); + } +} diff --git a/mea/src/keyed_fair_semaphore/tests.rs b/mea/src/keyed_fair_semaphore/tests.rs new file mode 100644 index 0000000..d6f8ff3 --- /dev/null +++ b/mea/src/keyed_fair_semaphore/tests.rs @@ -0,0 +1,192 @@ +// Copyright 2024 tison +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use std::future::Future; +use std::pin::pin; +use std::sync::Arc; +use std::sync::atomic::AtomicUsize; +use std::sync::atomic::Ordering; +use std::task::Context; +use std::task::Poll; +use std::task::Waker; + +use super::*; + +#[test] +#[should_panic(expected = "a keyed fair semaphore requires at least one permit")] +fn no_permits() { + KeyedFairSemaphore::::new(0); +} + +#[test] +fn try_acquire() { + let semaphore = KeyedFairSemaphore::new(1); + let permit = semaphore.try_acquire("a").unwrap(); + + assert_eq!(permit.key(), &"a"); + assert_eq!(semaphore.available_permits(), 0); + assert_eq!(semaphore.in_flight_for(&"a"), 1); + assert!(semaphore.try_acquire("b").is_none()); + + drop(permit); + assert_eq!(semaphore.available_permits(), 1); + assert!(semaphore.is_idle()); +} + +#[test] +fn balances_keys_with_fewer_in_flight_permits() { + let waker = Waker::noop(); + let mut context = Context::from_waker(waker); + + let semaphore = KeyedFairSemaphore::new(2); + let permit_a0 = semaphore.try_acquire("a").unwrap(); + let permit_a1 = semaphore.try_acquire("a").unwrap(); + + let waiter_a = semaphore.acquire("a"); + let mut waiter_a = pin!(waiter_a); + assert!(waiter_a.as_mut().poll(&mut context).is_pending()); + + let waiter_b = semaphore.acquire("b"); + let mut waiter_b = pin!(waiter_b); + assert!(waiter_b.as_mut().poll(&mut context).is_pending()); + assert_eq!(semaphore.queued_waiters(), 2); + + drop(permit_a0); + assert!(waiter_a.as_mut().poll(&mut context).is_pending()); + let permit_b = match waiter_b.as_mut().poll(&mut context) { + Poll::Ready(permit) => permit, + Poll::Pending => panic!("waiter for key b should be granted first"), + }; + assert_eq!(permit_b.key(), &"b"); + + drop(permit_a1); + let permit_a = match waiter_a.as_mut().poll(&mut context) { + Poll::Ready(permit) => permit, + Poll::Pending => panic!("waiter for key a should be granted second"), + }; + assert_eq!(permit_a.key(), &"a"); +} + +#[test] +fn preserves_fifo_within_key() { + let waker = Waker::noop(); + let mut context = Context::from_waker(waker); + + let semaphore = KeyedFairSemaphore::new(1); + let held = semaphore.try_acquire(7usize).unwrap(); + + let waiter1 = semaphore.acquire(7usize); + let mut waiter1 = pin!(waiter1); + assert!(waiter1.as_mut().poll(&mut context).is_pending()); + + let waiter2 = semaphore.acquire(7usize); + let mut waiter2 = pin!(waiter2); + assert!(waiter2.as_mut().poll(&mut context).is_pending()); + + drop(held); + let permit1 = match waiter1.as_mut().poll(&mut context) { + Poll::Ready(permit) => permit, + Poll::Pending => panic!("first waiter should be granted first"), + }; + assert!(waiter2.as_mut().poll(&mut context).is_pending()); + + drop(permit1); + let permit2 = match waiter2.as_mut().poll(&mut context) { + Poll::Ready(permit) => permit, + Poll::Pending => panic!("second waiter should be granted second"), + }; + assert_eq!(permit2.key(), &7); +} + +#[test] +fn cancelling_waiter_removes_it_from_queue() { + let waker = Waker::noop(); + let mut context = Context::from_waker(waker); + + let semaphore = KeyedFairSemaphore::new(1); + let held = semaphore.try_acquire(1usize).unwrap(); + + { + let waiter = semaphore.acquire(2usize); + let mut waiter = pin!(waiter); + assert!(waiter.as_mut().poll(&mut context).is_pending()); + assert_eq!(semaphore.queued_waiters(), 1); + } + + assert_eq!(semaphore.queued_waiters(), 0); + drop(held); + assert_eq!(semaphore.available_permits(), 1); + + let permit = semaphore.try_acquire(3usize).unwrap(); + assert_eq!(permit.key(), &3); +} + +#[test] +fn cancelling_granted_waiter_releases_permit() { + let waker = Waker::noop(); + let mut context = Context::from_waker(waker); + + let semaphore = KeyedFairSemaphore::new(1); + let held = semaphore.try_acquire(1usize).unwrap(); + + { + let waiter = semaphore.acquire(2usize); + let mut waiter = pin!(waiter); + assert!(waiter.as_mut().poll(&mut context).is_pending()); + + drop(held); + assert_eq!(semaphore.available_permits(), 0); + assert_eq!(semaphore.queued_waiters(), 0); + assert_eq!(semaphore.in_flight_for(&2), 1); + } + + assert_eq!(semaphore.available_permits(), 1); + assert_eq!(semaphore.in_flight_for(&2), 0); + + let permit = semaphore.try_acquire(3usize).unwrap(); + assert_eq!(permit.key(), &3); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn stress_test_preserves_capacity() { + let semaphore = Arc::new(KeyedFairSemaphore::new(3)); + let active = Arc::new(AtomicUsize::new(0)); + let max_active = Arc::new(AtomicUsize::new(0)); + let mut handles = vec![]; + + for key in 0..5usize { + for _ in 0..32usize { + let semaphore = semaphore.clone(); + let active = active.clone(); + let max_active = max_active.clone(); + handles.push(tokio::spawn(async move { + let _permit = semaphore.acquire_owned(key).await; + let now = active.fetch_add(1, Ordering::SeqCst) + 1; + max_active.fetch_max(now, Ordering::SeqCst); + tokio::task::yield_now().await; + active.fetch_sub(1, Ordering::SeqCst); + })); + } + } + + for handle in handles { + handle.await.unwrap(); + } + + assert_eq!(active.load(Ordering::SeqCst), 0); + assert!(max_active.load(Ordering::SeqCst) <= 3); + assert_eq!(semaphore.available_permits(), 3); + assert_eq!(semaphore.queued_waiters(), 0); + assert!(semaphore.is_idle()); +} diff --git a/mea/src/lib.rs b/mea/src/lib.rs index 7a4f1a6..056e9a7 100644 --- a/mea/src/lib.rs +++ b/mea/src/lib.rs @@ -24,6 +24,7 @@ //! * [`Barrier`]: A synchronization point where multiple tasks can wait until all participants //! arrive //! * [`Condvar`]: A condition variable that allows tasks to wait for a notification +//! * [`KeyedFairSemaphore`]: A semaphore that schedules waiters fairly across keys //! * [`Latch`]: A single-use barrier that allows one or more tasks to wait until a signal is given //! * [`Mutex`]: A mutual exclusion primitive for protecting shared data //! * [`Once`]: A primitive that ensures a one-time asynchronous operation runs at most once, even @@ -58,6 +59,7 @@ //! //! [`Barrier`]: barrier::Barrier //! [`Condvar`]: condvar::Condvar +//! [`KeyedFairSemaphore`]: keyed_fair_semaphore::KeyedFairSemaphore //! [`Latch`]: latch::Latch //! [`Mutex`]: mutex::Mutex //! [`Once`]: once::Once @@ -76,6 +78,7 @@ pub mod atomicbox; pub mod barrier; pub mod broadcast; pub mod condvar; +pub mod keyed_fair_semaphore; pub mod latch; pub mod mpsc; pub mod mutex; @@ -101,6 +104,9 @@ mod tests { use crate::barrier::Barrier; use crate::broadcast; use crate::condvar::Condvar; + use crate::keyed_fair_semaphore::KeyedFairSemaphore; + use crate::keyed_fair_semaphore::KeyedFairSemaphorePermit; + use crate::keyed_fair_semaphore::OwnedKeyedFairSemaphorePermit; use crate::latch::Latch; use crate::mpsc; use crate::mutex::Mutex; @@ -125,6 +131,9 @@ mod tests { fn do_assert_send_and_sync() {} do_assert_send_and_sync::(); do_assert_send_and_sync::(); + do_assert_send_and_sync::>(); + do_assert_send_and_sync::>(); + do_assert_send_and_sync::>(); do_assert_send_and_sync::(); do_assert_send_and_sync::>(); do_assert_send_and_sync::>(); @@ -165,6 +174,9 @@ mod tests { fn do_assert_unpin() {} do_assert_unpin::(); do_assert_unpin::(); + do_assert_unpin::>(); + do_assert_unpin::>(); + do_assert_unpin::>(); do_assert_unpin::(); do_assert_unpin::(); do_assert_unpin::>(); From d588b66d95ef3686df3eb98da21db64fa4890a3b Mon Sep 17 00:00:00 2001 From: tison Date: Sat, 6 Jun 2026 20:34:34 +0800 Subject: [PATCH 2/2] Fix keyed fair semaphore MSRV compatibility --- mea/src/keyed_fair_semaphore/mod.rs | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/mea/src/keyed_fair_semaphore/mod.rs b/mea/src/keyed_fair_semaphore/mod.rs index f8431b1..7406d1e 100644 --- a/mea/src/keyed_fair_semaphore/mod.rs +++ b/mea/src/keyed_fair_semaphore/mod.rs @@ -443,11 +443,11 @@ where let mut wakers = Vec::new(); { let mut state = self.semaphore.state.lock(); - if let Some(waiter) = state.remove_waiter(waiter_id) - && waiter.granted - { - state.release(&waiter.key); - state.grant_waiters(&mut wakers); + if let Some(waiter) = state.remove_waiter(waiter_id) { + if waiter.granted { + state.release(&waiter.key); + state.grant_waiters(&mut wakers); + } } } wake_all(wakers);