You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@asyncband.apache.org by ti...@apache.org on 2026/08/23 15:20:33 UTC
(asyncband) 01/01: feat(pool): port runtime-agnostic object pools
This is an automated email from the ASF dual-hosted git repository.
tisonkun pushed a commit to branch codex/port-fastpool
in repository https://gitbox.apache.org/repos/asf/asyncband.git
commit b43c95e20666717bb8ca198a9451323deaeddf63
Author: tison <wa...@gmail.com>
AuthorDate: Sun Aug 23 23:20:00 2026 +0800
feat(pool): port runtime-agnostic object pools
---
CHANGELOG.md | 1 +
Cargo.lock | 1 +
Cargo.toml | 1 +
HISTORY.md | 3 +-
README.md | 12 +-
asyncband/Cargo.toml | 4 +-
asyncband/src/lib.rs | 11 +-
asyncband/src/pool/bounded.rs | 614 ++++++++++++++++++++
asyncband/src/pool/common.rs | 117 ++++
asyncband/src/pool/mod.rs | 202 +++++++
asyncband/src/pool/mutex.rs | 57 ++
asyncband/src/pool/retain_spec.rs | 92 +++
asyncband/src/pool/unbounded.rs | 631 +++++++++++++++++++++
tests-integration/Cargo.toml | 1 +
.../tests/pool_recycle_cancelled_test.rs | 311 ++++++++++
tests-integration/tests/pool_replenish_test.rs | 78 +++
tests-integration/tests/traits_test.rs | 30 +
17 files changed, 2155 insertions(+), 11 deletions(-)
diff --git a/CHANGELOG.md b/CHANGELOG.md
index db9f27f..12c9357 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -9,6 +9,7 @@ All notable changes to this project will be documented in this file.
### New features
* Add an opt-in `asyncband::blocking::FutureExt` bridge with `block_on` and `wait_timeout` methods for waiting on runtime-agnostic futures from synchronous code.
+* Add opt-in bounded and unbounded runtime-agnostic object pools under `asyncband::pool`.
### Breaking changes
diff --git a/Cargo.lock b/Cargo.lock
index c72436f..496669e 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -57,6 +57,7 @@ name = "asyncband"
version = "0.6.7"
dependencies = [
"hashbrown",
+ "scopeguard",
"tokio",
]
diff --git a/Cargo.toml b/Cargo.toml
index e5c3ad8..5671f8a 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -33,6 +33,7 @@ asyncband = { path = "asyncband" }
# Optional runtime dependencies
hashbrown = { version = "0.17.1", default-features = false }
+scopeguard = { version = "1.2.0" }
# Dev dependencies
cargo_metadata = { version = "0.23.1" }
diff --git a/HISTORY.md b/HISTORY.md
index f570577..54143c7 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 primitives informed by several existing implementations. Only components that draw on external designs or code are listed here.
+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.
- `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.
@@ -13,6 +13,7 @@ Asyncband collects runtime-agnostic synchronization primitives informed by sever
- `once::OnceCell` is derived from [`tokio::sync::OnceCell`](https://docs.rs/tokio/latest/tokio/sync/struct.OnceCell.html), but uses Asyncband's semaphore implementation.
- `once::OnceMap` is inspired by [`uv-once-map`](https://github.com/astral-sh/uv/tree/main/crates/uv-once-map), with a redesigned interface and implementation.
- `oneshot::channel` is derived from the [`oneshot`](https://github.com/faern/oneshot) crate, with significant simplifications because Asyncband does not provide synchronized receive operations.
+- `pool` is ported from [`fastpool`](https://github.com/fast/fastpool), which is derived from [`deadpool`](https://github.com/bikeshedder/deadpool), while keeping Fastpool's runtime-agnostic design and caller-side timeout composition.
- `rwlock::RwLock` is derived from [`tokio::sync::RwLock`](https://docs.rs/tokio/latest/tokio/sync/struct.RwLock.html), but accepts any `NonZeroUsize` as `max_readers` instead of Tokio's restricted range.
- `semaphore::Semaphore` is derived from [`tokio::sync::Semaphore`](https://docs.rs/tokio/latest/tokio/sync/struct.Semaphore.html), but omits `close`, avoids Tokio's fixed maximum-permit constant, and adds operations such as `reduce_permits` for Asyncband's use cases.
- `waitgroup::WaitGroup` is inspired by [`waitgroup-rs`](https://github.com/laizy/waitgroup-rs), with a different API and an implementation based on the internal `CountdownState` primitive.
diff --git a/README.md b/README.md
index 8f59d4d..539c150 100644
--- a/README.md
+++ b/README.md
@@ -26,11 +26,11 @@
## Overview
-Asyncband is a runtime-agnostic library providing essential synchronization primitives for asynchronous Rust programming. The library offers a collection of well-tested, efficient synchronization tools that work with any async runtime.
+Asyncband is a runtime-agnostic library providing synchronization and coordination tools for asynchronous Rust programming. Its APIs work with any async runtime.
-## Available primitives
+## Available APIs
-The crate enables no primitives by default. Categories describe each primitive's primary purpose and do not add another module level, so public paths remain concise, such as `asyncband::mutex` and `asyncband::once::OnceCell`.
+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`.
| Category | Primitive | Feature | Purpose |
| ----------------------- | ------------------------------------------------------------------------------------ | -------------- | ----------------------------------------------------------------------- |
@@ -48,6 +48,8 @@ The crate enables no primitives by default. Categories describe each primitive's
| | [`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. |
| | [`broadcast::overflow`](https://docs.rs/asyncband/*/asyncband/broadcast/overflow/) | `broadcast` | Broadcast values and report when slow receivers miss overwritten items. |
+| 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. |
| | [`Group`](https://docs.rs/asyncband/*/asyncband/singleflight/struct.Group.html) | `singleflight` | Coalesce concurrent calls for the same key. |
@@ -59,7 +61,7 @@ Add the dependency to your `Cargo.toml` via:
cargo add asyncband --features mutex,oneshot
```
-List every primitive your application uses in `features`; a bare `cargo add asyncband` intentionally exposes no primitive modules.
+List every API your application uses in `features`; a bare `cargo add asyncband` intentionally exposes no optional modules.
## Synchronous interoperability
@@ -95,7 +97,7 @@ This is a minimal single-future executor, not a general-purpose async runtime. A
## Runtime Agnostic
-All synchronization primitives 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.
+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.
## Thread Safety
diff --git a/asyncband/Cargo.toml b/asyncband/Cargo.toml
index c36f13f..47496ac 100644
--- a/asyncband/Cargo.toml
+++ b/asyncband/Cargo.toml
@@ -20,7 +20,7 @@ name = "asyncband"
version = "0.6.7"
categories = ["asynchronous", "concurrency"]
-description = "A runtime-agnostic library providing essential synchronization primitives for asynchronous Rust programming."
+description = "Runtime-agnostic synchronization and coordination tools for asynchronous Rust."
documentation = "https://docs.rs/asyncband"
keywords = ["async", "concurrency", "synchronization", "waitgroup", "mutex"]
@@ -49,6 +49,7 @@ once = ["semaphore"]
once-cell = ["semaphore"]
once-map = ["dep:hashbrown", "once-cell"]
oneshot = []
+pool = ["dep:scopeguard", "semaphore"]
rwlock = []
semaphore = []
shutdown = ["latch", "waitgroup"]
@@ -59,6 +60,7 @@ waitgroup = []
hashbrown = { workspace = true, default-features = false, features = [
"inline-more",
], optional = true }
+scopeguard = { workspace = true, optional = true }
[dev-dependencies]
tokio = { workspace = true, features = ["full"] }
diff --git a/asyncband/src/lib.rs b/asyncband/src/lib.rs
index 3d810e9..851434f 100644
--- a/asyncband/src/lib.rs
+++ b/asyncband/src/lib.rs
@@ -18,11 +18,11 @@
#![cfg_attr(docsrs, feature(doc_cfg))]
#![deny(missing_docs)]
-//! Runtime-agnostic synchronization primitives for asynchronous Rust.
+//! Runtime-agnostic synchronization and coordination tools for asynchronous Rust.
//!
-//! `asyncband` provides locks, initialization tools, task coordination, channels, and workload
-//! controls without tying an application to a particular async runtime. The primitives use standard
-//! futures and wakers, so they can run on Tokio, async-std, smol, or a custom executor.
+//! `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
+//! standard futures and wakers, so they can run on Tokio, async-std, smol, or a custom executor.
//!
//! # Getting started
//!
@@ -57,6 +57,7 @@
//! | 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`], [`broadcast::overflow`] | `oneshot`, `mpsc`, `broadcast` |
+//! | Reuse managed objects | [`pool::bounded`], [`pool::unbounded`] | `pool` |
//! | Control workloads | [`semaphore::Semaphore`], [`singleflight::Group`] | `semaphore`, `singleflight` |
//! | Wait from synchronous code | [`blocking::FutureExt`] | `blocking` |
//!
@@ -107,6 +108,8 @@ pub mod mutex;
pub mod once;
#[cfg(feature = "oneshot")]
pub mod oneshot;
+#[cfg(feature = "pool")]
+pub mod pool;
#[cfg(feature = "rwlock")]
pub mod rwlock;
#[cfg(feature = "semaphore")]
diff --git a/asyncband/src/pool/bounded.rs b/asyncband/src/pool/bounded.rs
new file mode 100644
index 0000000..7642aa3
--- /dev/null
+++ b/asyncband/src/pool/bounded.rs
@@ -0,0 +1,614 @@
+// 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.
+
+//! Bounded object pools.
+//!
+//! A bounded pool creates and recycles objects with full management. You _cannot_ put an object to
+//! the pool 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.
+//!
+//! 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`]).
+//!
+//! 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;
+//! use asyncband::pool::bounded::Pool;
+//! use asyncband::pool::bounded::PoolConfig;
+//!
+//! struct Compute;
+//! impl Compute {
+//! async fn do_work(&self) -> i32 {
+//! 42
+//! }
+//! }
+//!
+//! struct Manager;
+//! impl ManageObject for Manager {
+//! type Object = Compute;
+//! type Error = ();
+//!
+//! async fn create(&self) -> Result<Self::Object, Self::Error> {
+//! Ok(Compute)
+//! }
+//!
+//! async fn is_recyclable(
+//! &self,
+//! o: &mut Self::Object,
+//! status: &ObjectStatus,
+//! ) -> Result<(), Self::Error> {
+//! Ok(())
+//! }
+//! }
+//!
+//! # #[tokio::main]
+//! # async fn main() {
+//! let pool = Pool::new(PoolConfig::new(16), Manager);
+//! let o = pool.get().await.unwrap();
+//! assert_eq!(o.do_work().await, 42);
+//! # }
+//! ```
+
+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::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::semaphore::OwnedSemaphorePermit;
+use crate::semaphore::Semaphore;
+
+/// The configuration of [`Pool`].
+#[derive(Clone, Copy, Debug)]
+#[non_exhaustive]
+pub struct PoolConfig {
+ /// Maximum size of the [`Pool`].
+ pub max_size: usize,
+
+ /// Queue strategy of the [`Pool`].
+ ///
+ /// Determines the order of objects being queued and dequeued.
+ pub queue_strategy: QueueStrategy,
+
+ /// Strategy when recycling object has been cancelled.
+ pub recycle_cancelled_strategy: RecycleCancelledStrategy,
+}
+
+impl PoolConfig {
+ /// Creates a new [`PoolConfig`].
+ pub fn new(max_size: usize) -> Self {
+ Self {
+ max_size,
+ queue_strategy: QueueStrategy::default(),
+ recycle_cancelled_strategy: RecycleCancelledStrategy::default(),
+ }
+ }
+
+ /// Returns a new [`PoolConfig`] with the specified queue strategy.
+ pub fn with_queue_strategy(mut self, queue_strategy: QueueStrategy) -> Self {
+ self.queue_strategy = queue_strategy;
+ self
+ }
+
+ /// Returns a new [`PoolConfig`] with the specified recycle cancelled strategy.
+ pub fn with_recycle_cancelled_strategy(
+ mut self,
+ recycle_cancelled_strategy: RecycleCancelledStrategy,
+ ) -> Self {
+ self.recycle_cancelled_strategy = recycle_cancelled_strategy;
+ self
+ }
+}
+
+/// The current pool status.
+///
+/// See [`Pool::status`].
+#[derive(Clone, Copy, Debug)]
+#[non_exhaustive]
+pub struct PoolStatus {
+ /// The maximum size of the pool.
+ pub max_size: usize,
+
+ /// The current size of the pool.
+ pub current_size: usize,
+
+ /// The number of idle objects in the pool.
+ 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.
+///
+/// See the [module level documentation](self) for more.
+pub struct Pool<M: ManageObject> {
+ 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.
+ permits: Arc<Semaphore>,
+ /// A deque that holds the objects.
+ slots: Mutex<PoolDeque<ObjectState<M::Object>>>,
+}
+
+#[derive(Debug)]
+struct PoolDeque<T> {
+ deque: VecDeque<T>,
+ current_size: usize,
+ max_size: usize,
+}
+
+impl<M> std::fmt::Debug for Pool<M>
+where
+ M: ManageObject,
+ M::Object: std::fmt::Debug,
+{
+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ f.debug_struct("Pool")
+ .field("slots", &self.slots)
+ .field("config", &self.config)
+ .field("users", &self.users)
+ .field("permits", &self.permits)
+ .finish()
+ }
+}
+
+impl<M: ManageObject> Pool<M> {
+ /// Creates a new [`Pool`].
+ pub fn new(config: PoolConfig, manager: M) -> Arc<Self> {
+ 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,
+ });
+
+ Arc::new(Self {
+ config,
+ manager,
+ users,
+ permits,
+ slots,
+ })
+ }
+
+ /// Replenishes the pool with at most `most` number of new 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.
+ ///
+ /// 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;
+ }
+ }
+ }
+ };
+
+ 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 mut replenished = 0;
+ for _ in 0..gap {
+ if let Ok(o) = self.manager.create().await {
+ let status = ObjectStatus::default();
+ let state = ObjectState { o, status };
+
+ 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);
+ }
+ }
+ }
+
+ 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.
+ pub async fn get(self: &Arc<Self>) -> Result<Object<M>, M::Error> {
+ self.users.fetch_add(1, Ordering::Relaxed);
+
+ // TODO(*) replace scopeguard with std DropGuard once stabilized
+ // https://github.com/rust-lang/rust/issues/144426
+ let guard = scopeguard::guard((), |()| {
+ self.users.fetch_sub(1, Ordering::Relaxed);
+ });
+
+ 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(),
+ };
+
+ match existing {
+ None => {
+ let object = self.manager.create().await?;
+ let state = ObjectState {
+ o: object,
+ status: ObjectStatus::default(),
+ };
+ self.slots.lock().current_size += 1;
+ break Object {
+ state: Some(state),
+ permit,
+ pool: Arc::downgrade(self),
+ };
+ }
+ Some(object) => {
+ let mut unready_object = UnreadyObject {
+ state: Some(object),
+ pool: Arc::downgrade(self),
+ recycle_cancelled_strategy: self.config.recycle_cancelled_strategy,
+ };
+
+ let state = unready_object.state();
+ let status = state.status;
+ if self
+ .manager
+ .is_recyclable(&mut state.o, &status)
+ .await
+ .is_ok()
+ {
+ state.status.recycle_count += 1;
+ state.status.recycled = Some(std::time::Instant::now());
+ break unready_object.ready(permit);
+ } else {
+ // We need to manually detach here as the drop implementation
+ // depends on the recycle cancelled strategy.
+ unready_object.detach();
+ }
+ }
+ };
+ };
+
+ scopeguard::ScopeGuard::into_inner(guard);
+ 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 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
+ /// terminate if the pool is dropped.
+ ///
+ /// ```rust,ignore
+ /// let interval = Duration::from_secs(30);
+ /// let max_age = Duration::from_secs(60);
+ ///
+ /// let weak_pool = Arc::downgrade(&pool);
+ /// tokio::spawn(async move {
+ /// loop {
+ /// tokio::time::sleep(interval).await;
+ /// if let Some(pool) = weak_pool.upgrade() {
+ /// pool.retain(|_, status| status.last_used().elapsed() < max_age);
+ /// } else {
+ /// break;
+ /// }
+ /// }
+ /// });
+ /// ```
+ pub fn retain(
+ &self,
+ f: impl FnMut(&mut M::Object, ObjectStatus) -> bool,
+ ) -> RetainResult<M::Object> {
+ let mut slots = self.slots.lock();
+ let result = retain_spec::do_vec_deque_retain(&mut slots.deque, f);
+ slots.current_size -= result.removed.len();
+ 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)
+ 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,
+ }
+ }
+
+ fn push_back(&self, o: ObjectState<M::Object>) {
+ self.return_to_pool(o);
+ self.users.fetch_sub(1, Ordering::Relaxed);
+ }
+
+ fn return_to_pool(&self, o: ObjectState<M::Object>) {
+ let mut slots = self.slots.lock();
+
+ assert!(
+ slots.current_size <= slots.max_size,
+ "invariant broken: current_size <= max_size (actual: {} <= {})",
+ slots.current_size,
+ slots.max_size,
+ );
+
+ slots.deque.push_back(o);
+ }
+
+ fn detach_object(&self, o: &mut M::Object, ready: bool) {
+ let mut slots = self.slots.lock();
+
+ assert!(
+ slots.current_size <= slots.max_size,
+ "invariant broken: current_size <= max_size (actual: {} <= {})",
+ slots.current_size,
+ slots.max_size,
+ );
+
+ slots.current_size -= 1;
+ 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);
+ }
+}
+
+/// A wrapper of the actual pooled object.
+///
+/// This object implements [`Deref`] and [`DerefMut`]. You can use it as if it was of type
+/// `M::Object`.
+///
+/// This object implements [`Drop`] that returns the underlying object to the pool on drop. You may
+/// call [`Object::detach`] to detach the object from the pool before dropping it.
+pub struct Object<M: ManageObject> {
+ state: Option<ObjectState<M::Object>>,
+ permit: OwnedSemaphorePermit,
+ pool: Weak<Pool<M>>,
+}
+
+impl<M> std::fmt::Debug for Object<M>
+where
+ M: ManageObject,
+ M::Object: std::fmt::Debug,
+{
+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ f.debug_struct("Object")
+ .field("state", &self.state)
+ .field("permit", &self.permit)
+ .finish()
+ }
+}
+
+impl<M: ManageObject> Drop for Object<M> {
+ fn drop(&mut self) {
+ if let Some(state) = self.state.take() {
+ if let Some(pool) = self.pool.upgrade() {
+ pool.push_back(state);
+ }
+ }
+ }
+}
+
+impl<M: ManageObject> Deref for Object<M> {
+ type Target = M::Object;
+ fn deref(&self) -> &M::Object {
+ // SAFETY: `state` is always `Some` when `Object` is owned.
+ &self.state.as_ref().unwrap().o
+ }
+}
+
+impl<M: ManageObject> DerefMut for Object<M> {
+ fn deref_mut(&mut self) -> &mut Self::Target {
+ // SAFETY: `state` is always `Some` when `Object` is owned.
+ &mut self.state.as_mut().unwrap().o
+ }
+}
+
+impl<M: ManageObject> AsRef<M::Object> for Object<M> {
+ fn as_ref(&self) -> &M::Object {
+ self
+ }
+}
+
+impl<M: ManageObject> AsMut<M::Object> for Object<M> {
+ fn as_mut(&mut self) -> &mut M::Object {
+ self
+ }
+}
+
+impl<M: ManageObject> Object<M> {
+ /// Detaches the object from the [`Pool`].
+ ///
+ /// This reduces the size of the pool by one.
+ pub fn detach(mut self) -> M::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);
+ }
+ o
+ }
+
+ /// Returns the status of the object.
+ pub fn status(&self) -> ObjectStatus {
+ // SAFETY: `state` is always `Some` when `Object` is owned.
+ self.state.as_ref().unwrap().status
+ }
+}
+
+/// A wrapper of ObjectState used during the `is_recyclable` check in `Pool::get`.
+///
+/// If the check passes, the object is converted to a ready `Object` via `ready()`.
+/// If the check fails, `detach()` should be called to permanently remove the object
+/// from the pool. If dropped without calling either method (due to being cancelled),
+/// the behavior depends on the pool's [`RecycleCancelledStrategy`] configuration.
+struct UnreadyObject<M: ManageObject> {
+ state: Option<ObjectState<M::Object>>,
+ pool: Weak<Pool<M>>,
+ recycle_cancelled_strategy: RecycleCancelledStrategy,
+}
+
+impl<M: ManageObject> Drop for UnreadyObject<M> {
+ fn drop(&mut self) {
+ if let Some(mut state) = self.state.take() {
+ if let Some(pool) = self.pool.upgrade() {
+ match self.recycle_cancelled_strategy {
+ RecycleCancelledStrategy::Detach => {
+ pool.detach_object(&mut state.o, false);
+ }
+ RecycleCancelledStrategy::ReturnToPool => {
+ pool.return_to_pool(state);
+ }
+ }
+ }
+ }
+ }
+}
+
+impl<M: ManageObject> UnreadyObject<M> {
+ fn ready(mut self, permit: OwnedSemaphorePermit) -> Object<M> {
+ // SAFETY: `state` is always `Some` when `UnreadyObject` is owned.
+ let state = Some(self.state.take().unwrap());
+ let pool = self.pool.clone();
+ Object {
+ state,
+ permit,
+ pool,
+ }
+ }
+
+ 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);
+ }
+ }
+ }
+
+ fn state(&mut self) -> &mut ObjectState<M::Object> {
+ // SAFETY: `state` is always `Some` when `UnreadyObject` is owned.
+ self.state.as_mut().unwrap()
+ }
+}
+
+#[derive(Debug)]
+struct ObjectState<T> {
+ o: T,
+ status: ObjectStatus,
+}
+
+impl<T> retain_spec::SealedState for ObjectState<T> {
+ 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
new file mode 100644
index 0000000..57f91aa
--- /dev/null
+++ b/asyncband/src/pool/common.rs
@@ -0,0 +1,117 @@
+// 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::future::Future;
+use std::time::Instant;
+
+/// Statistics regarding an object returned by the pool.
+#[derive(Debug, Clone, Copy)]
+pub struct ObjectStatus {
+ created: Instant,
+ pub(crate) recycled: Option<Instant>,
+ pub(crate) recycle_count: usize,
+}
+
+impl Default for ObjectStatus {
+ fn default() -> Self {
+ Self {
+ created: Instant::now(),
+ recycled: None,
+ recycle_count: 0,
+ }
+ }
+}
+
+impl ObjectStatus {
+ /// Returns the instant when this object was created.
+ pub fn created(&self) -> Instant {
+ self.created
+ }
+
+ /// Returns the instant when this object was last used.
+ pub fn last_used(&self) -> Instant {
+ self.recycled.unwrap_or(self.created)
+ }
+
+ /// Returns the number of times the object was recycled.
+ pub fn recycle_count(&self) -> usize {
+ self.recycle_count
+ }
+}
+
+/// A trait whose instance creates new objects and recycles existing ones.
+pub trait ManageObject: Send + Sync {
+ /// The type of objects that this instance creates and recycles.
+ type Object: Send;
+
+ /// The type of errors that this instance can return.
+ type Error: Send;
+
+ /// Creates a new object.
+ fn create(&self) -> impl Future<Output = Result<Self::Object, Self::Error>> + Send;
+
+ /// Whether the object `o` is recyclable.
+ ///
+ /// Returns `Ok(())` if the object is recyclable; otherwise, returns an error.
+ fn is_recyclable(
+ &self,
+ o: &mut Self::Object,
+ status: &ObjectStatus,
+ ) -> impl Future<Output = Result<(), Self::Error>> + Send;
+
+ /// A callback invoked when an object is detached from the pool.
+ ///
+ /// If this instance does not hold any references to the object, then the default
+ /// implementation can be used which does nothing.
+ fn on_detached(&self, _o: &mut Self::Object) {}
+}
+
+/// Queue strategy when dequeuing objects from the object pool.
+#[derive(Debug, Default, Clone, Copy)]
+pub enum QueueStrategy {
+ /// First in first out.
+ ///
+ /// This strategy behaves like a queue.
+ #[default]
+ Fifo,
+ /// Last in first out.
+ ///
+ /// This strategy behaves like a stack.
+ Lifo,
+}
+
+/// Strategy when recycling object has been cancelled.
+///
+/// This enum controls the behavior when the recycling process (specifically the
+/// [`ManageObject::is_recyclable`] check) is cancelled; for example, when the
+/// `get()` future is dropped.
+#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
+pub enum RecycleCancelledStrategy {
+ /// Detach the object from the pool.
+ ///
+ /// This is the safest option. If the recycling check is cancelled, we assume the object might
+ /// be in an unknown state or that the check was taking too long for a reason. The object will
+ /// detach from the pool.
+ #[default]
+ Detach,
+
+ /// Return the object to the pool for potential reuse.
+ ///
+ /// This assumes that interrupting the check does not invalidate the object. The object is put
+ /// back into the pool.
+ ReturnToPool,
+}
diff --git a/asyncband/src/pool/mod.rs b/asyncband/src/pool/mod.rs
new file mode 100644
index 0000000..7b40053
--- /dev/null
+++ b/asyncband/src/pool/mod.rs
@@ -0,0 +1,202 @@
+// 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.
+
+//! Runtime-agnostic object pools for async Rust.
+//!
+//! This module provides two implementations: [bounded pool](bounded::Pool) and
+//! [unbounded pool](unbounded::Pool).
+//!
+//! # Bounded pool
+//!
+//! A bounded pool creates and recycles objects with full management. You _cannot_ put an object to
+//! the pool 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.
+//!
+//! Bounded pools are useful for pooling database connections.
+//!
+//! ## Examples
+//!
+//! The following example shows the core managed-pool workflow.
+//!
+//! ```
+//! use asyncband::pool::ManageObject;
+//! use asyncband::pool::ObjectStatus;
+//! use asyncband::pool::bounded::Pool;
+//! use asyncband::pool::bounded::PoolConfig;
+//!
+//! struct Compute;
+//! impl Compute {
+//! async fn do_work(&self) -> i32 {
+//! 42
+//! }
+//! }
+//!
+//! struct Manager;
+//! impl ManageObject for Manager {
+//! type Object = Compute;
+//! type Error = ();
+//!
+//! async fn create(&self) -> Result<Self::Object, Self::Error> {
+//! Ok(Compute)
+//! }
+//!
+//! async fn is_recyclable(
+//! &self,
+//! o: &mut Self::Object,
+//! status: &ObjectStatus,
+//! ) -> Result<(), Self::Error> {
+//! Ok(())
+//! }
+//! }
+//!
+//! # #[tokio::main]
+//! # async fn main() {
+//! let pool = Pool::new(PoolConfig::new(16), Manager);
+//! let o = pool.get().await.unwrap();
+//! assert_eq!(o.do_work().await, 42);
+//! # }
+//! ```
+//!
+//! # 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).
+//!
+//! 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
+//! implementation of [`ManageObject`].
+//!
+//! ## Examples
+//!
+//! The following example shows a manually populated unbounded pool.
+//!
+//! ```
+//! use asyncband::pool::unbounded::Pool;
+//! use asyncband::pool::unbounded::PoolConfig;
+//!
+//! # #[tokio::main]
+//! # async fn main() {
+//! let pool = Pool::<Vec<u8>>::never_manage(PoolConfig::default());
+//!
+//! let result = pool.get().await;
+//! assert_eq!(result.unwrap_err().to_string(), "unbounded pool is empty");
+//!
+//! pool.extend_one(Vec::with_capacity(1024));
+//! let o = pool.get().await.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.
+//!
+//! 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:
+//!
+//! ```rust,ignore
+//! use std::sync::Arc;
+//! use std::time::Duration;
+//!
+//! use asyncband::pool::bounded::Object;
+//! use asyncband::pool::bounded::Pool;
+//!
+//! #[derive(Debug, Clone)]
+//! pub struct ConnectionPool {
+//! pool: Arc<Pool<ManageConnection>>,
+//! }
+//!
+//! impl ConnectionPool {
+//! pub async fn acquire(&self) -> Result<Object<ManageConnection>, Error> {
+//! const ACQUIRE_TIMEOUT: Duration = Duration::from_secs(60);
+//!
+//! // note that users can choose any timer implementation here
+//! let result = tokio::time::timeout(ACQUIRE_TIMEOUT, self.pool.get()).await;
+//!
+//! // ... processing the result
+//! }
+//! }
+//! ```
+//!
+//! ## Why are 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.
+//!
+//! 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:
+//!
+//! ```
+//! use asyncband::pool::ManageObject;
+//! use asyncband::pool::ObjectStatus;
+//!
+//! struct Manager;
+//! impl ManageObject for Manager {
+//! type Object = i32;
+//! type Error = std::convert::Infallible;
+//!
+//! async fn create(&self) -> Result<Self::Object, Self::Error> {
+//! let o = 42;
+//! // any post-create hooks
+//! Ok(o)
+//! }
+//!
+//! async fn is_recyclable(
+//! &self,
+//! _o: &mut Self::Object,
+//! _status: &ObjectStatus,
+//! ) -> Result<(), Self::Error> {
+//! // any pre-recycle hooks
+//! // determine whether the object is recyclable
+//! // any post-recycle hooks
+//! Ok(())
+//! }
+//! }
+//! ```
+
+pub use common::ManageObject;
+pub use common::ObjectStatus;
+pub use common::QueueStrategy;
+pub use common::RecycleCancelledStrategy;
+pub use retain_spec::RetainResult;
+
+mod common;
+mod mutex;
+mod retain_spec;
+
+pub mod bounded;
+pub mod unbounded;
diff --git a/asyncband/src/pool/mutex.rs b/asyncband/src/pool/mutex.rs
new file mode 100644
index 0000000..028f9d3
--- /dev/null
+++ b/asyncband/src/pool/mutex.rs
@@ -0,0 +1,57 @@
+// 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<T: ?Sized>(std::sync::Mutex<T>);
+
+impl<T: ?Sized + fmt::Debug> fmt::Debug for Mutex<T> {
+ fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+ self.0.fmt(f)
+ }
+}
+
+impl<T> Mutex<T> {
+ 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
new file mode 100644
index 0000000..f4bd08b
--- /dev/null
+++ b/asyncband/src/pool/retain_spec.rs
@@ -0,0 +1,92 @@
+// 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<T> {
+ /// The number of retained objects.
+ pub retained: usize,
+ /// The objects removed from the pool.
+ pub removed: Vec<T>,
+}
+
+/// 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<T, State: SealedState<Object = T>>(
+ deque: &mut VecDeque<State>,
+ mut f: impl FnMut(&mut T, ObjectStatus) -> bool,
+) -> RetainResult<T> {
+ 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/unbounded.rs b/asyncband/src/pool/unbounded.rs
new file mode 100644
index 0000000..7e59f22
--- /dev/null
+++ b/asyncband/src/pool/unbounded.rs
@@ -0,0 +1,631 @@
+// 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.
+
+//! 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).
+//!
+//! 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
+//! implementation of [`ManageObject`].
+//!
+//! ## Examples
+//!
+//! Read the following simple demos or more complex examples in the examples directory.
+//!
+//! 1. Create an unbounded pool with [`NeverManageObject`]:
+//!
+//! ```
+//! use asyncband::pool::unbounded::Pool;
+//! use asyncband::pool::unbounded::PoolConfig;
+//!
+//! # #[tokio::main]
+//! # async fn main() {
+//! let pool = Pool::<Vec<u8>>::never_manage(PoolConfig::default());
+//!
+//! let result = pool.get().await;
+//! assert_eq!(result.unwrap_err().to_string(), "unbounded pool is empty");
+//!
+//! pool.extend_one(Vec::with_capacity(1024));
+//! let o = pool.get().await.unwrap();
+//! assert_eq!(o.capacity(), 1024);
+//! drop(o);
+//! let o = pool.get().await.unwrap();
+//! assert_eq!(o.capacity(), 1024);
+//! let result = pool.get().await;
+//! assert_eq!(result.unwrap_err().to_string(), "unbounded pool is empty");
+//! # }
+//! ```
+//!
+//! 2. Create an unbounded pool with a custom [`ManageObject`] (object factory):
+//!
+//! ```
+//! use asyncband::pool::ManageObject;
+//! use asyncband::pool::ObjectStatus;
+//! use asyncband::pool::unbounded::Pool;
+//! use asyncband::pool::unbounded::PoolConfig;
+//!
+//! struct Compute;
+//! impl Compute {
+//! async fn do_work(&self) -> i32 {
+//! 42
+//! }
+//! }
+//!
+//! struct Manager;
+//! impl ManageObject for Manager {
+//! type Object = Compute;
+//! type Error = ();
+//!
+//! async fn create(&self) -> Result<Self::Object, Self::Error> {
+//! Ok(Compute)
+//! }
+//!
+//! async fn is_recyclable(
+//! &self,
+//! o: &mut Self::Object,
+//! status: &ObjectStatus,
+//! ) -> Result<(), Self::Error> {
+//! Ok(())
+//! }
+//! }
+//!
+//! # #[tokio::main]
+//! # async fn main() {
+//! let pool = Pool::new(PoolConfig::default(), Manager);
+//! let o = pool.get().await.unwrap();
+//! assert_eq!(o.do_work().await, 42);
+//! # }
+//! ```
+
+use std::collections::VecDeque;
+use std::future::Future;
+use std::ops::Deref;
+use std::ops::DerefMut;
+use std::sync::Arc;
+use std::sync::Weak;
+
+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;
+
+/// The configuration of [`Pool`].
+#[derive(Clone, Copy, Debug)]
+#[non_exhaustive]
+pub struct PoolConfig {
+ /// Queue strategy of the [`Pool`].
+ ///
+ /// Determines the order of objects being queued and dequeued.
+ pub queue_strategy: QueueStrategy,
+
+ /// Strategy when recycling object has been cancelled.
+ pub recycle_cancelled_strategy: RecycleCancelledStrategy,
+}
+
+impl Default for PoolConfig {
+ fn default() -> Self {
+ Self::new()
+ }
+}
+
+impl PoolConfig {
+ /// Creates a new [`PoolConfig`].
+ pub fn new() -> Self {
+ Self {
+ queue_strategy: QueueStrategy::default(),
+ recycle_cancelled_strategy: RecycleCancelledStrategy::default(),
+ }
+ }
+
+ /// Returns a new [`PoolConfig`] with the specified queue strategy.
+ pub fn with_queue_strategy(mut self, queue_strategy: QueueStrategy) -> Self {
+ self.queue_strategy = queue_strategy;
+ self
+ }
+
+ /// Returns a new [`PoolConfig`] with the specified recycle cancelled strategy.
+ pub fn with_recycle_cancelled_strategy(
+ mut self,
+ recycle_cancelled_strategy: RecycleCancelledStrategy,
+ ) -> Self {
+ self.recycle_cancelled_strategy = recycle_cancelled_strategy;
+ self
+ }
+}
+
+/// The current pool status.
+///
+/// See [`Pool::status`].
+#[derive(Clone, Copy, Debug)]
+#[non_exhaustive]
+pub struct PoolStatus {
+ /// The current size of the pool.
+ pub current_size: usize,
+
+ /// The number of idle objects in the pool.
+ pub idle_count: usize,
+}
+
+/// The default [`ManageObject`] implementation for unbounded pool.
+///
+/// * [`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<T: Send + Sync> {
+ _marker: std::marker::PhantomData<T>,
+}
+
+impl<T: Send + Sync> Default for NeverManageObject<T> {
+ fn default() -> Self {
+ Self {
+ _marker: std::marker::PhantomData,
+ }
+ }
+}
+
+/// The error returned by [`NeverManageObject::create`].
+pub struct PoolIsEmpty(());
+
+impl std::fmt::Debug for PoolIsEmpty {
+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ write!(f, "unbounded pool is empty")
+ }
+}
+
+impl std::fmt::Display for PoolIsEmpty {
+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ std::fmt::Debug::fmt(self, f)
+ }
+}
+
+impl std::error::Error for PoolIsEmpty {}
+
+impl<T: Send + Sync> ManageObject for NeverManageObject<T> {
+ type Object = T;
+ type Error = PoolIsEmpty;
+
+ fn create(&self) -> impl Future<Output = Result<Self::Object, Self::Error>> + Send {
+ std::future::ready(Err(PoolIsEmpty(())))
+ }
+
+ fn is_recyclable(
+ &self,
+ _: &mut Self::Object,
+ _: &ObjectStatus,
+ ) -> impl Future<Output = Result<(), Self::Error>> + Send {
+ std::future::ready(Ok(()))
+ }
+}
+
+/// Generic runtime-agnostic unbounded object pool.
+///
+/// See the [module level documentation](self) for more.
+pub struct Pool<T, M: ManageObject<Object = T> = NeverManageObject<T>> {
+ config: PoolConfig,
+ manager: M,
+
+ /// A deque that holds the objects.
+ slots: Mutex<PoolDeque<ObjectState<T>>>,
+}
+
+#[derive(Debug)]
+struct PoolDeque<T> {
+ deque: VecDeque<T>,
+ current_size: usize,
+}
+
+impl<T, M> std::fmt::Debug for Pool<T, M>
+where
+ T: std::fmt::Debug,
+ M: ManageObject<Object = T>,
+{
+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ f.debug_struct("Pool")
+ .field("slots", &self.slots)
+ .field("config", &self.config)
+ .finish()
+ }
+}
+
+// Methods for `Pool` with `NeverManageObject`.
+impl<T: Send + Sync> Pool<T> {
+ /// Creates a new [`Pool`] from config and the [`NeverManageObject`].
+ pub fn never_manage(config: PoolConfig) -> Arc<Self> {
+ Self::new(config, NeverManageObject::<T>::default())
+ }
+
+ /// 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.
+ pub async fn get_or_create<E, F>(self: &Arc<Self>, f: F) -> Result<Object<T>, E>
+ where
+ F: AsyncFnOnce() -> Result<T, E> + 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),
+ })
+ }
+ }
+ }
+}
+
+impl<T, M: ManageObject<Object = T>> Pool<T, M> {
+ /// Creates a new [`Pool`] with config and the specified [`ManageObject`].
+ pub fn new(config: PoolConfig, manager: M) -> Arc<Self> {
+ let slots = Mutex::new(PoolDeque {
+ deque: VecDeque::new(),
+ current_size: 0,
+ });
+
+ Arc::new(Self {
+ config,
+ manager,
+ slots,
+ })
+ }
+
+ /// Retrieves an [`Object`] from this [`Pool`].
+ ///
+ /// This method should be called with a pool wrapped in an [`Arc`].
+ pub async fn get(self: &Arc<Self>) -> Result<Object<T, M>, 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(),
+ };
+
+ match existing {
+ None => {
+ let object = self.manager.create().await?;
+ let state = ObjectState {
+ o: object,
+ status: ObjectStatus::default(),
+ };
+ self.slots.lock().current_size += 1;
+ break Object {
+ state: Some(state),
+ pool: Arc::downgrade(self),
+ };
+ }
+ Some(object) => {
+ let mut unready_object = UnreadyObject {
+ state: Some(object),
+ pool: Arc::downgrade(self),
+ recycle_cancelled_strategy: self.config.recycle_cancelled_strategy,
+ };
+
+ let state = unready_object.state();
+ let status = state.status;
+ if self
+ .manager
+ .is_recyclable(&mut state.o, &status)
+ .await
+ .is_ok()
+ {
+ state.status.recycle_count += 1;
+ state.status.recycled = Some(std::time::Instant::now());
+ break unready_object.ready();
+ } else {
+ // We need to manually detach here as the drop implementation
+ // depends on the recycle cancelled strategy.
+ unready_object.detach();
+ }
+ }
+ };
+ };
+
+ Ok(object)
+ }
+
+ /// Extends the pool with exactly one object.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use asyncband::pool::unbounded::Pool;
+ /// use asyncband::pool::unbounded::PoolConfig;
+ ///
+ /// let config = PoolConfig::default();
+ /// let pool = Pool::never_manage(config);
+ ///
+ /// pool.extend_one(Vec::<i64>::with_capacity(1024));
+ /// ```
+ pub fn extend_one(&self, o: T) {
+ self.extend(Some(o));
+ }
+
+ /// Extends the pool with the objects of an iterator.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use asyncband::pool::unbounded::Pool;
+ /// use asyncband::pool::unbounded::PoolConfig;
+ ///
+ /// let config = PoolConfig::default();
+ /// let pool = Pool::never_manage(config);
+ ///
+ /// pool.extend([
+ /// Vec::<i64>::with_capacity(1024),
+ /// Vec::<i64>::with_capacity(512),
+ /// Vec::<i64>::with_capacity(256),
+ /// ]);
+ /// ```
+ pub fn extend(&self, iter: impl IntoIterator<Item = T>) {
+ let mut slots = self.slots.lock();
+ for o in iter {
+ slots.current_size += 1;
+ slots.deque.push_back(ObjectState {
+ o,
+ status: ObjectStatus::default(),
+ });
+ }
+ }
+
+ /// Retains only the objects that pass the given predicate.
+ ///
+ /// This function blocks the entire pool. Therefore, the given function should not block.
+ ///
+ /// 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
+ /// terminate if the pool is dropped.
+ ///
+ /// ```rust,ignore
+ /// let interval = Duration::from_secs(30);
+ /// let max_age = Duration::from_secs(60);
+ ///
+ /// let weak_pool = Arc::downgrade(&pool);
+ /// tokio::spawn(async move {
+ /// loop {
+ /// tokio::time::sleep(interval).await;
+ /// if let Some(pool) = weak_pool.upgrade() {
+ /// pool.retain(|_, status| status.last_used().elapsed() < max_age);
+ /// } else {
+ /// break;
+ /// }
+ /// }
+ /// });
+ /// ```
+ pub fn retain(
+ &self,
+ f: impl FnMut(&mut M::Object, ObjectStatus) -> bool,
+ ) -> RetainResult<M::Object> {
+ let mut slots = self.slots.lock();
+ let result = retain_spec::do_vec_deque_retain(&mut slots.deque, f);
+ slots.current_size -= result.removed.len();
+ 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)
+ 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,
+ }
+ }
+
+ fn push_back(&self, o: ObjectState<T>) {
+ let mut slots = self.slots.lock();
+ slots.deque.push_back(o);
+ drop(slots);
+ }
+
+ fn detach_object(&self, o: &mut T) {
+ let mut slots = self.slots.lock();
+ slots.current_size -= 1;
+ drop(slots);
+ self.manager.on_detached(o);
+ }
+}
+
+/// A wrapper of the actual pooled object.
+///
+/// This object implements [`Deref`] and [`DerefMut`]. You can use it as if it was of type `T`.
+///
+/// This object implements [`Drop`] that returns the underlying object to the pool on drop. You may
+/// call [`Object::detach`] to detach the object from the pool before dropping it.
+pub struct Object<T, M: ManageObject<Object = T> = NeverManageObject<T>> {
+ state: Option<ObjectState<T>>,
+ pool: Weak<Pool<T, M>>,
+}
+
+impl<T, M> std::fmt::Debug for Object<T, M>
+where
+ T: std::fmt::Debug,
+ M: ManageObject<Object = T>,
+{
+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ f.debug_struct("Object")
+ .field("state", &self.state)
+ .finish()
+ }
+}
+
+impl<T, M: ManageObject<Object = T>> Drop for Object<T, M> {
+ fn drop(&mut self) {
+ if let Some(state) = self.state.take() {
+ if let Some(pool) = self.pool.upgrade() {
+ pool.push_back(state);
+ }
+ }
+ }
+}
+
+impl<T, M: ManageObject<Object = T>> Deref for Object<T, M> {
+ type Target = T;
+ fn deref(&self) -> &T {
+ // SAFETY: `state` is always `Some` when `Object` is owned.
+ &self.state.as_ref().unwrap().o
+ }
+}
+
+impl<T, M: ManageObject<Object = T>> DerefMut for Object<T, M> {
+ fn deref_mut(&mut self) -> &mut Self::Target {
+ // SAFETY: `state` is always `Some` when `Object` is owned.
+ &mut self.state.as_mut().unwrap().o
+ }
+}
+
+impl<T, M: ManageObject<Object = T>> AsRef<M::Object> for Object<T, M> {
+ fn as_ref(&self) -> &M::Object {
+ self
+ }
+}
+
+impl<T, M: ManageObject<Object = T>> AsMut<M::Object> for Object<T, M> {
+ fn as_mut(&mut self) -> &mut M::Object {
+ self
+ }
+}
+
+impl<T, M: ManageObject<Object = T>> Object<T, M> {
+ /// Detaches the object from the [`Pool`].
+ ///
+ /// This reduces the size of the pool by one.
+ pub fn detach(mut self) -> M::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);
+ }
+ o
+ }
+
+ /// Returns the status of the object.
+ pub fn status(&self) -> ObjectStatus {
+ // SAFETY: `state` is always `Some` when `Object` is owned.
+ self.state.as_ref().unwrap().status
+ }
+}
+
+/// A wrapper of ObjectState used during the `is_recyclable` check in `Pool::get`.
+///
+/// If the check passes, the object is converted to a ready `Object` via `ready()`.
+/// If the check fails, `detach()` should be called to permanently remove the object
+/// from the pool. If dropped without calling either method (due to being cancelled),
+/// the behavior depends on the pool's [`RecycleCancelledStrategy`] configuration.
+struct UnreadyObject<T, M: ManageObject<Object = T> = NeverManageObject<T>> {
+ state: Option<ObjectState<T>>,
+ pool: Weak<Pool<T, M>>,
+ recycle_cancelled_strategy: RecycleCancelledStrategy,
+}
+
+impl<T, M: ManageObject<Object = T>> Drop for UnreadyObject<T, M> {
+ fn drop(&mut self) {
+ if let Some(mut state) = self.state.take() {
+ if let Some(pool) = self.pool.upgrade() {
+ match self.recycle_cancelled_strategy {
+ RecycleCancelledStrategy::Detach => {
+ pool.detach_object(&mut state.o);
+ }
+ RecycleCancelledStrategy::ReturnToPool => {
+ pool.push_back(state);
+ }
+ }
+ }
+ }
+ }
+}
+
+impl<T, M: ManageObject<Object = T>> UnreadyObject<T, M> {
+ fn ready(mut self) -> Object<T, M> {
+ // SAFETY: `state` is always `Some` when `UnreadyObject` is owned.
+ let state = Some(self.state.take().unwrap());
+ let pool = self.pool.clone();
+ Object { state, pool }
+ }
+
+ 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);
+ }
+ }
+ }
+
+ fn state(&mut self) -> &mut ObjectState<T> {
+ // SAFETY: `state` is always `Some` when `UnreadyObject` is owned.
+ self.state.as_mut().unwrap()
+ }
+}
+
+#[derive(Debug)]
+struct ObjectState<T> {
+ o: T,
+ status: ObjectStatus,
+}
+
+impl<T> retain_spec::SealedState for ObjectState<T> {
+ 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/tests-integration/Cargo.toml b/tests-integration/Cargo.toml
index 104171c..658dba3 100644
--- a/tests-integration/Cargo.toml
+++ b/tests-integration/Cargo.toml
@@ -38,6 +38,7 @@ asyncband = { workspace = true, features = [
"once-cell",
"once-map",
"oneshot",
+ "pool",
"rwlock",
"semaphore",
"shutdown",
diff --git a/tests-integration/tests/pool_recycle_cancelled_test.rs b/tests-integration/tests/pool_recycle_cancelled_test.rs
new file mode 100644
index 0000000..e527f20
--- /dev/null
+++ b/tests-integration/tests/pool_recycle_cancelled_test.rs
@@ -0,0 +1,311 @@
+// 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::sync::Arc;
+use std::sync::atomic::AtomicUsize;
+use std::sync::atomic::Ordering;
+use std::time::Duration;
+
+use asyncband::pool::ManageObject;
+use asyncband::pool::ObjectStatus;
+use asyncband::pool::RecycleCancelledStrategy;
+
+struct SlowRecycleManager {
+ created_count: Arc<AtomicUsize>,
+ recycle_delay: Duration,
+}
+
+impl SlowRecycleManager {
+ fn new(created_count: Arc<AtomicUsize>, recycle_delay: Duration) -> Self {
+ Self {
+ created_count,
+ recycle_delay,
+ }
+ }
+}
+
+impl ManageObject for SlowRecycleManager {
+ type Object = usize;
+ type Error = Infallible;
+
+ async fn create(&self) -> Result<Self::Object, Self::Error> {
+ let id = self.created_count.fetch_add(1, Ordering::SeqCst);
+ Ok(id)
+ }
+
+ async fn is_recyclable(
+ &self,
+ _o: &mut Self::Object,
+ _status: &ObjectStatus,
+ ) -> Result<(), Self::Error> {
+ tokio::time::sleep(self.recycle_delay).await;
+ Ok(())
+ }
+}
+
+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");
+
+ tokio::time::sleep(Duration::from_millis(10)).await;
+
+ let status = pool.status();
+ assert_eq!(
+ status.current_size, 0,
+ "Pool size should be 0 after cancelled get() with Detach behavior"
+ );
+
+ 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"
+ );
+ }
+
+ /// 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)
+ .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);
+
+ 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 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");
+ }
+
+ /// 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)
+ .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);
+
+ for _ in 0..5 {
+ let _ = tokio::time::timeout(Duration::from_millis(10), pool.get()).await;
+ tokio::time::sleep(Duration::from_millis(5)).await;
+ }
+
+ let status = pool.status();
+ assert_eq!(
+ status.current_size, 3,
+ "Pool size should be preserved after multiple cancelled gets"
+ );
+ }
+
+ /// 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);
+ 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"
+ );
+ }
+}
+
+mod unbounded_tests {
+ use asyncband::pool::unbounded::Pool;
+ use asyncband::pool::unbounded::PoolConfig;
+
+ 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));
+ 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 status = pool.status();
+ assert_eq!(
+ status.current_size, 0,
+ "Pool size should be 0 after cancelled get() with Detach behavior"
+ );
+
+ 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"
+ );
+ }
+
+ /// 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));
+ 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);
+
+ 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 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");
+ }
+
+ /// 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));
+ 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);
+
+ for _ in 0..5 {
+ let _ = tokio::time::timeout(Duration::from_millis(10), pool.get()).await;
+ tokio::time::sleep(Duration::from_millis(5)).await;
+ }
+
+ let status = pool.status();
+ assert_eq!(
+ status.current_size, 3,
+ "Pool size should be preserved after multiple cancelled gets"
+ );
+ }
+}
diff --git a/tests-integration/tests/pool_replenish_test.rs b/tests-integration/tests/pool_replenish_test.rs
new file mode 100644
index 0000000..7f0535e
--- /dev/null
+++ b/tests-integration/tests/pool_replenish_test.rs
@@ -0,0 +1,78 @@
+// 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::sync::Arc;
+
+use asyncband::pool::ManageObject;
+use asyncband::pool::ObjectStatus;
+use asyncband::pool::bounded::Pool;
+use asyncband::pool::bounded::PoolConfig;
+
+#[tokio::test]
+async fn test_replenish() {
+ #[derive(Default)]
+ struct Manager;
+
+ impl ManageObject for Manager {
+ type Object = ();
+ type Error = Infallible;
+
+ async fn create(&self) -> Result<Self::Object, Self::Error> {
+ Ok(())
+ }
+
+ async fn is_recyclable(
+ &self,
+ _o: &mut Self::Object,
+ _status: &ObjectStatus,
+ ) -> Result<(), Self::Error> {
+ Ok(())
+ }
+ }
+
+ const MAX_SIZE: usize = 2;
+
+ fn make_default() -> Arc<Pool<Manager>> {
+ Pool::new(PoolConfig::new(MAX_SIZE), Manager)
+ }
+
+ for i in 0..5 {
+ let pool = make_default();
+ let n = pool.replenish(i).await;
+ assert_eq!(n, i.min(MAX_SIZE));
+ }
+
+ // stage one idle object
+ {
+ let pool = make_default();
+ pool.get().await.unwrap();
+ let n = pool.replenish(2).await;
+ assert_eq!(n, 1);
+ }
+
+ // stage two idle objects
+ {
+ let pool = make_default();
+ let o1 = pool.get().await.unwrap();
+ let o2 = pool.get().await.unwrap();
+ drop((o1, o2));
+
+ let n = pool.replenish(2).await;
+ assert_eq!(n, 0);
+ }
+}
diff --git a/tests-integration/tests/traits_test.rs b/tests-integration/tests/traits_test.rs
index e894246..f2e4a99 100644
--- a/tests-integration/tests/traits_test.rs
+++ b/tests-integration/tests/traits_test.rs
@@ -26,6 +26,9 @@ use asyncband::once::Once;
use asyncband::once::OnceCell;
use asyncband::once::OnceMap;
use asyncband::oneshot;
+use asyncband::pool;
+use asyncband::pool::ManageObject;
+use asyncband::pool::ObjectStatus;
use asyncband::rwlock::OwnedRwLockReadGuard;
use asyncband::rwlock::RwLock;
use asyncband::rwlock::RwLockReadGuard;
@@ -38,6 +41,25 @@ use asyncband::singleflight;
use asyncband::waitgroup::Wait;
use asyncband::waitgroup::WaitGroup;
+struct PoolManager;
+
+impl ManageObject for PoolManager {
+ type Object = i64;
+ type Error = std::convert::Infallible;
+
+ async fn create(&self) -> Result<Self::Object, Self::Error> {
+ Ok(0)
+ }
+
+ async fn is_recyclable(
+ &self,
+ _object: &mut Self::Object,
+ _status: &ObjectStatus,
+ ) -> Result<(), Self::Error> {
+ Ok(())
+ }
+}
+
#[test]
fn public_types_are_send_and_sync() {
fn assert_send_and_sync<T: Send + Sync>() {}
@@ -66,6 +88,10 @@ fn public_types_are_send_and_sync() {
assert_send_and_sync::<broadcast::overflow::TryRecvError>();
assert_send_and_sync::<oneshot::SendError<i64>>();
assert_send_and_sync::<oneshot::Sender<i64>>();
+ assert_send_and_sync::<pool::bounded::Pool<PoolManager>>();
+ assert_send_and_sync::<pool::bounded::Object<PoolManager>>();
+ assert_send_and_sync::<pool::unbounded::Pool<i64>>();
+ assert_send_and_sync::<pool::unbounded::Object<i64>>();
assert_send_and_sync::<mpsc::SendError<i64>>();
assert_send_and_sync::<mpsc::UnboundedSender<i64>>();
assert_send_and_sync::<mpsc::UnboundedReceiver<i64>>();
@@ -112,6 +138,10 @@ fn public_types_are_unpin() {
assert_unpin::<oneshot::SendError<i64>>();
assert_unpin::<oneshot::Receiver<i64>>();
assert_unpin::<oneshot::Recv<i64>>();
+ assert_unpin::<pool::bounded::Pool<PoolManager>>();
+ assert_unpin::<pool::bounded::Object<PoolManager>>();
+ assert_unpin::<pool::unbounded::Pool<i64>>();
+ assert_unpin::<pool::unbounded::Object<i64>>();
assert_unpin::<mpsc::SendError<i64>>();
assert_unpin::<mpsc::UnboundedSender<i64>>();
assert_unpin::<mpsc::UnboundedReceiver<i64>>();
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@asyncband.apache.org
For additional commands, e-mail: commits-help@asyncband.apache.org