From 1b3d1daee37afe5ade62c84dfc042c932486fe76 Mon Sep 17 00:00:00 2001 From: tison Date: Mon, 24 Aug 2026 00:56:47 +0800 Subject: [PATCH 1/5] refactor(pool): simplify lifecycle and state tracking --- asyncband/src/pool/bounded.rs | 298 ++++---------- asyncband/src/pool/common.rs | 47 ++- asyncband/src/pool/mod.rs | 78 ++-- asyncband/src/pool/mutex.rs | 57 --- asyncband/src/pool/retain_spec.rs | 92 ----- asyncband/src/pool/state.rs | 136 +++++++ asyncband/src/pool/unbounded.rs | 231 +++++------ asyncband/src/semaphore/mod.rs | 19 + tests-integration/tests/pool_behavior_test.rs | 171 ++++++++ .../tests/pool_recycle_cancelled_test.rs | 371 ++++++------------ .../tests/pool_replenish_test.rs | 132 ++++++- tests-integration/tests/traits_test.rs | 13 + 12 files changed, 833 insertions(+), 812 deletions(-) delete mode 100644 asyncband/src/pool/mutex.rs delete mode 100644 asyncband/src/pool/retain_spec.rs create mode 100644 asyncband/src/pool/state.rs create mode 100644 tests-integration/tests/pool_behavior_test.rs diff --git a/asyncband/src/pool/bounded.rs b/asyncband/src/pool/bounded.rs index dae5c45..25ea11f 100644 --- a/asyncband/src/pool/bounded.rs +++ b/asyncband/src/pool/bounded.rs @@ -17,23 +17,21 @@ //! Bounded object pools. //! -//! A bounded pool creates and recycles objects with full management. You _cannot_ put an object to -//! the pool manually. +//! A bounded pool uses a [`ManageObject`] implementation to create, validate, and detach objects. +//! Objects cannot be inserted manually. //! //! The pool is bounded by the `max_size` config option of [`PoolConfig`]. If the pool reaches the -//! maximum size, it will block all the [`Pool::get`] calls until an object is returned to the pool -//! or an object is detached from the pool. +//! maximum size, additional [`Pool::get`] calls wait until an object is returned to or detached +//! from the pool. //! -//! Typically, a bounded pool is used wrapped in an [`Arc`] in order to call [`Pool::get`]. -//! This is intended so that users can leverage [`Arc::downgrade`] for running background -//! maintenance tasks (e.g., [`Pool::retain`]). +//! [`Pool::new`] returns an [`Arc`], allowing background maintenance code to hold a [`Weak`] and +//! terminate naturally when application owners drop the pool. Scheduling that maintenance remains +//! the caller's responsibility. //! //! Bounded pools are useful for pooling database connections. //! //! ## Examples //! -//! Read the following simple demo or more complex examples in the examples directory. -//! //! ``` //! use asyncband::pool::ManageObject; //! use asyncband::pool::ObjectStatus; @@ -58,8 +56,8 @@ //! //! async fn is_recyclable( //! &self, -//! o: &mut Self::Object, -//! status: &ObjectStatus, +//! _object: &mut Self::Object, +//! _status: &ObjectStatus, //! ) -> Result<(), Self::Error> { //! Ok(()) //! } @@ -73,21 +71,19 @@ //! # } //! ``` -use std::collections::VecDeque; use std::ops::Deref; use std::ops::DerefMut; use std::sync::Arc; use std::sync::Weak; -use std::sync::atomic::AtomicUsize; -use std::sync::atomic::Ordering; +use crate::internal::mutex::Mutex; use crate::pool::ManageObject; use crate::pool::ObjectStatus; use crate::pool::QueueStrategy; use crate::pool::RecycleCancelledStrategy; use crate::pool::RetainResult; -use crate::pool::mutex::Mutex; -use crate::pool::retain_spec; +use crate::pool::state::ObjectState; +use crate::pool::state::PoolState; use crate::semaphore::OwnedSemaphorePermit; use crate::semaphore::Semaphore; @@ -103,7 +99,7 @@ pub struct PoolConfig { /// Determines the order of objects being queued and dequeued. pub queue_strategy: QueueStrategy, - /// Strategy when recycling object has been cancelled. + /// Strategy to apply when object recycling is cancelled. pub recycle_cancelled_strategy: RecycleCancelledStrategy, } @@ -142,14 +138,11 @@ pub struct PoolStatus { /// The maximum size of the pool. pub max_size: usize, - /// The current size of the pool. + /// The number of successfully created objects that have not been detached. pub current_size: usize, - /// The number of idle objects in the pool. + /// The number of objects currently available for checkout. pub idle_count: usize, - - /// The number of futures waiting for an object. - pub wait_count: usize, } /// Generic runtime-agnostic object pool with a maximum size. @@ -159,48 +152,10 @@ pub struct Pool { config: PoolConfig, manager: M, - /// A counter that tracks the sum of waiters + obtained objects. - users: AtomicUsize, - /// A semaphore that limits the maximum of users of the pool. + /// A semaphore that reserves capacity for checkouts and object creation. permits: Arc, - /// A deque that holds the objects. - slots: Mutex>>, -} - -/// Restores the pool's user count when a `get` attempt fails or is cancelled. -/// -/// A successful `get` transfers responsibility for decrementing the count to the returned -/// [`Object`]. -// TODO: Replace this pool-specific guard with the standard library's `DropGuard` once -// https://github.com/rust-lang/rust/issues/144426 stabilizes. -struct UserCountGuard<'a> { - users: Option<&'a AtomicUsize>, -} - -impl<'a> UserCountGuard<'a> { - fn new(users: &'a AtomicUsize) -> Self { - users.fetch_add(1, Ordering::Relaxed); - Self { users: Some(users) } - } - - fn commit(mut self) { - self.users = None; - } -} - -impl Drop for UserCountGuard<'_> { - fn drop(&mut self) { - if let Some(users) = self.users { - users.fetch_sub(1, Ordering::Relaxed); - } - } -} - -#[derive(Debug)] -struct PoolDeque { - deque: VecDeque, - current_size: usize, - max_size: usize, + /// The objects tracked by the pool. + slots: Mutex>, } impl std::fmt::Debug for Pool @@ -212,7 +167,6 @@ where f.debug_struct("Pool") .field("slots", &self.slots) .field("config", &self.config) - .field("users", &self.users) .field("permits", &self.permits) .finish() } @@ -221,121 +175,65 @@ where impl Pool { /// Creates a new [`Pool`]. pub fn new(config: PoolConfig, manager: M) -> Arc { - let users = AtomicUsize::new(0); let permits = Arc::new(Semaphore::new(config.max_size)); - let slots = Mutex::new(PoolDeque { - deque: VecDeque::with_capacity(config.max_size), - current_size: 0, - max_size: config.max_size, - }); + let slots = Mutex::new(PoolState::new()); Arc::new(Self { config, manager, - users, permits, slots, }) } - /// Replenishes the pool with at most `most` number of new objects: + /// Creates objects until the pool approaches `target_idle` idle objects. /// - /// 1. If the pool has fewer slots to fill than `most`, narrow `most` to the number of slots. - /// 2. If there is already any idle object in the pool, decrease `most` by the number of idle - /// objects. - /// 3. If [`ManageObject::create`] returns `Err`, reduces `most` by 1 and continues to the next. + /// The pool reserves only capacity that is immediately available and never waits for + /// checked-out objects. Existing idle objects count toward the target, and the pool's + /// maximum size is never exceeded. Concurrent calls and checkouts can change the observed + /// idle count while this method is running, so the target is best effort rather than a + /// postcondition. /// - /// Returns the number of objects that are actually replenished to the pool. This method is - /// suitable to implement functionalities like minimal idle connections in a connection - /// pool. - pub async fn replenish(&self, most: usize) -> usize { - let mut permit = { - let mut n = most; - loop { - match self.permits.try_acquire(n) { - Some(permit) => break permit, - None => { - n = n.min(self.permits.available_permits()); - continue; - } - } - } + /// Returns the number of objects created. If [`ManageObject::create`] fails, objects created by + /// this call before the failure remain in the pool and the error is returned. + pub async fn replenish_to(&self, target_idle: usize) -> Result { + let Some(mut permit) = self.permits.clone().try_acquire_up_to_owned(target_idle) else { + return Ok(0); }; - if permit.permits() == 0 { - return 0; - } - - let gap = { - let idles = self.slots.lock().deque.len(); - if idles >= permit.permits() { - return 0; - } - - match permit.split(idles) { - None => unreachable!( - "idles ({}) should be less than permits ({})", - idles, - permit.permits() - ), - Some(p) => { - // reduced by existing idle objects and release the corresponding permits - drop(p); - } - } - - permit.permits() - }; + let idle_count = self.slots.lock().idle_count(); + let to_create = target_idle.saturating_sub(idle_count).min(permit.permits()); + permit.release(permit.permits() - to_create); let mut replenished = 0; - for _ in 0..gap { - if let Ok(o) = self.manager.create().await { - let status = ObjectStatus::default(); - let state = ObjectState { o, status }; - + for _ in 0..to_create { + let object = self.manager.create().await?; + { let mut slots = self.slots.lock(); - slots.current_size += 1; - slots.deque.push_back(state); - drop(slots); - - replenished += 1; - } - - match permit.split(1) { - None => unreachable!("permit must be greater than 0 at this point"), - Some(p) => { - // always release one permit to unblock other waiters - drop(p); - } + slots.add_idle(ObjectState::new(object)); } + replenished += 1; + permit.release(1); } - replenished + Ok(replenished) } /// Retrieves an [`Object`] from this [`Pool`]. /// - /// This method should be called with a pool wrapped in an [`Arc`]. If the pool reaches the - /// maximum size, this method waits until an object is returned to the pool or detached from it. + /// If the pool has reached its maximum size and has no idle object, this method waits until an + /// object is returned to or detached from the pool. pub async fn get(self: &Arc) -> Result, M::Error> { - let user_count_guard = UserCountGuard::new(&self.users); - let permit = self.permits.clone().acquire_owned(1).await; let object = loop { - let existing = match self.config.queue_strategy { - QueueStrategy::Fifo => self.slots.lock().deque.pop_front(), - QueueStrategy::Lifo => self.slots.lock().deque.pop_back(), - }; + let existing = self.slots.lock().pop(self.config.queue_strategy); match existing { None => { let object = self.manager.create().await?; - let state = ObjectState { - o: object, - status: ObjectStatus::default(), - }; - self.slots.lock().current_size += 1; + let state = ObjectState::new(object); + self.slots.lock().add_active(); break Object { state: Some(state), permit, @@ -357,8 +255,7 @@ impl Pool { .await .is_ok() { - state.status.recycle_count += 1; - state.status.recycled = Some(std::time::Instant::now()); + state.status.mark_recycled(); break unready_object.ready(permit); } else { // We need to manually detach here as the drop implementation @@ -369,13 +266,13 @@ impl Pool { }; }; - user_count_guard.commit(); Ok(object) } /// Retains only the objects that pass the given predicate. /// - /// This function blocks the entire pool. Therefore, the given function should not block. + /// The predicate runs while the idle-object lock is held and therefore must not block or call + /// back into the pool. Detachment hooks for removed objects run after the lock is released. /// /// The following example starts a background task that runs every 30 seconds and removes /// objects from the pool that have not been used for more than one minute. The task will @@ -401,77 +298,58 @@ impl Pool { &self, f: impl FnMut(&mut M::Object, ObjectStatus) -> bool, ) -> RetainResult { - let mut slots = self.slots.lock(); - let result = retain_spec::do_vec_deque_retain(&mut slots.deque, f); - slots.current_size -= result.removed.len(); + let mut result = { + let mut slots = self.slots.lock(); + slots.retain(f) + }; + for object in &mut result.removed { + self.manager.on_detached(object); + } result } - /// Returns the current status of the pool. - /// - /// The status returned by the pool is not guaranteed to be consistent. - /// - /// Although this status provides [eventual consistency], the numbers can be - /// temporarily inaccurate under heavy load. They are intended as an overall insight. - /// - /// [eventual consistency]: (https://en.wikipedia.org/wiki/Eventual_consistency) + /// Returns a consistent snapshot of the objects currently tracked by the pool. pub fn status(&self) -> PoolStatus { let slots = self.slots.lock(); - let (current_size, max_size) = (slots.current_size, slots.max_size); - drop(slots); - - let users = self.users.load(Ordering::Relaxed); - let (idle_count, wait_count) = if users < current_size { - (current_size - users, 0) - } else { - (0, users - current_size) - }; PoolStatus { - max_size, - current_size, - idle_count, - wait_count, + max_size: self.config.max_size, + current_size: slots.current_size(), + idle_count: slots.idle_count(), } } - fn push_back(&self, o: ObjectState) { - self.return_to_pool(o); - self.users.fetch_sub(1, Ordering::Relaxed); + fn return_object(&self, mut state: ObjectState) { + state.status.mark_returned(); + self.restore_idle(state); } - fn return_to_pool(&self, o: ObjectState) { + fn restore_idle(&self, state: ObjectState) { let mut slots = self.slots.lock(); assert!( - slots.current_size <= slots.max_size, + slots.current_size() <= self.config.max_size, "invariant broken: current_size <= max_size (actual: {} <= {})", - slots.current_size, - slots.max_size, + slots.current_size(), + self.config.max_size, ); - slots.deque.push_back(o); + slots.return_idle(state); } - fn detach_object(&self, o: &mut M::Object, ready: bool) { + fn detach_object(&self, o: &mut M::Object) { let mut slots = self.slots.lock(); assert!( - slots.current_size <= slots.max_size, + slots.current_size() <= self.config.max_size, "invariant broken: current_size <= max_size (actual: {} <= {})", - slots.current_size, - slots.max_size, + slots.current_size(), + self.config.max_size, ); - slots.current_size -= 1; + slots.detach(); drop(slots); - if ready { - self.users.fetch_sub(1, Ordering::Relaxed); - } else { - // if the object is not ready, users count decrement is handled in the caller side, - // that is, on exiting the `Pool::get` method. - } self.manager.on_detached(o); } } @@ -506,7 +384,7 @@ impl Drop for Object { fn drop(&mut self) { if let Some(state) = self.state.take() { if let Some(pool) = self.pool.upgrade() { - pool.push_back(state); + pool.return_object(state); } } } @@ -547,7 +425,7 @@ impl Object { // SAFETY: `state` is always `Some` when `Object` is owned. let mut o = self.state.take().unwrap().o; if let Some(pool) = self.pool.upgrade() { - pool.detach_object(&mut o, true); + pool.detach_object(&mut o); } o } @@ -577,10 +455,10 @@ impl Drop for UnreadyObject { if let Some(pool) = self.pool.upgrade() { match self.recycle_cancelled_strategy { RecycleCancelledStrategy::Detach => { - pool.detach_object(&mut state.o, false); + pool.detach_object(&mut state.o); } RecycleCancelledStrategy::ReturnToPool => { - pool.return_to_pool(state); + pool.restore_idle(state); } } } @@ -603,7 +481,7 @@ impl UnreadyObject { fn detach(&mut self) { if let Some(mut state) = self.state.take() { if let Some(pool) = self.pool.upgrade() { - pool.detach_object(&mut state.o, false); + pool.detach_object(&mut state.o); } } } @@ -613,25 +491,3 @@ impl UnreadyObject { self.state.as_mut().unwrap() } } - -#[derive(Debug)] -struct ObjectState { - o: T, - status: ObjectStatus, -} - -impl retain_spec::SealedState for ObjectState { - type Object = T; - - fn status(&self) -> ObjectStatus { - self.status - } - - fn mut_object(&mut self) -> &mut Self::Object { - &mut self.o - } - - fn take_object(self) -> Self::Object { - self.o - } -} diff --git a/asyncband/src/pool/common.rs b/asyncband/src/pool/common.rs index 57f91aa..d6f54f2 100644 --- a/asyncband/src/pool/common.rs +++ b/asyncband/src/pool/common.rs @@ -18,19 +18,19 @@ use std::future::Future; use std::time::Instant; -/// Statistics regarding an object returned by the pool. +/// Lifecycle metadata for a pooled object. #[derive(Debug, Clone, Copy)] pub struct ObjectStatus { created: Instant, - pub(crate) recycled: Option, - pub(crate) recycle_count: usize, + last_returned: Option, + recycle_count: usize, } impl Default for ObjectStatus { fn default() -> Self { Self { created: Instant::now(), - recycled: None, + last_returned: None, recycle_count: 0, } } @@ -42,15 +42,36 @@ impl ObjectStatus { self.created } - /// Returns the instant when this object was last used. + /// Returns the instant when this object was last returned to the pool. + /// + /// If the object has not been returned yet, this returns its creation time. While an object is + /// checked out, the value therefore describes the end of its previous use, if any. pub fn last_used(&self) -> Instant { - self.recycled.unwrap_or(self.created) + self.last_returned.unwrap_or(self.created) } - /// Returns the number of times the object was recycled. + /// Returns the number of successful checkouts from the idle queue. pub fn recycle_count(&self) -> usize { self.recycle_count } + + pub(crate) fn mark_recycled(&mut self) { + self.recycle_count += 1; + } + + pub(crate) fn mark_returned(&mut self) { + self.last_returned = Some(Instant::now()); + } +} + +/// The result returned by a pool's `retain` method. +#[derive(Debug)] +#[non_exhaustive] +pub struct RetainResult { + /// The number of retained objects. + pub retained: usize, + /// The objects removed from the pool, after the manager's detachment hook has run. + pub removed: Vec, } /// A trait whose instance creates new objects and recycles existing ones. @@ -73,10 +94,14 @@ pub trait ManageObject: Send + Sync { status: &ObjectStatus, ) -> impl Future> + Send; - /// A callback invoked when an object is detached from the pool. + /// A callback invoked when an object is detached from a live pool. + /// + /// This includes explicit detachment, failed or cancelled recycling, and removal through + /// `retain`. The callback runs without the pool's internal lock held. It is not invoked when + /// the pool itself is dropped or when an object can no longer reach its pool. /// - /// If this instance does not hold any references to the object, then the default - /// implementation can be used which does nothing. + /// If this instance does not hold any references to the object, the default implementation can + /// be used, which does nothing. fn on_detached(&self, _o: &mut Self::Object) {} } @@ -94,7 +119,7 @@ pub enum QueueStrategy { Lifo, } -/// Strategy when recycling object has been cancelled. +/// Strategy to apply when object recycling is cancelled. /// /// This enum controls the behavior when the recycling process (specifically the /// [`ManageObject::is_recyclable`] check) is cancelled; for example, when the diff --git a/asyncband/src/pool/mod.rs b/asyncband/src/pool/mod.rs index 7b40053..46e5567 100644 --- a/asyncband/src/pool/mod.rs +++ b/asyncband/src/pool/mod.rs @@ -17,17 +17,21 @@ //! Runtime-agnostic object pools for async Rust. //! -//! This module provides two implementations: [bounded pool](bounded::Pool) and -//! [unbounded pool](unbounded::Pool). +//! This module provides a manager-created [bounded pool](bounded::Pool) and an +//! [unbounded pool](unbounded::Pool) that can also accept objects supplied by callers. +//! +//! Both implementations provide resource reuse without taking ownership of runtime policy. They do +//! not start maintenance tasks or install timers. Callers decide how to schedule maintenance and +//! can wrap operations such as [`bounded::Pool::get`] in the deadline mechanism of their runtime. //! //! # Bounded pool //! -//! A bounded pool creates and recycles objects with full management. You _cannot_ put an object to -//! the pool manually. +//! A bounded pool uses a [`ManageObject`] implementation to create, validate, and detach objects. +//! Objects cannot be inserted manually. //! //! The pool is bounded by the `max_size` config option of [`PoolConfig`](bounded::PoolConfig). If -//! the pool reaches the maximum size, it will block all the [`Pool::get`](bounded::Pool::get) calls -//! until an object is returned to the pool or an object is detached from the pool. +//! the pool reaches the maximum size, additional [`Pool::get`](bounded::Pool::get) calls wait until +//! an object is returned to or detached from the pool. //! //! Bounded pools are useful for pooling database connections. //! @@ -76,8 +80,8 @@ //! //! # Unbounded pool //! -//! An unbounded pool, on the other hand, allows you to put objects to the pool manually. You can -//! use it like Go's [`sync.Pool`](https://pkg.go.dev/sync#Pool). +//! An unbounded pool accepts manually supplied objects and can be used like Go's +//! [`sync.Pool`](https://pkg.go.dev/sync#Pool). //! //! To configure a factory for creating objects when the pool is empty, like `sync.Pool`'s `New`, //! you can create the unbounded pool via [`Pool::new`](unbounded::Pool::new) with an @@ -91,38 +95,23 @@ //! use asyncband::pool::unbounded::Pool; //! use asyncband::pool::unbounded::PoolConfig; //! -//! # #[tokio::main] -//! # async fn main() { //! let pool = Pool::>::never_manage(PoolConfig::default()); //! -//! let result = pool.get().await; -//! assert_eq!(result.unwrap_err().to_string(), "unbounded pool is empty"); +//! assert!(pool.try_get().is_none()); //! //! pool.extend_one(Vec::with_capacity(1024)); -//! let o = pool.get().await.unwrap(); +//! let o = pool.try_get().unwrap(); //! assert_eq!(o.capacity(), 1024); -//! # } //! ``` //! //! # FAQ //! -//! ## Why is timeout configuration outside the pool? -//! -//! Many async object pool implementations allow multiple timeout settings, such as wait, create, -//! and recycle timeouts. +//! ## Why does the caller control timeouts? //! -//! This introduces two major problems: -//! -//! First, implementing timeouts inside the pool requires a timer implementation such as -//! `tokio::time`. This would prevent the pool from being runtime-agnostic. The pool could depend on -//! a timer trait, but the Rust ecosystem does not yet have a standard one. -//! -//! Second, timeout options add configuration complexity without necessarily expressing the caller's -//! actual deadline. For example, end users often care about the total time used to obtain an -//! object. This is not solely a wait, create, or recycle timeout, but a conditional composition of -//! all internal operations. -//! -//! Thus, we propose a caller-side timeout solution: +//! A timer inside the pool would couple it to a runtime or require one adapter per timer ecosystem. +//! Separate wait, create, and recycle timeouts also do not necessarily express the caller's actual +//! deadline for the complete checkout operation. Asyncband therefore returns an ordinary future so +//! the caller can apply one end-to-end deadline with its chosen timer: //! //! ```rust,ignore //! use std::sync::Arc; @@ -140,7 +129,7 @@ //! pub async fn acquire(&self) -> Result, Error> { //! const ACQUIRE_TIMEOUT: Duration = Duration::from_secs(60); //! -//! // note that users can choose any timer implementation here +//! // Callers can use the timer implementation of their runtime. //! let result = tokio::time::timeout(ACQUIRE_TIMEOUT, self.pool.get()).await; //! //! // ... processing the result @@ -148,17 +137,13 @@ //! } //! ``` //! -//! ## Why are before/after hooks outside the pool? +//! ## Why are general before/after hooks outside the pool? //! -//! Similar to the second point above, before/after hooks are hard to configure generally. Small -//! operations are easy to write in place, while passing larger blocks as closures can introduce -//! lifetime and ownership constraints. Error handling also depends on the surrounding application. +//! Before/after behavior is application policy. Small operations are clearer at the call site, +//! while larger behavior can live in the manager or an application wrapper without forcing a +//! general closure and error model into the pool. //! -//! The module provides an ordinary object-pool interface, so callers can add before/after logic in -//! the manager implementation or a wrapper. -//! -//! For example, all the "post-create", "pre-recycle", and "post-recycle" hooks can be implemented -//! as: +//! For example, create and recycle behavior can be expressed directly by [`ManageObject`]: //! //! ``` //! use asyncband::pool::ManageObject; @@ -177,12 +162,10 @@ //! //! async fn is_recyclable( //! &self, -//! _o: &mut Self::Object, -//! _status: &ObjectStatus, +//! object: &mut Self::Object, +//! status: &ObjectStatus, //! ) -> Result<(), Self::Error> { -//! // any pre-recycle hooks -//! // determine whether the object is recyclable -//! // any post-recycle hooks +//! // Validate or refresh `object`, using `status` when useful. //! Ok(()) //! } //! } @@ -192,11 +175,10 @@ pub use common::ManageObject; pub use common::ObjectStatus; pub use common::QueueStrategy; pub use common::RecycleCancelledStrategy; -pub use retain_spec::RetainResult; +pub use common::RetainResult; mod common; -mod mutex; -mod retain_spec; +mod state; pub mod bounded; pub mod unbounded; diff --git a/asyncband/src/pool/mutex.rs b/asyncband/src/pool/mutex.rs deleted file mode 100644 index 028f9d3..0000000 --- a/asyncband/src/pool/mutex.rs +++ /dev/null @@ -1,57 +0,0 @@ -// Licensed to the Apache Software Foundation (ASF) under one -// or more contributor license agreements. See the NOTICE file -// distributed with this work for additional information -// regarding copyright ownership. The ASF licenses this file -// to you 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::fmt; -use std::sync::PoisonError; - -pub(crate) struct Mutex(std::sync::Mutex); - -impl fmt::Debug for Mutex { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - self.0.fmt(f) - } -} - -impl Mutex { - pub(crate) const fn new(t: T) -> Self { - Self(std::sync::Mutex::new(t)) - } - - pub(crate) fn lock(&self) -> std::sync::MutexGuard<'_, T> { - self.0.lock().unwrap_or_else(PoisonError::into_inner) - } -} - -#[cfg(test)] -mod tests { - use std::sync::Arc; - - use super::*; - - #[test] - fn test_poison_mutex() { - let mutex = Arc::new(Mutex::new(42)); - let m = mutex.clone(); - let handle = std::thread::spawn(move || { - let _guard = m.lock(); - panic!("poison"); - }); - let _ = handle.join(); - let guard = mutex.lock(); - assert_eq!(*guard, 42); - } -} diff --git a/asyncband/src/pool/retain_spec.rs b/asyncband/src/pool/retain_spec.rs deleted file mode 100644 index f4bd08b..0000000 --- a/asyncband/src/pool/retain_spec.rs +++ /dev/null @@ -1,92 +0,0 @@ -// Licensed to the Apache Software Foundation (ASF) under one -// or more contributor license agreements. See the NOTICE file -// distributed with this work for additional information -// regarding copyright ownership. The ASF licenses this file -// to you 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::collections::VecDeque; - -use crate::pool::ObjectStatus; - -/// The result returned by `Pool::retain`. -#[derive(Debug)] -#[non_exhaustive] -pub struct RetainResult { - /// The number of retained objects. - pub retained: usize, - /// The objects removed from the pool. - pub removed: Vec, -} - -/// An internal trait that abstracts over unbounded and bounded `ObjectState`. -pub(crate) trait SealedState { - /// The type of the object. - type Object; - - /// Returns the status of the object. - fn status(&self) -> ObjectStatus; - /// Returns a mutable reference to the object. - fn mut_object(&mut self) -> &mut Self::Object; - /// Returns the owned object, consuming the state. - fn take_object(self) -> Self::Object; -} - -/// Shared `VecDeque`'s retain (`extract_if`) implementation for both bounded and unbounded pools. -pub(crate) fn do_vec_deque_retain>( - deque: &mut VecDeque, - mut f: impl FnMut(&mut T, ObjectStatus) -> bool, -) -> RetainResult { - let len = deque.len(); - let mut idx = 0; - let mut cur = 0; - - // Stage 1: All values are retained. - while cur < len { - let state = &mut deque[cur]; - let status = state.status(); - if !f(state.mut_object(), status) { - cur += 1; - break; - } - cur += 1; - idx += 1; - } - - // Stage 2: Swap retained value into current idx. - while cur < len { - let state = &mut deque[cur]; - let status = state.status(); - if !f(state.mut_object(), status) { - cur += 1; - continue; - } - - deque.swap(idx, cur); - cur += 1; - idx += 1; - } - - // Stage 3: Truncate all values after idx. - let removed = if cur != idx { - let removed = deque.split_off(idx); - removed.into_iter().map(State::take_object).collect() - } else { - Vec::new() - }; - - RetainResult { - retained: idx, - removed, - } -} diff --git a/asyncband/src/pool/state.rs b/asyncband/src/pool/state.rs new file mode 100644 index 0000000..1a57427 --- /dev/null +++ b/asyncband/src/pool/state.rs @@ -0,0 +1,136 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you 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::collections::VecDeque; + +use crate::pool::ObjectStatus; +use crate::pool::QueueStrategy; +use crate::pool::RetainResult; + +#[derive(Debug)] +pub(crate) struct ObjectState { + pub(crate) o: T, + pub(crate) status: ObjectStatus, +} + +impl ObjectState { + pub(crate) fn new(o: T) -> Self { + Self { + o, + status: ObjectStatus::default(), + } + } +} + +#[derive(Debug)] +pub(crate) struct PoolState { + idle: VecDeque>, + current_size: usize, +} + +impl PoolState { + pub(crate) const fn new() -> Self { + Self { + idle: VecDeque::new(), + current_size: 0, + } + } + + pub(crate) fn current_size(&self) -> usize { + self.current_size + } + + pub(crate) fn idle_count(&self) -> usize { + self.idle.len() + } + + pub(crate) fn pop(&mut self, strategy: QueueStrategy) -> Option> { + match strategy { + QueueStrategy::Fifo => self.idle.pop_front(), + QueueStrategy::Lifo => self.idle.pop_back(), + } + } + + pub(crate) fn add_idle(&mut self, state: ObjectState) { + self.current_size += 1; + self.idle.push_back(state); + } + + pub(crate) fn add_active(&mut self) { + self.current_size += 1; + } + + pub(crate) fn return_idle(&mut self, state: ObjectState) { + self.idle.push_back(state); + } + + pub(crate) fn detach(&mut self) { + self.current_size = self + .current_size + .checked_sub(1) + .expect("detached object must belong to the pool"); + } + + /// Retains matching idle objects without losing any object if the predicate panics. + pub(crate) fn retain( + &mut self, + mut f: impl FnMut(&mut T, ObjectStatus) -> bool, + ) -> RetainResult { + let len = self.idle.len(); + let mut retained = 0; + let mut current = 0; + + // Leave the deque untouched until the first object to remove is found. If the predicate + // panics, every object remains owned by the pool. + while current < len { + let state = &mut self.idle[current]; + if !f(&mut state.o, state.status) { + current += 1; + break; + } + current += 1; + retained += 1; + } + + // Compact retained objects in place. A panic may change their order, but no object is + // removed until every predicate call has completed. + while current < len { + let state = &mut self.idle[current]; + if !f(&mut state.o, state.status) { + current += 1; + continue; + } + + self.idle.swap(retained, current); + current += 1; + retained += 1; + } + + let removed = if current == retained { + Vec::new() + } else { + self.idle + .split_off(retained) + .into_iter() + .map(|state| state.o) + .collect::>() + }; + self.current_size -= removed.len(); + + RetainResult { retained, removed } + } +} diff --git a/asyncband/src/pool/unbounded.rs b/asyncband/src/pool/unbounded.rs index 7e59f22..64f7525 100644 --- a/asyncband/src/pool/unbounded.rs +++ b/asyncband/src/pool/unbounded.rs @@ -17,8 +17,8 @@ //! Unbounded object pools. //! -//! An unbounded pool, on the other hand, allows you to put objects to the pool manually. You can -//! use it like Go's [`sync.Pool`](https://pkg.go.dev/sync#Pool). +//! An unbounded pool accepts objects supplied by callers and can be used like Go's +//! [`sync.Pool`](https://pkg.go.dev/sync#Pool). //! //! To configure a factory for creating objects when the pool is empty, like `sync.Pool`'s `New`, //! you can create the unbounded pool via [`Pool::new`](Pool::new) with an @@ -26,30 +26,23 @@ //! //! ## Examples //! -//! Read the following simple demos or more complex examples in the examples directory. -//! -//! 1. Create an unbounded pool with [`NeverManageObject`]: +//! 1. Create a manually populated pool with [`NeverManageObject`]: //! //! ``` //! use asyncband::pool::unbounded::Pool; //! use asyncband::pool::unbounded::PoolConfig; //! -//! # #[tokio::main] -//! # async fn main() { //! let pool = Pool::>::never_manage(PoolConfig::default()); //! -//! let result = pool.get().await; -//! assert_eq!(result.unwrap_err().to_string(), "unbounded pool is empty"); +//! assert!(pool.try_get().is_none()); //! //! pool.extend_one(Vec::with_capacity(1024)); -//! let o = pool.get().await.unwrap(); +//! let o = pool.try_get().unwrap(); //! assert_eq!(o.capacity(), 1024); //! drop(o); -//! let o = pool.get().await.unwrap(); +//! let o = pool.try_get().unwrap(); //! assert_eq!(o.capacity(), 1024); -//! let result = pool.get().await; -//! assert_eq!(result.unwrap_err().to_string(), "unbounded pool is empty"); -//! # } +//! assert!(pool.try_get().is_none()); //! ``` //! //! 2. Create an unbounded pool with a custom [`ManageObject`] (object factory): @@ -78,8 +71,8 @@ //! //! async fn is_recyclable( //! &self, -//! o: &mut Self::Object, -//! status: &ObjectStatus, +//! _object: &mut Self::Object, +//! _status: &ObjectStatus, //! ) -> Result<(), Self::Error> { //! Ok(()) //! } @@ -93,20 +86,21 @@ //! # } //! ``` -use std::collections::VecDeque; use std::future::Future; +use std::marker::PhantomData; use std::ops::Deref; use std::ops::DerefMut; use std::sync::Arc; use std::sync::Weak; +use crate::internal::mutex::Mutex; use crate::pool::ManageObject; use crate::pool::ObjectStatus; use crate::pool::QueueStrategy; use crate::pool::RecycleCancelledStrategy; use crate::pool::RetainResult; -use crate::pool::mutex::Mutex; -use crate::pool::retain_spec; +use crate::pool::state::ObjectState; +use crate::pool::state::PoolState; /// The configuration of [`Pool`]. #[derive(Clone, Copy, Debug)] @@ -117,7 +111,7 @@ pub struct PoolConfig { /// Determines the order of objects being queued and dequeued. pub queue_strategy: QueueStrategy, - /// Strategy when recycling object has been cancelled. + /// Strategy to apply when object recycling is cancelled. pub recycle_cancelled_strategy: RecycleCancelledStrategy, } @@ -158,28 +152,40 @@ impl PoolConfig { #[derive(Clone, Copy, Debug)] #[non_exhaustive] pub struct PoolStatus { - /// The current size of the pool. + /// The number of objects that have not been detached. pub current_size: usize, - /// The number of idle objects in the pool. + /// The number of objects currently available for checkout. pub idle_count: usize, } -/// The default [`ManageObject`] implementation for unbounded pool. +/// The [`ManageObject`] implementation used by manually populated unbounded pools. /// -/// * [`NeverManageObject::create`] always returns [`PoolIsEmpty`] so that [`Pool::get`] would get -/// the error if no object is in the pool. -/// * [`NeverManageObject::is_recyclable`] always returns `Ok(())` so that any object is always -/// recyclable. -#[derive(Debug, Copy, Clone)] -pub struct NeverManageObject { - _marker: std::marker::PhantomData, +/// [`NeverManageObject::create`] returns [`PoolIsEmpty`] and +/// [`NeverManageObject::is_recyclable`] accepts every object. Prefer the synchronous +/// [`Pool::try_get`] method when the pool has no factory. +pub struct NeverManageObject { + _marker: PhantomData T>, } -impl Default for NeverManageObject { +impl Copy for NeverManageObject {} + +impl Clone for NeverManageObject { + fn clone(&self) -> Self { + *self + } +} + +impl std::fmt::Debug for NeverManageObject { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str("NeverManageObject") + } +} + +impl Default for NeverManageObject { fn default() -> Self { Self { - _marker: std::marker::PhantomData, + _marker: PhantomData, } } } @@ -201,7 +207,7 @@ impl std::fmt::Display for PoolIsEmpty { impl std::error::Error for PoolIsEmpty {} -impl ManageObject for NeverManageObject { +impl ManageObject for NeverManageObject { type Object = T; type Error = PoolIsEmpty; @@ -225,14 +231,8 @@ pub struct Pool = NeverManageObject> { config: PoolConfig, manager: M, - /// A deque that holds the objects. - slots: Mutex>>, -} - -#[derive(Debug)] -struct PoolDeque { - deque: VecDeque, - current_size: usize, + /// The objects tracked by the pool. + slots: Mutex>, } impl std::fmt::Debug for Pool @@ -249,17 +249,28 @@ where } // Methods for `Pool` with `NeverManageObject`. -impl Pool { - /// Creates a new [`Pool`] from config and the [`NeverManageObject`]. +impl Pool { + /// Creates a manually populated [`Pool`] with no object factory. pub fn never_manage(config: PoolConfig) -> Arc { Self::new(config, NeverManageObject::::default()) } + /// Retrieves an idle [`Object`] without waiting or creating a new one. + /// + /// This method only exists for [`NeverManageObject`] pools. It returns `None` when the pool is + /// empty. + pub fn try_get(self: &Arc) -> Option> { + let mut state = self.slots.lock().pop(self.config.queue_strategy)?; + state.status.mark_recycled(); + Some(Object { + state: Some(state), + pool: Arc::downgrade(self), + }) + } + /// Retrieves an [`Object`] from this [`Pool`], or creates a new one with the passed-in async /// closure, if the pool is empty. /// - /// This method should be called with a pool wrapped in an [`Arc`]. - /// /// This method only exists for [`NeverManageObject`] pools. If you provide a custom /// [`ManageObject`] implementation, you should use [`Pool::get`] instead, and it will call /// [`ManageObject::create`] to create a new object if the pool is empty. @@ -267,43 +278,24 @@ impl Pool { where F: AsyncFnOnce() -> Result + Send, { - let existing = match self.config.queue_strategy { - QueueStrategy::Fifo => self.slots.lock().deque.pop_front(), - QueueStrategy::Lifo => self.slots.lock().deque.pop_back(), - }; - - match existing { - None => { - let object = f().await?; - let state = ObjectState { - o: object, - status: ObjectStatus::default(), - }; - self.slots.lock().current_size += 1; - Ok(Object { - state: Some(state), - pool: Arc::downgrade(self), - }) - } - Some(mut state) => { - state.status.recycle_count += 1; - state.status.recycled = Some(std::time::Instant::now()); - Ok(Object { - state: Some(state), - pool: Arc::downgrade(self), - }) - } + if let Some(object) = self.try_get() { + return Ok(object); } + + let object = f().await?; + let state = ObjectState::new(object); + self.slots.lock().add_active(); + Ok(Object { + state: Some(state), + pool: Arc::downgrade(self), + }) } } impl> Pool { /// Creates a new [`Pool`] with config and the specified [`ManageObject`]. pub fn new(config: PoolConfig, manager: M) -> Arc { - let slots = Mutex::new(PoolDeque { - deque: VecDeque::new(), - current_size: 0, - }); + let slots = Mutex::new(PoolState::new()); Arc::new(Self { config, @@ -314,22 +306,16 @@ impl> Pool { /// Retrieves an [`Object`] from this [`Pool`]. /// - /// This method should be called with a pool wrapped in an [`Arc`]. + /// If no idle object is available, this method calls [`ManageObject::create`]. pub async fn get(self: &Arc) -> Result, M::Error> { let object = loop { - let existing = match self.config.queue_strategy { - QueueStrategy::Fifo => self.slots.lock().deque.pop_front(), - QueueStrategy::Lifo => self.slots.lock().deque.pop_back(), - }; + let existing = self.slots.lock().pop(self.config.queue_strategy); match existing { None => { let object = self.manager.create().await?; - let state = ObjectState { - o: object, - status: ObjectStatus::default(), - }; - self.slots.lock().current_size += 1; + let state = ObjectState::new(object); + self.slots.lock().add_active(); break Object { state: Some(state), pool: Arc::downgrade(self), @@ -350,8 +336,7 @@ impl> Pool { .await .is_ok() { - state.status.recycle_count += 1; - state.status.recycled = Some(std::time::Instant::now()); + state.status.mark_recycled(); break unready_object.ready(); } else { // We need to manually detach here as the drop implementation @@ -402,17 +387,14 @@ impl> Pool { pub fn extend(&self, iter: impl IntoIterator) { let mut slots = self.slots.lock(); for o in iter { - slots.current_size += 1; - slots.deque.push_back(ObjectState { - o, - status: ObjectStatus::default(), - }); + slots.add_idle(ObjectState::new(o)); } } /// Retains only the objects that pass the given predicate. /// - /// This function blocks the entire pool. Therefore, the given function should not block. + /// The predicate runs while the idle-object lock is held and therefore must not block or call + /// back into the pool. Detachment hooks for removed objects run after the lock is released. /// /// The following example starts a background task that runs every 30 seconds and removes /// objects from the pool that have not been used for more than one minute. The task will @@ -438,40 +420,39 @@ impl> Pool { &self, f: impl FnMut(&mut M::Object, ObjectStatus) -> bool, ) -> RetainResult { - let mut slots = self.slots.lock(); - let result = retain_spec::do_vec_deque_retain(&mut slots.deque, f); - slots.current_size -= result.removed.len(); + let mut result = { + let mut slots = self.slots.lock(); + slots.retain(f) + }; + for object in &mut result.removed { + self.manager.on_detached(object); + } result } - /// Returns the current status of the pool. - /// - /// The status returned by the pool is not guaranteed to be consistent. - /// - /// Although this status provides [eventual consistency], the numbers can be - /// temporarily inaccurate under heavy load. They are intended as an overall insight. - /// - /// [eventual consistency]: (https://en.wikipedia.org/wiki/Eventual_consistency) + /// Returns a consistent snapshot of the objects currently tracked by the pool. pub fn status(&self) -> PoolStatus { let slots = self.slots.lock(); - let (current_size, idle_count) = (slots.current_size, slots.deque.len()); - drop(slots); PoolStatus { - current_size, - idle_count, + current_size: slots.current_size(), + idle_count: slots.idle_count(), } } - fn push_back(&self, o: ObjectState) { + fn return_object(&self, mut state: ObjectState) { + state.status.mark_returned(); + self.restore_idle(state); + } + + fn restore_idle(&self, state: ObjectState) { let mut slots = self.slots.lock(); - slots.deque.push_back(o); - drop(slots); + slots.return_idle(state); } fn detach_object(&self, o: &mut T) { let mut slots = self.slots.lock(); - slots.current_size -= 1; + slots.detach(); drop(slots); self.manager.on_detached(o); } @@ -504,7 +485,7 @@ impl> Drop for Object { fn drop(&mut self) { if let Some(state) = self.state.take() { if let Some(pool) = self.pool.upgrade() { - pool.push_back(state); + pool.return_object(state); } } } @@ -578,7 +559,7 @@ impl> Drop for UnreadyObject { pool.detach_object(&mut state.o); } RecycleCancelledStrategy::ReturnToPool => { - pool.push_back(state); + pool.restore_idle(state); } } } @@ -607,25 +588,3 @@ impl> UnreadyObject { self.state.as_mut().unwrap() } } - -#[derive(Debug)] -struct ObjectState { - o: T, - status: ObjectStatus, -} - -impl retain_spec::SealedState for ObjectState { - type Object = T; - - fn status(&self) -> ObjectStatus { - self.status - } - - fn mut_object(&mut self) -> &mut Self::Object { - &mut self.o - } - - fn take_object(self) -> Self::Object { - self.o - } -} diff --git a/asyncband/src/semaphore/mod.rs b/asyncband/src/semaphore/mod.rs index c812f51..b6fe842 100644 --- a/asyncband/src/semaphore/mod.rs +++ b/asyncband/src/semaphore/mod.rs @@ -310,6 +310,15 @@ impl Semaphore { } } + #[cfg(feature = "pool")] + pub(crate) fn try_acquire_up_to_owned( + self: Arc, + up_to: usize, + ) -> Option { + let permits = self.s.drain_permits(up_to); + (permits != 0).then_some(OwnedSemaphorePermit { sem: self, permits }) + } + /// Acquires `n` permits from the semaphore. /// /// The semaphore must be wrapped in an [`Arc`] to call this method. @@ -509,6 +518,16 @@ pub struct OwnedSemaphorePermit { } impl OwnedSemaphorePermit { + #[cfg(feature = "pool")] + pub(crate) fn release(&mut self, permits: usize) { + assert!( + permits <= self.permits, + "cannot release more permits than this permit holds" + ); + self.permits -= permits; + self.sem.release(permits); + } + /// Forgets the permit **without** releasing it back to the semaphore. /// /// This can be used to permanently reduce the number of permits available diff --git a/tests-integration/tests/pool_behavior_test.rs b/tests-integration/tests/pool_behavior_test.rs new file mode 100644 index 0000000..864ce28 --- /dev/null +++ b/tests-integration/tests/pool_behavior_test.rs @@ -0,0 +1,171 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you 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::convert::Infallible; +use std::panic::AssertUnwindSafe; +use std::panic::catch_unwind; +use std::sync::Arc; +use std::sync::atomic::AtomicUsize; +use std::sync::atomic::Ordering; +use std::time::Instant; + +use asyncband::pool::ManageObject; +use asyncband::pool::ObjectStatus; +use asyncband::pool::QueueStrategy; +use asyncband::pool::bounded; +use asyncband::pool::unbounded; + +struct CountingManager { + next: Arc, + detached: Arc, +} + +impl ManageObject for CountingManager { + type Object = usize; + type Error = Infallible; + + async fn create(&self) -> Result { + Ok(self.next.fetch_add(1, Ordering::Relaxed)) + } + + async fn is_recyclable( + &self, + _object: &mut Self::Object, + _status: &ObjectStatus, + ) -> Result<(), Self::Error> { + Ok(()) + } + + fn on_detached(&self, object: &mut Self::Object) { + self.detached.fetch_add(1, Ordering::Relaxed); + *object += 1000; + } +} + +#[test] +fn bounded_construction_allocates_idle_storage_lazily() { + let pool = bounded::Pool::new( + bounded::PoolConfig::new(usize::MAX), + CountingManager { + next: Arc::new(AtomicUsize::new(0)), + detached: Arc::new(AtomicUsize::new(0)), + }, + ); + + assert_eq!(pool.status().max_size, usize::MAX); + assert_eq!(pool.status().current_size, 0); + assert_eq!(pool.status().idle_count, 0); +} + +#[tokio::test] +async fn bounded_last_used_tracks_the_end_of_a_checkout() { + let pool = bounded::Pool::new( + bounded::PoolConfig::new(1), + CountingManager { + next: Arc::new(AtomicUsize::new(0)), + detached: Arc::new(AtomicUsize::new(0)), + }, + ); + + let object = pool.get().await.unwrap(); + let before_return = Instant::now(); + drop(object); + + let object = pool.get().await.unwrap(); + assert!(object.status().last_used() >= before_return); + assert_eq!(object.status().recycle_count(), 1); +} + +#[tokio::test] +async fn retain_invokes_detachment_hook_once_per_removed_object() { + let detached = Arc::new(AtomicUsize::new(0)); + let pool = bounded::Pool::new( + bounded::PoolConfig::new(4), + CountingManager { + next: Arc::new(AtomicUsize::new(0)), + detached: detached.clone(), + }, + ); + + let mut objects = Vec::new(); + for _ in 0..4 { + objects.push(pool.get().await.unwrap()); + } + drop(objects); + + let mut result = pool.retain(|object, _status| *object % 2 == 0); + result.removed.sort_unstable(); + + assert_eq!(result.retained, 2); + assert_eq!(result.removed, [1001, 1003]); + assert_eq!(detached.load(Ordering::Relaxed), 2); + assert_eq!(pool.status().current_size, 2); + assert_eq!(pool.status().idle_count, 2); +} + +#[test] +fn manual_pool_try_get_tracks_return_time() { + let pool = unbounded::Pool::::never_manage(unbounded::PoolConfig::default()); + assert!(pool.try_get().is_none()); + + pool.extend_one(42); + let object = pool.try_get().unwrap(); + assert_eq!(object.status().recycle_count(), 1); + + let before_return = Instant::now(); + drop(object); + + let object = pool.try_get().unwrap(); + assert!(object.status().last_used() >= before_return); + assert_eq!(object.status().recycle_count(), 2); +} + +#[test] +fn manual_pool_honors_fifo_and_lifo_order() { + fn drain(strategy: QueueStrategy) -> Vec { + let config = unbounded::PoolConfig::new().with_queue_strategy(strategy); + let pool = unbounded::Pool::::never_manage(config); + pool.extend([1, 2, 3]); + + (0..3).map(|_| pool.try_get().unwrap().detach()).collect() + } + + assert_eq!(drain(QueueStrategy::Fifo), [1, 2, 3]); + assert_eq!(drain(QueueStrategy::Lifo), [3, 2, 1]); +} + +#[test] +fn retain_predicate_panic_preserves_pool_ownership() { + let pool = unbounded::Pool::::never_manage(unbounded::PoolConfig::default()); + pool.extend([1, 2, 3, 4]); + + let result = catch_unwind(AssertUnwindSafe(|| { + pool.retain(|object, _status| { + assert_ne!(*object, 3, "predicate panic"); + *object % 2 == 0 + }); + })); + assert!(result.is_err()); + assert_eq!(pool.status().current_size, 4); + assert_eq!(pool.status().idle_count, 4); + + let mut objects = (0..4) + .map(|_| pool.try_get().unwrap().detach()) + .collect::>(); + objects.sort_unstable(); + assert_eq!(objects, [1, 2, 3, 4]); +} diff --git a/tests-integration/tests/pool_recycle_cancelled_test.rs b/tests-integration/tests/pool_recycle_cancelled_test.rs index eeeac3a..1bc8c60 100644 --- a/tests-integration/tests/pool_recycle_cancelled_test.rs +++ b/tests-integration/tests/pool_recycle_cancelled_test.rs @@ -15,235 +15,164 @@ // specific language governing permissions and limitations // under the License. -use std::convert::Infallible; +use std::future::Future; +use std::future::poll_fn; +use std::pin::pin; use std::sync::Arc; +use std::sync::atomic::AtomicBool; use std::sync::atomic::AtomicUsize; use std::sync::atomic::Ordering; -use std::time::Duration; +use std::task::Poll; use asyncband::pool::ManageObject; use asyncband::pool::ObjectStatus; use asyncband::pool::RecycleCancelledStrategy; -struct SlowRecycleManager { - created_count: Arc, - recycle_delay: Duration, +#[derive(Default)] +struct Controls { + created: AtomicUsize, + recycle_ready: AtomicBool, + reject_recycle: AtomicBool, } -impl SlowRecycleManager { - fn new(created_count: Arc, recycle_delay: Duration) -> Self { - Self { - created_count, - recycle_delay, - } - } +struct ControlledRecycleManager { + controls: Arc, } -impl ManageObject for SlowRecycleManager { +impl ManageObject for ControlledRecycleManager { type Object = usize; - type Error = Infallible; + type Error = (); async fn create(&self) -> Result { - let id = self.created_count.fetch_add(1, Ordering::SeqCst); - Ok(id) + Ok(self.controls.created.fetch_add(1, Ordering::Relaxed)) } async fn is_recyclable( &self, - _o: &mut Self::Object, + _object: &mut Self::Object, _status: &ObjectStatus, ) -> Result<(), Self::Error> { - tokio::time::sleep(self.recycle_delay).await; - Ok(()) + poll_fn(|_| { + if !self.controls.recycle_ready.load(Ordering::Acquire) { + Poll::Pending + } else if self.controls.reject_recycle.load(Ordering::Relaxed) { + Poll::Ready(Err(())) + } else { + Poll::Ready(Ok(())) + } + }) + .await } } +fn manager() -> (ControlledRecycleManager, Arc) { + let controls = Arc::new(Controls::default()); + ( + ControlledRecycleManager { + controls: controls.clone(), + }, + controls, + ) +} + +fn poll_and_cancel(future: impl Future) { + let mut future = pin!(future); + assert!(tests_integration::poll_once(future.as_mut()).is_pending()); +} + mod bounded_tests { use asyncband::pool::bounded::Pool; use asyncband::pool::bounded::PoolConfig; use super::*; - /// Test default behavior (Detach): cancelled get() calls detach objects from the pool. #[tokio::test] - async fn test_default_detach_behavior() { - const MAX_SIZE: usize = 1; - let created_count = Arc::new(AtomicUsize::new(0)); - let manager = SlowRecycleManager::new(created_count.clone(), Duration::from_millis(100)); - let pool = Pool::new(PoolConfig::new(MAX_SIZE), manager); - - let obj = pool.get().await.unwrap(); - assert_eq!(*obj, 0); - assert_eq!(pool.status().current_size, 1); - - drop(obj); - assert_eq!(pool.status().current_size, 1); - assert_eq!(pool.status().idle_count, 1); - - let timeout_result = tokio::time::timeout(Duration::from_millis(10), pool.get()).await; - assert!(timeout_result.is_err(), "Should have timed out"); + async fn cancellation_detaches_by_default() { + let (manager, controls) = manager(); + let pool = Pool::new(PoolConfig::new(1), manager); - tokio::time::sleep(Duration::from_millis(10)).await; + let object = pool.get().await.unwrap(); + assert_eq!(*object, 0); + drop(object); - let status = pool.status(); - assert_eq!( - status.current_size, 0, - "Pool size should be 0 after cancelled get() with Detach behavior" - ); + poll_and_cancel(pool.get()); + assert_eq!(pool.status().current_size, 0); + assert_eq!(pool.status().idle_count, 0); - let obj = pool.get().await.unwrap(); - assert_eq!(*obj, 1, "Should be a new object (id=1)"); - assert_eq!( - created_count.load(Ordering::SeqCst), - 2, - "Two objects should have been created" - ); + let object = pool.get().await.unwrap(); + assert_eq!(*object, 1); + assert_eq!(controls.created.load(Ordering::Relaxed), 2); } - /// Test ReturnToPool behavior: cancelled get() calls return objects to the pool. #[tokio::test] - async fn test_return_to_pool_behavior() { - const MAX_SIZE: usize = 1; - let created_count = Arc::new(AtomicUsize::new(0)); - let manager = SlowRecycleManager::new(created_count.clone(), Duration::from_millis(100)); - let config = PoolConfig::new(MAX_SIZE) + async fn cancellation_can_restore_the_idle_object() { + let (manager, controls) = manager(); + let config = PoolConfig::new(1) .with_recycle_cancelled_strategy(RecycleCancelledStrategy::ReturnToPool); let pool = Pool::new(config, manager); - let obj = pool.get().await.unwrap(); - assert_eq!(*obj, 0); - assert_eq!(pool.status().current_size, 1); - assert_eq!(created_count.load(Ordering::SeqCst), 1); + let object = pool.get().await.unwrap(); + drop(object); + let mut last_used_before = None; + pool.retain(|_, status| { + last_used_before = Some(status.last_used()); + true + }); - drop(obj); + poll_and_cancel(pool.get()); assert_eq!(pool.status().current_size, 1); assert_eq!(pool.status().idle_count, 1); - let timeout_result = tokio::time::timeout(Duration::from_millis(10), pool.get()).await; - assert!(timeout_result.is_err(), "Should have timed out"); - - tokio::time::sleep(Duration::from_millis(10)).await; - - let status = pool.status(); - assert_eq!( - status.current_size, 1, - "Pool size should be preserved after cancelled get() with ReturnToPool behavior" - ); - assert_eq!( - status.idle_count, 1, - "Object should be back in idle state after cancelled get()" - ); - - assert_eq!( - created_count.load(Ordering::SeqCst), - 1, - "No extra objects should be created" - ); - - let obj = pool.get().await.unwrap(); - assert_eq!(*obj, 0, "Should get the same object back"); + let mut last_used_after = None; + pool.retain(|_, status| { + last_used_after = Some(status.last_used()); + true + }); + assert_eq!(last_used_after, last_used_before); + + controls.recycle_ready.store(true, Ordering::Release); + let object = pool.get().await.unwrap(); + assert_eq!(*object, 0); + assert_eq!(controls.created.load(Ordering::Relaxed), 1); } - /// Test that multiple cancelled get() calls with ReturnToPool don't shrink the pool. #[tokio::test] - async fn test_multiple_cancelled_gets_with_return_to_pool() { - const MAX_SIZE: usize = 3; - let created_count = Arc::new(AtomicUsize::new(0)); - let manager = SlowRecycleManager::new(created_count.clone(), Duration::from_millis(100)); - let config = PoolConfig::new(MAX_SIZE) + async fn repeated_cancellation_does_not_shrink_a_restoring_pool() { + let (manager, _) = manager(); + let config = PoolConfig::new(3) .with_recycle_cancelled_strategy(RecycleCancelledStrategy::ReturnToPool); let pool = Pool::new(config, manager); - let obj1 = pool.get().await.unwrap(); - let obj2 = pool.get().await.unwrap(); - let obj3 = pool.get().await.unwrap(); - - drop((obj1, obj2, obj3)); - assert_eq!(pool.status().current_size, 3); - assert_eq!(pool.status().idle_count, 3); + let objects = [ + pool.get().await.unwrap(), + pool.get().await.unwrap(), + pool.get().await.unwrap(), + ]; + drop(objects); for _ in 0..5 { - let _ = tokio::time::timeout(Duration::from_millis(10), pool.get()).await; - tokio::time::sleep(Duration::from_millis(5)).await; + poll_and_cancel(pool.get()); } - - let status = pool.status(); - assert_eq!( - status.current_size, 3, - "Pool size should be preserved after multiple cancelled gets" - ); + assert_eq!(pool.status().current_size, 3); + assert_eq!(pool.status().idle_count, 3); } - /// Test that cancelling `get()` releases its user-count registration. #[tokio::test] - async fn test_cancelled_get_releases_user_count() { - let created_count = Arc::new(AtomicUsize::new(0)); - let manager = SlowRecycleManager::new(created_count, Duration::from_millis(100)); - let pool = Pool::new(PoolConfig::new(1), manager); + async fn rejected_recycle_detaches_even_when_cancellation_would_restore() { + let (manager, controls) = manager(); + let config = PoolConfig::new(1) + .with_recycle_cancelled_strategy(RecycleCancelledStrategy::ReturnToPool); + let pool = Pool::new(config, manager); let object = pool.get().await.unwrap(); drop(object); + controls.reject_recycle.store(true, Ordering::Relaxed); + controls.recycle_ready.store(true, Ordering::Release); - let mut get = Box::pin(pool.get()); - assert!(tests_integration::poll_once(get.as_mut()).is_pending()); - drop(get); - - assert_eq!(pool.status().wait_count, 0); - } - - /// Test that a failed create call releases its user-count registration. - #[tokio::test] - async fn test_failed_create_releases_user_count() { - struct FailingManager; - - impl ManageObject for FailingManager { - type Object = (); - type Error = (); - - async fn create(&self) -> Result { - Err(()) - } - - async fn is_recyclable( - &self, - _o: &mut Self::Object, - _status: &ObjectStatus, - ) -> Result<(), Self::Error> { - Ok(()) - } - } - - let pool = Pool::new(PoolConfig::new(1), FailingManager); - assert!(pool.get().await.is_err()); - - let status = pool.status(); - assert_eq!(status.current_size, 0); - assert_eq!(status.wait_count, 0); - } - - /// Test that failed is_recyclable always properly detaches objects. - #[tokio::test] - async fn test_failed_recyclable_still_detaches() { - const MAX_SIZE: usize = 1; - let created_count = Arc::new(AtomicUsize::new(0)); - let manager = SlowRecycleManager::new(created_count.clone(), Duration::from_millis(10)); - let config = PoolConfig::new(MAX_SIZE) - .with_recycle_cancelled_strategy(RecycleCancelledStrategy::ReturnToPool); - let pool = Pool::new(config, manager); - - let obj = pool.get().await.unwrap(); - assert_eq!(*obj, 0); - drop(obj); + let object = pool.get().await.unwrap(); + assert_eq!(*object, 1); + assert_eq!(controls.created.load(Ordering::Relaxed), 2); assert_eq!(pool.status().current_size, 1); - assert_eq!(pool.status().idle_count, 1); - - let obj = pool.get().await.unwrap(); - assert_eq!(*obj, 0, "Should get the recycled object"); - assert_eq!( - obj.status().recycle_count(), - 1, - "Should have been recycled once" - ); } } @@ -253,106 +182,62 @@ mod unbounded_tests { use super::*; - /// Test default behavior (Detach): cancelled get() calls detach objects from the unbounded - /// pool. #[tokio::test] - async fn test_default_detach_behavior() { - let created_count = Arc::new(AtomicUsize::new(0)); - let manager = SlowRecycleManager::new(created_count.clone(), Duration::from_millis(100)); + async fn cancellation_detaches_by_default() { + let (manager, controls) = manager(); let pool = Pool::new(PoolConfig::default(), manager); - let obj = pool.get().await.unwrap(); - assert_eq!(*obj, 0); - assert_eq!(pool.status().current_size, 1); - - drop(obj); - assert_eq!(pool.status().current_size, 1); - assert_eq!(pool.status().idle_count, 1); - - let timeout_result = tokio::time::timeout(Duration::from_millis(10), pool.get()).await; - assert!(timeout_result.is_err(), "Should have timed out"); - - tokio::time::sleep(Duration::from_millis(10)).await; + let object = pool.get().await.unwrap(); + assert_eq!(*object, 0); + drop(object); - let status = pool.status(); - assert_eq!( - status.current_size, 0, - "Pool size should be 0 after cancelled get() with Detach behavior" - ); + poll_and_cancel(pool.get()); + assert_eq!(pool.status().current_size, 0); + assert_eq!(pool.status().idle_count, 0); - let obj = pool.get().await.unwrap(); - assert_eq!(*obj, 1, "Should be a new object (id=1)"); - assert_eq!( - created_count.load(Ordering::SeqCst), - 2, - "Two objects should have been created" - ); + let object = pool.get().await.unwrap(); + assert_eq!(*object, 1); + assert_eq!(controls.created.load(Ordering::Relaxed), 2); } - /// Test ReturnToPool behavior: cancelled get() calls return objects to the unbounded pool. #[tokio::test] - async fn test_return_to_pool_behavior() { - let created_count = Arc::new(AtomicUsize::new(0)); - let manager = SlowRecycleManager::new(created_count.clone(), Duration::from_millis(100)); + async fn cancellation_can_restore_the_idle_object() { + let (manager, controls) = manager(); let config = PoolConfig::new() .with_recycle_cancelled_strategy(RecycleCancelledStrategy::ReturnToPool); let pool = Pool::new(config, manager); - let obj = pool.get().await.unwrap(); - assert_eq!(*obj, 0); - assert_eq!(pool.status().current_size, 1); - assert_eq!(created_count.load(Ordering::SeqCst), 1); + let object = pool.get().await.unwrap(); + drop(object); + poll_and_cancel(pool.get()); - drop(obj); assert_eq!(pool.status().current_size, 1); assert_eq!(pool.status().idle_count, 1); + controls.recycle_ready.store(true, Ordering::Release); - let timeout_result = tokio::time::timeout(Duration::from_millis(10), pool.get()).await; - assert!(timeout_result.is_err(), "Should have timed out"); - - tokio::time::sleep(Duration::from_millis(10)).await; - - let status = pool.status(); - assert_eq!( - status.current_size, 1, - "Pool size should be preserved after cancelled get() with ReturnToPool behavior" - ); - assert_eq!( - status.idle_count, 1, - "Object should be back in idle state after cancelled get()" - ); - - // Verify we can still get the same object - let obj = pool.get().await.unwrap(); - assert_eq!(*obj, 0, "Should get the same object back"); + let object = pool.get().await.unwrap(); + assert_eq!(*object, 0); + assert_eq!(controls.created.load(Ordering::Relaxed), 1); } - /// Test that multiple cancelled get() calls with ReturnToPool don't shrink the unbounded pool. #[tokio::test] - async fn test_multiple_cancelled_gets_with_return_to_pool() { - let created_count = Arc::new(AtomicUsize::new(0)); - let manager = SlowRecycleManager::new(created_count.clone(), Duration::from_millis(100)); + async fn repeated_cancellation_does_not_shrink_a_restoring_pool() { + let (manager, _) = manager(); let config = PoolConfig::new() .with_recycle_cancelled_strategy(RecycleCancelledStrategy::ReturnToPool); let pool = Pool::new(config, manager); - let obj1 = pool.get().await.unwrap(); - let obj2 = pool.get().await.unwrap(); - let obj3 = pool.get().await.unwrap(); - - drop((obj1, obj2, obj3)); - assert_eq!(pool.status().current_size, 3); - assert_eq!(pool.status().idle_count, 3); + let objects = [ + pool.get().await.unwrap(), + pool.get().await.unwrap(), + pool.get().await.unwrap(), + ]; + drop(objects); for _ in 0..5 { - let _ = tokio::time::timeout(Duration::from_millis(10), pool.get()).await; - tokio::time::sleep(Duration::from_millis(5)).await; + poll_and_cancel(pool.get()); } - - let status = pool.status(); - assert_eq!( - status.current_size, 3, - "Pool size should be preserved after multiple cancelled gets" - ); + assert_eq!(pool.status().current_size, 3); + assert_eq!(pool.status().idle_count, 3); } } diff --git a/tests-integration/tests/pool_replenish_test.rs b/tests-integration/tests/pool_replenish_test.rs index 7f0535e..08c8e22 100644 --- a/tests-integration/tests/pool_replenish_test.rs +++ b/tests-integration/tests/pool_replenish_test.rs @@ -16,7 +16,12 @@ // under the License. use std::convert::Infallible; +use std::future::poll_fn; use std::sync::Arc; +use std::sync::atomic::AtomicBool; +use std::sync::atomic::AtomicUsize; +use std::sync::atomic::Ordering; +use std::task::Poll; use asyncband::pool::ManageObject; use asyncband::pool::ObjectStatus; @@ -24,7 +29,7 @@ use asyncband::pool::bounded::Pool; use asyncband::pool::bounded::PoolConfig; #[tokio::test] -async fn test_replenish() { +async fn test_replenish_to() { #[derive(Default)] struct Manager; @@ -53,7 +58,7 @@ async fn test_replenish() { for i in 0..5 { let pool = make_default(); - let n = pool.replenish(i).await; + let n = pool.replenish_to(i).await.unwrap(); assert_eq!(n, i.min(MAX_SIZE)); } @@ -61,7 +66,7 @@ async fn test_replenish() { { let pool = make_default(); pool.get().await.unwrap(); - let n = pool.replenish(2).await; + let n = pool.replenish_to(2).await.unwrap(); assert_eq!(n, 1); } @@ -72,7 +77,126 @@ async fn test_replenish() { let o2 = pool.get().await.unwrap(); drop((o1, o2)); - let n = pool.replenish(2).await; + let n = pool.replenish_to(2).await.unwrap(); assert_eq!(n, 0); } } + +#[derive(Debug, PartialEq, Eq)] +struct CreateError; + +struct FailingManager { + calls: Arc, +} + +impl ManageObject for FailingManager { + type Object = usize; + type Error = CreateError; + + async fn create(&self) -> Result { + let call = self.calls.fetch_add(1, Ordering::Relaxed); + if call == 1 { + Err(CreateError) + } else { + Ok(call) + } + } + + async fn is_recyclable( + &self, + _object: &mut Self::Object, + _status: &ObjectStatus, + ) -> Result<(), Self::Error> { + Ok(()) + } +} + +#[tokio::test] +async fn replenish_to_reports_create_errors_and_keeps_prior_objects() { + let pool = Pool::new( + PoolConfig::new(2), + FailingManager { + calls: Arc::new(AtomicUsize::new(0)), + }, + ); + + assert_eq!(pool.replenish_to(2).await, Err(CreateError)); + assert_eq!(pool.status().current_size, 1); + assert_eq!(pool.status().idle_count, 1); + + assert_eq!(pool.replenish_to(2).await, Ok(1)); + assert_eq!(pool.status().current_size, 2); + assert_eq!(pool.status().idle_count, 2); +} + +struct ControlledManager { + calls: Arc, + allow_create: Arc, +} + +impl ManageObject for ControlledManager { + type Object = usize; + type Error = Infallible; + + async fn create(&self) -> Result { + let call = self.calls.fetch_add(1, Ordering::Relaxed); + if call != 0 { + poll_fn(|_| { + if self.allow_create.load(Ordering::Acquire) { + Poll::Ready(()) + } else { + Poll::Pending + } + }) + .await; + } + Ok(call) + } + + async fn is_recyclable( + &self, + _object: &mut Self::Object, + _status: &ObjectStatus, + ) -> Result<(), Self::Error> { + Ok(()) + } +} + +#[tokio::test] +async fn concurrent_get_and_replenish_to_respect_capacity() { + let calls = Arc::new(AtomicUsize::new(0)); + let allow_create = Arc::new(AtomicBool::new(false)); + let pool = Pool::new( + PoolConfig::new(2), + ControlledManager { + calls: calls.clone(), + allow_create: allow_create.clone(), + }, + ); + + let first = pool.get().await.unwrap(); + + let mut replenish = Box::pin(pool.replenish_to(2)); + assert!(tests_integration::poll_once(replenish.as_mut()).is_pending()); + + let mut get = Box::pin(pool.get()); + assert!(tests_integration::poll_once(get.as_mut()).is_pending()); + assert_eq!(pool.status().current_size, 1); + + allow_create.store(true, Ordering::Release); + assert_eq!( + tests_integration::poll_once(replenish.as_mut()), + Poll::Ready(Ok(1)) + ); + + let second = match tests_integration::poll_once(get.as_mut()) { + Poll::Ready(Ok(object)) => object, + _ => panic!("get should consume the replenished object"), + }; + assert_eq!(calls.load(Ordering::Relaxed), 2); + assert_eq!(pool.status().current_size, 2); + assert_eq!(pool.status().idle_count, 0); + + drop((first, second)); + assert_eq!(pool.status().idle_count, 2); +} diff --git a/tests-integration/tests/traits_test.rs b/tests-integration/tests/traits_test.rs index 14d4f1a..4419543 100644 --- a/tests-integration/tests/traits_test.rs +++ b/tests-integration/tests/traits_test.rs @@ -15,6 +15,8 @@ // specific language governing permissions and limitations // under the License. +use std::cell::Cell; + use asyncband::barrier::Barrier; use asyncband::condvar::Condvar; use asyncband::latch::Latch; @@ -87,6 +89,7 @@ fn public_types_are_send_and_sync() { assert_send_and_sync::>(); assert_send_and_sync::>(); assert_send_and_sync::>(); + assert_send_and_sync::>>(); assert_send_and_sync::>(); assert_send_and_sync::>(); assert_send_and_sync::>(); @@ -101,6 +104,7 @@ fn movable_public_types_are_send() { assert_send::>>(); assert_send::>(); assert_send::>(); + assert_send::>>(); } #[test] @@ -139,3 +143,12 @@ fn public_types_are_unpin() { assert_unpin::>(); assert_unpin::>(); } + +#[test] +fn unbounded_manual_manager_traits_do_not_depend_on_the_object() { + fn assert_copy() {} + fn assert_debug() {} + + assert_copy::>(); + assert_debug::>(); +} From ed72a372bf99a3fb08957f0d2cde555ec024f1b4 Mon Sep 17 00:00:00 2001 From: tison Date: Mon, 24 Aug 2026 00:56:55 +0800 Subject: [PATCH 2/5] bench(pool): cover core checkout paths --- benchmarks/Cargo.toml | 1 + benchmarks/main.rs | 1 + benchmarks/pool.rs | 121 ++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 123 insertions(+) create mode 100644 benchmarks/pool.rs diff --git a/benchmarks/Cargo.toml b/benchmarks/Cargo.toml index 14f1453..6905629 100644 --- a/benchmarks/Cargo.toml +++ b/benchmarks/Cargo.toml @@ -34,6 +34,7 @@ asyncband = { workspace = true, features = [ "once-cell", "once-map", "oneshot", + "pool", "rwlock", "semaphore", "shutdown", diff --git a/benchmarks/main.rs b/benchmarks/main.rs index 9c3e2eb..cb8ac1a 100644 --- a/benchmarks/main.rs +++ b/benchmarks/main.rs @@ -24,6 +24,7 @@ mod mutex; mod once; mod once_map; mod oneshot; +mod pool; mod rwlock; mod semaphore; mod shutdown; diff --git a/benchmarks/pool.rs b/benchmarks/pool.rs new file mode 100644 index 0000000..42e3a82 --- /dev/null +++ b/benchmarks/pool.rs @@ -0,0 +1,121 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you 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::convert::Infallible; +use std::pin::pin; + +use asyncband::pool::ManageObject; +use asyncband::pool::ObjectStatus; +use asyncband::pool::bounded; +use asyncband::pool::unbounded; +use divan::Bencher; +use divan::black_box; + +use super::support::bench_context; +use super::support::poll_pending; +use super::support::poll_pinned_ready; +use super::support::poll_ready; + +const CAPACITIES: &[usize] = &[1, 32, 1024, usize::MAX]; + +struct Manager; + +impl ManageObject for Manager { + type Object = usize; + type Error = Infallible; + + async fn create(&self) -> Result { + Ok(0) + } + + async fn is_recyclable( + &self, + _object: &mut Self::Object, + _status: &ObjectStatus, + ) -> Result<(), Self::Error> { + Ok(()) + } +} + +#[divan::bench(args = CAPACITIES)] +fn construct_bounded(bencher: Bencher, capacity: usize) { + bencher.bench_local(|| { + black_box(bounded::Pool::new( + bounded::PoolConfig::new(black_box(capacity)), + Manager, + )) + }); +} + +#[divan::bench] +fn bounded_warm_get_and_return(bencher: Bencher) { + let pool = bounded::Pool::new(bounded::PoolConfig::new(1), Manager); + let mut context = bench_context(); + drop(poll_ready(pool.get(), &mut context).unwrap()); + + bencher.bench_local(|| { + let object = poll_ready(pool.get(), &mut context).unwrap(); + black_box(*object); + drop(object); + }); +} + +#[divan::bench] +fn unbounded_warm_try_get_and_return(bencher: Bencher) { + let pool = unbounded::Pool::::never_manage(unbounded::PoolConfig::default()); + pool.extend_one(0); + + bencher.bench_local(|| { + let object = pool.try_get().unwrap(); + black_box(*object); + drop(object); + }); +} + +#[divan::bench] +fn bounded_contended_handoff(bencher: Bencher) { + let pool = bounded::Pool::new(bounded::PoolConfig::new(1), Manager); + let mut context = bench_context(); + drop(poll_ready(pool.get(), &mut context).unwrap()); + + bencher.bench_local(|| { + let held = poll_ready(pool.get(), &mut context).unwrap(); + let mut waiter = pin!(pool.get()); + poll_pending(waiter.as_mut(), &mut context); + + drop(held); + let object = poll_pinned_ready(waiter.as_mut(), &mut context).unwrap(); + black_box(*object); + drop(object); + }); +} + +#[divan::bench] +fn cancel_bounded_waiter(bencher: Bencher) { + let pool = bounded::Pool::new(bounded::PoolConfig::new(1), Manager); + let mut context = bench_context(); + drop(poll_ready(pool.get(), &mut context).unwrap()); + + bencher.bench_local(|| { + let held = poll_ready(pool.get(), &mut context).unwrap(); + { + let mut waiter = pin!(pool.get()); + poll_pending(waiter.as_mut(), &mut context); + } + drop(held); + }); +} From 13807452a2ae7bf7065c24b6ddfdc949c75c1909 Mon Sep 17 00:00:00 2001 From: tison Date: Mon, 24 Aug 2026 00:57:05 +0800 Subject: [PATCH 3/5] docs: clarify asyncband scope and composition --- CHANGELOG.md | 2 ++ HISTORY.md | 2 +- README.md | 83 +++++++++++++++++++++++--------------------- asyncband/Cargo.toml | 10 ++++-- asyncband/src/lib.rs | 28 +++++++++------ 5 files changed, 71 insertions(+), 54 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index a0d21b9..9731ae0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -29,3 +29,5 @@ All notable changes to this project will be documented in this file. ### Improvements * Remove the `slab` dependency in favor of a focused internal waiter arena. +* Refine object-pool lifecycle and maintenance APIs with exact idle status, return-time usage metadata, detachment hooks during retention, fallible `replenish_to`, and synchronous `try_get` for manually populated pools. +* Clarify Asyncband's scope as composable, runtime-agnostic concurrency building blocks that keep execution and timing policy with callers. diff --git a/HISTORY.md b/HISTORY.md index 55a9cc2..9ffeb03 100644 --- a/HISTORY.md +++ b/HISTORY.md @@ -2,7 +2,7 @@ > Apache Asyncband (Incubating) is an effort undergoing incubation at the Apache Software Foundation (ASF), sponsored by the Apache Incubator PMC. Please read the [DISCLAIMER](DISCLAIMER). -Asyncband collects runtime-agnostic synchronization and coordination tools informed by several existing implementations. Only components that draw on external designs or code are listed here. +Asyncband collects composable, runtime-agnostic concurrency building blocks informed by several existing implementations. Only components that draw on external designs or code are listed here. - `barrier::Barrier` is inspired by [`std::sync::Barrier`](https://doc.rust-lang.org/std/sync/struct.Barrier.html) and [`tokio::sync::Barrier`](https://docs.rs/tokio/latest/tokio/sync/struct.Barrier.html), with a different implementation based on the internal `WaitSet` primitive. - The single-future polling loop in `blocking` is adapted from [`pollster`](https://github.com/zesterer/pollster), its parker caching strategy follows [`futures-lite`](https://github.com/smol-rs/futures-lite), and its private parker state machine is adapted from [`parking`](https://github.com/smol-rs/parking) 2.2.1. diff --git a/README.md b/README.md index 674c87c..3a1d45b 100644 --- a/README.md +++ b/README.md @@ -26,18 +26,44 @@ ## Overview -Asyncband is a runtime-agnostic library providing synchronization and coordination tools for asynchronous Rust programming. Its APIs work with any async runtime. +Asyncband is a focused collection of composable, runtime-agnostic concurrency building blocks for async Rust. It provides synchronization, initialization, task coordination, channels, resource reuse, and workload control without choosing an executor for the application. -## Available APIs +Asyncband's async APIs are built on standard futures and wakers. The library does not spawn tasks, own worker threads, install timers, or require a reactor or I/O driver. Applications can poll its futures with Tokio, async-std, smol, a custom executor, or any other standards-based runtime, and compose runtime services such as deadlines around them. -The crate enables no APIs by default. Categories describe each API's primary purpose and do not add another module level, so public paths remain concise, such as `asyncband::mutex`, `asyncband::pool`, and `asyncband::once::OnceCell`. +### Project scope -| Category | Primitive | Feature | Purpose | +The project is not limited to small or stateless synchronization primitives. Stateful utilities such as a singleflight group or an object pool fit when they provide a generally reusable coordination mechanism and remain independent of executor policy. + +The boundary is mechanism versus policy. Task placement, timers, deadlines, retries, periodic maintenance, and application lifecycle orchestration stay with the caller and its runtime. Potential future-concurrency or scheduling APIs are evaluated against the same boundary: they must remain executor-independent and compose with caller-owned execution and timing. + +## Getting started + +The crate enables no APIs by default. Enable only the features your application uses: + +```shell +cargo add asyncband --features mutex,oneshot +``` + +```rust +use asyncband::mutex::Mutex; + +async fn increment() { + let counter = Mutex::new(0); + *counter.lock().await += 1; + assert_eq!(*counter.lock().await, 1); +} +``` + +Public paths stay direct—such as `asyncband::mutex`, `asyncband::pool`, and `asyncband::once::OnceCell`—while Cargo features keep unused implementations out of the build. + +## API map + +| Area | API | Feature | Use | | ----------------------- | ------------------------------------------------------------------------------------ | -------------- | ----------------------------------------------------------------------- | | Shared state | [`Mutex`](https://docs.rs/asyncband/*/asyncband/mutex/struct.Mutex.html) | `mutex` | Protect shared data with asynchronous mutual exclusion. | | | [`RwLock`](https://docs.rs/asyncband/*/asyncband/rwlock/struct.RwLock.html) | `rwlock` | Allow multiple readers or one writer. | | | [`Condvar`](https://docs.rs/asyncband/*/asyncband/condvar/struct.Condvar.html) | `condvar` | Wait for notifications while releasing a mutex. | -| One-time initialization | [`Once`](https://docs.rs/asyncband/*/asyncband/once/struct.Once.html) | `once` | Run asynchronous initialization exactly once. | +| Initialization | [`Once`](https://docs.rs/asyncband/*/asyncband/once/struct.Once.html) | `once` | Run asynchronous initialization exactly once. | | | [`OnceCell`](https://docs.rs/asyncband/*/asyncband/once/struct.OnceCell.html) | `once-cell` | Initialize and store one asynchronous value. | | | [`OnceMap`](https://docs.rs/asyncband/*/asyncband/once/struct.OnceMap.html) | `once-map` | Initialize and store one value per key. | | Task coordination | [`Barrier`](https://docs.rs/asyncband/*/asyncband/barrier/struct.Barrier.html) | `barrier` | Wait until all participants reach a synchronization point. | @@ -47,24 +73,15 @@ The crate enables no APIs by default. Categories describe each API's primary pur | Channels | [`oneshot::channel`](https://docs.rs/asyncband/*/asyncband/oneshot/fn.channel.html) | `oneshot` | Send one value between two tasks. | | | [`mpsc::bounded`](https://docs.rs/asyncband/*/asyncband/mpsc/fn.bounded.html) | `mpsc` | Send values from multiple producers through a bounded channel. | | | [`mpsc::unbounded`](https://docs.rs/asyncband/*/asyncband/mpsc/fn.unbounded.html) | `mpsc` | Send values from multiple producers through an unbounded channel. | -| Resource reuse | [`pool::bounded`](https://docs.rs/asyncband/*/asyncband/pool/bounded/) | `pool` | Reuse managed objects up to a configured capacity. | -| | [`pool::unbounded`](https://docs.rs/asyncband/*/asyncband/pool/unbounded/) | `pool` | Reuse manually supplied or manager-created objects. | -| Workload control | [`Semaphore`](https://docs.rs/asyncband/*/asyncband/semaphore/struct.Semaphore.html) | `semaphore` | Control concurrent access with permits. | +| Resource reuse | [`pool::bounded`](https://docs.rs/asyncband/*/asyncband/pool/bounded/) | `pool` | Reuse managed objects up to a configured capacity. | +| | [`pool::unbounded`](https://docs.rs/asyncband/*/asyncband/pool/unbounded/) | `pool` | Reuse manually supplied or manager-created objects. | +| Workload coordination | [`Semaphore`](https://docs.rs/asyncband/*/asyncband/semaphore/struct.Semaphore.html) | `semaphore` | Control concurrent access with permits. | | | [`Group`](https://docs.rs/asyncband/*/asyncband/singleflight/struct.Group.html) | `singleflight` | Coalesce concurrent calls for the same key. | - -## Installation - -Add the dependency to your `Cargo.toml` via: - -```shell -cargo add asyncband --features mutex,oneshot -``` - -List every API your application uses in `features`; a bare `cargo add asyncband` intentionally exposes no optional modules. +| Synchronous interop | [`FutureExt`](https://docs.rs/asyncband/*/asyncband/blocking/trait.FutureExt.html) | `blocking` | Drive one runtime-agnostic future from a blocking thread. | ## Synchronous interoperability -The optional `blocking` module bridges synchronous Rust code to runtime-agnostic futures. It is an interoperability utility rather than another async primitive, so it is documented separately from the table above. +The optional `blocking` module is a boundary adapter for synchronous callers. It parks the calling thread while driving one future; it is not a general-purpose executor. ```shell cargo add asyncband --features blocking @@ -82,31 +99,17 @@ let value = async { 42 }.wait_timeout(Duration::ZERO); assert_eq!(value, Some(42)); ``` -`asyncband::blocking::FutureExt::block_on(future)` is the equivalent UFCS spelling when function syntax is preferred; it calls the same trait method rather than a separate free function. - -### Async first, blocking by adaptation - -Async and synchronous synchronization primitives have different optimization constraints. Once an async primitive is runtime-agnostic, synchronous code can usually drive its future with a `block_on` adapter. Asyncband's `blocking` feature provides this adapter with a lightweight, thread-parking single-future executor: pending work parks the calling thread and its waker resumes it, providing practical blocking interoperability without busy-waiting or a full async runtime. - -A sync-first implementation can still exploit OS- or platform-specific facilities for better performance. Asyncband therefore optimizes its primitives for async code and keeps blocking as a boundary adapter instead of duplicating sync and async methods across every type. This keeps the public API focused while leaving sync-oriented optimizations to dedicated libraries. - -### Execution constraints - -This is a minimal single-future executor, not a general-purpose async runtime. A timed-out `wait_timeout` drops the future. The implementation uses a private parker, so it does not consume wake-ups belonging to other parking operations on the same thread; recursive calls use a separate parker. Futures depending on a runtime-specific timer or I/O driver may not make progress, and blocking an executor thread can cause starvation or deadlocks. See [`asyncband::blocking`](https://docs.rs/asyncband/*/asyncband/blocking/index.html) for details. - -## Runtime Agnostic - -All asynchronous APIs in this library are runtime-agnostic, meaning they can be used with any async runtime like Tokio, async-std, or others. This makes the library highly versatile and portable. +`wait_timeout` drops the future on timeout. Futures that depend on a runtime-specific timer or I/O driver still need that runtime's driver to make progress, and blocking an executor thread can cause starvation or deadlocks. See the [`blocking` module documentation](https://docs.rs/asyncband/*/asyncband/blocking/) for the full contract. -## Thread Safety +## Thread safety -Asyncband primitives and guards implement `Send` and `Sync` only when the protected or transferred value satisfies the necessary bounds. In particular, owned read guards that may move destruction to another thread require the protected value to be `Send` as well as `Sync`. See each type's documentation for its exact bounds. +Asyncband types implement `Send` and `Sync` only when the protected, transferred, or managed value satisfies the necessary bounds. See each API's documentation for its exact contract. -## Minimum Supported Rust Version (MSRV) +## Minimum supported Rust version (MSRV) -This crate is built against the latest stable release, and its minimum supported rustc version is 1.86.0. +The minimum supported Rust version is 1.86.0 and is checked in CI. -The policy is that the minimum Rust version required to use this crate can be increased in minor version updates. For example, if Asyncband 1.0 requires Rust 1.20.0, then Asyncband 1.0.z for all values of z will also require Rust 1.20.0 or newer. However, Asyncband 1.y for y > 0 may require a newer minimum version of Rust. +The MSRV may increase in a minor release, but patch releases within the same minor line will not raise it. ## License and Trademarks @@ -116,4 +119,4 @@ Apache Asyncband, Asyncband, and Apache are either registered trademarks or trad ## History -See [HISTORY.md](HISTORY.md) for the external implementations that informed Asyncband's primitives. +See [HISTORY.md](HISTORY.md) for the external implementations that informed Asyncband's APIs. diff --git a/asyncband/Cargo.toml b/asyncband/Cargo.toml index ce31add..af8e642 100644 --- a/asyncband/Cargo.toml +++ b/asyncband/Cargo.toml @@ -20,9 +20,15 @@ name = "asyncband" version = "0.6.7" categories = ["asynchronous", "concurrency"] -description = "Runtime-agnostic synchronization and coordination tools for asynchronous Rust." +description = "Composable, runtime-agnostic concurrency building blocks for async Rust." documentation = "https://docs.rs/asyncband" -keywords = ["async", "concurrency", "synchronization", "waitgroup", "mutex"] +keywords = [ + "async", + "concurrency", + "synchronization", + "coordination", + "runtime-agnostic", +] edition.workspace = true homepage.workspace = true diff --git a/asyncband/src/lib.rs b/asyncband/src/lib.rs index ea458cd..917c86d 100644 --- a/asyncband/src/lib.rs +++ b/asyncband/src/lib.rs @@ -18,10 +18,10 @@ #![cfg_attr(docsrs, feature(doc_cfg))] #![deny(missing_docs)] -//! Runtime-agnostic synchronization and coordination tools for asynchronous Rust. +//! Composable, runtime-agnostic concurrency building blocks for async Rust. //! -//! `asyncband` provides locks, initialization tools, task coordination, channels, object pools, and -//! workload controls without tying an application to a particular async runtime. The APIs use +//! `asyncband` provides synchronization, initialization, task coordination, channels, resource +//! reuse, and workload control without choosing an executor for the application. Its async APIs use //! standard futures and wakers, so they can run on Tokio, async-std, smol, or a custom executor. //! //! # Getting started @@ -33,7 +33,7 @@ //! asyncband = { version = "0.7", features = ["mutex", "oneshot"] } //! ``` //! -//! Then use the selected primitives directly: +//! Then use the selected APIs directly: //! //! ``` //! # #[tokio::main] @@ -57,14 +57,20 @@ //! | Initialize values once | [`once::Once`], [`once::OnceCell`], [`once::OnceMap`] | `once`, `once-cell`, `once-map` | //! | Coordinate tasks | [`barrier::Barrier`], [`latch::Latch`], [`waitgroup::WaitGroup`], [`shutdown`] | `barrier`, `latch`, `waitgroup`, `shutdown` | //! | Send values | [`oneshot::channel`], [`mpsc::bounded`], [`mpsc::unbounded`] | `oneshot`, `mpsc` | -//! | Reuse managed objects | [`pool::bounded`], [`pool::unbounded`] | `pool` | -//! | Control workloads | [`semaphore::Semaphore`], [`singleflight::Group`] | `semaphore`, `singleflight` | +//! | Reuse objects | [`pool::bounded`], [`pool::unbounded`] | `pool` | +//! | Coordinate workloads | [`semaphore::Semaphore`], [`singleflight::Group`] | `semaphore`, `singleflight` | //! | Wait from synchronous code | [`blocking::FutureExt`] | `blocking` | //! -//! # Runtime and blocking model +//! # Scope and runtime model //! -//! The async primitives do not start threads, spawn tasks, or require a runtime-specific reactor. -//! Await them inside any executor that polls standard Rust futures. +//! The project is not limited to small or stateless primitives. Stateful tools such as +//! [`singleflight::Group`] and the [`pool`] module fit when they provide reusable coordination and +//! remain independent of executor policy. +//! +//! The async APIs do not start threads, spawn tasks, install timers, or require a runtime-specific +//! reactor. Task placement, deadlines, retries, periodic maintenance, and lifecycle orchestration +//! remain with the caller. Await Asyncband futures inside any executor that polls standard Rust +//! futures, and compose those runtime services around them. //! //! Async APIs are the primary interface. The optional [`blocking`] module is a boundary adapter for //! synchronous callers: its single-future executor parks the calling thread and resumes it through @@ -74,8 +80,8 @@ //! //! # Thread safety //! -//! Primitives and guards implement `Send` and `Sync` only when their protected or transferred -//! values satisfy the required bounds. Consult each type's documentation for its exact contract. +//! Asyncband types implement `Send` and `Sync` only when their protected, transferred, or managed +//! values satisfy the required bounds. Consult each API's documentation for its exact contract. //! //! # Disclaimer //! From 60c29c8bf358aee7e43c90f8a8eac00a89f30d0c Mon Sep 17 00:00:00 2001 From: tison Date: Mon, 24 Aug 2026 01:22:16 +0800 Subject: [PATCH 4/5] docs: preserve async-first design principle --- CHANGELOG.md | 2 -- README.md | 10 +++++++++- asyncband/src/lib.rs | 12 +++++++++--- 3 files changed, 18 insertions(+), 6 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 9731ae0..a0d21b9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -29,5 +29,3 @@ All notable changes to this project will be documented in this file. ### Improvements * Remove the `slab` dependency in favor of a focused internal waiter arena. -* Refine object-pool lifecycle and maintenance APIs with exact idle status, return-time usage metadata, detachment hooks during retention, fallible `replenish_to`, and synchronous `try_get` for manually populated pools. -* Clarify Asyncband's scope as composable, runtime-agnostic concurrency building blocks that keep execution and timing policy with callers. diff --git a/README.md b/README.md index 3a1d45b..bacc65d 100644 --- a/README.md +++ b/README.md @@ -99,7 +99,15 @@ let value = async { 42 }.wait_timeout(Duration::ZERO); assert_eq!(value, Some(42)); ``` -`wait_timeout` drops the future on timeout. Futures that depend on a runtime-specific timer or I/O driver still need that runtime's driver to make progress, and blocking an executor thread can cause starvation or deadlocks. See the [`blocking` module documentation](https://docs.rs/asyncband/*/asyncband/blocking/) for the full contract. +### Async first, blocking by adaptation + +Async and synchronous synchronization primitives have different optimization constraints. Once an async operation is exposed as a runtime-agnostic future, synchronous code can usually drive that future through a `block_on` adapter. Asyncband therefore designs its primitives for async use and provides blocking interoperability at the boundary instead of duplicating synchronous methods across every type. + +A sync-first implementation can exploit OS- or platform-specific facilities that an async implementation cannot assume. Libraries focused on synchronous code can therefore make different and sometimes better tradeoffs. Asyncband leaves those optimizations to dedicated libraries rather than treating blocking adaptation as a second family of primitives. + +### Execution constraints + +The `blocking` module is a lightweight, thread-parking single-future executor, not a general-purpose async runtime. `wait_timeout` drops the future on timeout. Futures that depend on a runtime-specific timer or I/O driver still need that runtime's driver to make progress, and blocking an executor thread can cause starvation or deadlocks. See the [`blocking` module documentation](https://docs.rs/asyncband/*/asyncband/blocking/) for the full contract. ## Thread safety diff --git a/asyncband/src/lib.rs b/asyncband/src/lib.rs index 917c86d..1c3ffa5 100644 --- a/asyncband/src/lib.rs +++ b/asyncband/src/lib.rs @@ -72,9 +72,15 @@ //! remain with the caller. Await Asyncband futures inside any executor that polls standard Rust //! futures, and compose those runtime services around them. //! -//! Async APIs are the primary interface. The optional [`blocking`] module is a boundary adapter for -//! synchronous callers: its single-future executor parks the calling thread and resumes it through -//! the future's waker. It is not a general-purpose async runtime, and futures that depend on a +//! # Async first, blocking by adaptation +//! +//! Async and synchronous primitives have different optimization constraints. Asyncband designs its +//! primitives for async use and provides the optional [`blocking`] module as a boundary adapter +//! instead of duplicating synchronous methods across every type. Sync-first implementations can +//! exploit OS- or platform-specific facilities and remain the domain of dedicated libraries. +//! +//! The adapter's single-future executor parks the calling thread and resumes it through the +//! future's waker. It is not a general-purpose async runtime, and futures that depend on a //! runtime-specific timer or I/O driver may not make progress. See the module documentation for the //! full execution constraints. //! From 1394d5c31e5ae4f18deddc190af9b5e6889f965a Mon Sep 17 00:00:00 2001 From: tison Date: Mon, 24 Aug 2026 03:12:22 +0800 Subject: [PATCH 5/5] docs: restore msrv policy wording --- README.md | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/README.md b/README.md index bacc65d..44ecbb6 100644 --- a/README.md +++ b/README.md @@ -113,11 +113,11 @@ The `blocking` module is a lightweight, thread-parking single-future executor, n Asyncband types implement `Send` and `Sync` only when the protected, transferred, or managed value satisfies the necessary bounds. See each API's documentation for its exact contract. -## Minimum supported Rust version (MSRV) +## Minimum Supported Rust Version (MSRV) -The minimum supported Rust version is 1.86.0 and is checked in CI. +This crate is built against the latest stable release, and its minimum supported rustc version is 1.86.0. -The MSRV may increase in a minor release, but patch releases within the same minor line will not raise it. +The policy is that the minimum Rust version required to use this crate can be increased in minor version updates. For example, if Asyncband 1.0 requires Rust 1.20.0, then Asyncband 1.0.z for all values of z will also require Rust 1.20.0 or newer. However, Asyncband 1.y for y > 0 may require a newer minimum version of Rust. ## License and Trademarks