diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index 364b65b..e8eb8bf 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -20,3 +20,14 @@ jobs: - run: cargo build --all-features --verbose - run: cargo test --all - run: cargo test --all --all-features + + msrv: + runs-on: ubuntu-latest + strategy: + matrix: + features: ["--no-default-features", "--all-features"] + steps: + - uses: actions/checkout@v4 + - uses: dtolnay/rust-toolchain@1.85.0 + # Dev targets use current stable; verify the consumer library graph here. + - run: cargo check -p auto_pool --lib ${{ matrix.features }} diff --git a/Cargo.toml b/Cargo.toml index 057f165..74d8020 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -10,7 +10,7 @@ edition = "2021" authors = ["Sild >`, `pool::AutoPool`, `config::AutoPoolConfig`, + `config::PickStrategy`, and `pool_object::PoolObject` as the canonical paths. + No factories, background workers, unsafe storage, or alternate public APIs. +- Every successful checkout removes exactly one item. Drop returns its mutated + value exactly once; `release()` removes it permanently. The wrapper borrows + the pool. Never hold a storage guard across an await or user work. +- Async waiters register then recheck storage. Each return sends an additional + notification; cancellation must forward an unconsumed notification. Sync + and async consumers can race for items, with no promised waiter fairness. +- Use one deadline across retries. Zero attempts immediately; an unrepresentable + deadline means unlimited waiting. Document scheduling/short mutex limitations. + Do not reintroduce polling via the ignored public `lock_duration` or + `sleep_duration` fields or remove these fields in a compatible release. +- The `async` feature is opt-in and executor-independent. Do not add a self + dev-dependency that enables it. Gate async examples, docs, and tests explicitly. +- Library MSRV is 1.85 because rand 0.10 requires it; dev targets use current + stable (Criterion needs 1.86+). Keep the workspace manifest and README aligned. + Cargo.lock is intentionally ignored. Release-plz owns version changes. +- Extend existing config/methods only for a demonstrated contract. Public fields, + paths, bounds and feature names are compatibility commitments. Update README, + rustdoc, examples, tests and this guide together when contracts change. +- Fast checks: `cargo test -p auto_pool --no-default-features` and + `cargo test -p auto_pool --no-default-features --features async`. +- Full affected checks, from the workspace root: + ```sh + cargo test --workspace + cargo test --workspace --all-features + cargo test -p auto_pool --examples --all-features + cargo run -p auto_pool --example main + cargo run -p auto_pool --example main --features async + cargo +nightly fmt --check + cargo clippy --workspace --all-targets --all-features -- -D warnings + RUSTDOCFLAGS="-D warnings" cargo doc -p auto_pool --no-deps --all-features + RUSTDOCFLAGS="-D warnings" cargo doc -p auto_pool --no-deps --no-default-features + cargo +1.85.0 check -p auto_pool --lib --no-default-features + cargo +1.85.0 check -p auto_pool --lib --all-features + cargo package -p auto_pool --list + cargo package -p auto_pool + git diff --check + ``` +- For dependency/package changes, extract `target/package/auto_pool-*.crate` + into a temporary directory outside the workspace. Run a consumer on Rust + 1.85 with default dependency features, `default-features = false`, and that + setting plus `features = ["async"]`. Exercise struct literals, checkout, + mutation, drop, add and release. Inspect the normalized manifest and package + inventory. Remove the temporary consumer afterward. +- Require one independent review for substantive concurrency/dependency changes. + Benchmarks in BENCHMARKS.md are scoped measurements, not universal speed claims. diff --git a/auto_pool/BENCHMARKS.md b/auto_pool/BENCHMARKS.md new file mode 100644 index 0000000..a47bdbe --- /dev/null +++ b/auto_pool/BENCHMARKS.md @@ -0,0 +1,59 @@ +# Benchmark workloads + +Run a bounded local sample from the workspace root: + +```sh +cargo bench -p auto_pool --bench multithread_push_pop --all-features -- \ + --warm-up-time 0.2 --measurement-time 0.5 --sample-size 10 --noplot +``` + +- `uncontended/{lifo,random,mutex_stack,allocate_1k}` measures checkout, a + mutation of a reused 1 KiB byte buffer, and return. Pools have 64 objects. + The mutex stack takes two locks and has no waiting or RAII wrapper. Allocation + includes zero-initialization, mutation and deallocation of a fresh buffer. +- `contention/auto_pool/{1,4}` uses four synchronized workers. One item forces + consumers to compete for availability; four items isolate storage contention + without item exhaustion. Each measured round is 400 operations (100 per + worker). The mutex-stack baseline has four objects; allocation has no shared + state. Thread creation and join are excluded from the returned measurement. + Start and finish barrier overhead is included once per sample. +- `exhaustion/try_empty` measures an unsuccessful zero-budget checkout. + `exhaustion/wait_1ms` includes the configured timeout and scheduler latency. +- `async/handoff_from_thread` registers an empty-pool future before requesting + an item from a producer thread. It sums the interval from just before `add()` + to consumer resumption and `release()`. Request-channel delay, listener setup, + and thread creation/join are excluded. It uses smol's local `block_on`, not + a loaded multi-task executor, and has no configured timeout. + +The harness uses `std::hint::black_box`. Grow-on-demand third-party pools are +not compared because their exhaustion behavior differs. Compare only matching +workloads and report the feature set, machine, toolchain and sample settings. +These measurements do not establish a universal speedup, waiter fairness, +tail latency under load, or superiority over other pool implementations. +`swap_remove` is a readability simplification, not a measured optimization. + +## Local sample: 2026-09-10 + +Apple M3 Max, macOS arm64, rustc 1.97.1, `--all-features`, Criterion 0.8.2; +10 samples, 0.2 s warmup and 0.5 s measurement per workload. Point estimates +from bounded runs (contention rerun after adding the untimed readiness +barrier; not a before/after comparison): + +| Workload | Estimate | +| --- | ---: | +| Uncontended LIFO | 32.6 ns / operation | +| Uncontended random | 37.8 ns / operation | +| Uncontended mutex stack | 4.94 ns / operation | +| Uncontended allocate 1 KiB | 25.5 ns / operation | +| Four workers, one pooled item | 57.3 us / 400 operations | +| Four workers, four pooled items | 59.8 us / 400 operations | +| Four workers, mutex stack | 4.50 us / 400 operations | +| Four workers, allocation | 17.6 us / 400 operations | +| Empty, zero budget | 39.6 ns / attempt | +| Empty, 1 ms budget | 1.25 ms / attempt | +| Async handoff from a thread | 2.84 us / handoff | + +The minimal stack and allocation baselines are cheaper in this deliberately +small workload. They omit waiting, notifications and borrowed RAII semantics. +The one-item and four-item cases also differ in scheduling and availability; +these results do not predict performance for a particular application. diff --git a/auto_pool/Cargo.toml b/auto_pool/Cargo.toml index 7cdbc83..6234063 100644 --- a/auto_pool/Cargo.toml +++ b/auto_pool/Cargo.toml @@ -12,20 +12,18 @@ rust-version.workspace = true publish = true [features] -async = ["dep:smol"] +async = ["dep:smol", "dep:event-listener"] [dependencies] parking_lot = "0.12" +event-listener = { version = "5.4", optional = true } smol = { version = "2.0", optional = true } rand = "0.10.0" [dev-dependencies] criterion = { version = "0.8" } -lockfree-object-pool = "0.1.6" -object-pool = "0.6.0" anyhow = "1.0" tokio = { version = "1.43.0", features = ["rt", "macros"] } -auto_pool = {path = "", features = ["async"]} [[bench]] name = "multithread_push_pop" diff --git a/auto_pool/README.md b/auto_pool/README.md index e951bcd..330a614 100644 --- a/auto_pool/README.md +++ b/auto_pool/README.md @@ -1,47 +1,101 @@ # auto_pool -This pool automatically manages the return of objects. +A small, thread-safe pool of caller-supplied objects. `AutoPool` stores a +`Mutex>`; checkout returns a borrowed `PoolObject<'_, T>` that implements +`Deref` and `DerefMut`. Dropping the wrapper returns the object, including any +mutations. Calling `release()` takes ownership permanently instead. -The primary difference from competitors is the interface - implementation follows the RAII pattern, meaning the user must provide objects when creating the pool. +```toml +[dependencies] +auto_pool = "0.3.3" +# Enable asynchronous checkout when needed: +# auto_pool = { version = "0.3.3", features = ["async"] } +``` -The `Pool` class has only three public methods: -- Constructor `with_config(config: Config, items)`: Creates a new pool with a custom configuration. -- Constructor `new(items)`: An alias for `with_config` that uses the default configuration. -- `take(&self)`: Extracts an object from the pool. +```rust +use auto_pool::config::AutoPoolConfig; +use auto_pool::pool::AutoPool; +use std::time::Duration; -The configuration allows you to set the wait duration for the `take` method (default is `Duration::MAX`, which essentially means "Wait until you receive it"). +# fn main() -> Result<(), Box> { +let pool = AutoPool::new_with_config( + AutoPoolConfig { + wait_duration: Duration::from_millis(20), + ..Default::default() + }, + [String::with_capacity(1024)], +); +{ + let mut buffer = pool.get().ok_or("pool exhausted")?; + buffer.push_str("reused allocation"); +} // the buffer returns here +let buffer = pool.get().ok_or("pool exhausted")?.release(); +assert_eq!(buffer, "reused allocation"); +assert_eq!(pool.size(), 0); +pool.add(buffer); +# Ok(()) +# } +``` +`AutoPool::new(items)` uses the default configuration. Constructors are +infallible. `add(item)` supplies or returns an item; `size()` counts only +available items; `shrink_to_fit()` shrinks available storage, not checked-out +objects. Items must be `Send + 'static`. The wrapper borrows its pool, so the +pool must outlive every checkout. Share the pool across threads with `Arc` or +scoped threads. The pool never creates replacement objects. -Examples: -```rust +## Waiting and selection -fn main() -> Result<(), autoreturn_pool::Error> { - // basic usage - let pool = autoreturn_pool::Pool::new([1, 2])?; - let item = pool.take()?.unwrap(); - - // with custom config - let config = autoreturn_pool::Config { - wait_duration: std::time::Duration::from_millis(5), - }; - let pool = autoreturn_pool::Pool::with_config(config, [1, 2])?; - let item = pool.take()?.unwrap(); -} -``` +`get()` waits up to `wait_duration` and returns `None` on timeout. Retries use +one overall deadline, including initial synchronous mutex acquisition. +`Duration::ZERO` performs an immediate attempt; `Duration::MAX` (and other +values that overflow the platform's `Instant`) means an unlimited wait. + +Timeouts are not hard real-time guarantees: scheduling and mutex reacquisition +can delay completion. An object available when a waiter resumes may win a race +with its deadline. There is no FIFO fairness guarantee between consumers. +Avoid holding all objects while requesting another with an unlimited wait. + +`PickStrategy::LIFO` is the default. `PickStrategy::RANDOM` selects from the +available items using a random index; it changes object selection, not waiter +priority. + +## Async feature + +The optional `async` feature adds `get_async()`. It works on any executor; +finite deadlines use smol's timer. Exhausted pools suspend through +[event-listener](https://docs.rs/event-listener/5.4.2/event_listener/struct.Event.html) +notifications, with no blocking item wait or sleep polling. Checkout still +briefly locks the storage mutex, as do `add()` and wrapper drop. Avoid expensive +pool maintenance on latency-sensitive executor threads. ```rust -// with custom object: -#[derive(Default)] -struct MyObject { - value: i32, -} -fn main() -> Result<(), autoreturn_pool::Error> { - let pool_objects = [ - MyObject::default(), - MyObject::default() - ]; - let pool = autoreturn_pool::Pool::new(pool_objects)?; - let mut item = pool.take()?.unwrap(); - Ok(()) -} +# #[cfg(feature = "async")] +# fn main() -> Result<(), Box> { +# smol::block_on(async { +use auto_pool::pool::AutoPool; +let pool = AutoPool::new([vec![0u8; 1024]]); +let mut buffer = pool.get_async().await.ok_or("pool exhausted")?; +buffer[0] = 42; +drop(buffer); // also wakes waiting consumers +# Ok(()) +# }) +# } +# #[cfg(not(feature = "async"))] +# fn main() {} ``` + +Dropping a pending checkout cancels it without consuming an item. Sync and +async consumers can use the same pool. `lock_duration` and `sleep_duration` +remain public for source compatibility but are ignored; `wait_duration` is +the only waiting setting. Their old defaults remain unchanged. + +## Toolchains and development + +The library requires **Rust 1.85**, matching its existing `rand 0.10` +dependency. The old 1.81 declaration was inaccurate. Development tests and +benchmarks use current stable Rust (Criterion 0.8 requires at least 1.86), and +formatting uses nightly. Release-plz manages crate versions. + +See [AGENTS.md](AGENTS.md) for maintainer checks and +[BENCHMARKS.md](BENCHMARKS.md) for benchmark workloads and measurement limits. diff --git a/auto_pool/benches/multithread_push_pop.rs b/auto_pool/benches/multithread_push_pop.rs index 967c576..cd37aff 100644 --- a/auto_pool/benches/multithread_push_pop.rs +++ b/auto_pool/benches/multithread_push_pop.rs @@ -1,78 +1,166 @@ +use auto_pool::config::{AutoPoolConfig, PickStrategy}; use auto_pool::pool::AutoPool; -use criterion::{black_box, criterion_group, criterion_main, Criterion}; -use std::sync::Arc; +use criterion::{criterion_group, criterion_main, BenchmarkId, Criterion, Throughput}; +use parking_lot::Mutex; +use std::hint::black_box; +use std::sync::Barrier; +use std::time::{Duration, Instant}; -struct DummyObject { - id: usize, - value: usize, -} +const WORKERS: usize = 4; +const BATCH: u64 = 100; +const BUFFER_SIZE: usize = 1024; -const POOL_SIZE: usize = 1000; -const OPERATIONS_PER_THREAD: usize = 100; -const THREADS_COUNT: usize = 64; +fn buffer() -> Vec { vec![0; BUFFER_SIZE] } -fn run_tests(arc_pool: Arc

, pool_op: fn(&P)) { - let threads: Vec<_> = (0..THREADS_COUNT) - .map(|_| { - let pool = arc_pool.clone(); - std::thread::spawn(move || { - for _ in 0..OPERATIONS_PER_THREAD { - pool_op(&pool); - } - }) - }) - .collect(); - for thread in threads { - thread.join().unwrap(); - } +fn touch(buffer: &mut [u8]) { + buffer[0] = buffer[0].wrapping_add(1); + black_box(buffer); } -fn perf_auto_pool(pool: Arc>) { - run_tests(pool, |pool| { - let obj = pool.get().unwrap(); - let _id = &obj.id; - let _val = &obj.value; - }); +// Thread startup and joining are excluded from the returned duration. The start +// and finish barriers are measured once per sample, amortized over all rounds. +fn parallel_rounds(rounds: u64, operation: impl Fn() + Sync) -> Duration { + let ready = Barrier::new(WORKERS + 1); + let start = Barrier::new(WORKERS + 1); + let finish = Barrier::new(WORKERS + 1); + std::thread::scope(|scope| { + let workers: Vec<_> = (0..WORKERS) + .map(|_| { + let (ready, start, finish, operation) = (&ready, &start, &finish, &operation); + scope.spawn(move || { + ready.wait(); + start.wait(); + for _ in 0..rounds { + for _ in 0..BATCH { + operation(); + } + } + finish.wait(); + }) + }) + .collect(); + ready.wait(); + let before = Instant::now(); + start.wait(); + finish.wait(); + let elapsed = before.elapsed(); + for worker in workers { + worker.join().unwrap(); + } + elapsed + }) } -fn perf_lockfree_pool(pool: Arc>) { - run_tests(pool, |pool| { - let obj = pool.pull(); - let _id = &obj.id; - let _val = &obj.value; +fn uncontended(c: &mut Criterion) { + let mut group = c.benchmark_group("uncontended"); + for (name, strategy) in [("lifo", PickStrategy::LIFO), ("random", PickStrategy::RANDOM)] { + let pool = AutoPool::new_with_config( + AutoPoolConfig { + pick_strategy: strategy, + ..Default::default() + }, + (0..64).map(|_| buffer()), + ); + group.bench_function(name, |b| b.iter(|| touch(&mut pool.get().unwrap()))); + } + let stack = Mutex::new((0..64).map(|_| buffer()).collect::>()); + group.bench_function("mutex_stack", |b| { + b.iter(|| { + let mut item = stack.lock().pop().unwrap(); + touch(&mut item); + stack.lock().push(item); + }); }); + group.bench_function("allocate_1k", |b| b.iter(|| touch(&mut black_box(buffer())))); + group.finish(); } -fn perf_object_pool(pool: Arc>) { - run_tests(pool, |pool| { - let obj = pool.pull(|| DummyObject { id: 0, value: 1 }); - let _id = &obj.id; - let _val = &obj.value; +fn contention(c: &mut Criterion) { + let mut group = c.benchmark_group("contention"); + group.throughput(Throughput::Elements(WORKERS as u64 * BATCH)); + for size in [1, WORKERS] { + let pool = AutoPool::new((0..size).map(|_| buffer())); + group.bench_with_input(BenchmarkId::new("auto_pool", size), &size, |b, _| { + b.iter_custom(|rounds| parallel_rounds(rounds, || touch(&mut pool.get().unwrap()))); + }); + } + // Enough items for every worker; this baseline has no exhaustion policy. + let stack = Mutex::new((0..WORKERS).map(|_| buffer()).collect::>()); + group.bench_function("mutex_stack_4", |b| { + b.iter_custom(|rounds| { + parallel_rounds(rounds, || { + let mut item = stack.lock().pop().unwrap(); + touch(&mut item); + stack.lock().push(item); + }) + }); }); + group.bench_function("allocate_1k", |b| { + b.iter_custom(|rounds| parallel_rounds(rounds, || touch(&mut black_box(buffer())))); + }); + group.finish(); } -fn benchmark_functions(c: &mut Criterion) { - let auto_pool = Arc::new(AutoPool::new((0..POOL_SIZE).map(|id| DummyObject { id, value: 1 }))); - c.bench_function("auto_pool", |b| b.iter(|| perf_auto_pool(black_box(auto_pool.clone())))); +fn exhaustion(c: &mut Criterion) { + let pool = AutoPool::>::new_with_config( + AutoPoolConfig { + wait_duration: Duration::ZERO, + ..Default::default() + }, + [], + ); + c.bench_function("exhaustion/try_empty", |b| b.iter(|| assert!(black_box(pool.get()).is_none()))); + let pool = AutoPool::>::new_with_config( + AutoPoolConfig { + wait_duration: Duration::from_millis(1), + ..Default::default() + }, + [], + ); + c.bench_function("exhaustion/wait_1ms", |b| b.iter(|| assert!(black_box(pool.get()).is_none()))); +} - // Nice interface... - let lockfree_pool = { - let pool = lockfree_object_pool::LinearObjectPool::new(|| DummyObject { id: 0, value: 1 }, |_| {}); - { - let mut items = Vec::with_capacity(POOL_SIZE); - for _ in 0..POOL_SIZE { - items.push(pool.pull()); +#[cfg(feature = "async")] +fn async_handoff(c: &mut Criterion) { + // Poll to Pending before asking the producer for an item. Measure from just + // before add() to consumer resumption, excluding the request-channel delay. + let pool = AutoPool::new([]); + let (request, requests) = std::sync::mpsc::channel(); + std::thread::scope(|scope| { + let pool = &pool; + let producer = scope.spawn(move || { + while requests.recv().is_ok() { + pool.add(Instant::now()); } - } - pool - }; - let lockfree_pool = Arc::new(lockfree_pool); - c.bench_function("lockfree_pool", |b| b.iter(|| perf_lockfree_pool(black_box(lockfree_pool.clone())))); + }); + c.bench_function("async/handoff_from_thread", |b| { + b.iter_custom(|iterations| { + smol::block_on(async { + let mut elapsed = Duration::ZERO; + for _ in 0..iterations { + let mut get = Box::pin(pool.get_async()); + assert!(smol::future::poll_once(&mut get).await.is_none()); + request.send(()).unwrap(); + let sent = get.await.unwrap().release(); + elapsed += sent.elapsed(); + black_box(sent); + } + elapsed + }) + }); + }); + drop(request); + producer.join().unwrap(); + }); +} - let object_pool = object_pool::Pool::new(POOL_SIZE, || DummyObject { id: 0, value: 1 }); - let object_pool = Arc::new(object_pool); - c.bench_function("object_pool", |b| b.iter(|| perf_object_pool(black_box(object_pool.clone())))); +fn benchmarks(c: &mut Criterion) { + uncontended(c); + contention(c); + exhaustion(c); + #[cfg(feature = "async")] + async_handoff(c); } -criterion_group!(benches, benchmark_functions); +criterion_group!(benches, benchmarks); criterion_main!(benches); diff --git a/auto_pool/examples/main.rs b/auto_pool/examples/main.rs index 689ba38..d29220f 100644 --- a/auto_pool/examples/main.rs +++ b/auto_pool/examples/main.rs @@ -8,6 +8,7 @@ async fn main() -> anyhow::Result<()> { single_thread()?; multi_thread()?; add_release()?; + #[cfg(feature = "async")] add_release_async().await?; Ok(()) } @@ -92,6 +93,7 @@ fn add_release() -> anyhow::Result<()> { Ok(()) } +#[cfg(feature = "async")] async fn add_release_async() -> anyhow::Result<()> { let objects = [MyObject { value: 1 }]; let pool = AutoPool::new(objects); diff --git a/auto_pool/src/async_tests.rs b/auto_pool/src/async_tests.rs new file mode 100644 index 0000000..c84fa46 --- /dev/null +++ b/auto_pool/src/async_tests.rs @@ -0,0 +1,186 @@ +use crate::config::AutoPoolConfig; +use crate::pool::AutoPool; +use std::future::Future; +use std::pin::Pin; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::Arc; +use std::task::{Context, Poll, Wake, Waker}; +use std::time::{Duration, Instant}; + +#[derive(Default)] +struct WakeCount(AtomicUsize); + +impl Wake for WakeCount { + fn wake(self: Arc) { self.wake_by_ref(); } + fn wake_by_ref(self: &Arc) { self.0.fetch_add(1, Ordering::SeqCst); } +} + +fn poll(future: Pin<&mut F>, wake: &Arc) -> Poll { + future.poll(&mut Context::from_waker(&Waker::from(wake.clone()))) +} + +#[test] +fn test_multiple_returns_wake_multiple_waiters() { + let pool = AutoPool::new([]); + let wakes: Vec<_> = (0..3).map(|_| Arc::new(WakeCount::default())).collect(); + let mut gets: Vec<_> = (0..3).map(|_| Box::pin(pool.get_async())).collect(); + for (get, wake) in gets.iter_mut().zip(&wakes) { + assert!(poll(get.as_mut(), wake).is_pending()); + } + for value in 0..3 { + pool.add(value); + } + let mut values = Vec::new(); + for (get, wake) in gets.iter_mut().zip(&wakes) { + assert!(wake.0.load(Ordering::SeqCst) > 0); + match poll(get.as_mut(), wake) { + Poll::Ready(Some(item)) => values.push(item.release()), + _ => panic!("a returned object was stranded"), + } + } + values.sort_unstable(); + assert_eq!(values, [0, 1, 2]); + assert_eq!(pool.size(), 0); +} + +#[test] +fn test_cancelling_notified_waiter_forwards_wakeup() { + let pool = AutoPool::new([]); + let wake1 = Arc::new(WakeCount::default()); + let wake2 = Arc::new(WakeCount::default()); + let mut first = Box::pin(pool.get_async()); + let mut second = Box::pin(pool.get_async()); + assert!(poll(first.as_mut(), &wake1).is_pending()); + assert!(poll(second.as_mut(), &wake2).is_pending()); + pool.add(42); + assert!(wake1.0.load(Ordering::SeqCst) > 0); + drop(first); + assert!(wake2.0.load(Ordering::SeqCst) > 0); + assert_eq!(smol::block_on(second).unwrap().release(), 42); + assert_eq!(pool.size(), 0); +} + +#[test] +fn test_cancelling_unnotified_waiter_preserves_future_returns() { + let pool = AutoPool::new([]); + let mut first = Box::pin(pool.get_async()); + assert!(smol::block_on(smol::future::poll_once(&mut first)).is_none()); + drop(first); + pool.add(42); + assert_eq!(smol::block_on(pool.get_async()).unwrap().release(), 42); +} + +#[test] +fn test_sync_consumer_can_win_async_notification() { + let pool = AutoPool::new([]); + let wake = Arc::new(WakeCount::default()); + let mut get = Box::pin(pool.get_async()); + assert!(poll(get.as_mut(), &wake).is_pending()); + pool.add(1); + assert_eq!(pool.get().unwrap().release(), 1); + assert!(poll(get.as_mut(), &wake).is_pending()); + let prior_wakes = wake.0.load(Ordering::SeqCst); + pool.add(2); + assert!(wake.0.load(Ordering::SeqCst) > prior_wakes); + assert_eq!(smol::block_on(get).unwrap().release(), 2); +} + +#[test] +fn test_async_zero_and_max_timeouts() { + for timeout in [Duration::ZERO, Duration::MAX] { + let pool = AutoPool::new_with_config( + AutoPoolConfig { + wait_duration: timeout, + ..Default::default() + }, + [1], + ); + assert_eq!(smol::block_on(pool.get_async()).unwrap().release(), 1); + let mut get = Box::pin(pool.get_async()); + if timeout.is_zero() { + assert!(smol::block_on(get).is_none()); + } else { + assert!(smol::block_on(smol::future::poll_once(&mut get)).is_none()); + pool.add(2); + assert_eq!(smol::block_on(get).unwrap().release(), 2); + } + } +} + +#[test] +fn test_async_timeout_wakes_executor() { + let timeout = Duration::from_millis(20); + let pool = AutoPool::::new_with_config( + AutoPoolConfig { + wait_duration: timeout, + sleep_duration: Duration::from_secs(60), + ..Default::default() + }, + [], + ); + let start = Instant::now(); + assert!(smol::block_on(pool.get_async()).is_none()); + assert!(start.elapsed() >= timeout); + assert!(start.elapsed() < Duration::from_millis(200)); +} + +#[test] +fn test_async_lost_handoffs_do_not_restart_timeout() { + let timeout = Duration::from_millis(40); + let pool = AutoPool::new_with_config( + AutoPoolConfig { + wait_duration: timeout, + ..Default::default() + }, + [], + ); + let wake = Arc::new(WakeCount::default()); + let mut get = Box::pin(pool.get_async()); + let start = Instant::now(); + assert!(poll(get.as_mut(), &wake).is_pending()); + loop { + pool.add(1); + assert_eq!(pool.get().unwrap().release(), 1); + match poll(get.as_mut(), &wake) { + Poll::Ready(None) => break, + Poll::Ready(Some(_)) => panic!("all returned items were already consumed"), + Poll::Pending => {} + } + assert!(start.elapsed() < Duration::from_millis(200)); + std::thread::sleep(Duration::from_millis(5)); + } + assert!(start.elapsed() >= timeout); +} + +#[test] +fn test_contended_sync_async_return_handoff() { + let pool = AutoPool::new_with_config( + AutoPoolConfig { + wait_duration: Duration::from_secs(2), + ..Default::default() + }, + [0usize], + ); + let start = std::sync::Barrier::new(4); + std::thread::scope(|scope| { + for worker in 0..4 { + let pool = &pool; + let start = &start; + scope.spawn(move || { + start.wait(); + for _ in 0..200 { + let mut item = if worker % 2 == 0 { + pool.get() + } else { + smol::block_on(pool.get_async()) + } + .unwrap(); + *item += 1; + std::thread::yield_now(); + } + }); + } + }); + assert_eq!(pool.size(), 1); + assert_eq!(pool.get().unwrap().release(), 800); +} diff --git a/auto_pool/src/config.rs b/auto_pool/src/config.rs index 3268083..0edc556 100644 --- a/auto_pool/src/config.rs +++ b/auto_pool/src/config.rs @@ -1,22 +1,25 @@ use std::time::Duration; +/// How an available object is selected; this does not order waiting consumers. #[derive(Clone, Debug, Copy)] pub enum PickStrategy { - /// stack - always pick the object which was added last + /// Select the object that was added or returned last. LIFO, - /// pick the object from the pool randomly + /// Select an available object using a random index. RANDOM, } +/// Checkout policy. Public fields support struct literals and update syntax. #[derive(Clone, Debug, Copy)] pub struct AutoPoolConfig { - /// Duration to wait for an object to be available + /// Overall checkout budget. Zero tries immediately; `Duration::MAX` and + /// durations that overflow the platform's `Instant` wait indefinitely. pub wait_duration: Duration, - /// For async operations, how long to keep the lock on the pool + /// Legacy async polling setting, retained for compatibility and ignored. pub lock_duration: Duration, - /// For async operations, how long to sleep between retries + /// Legacy async polling setting, retained for compatibility and ignored. pub sleep_duration: Duration, - + /// Selection among available objects, independent of waiter scheduling. pub pick_strategy: PickStrategy, } diff --git a/auto_pool/src/lib.rs b/auto_pool/src/lib.rs index f1028a4..b0836d8 100644 --- a/auto_pool/src/lib.rs +++ b/auto_pool/src/lib.rs @@ -1,5 +1,13 @@ +#![doc = include_str!("../README.md")] + +/// Checkout timing and object selection. pub mod config; +/// Pool storage and checkout operations. pub mod pool; +/// Borrowed wrappers that return objects on drop. pub mod pool_object; #[cfg(test)] mod test; + +#[cfg(all(test, feature = "async"))] +mod async_tests; diff --git a/auto_pool/src/pool.rs b/auto_pool/src/pool.rs index 8b81c75..c16a131 100644 --- a/auto_pool/src/pool.rs +++ b/auto_pool/src/pool.rs @@ -3,90 +3,121 @@ use crate::config::{AutoPoolConfig, PickStrategy}; use parking_lot::lock_api::{MutexGuard, RawMutex}; use parking_lot::{Condvar, Mutex}; use rand::Rng; -use std::time::Duration; +use std::time::{Duration, Instant}; -/// A pool of objects. -/// After an object is taken from the pool, it is returned to the pool when it is dropped. -/// Pool items must be passed on creation or added later: -/// # Examples -/// Basic usage: -/// ``` -/// async fn test() { -/// use auto_pool::pool::AutoPool; -/// let pool = AutoPool::new([1, 2]); -/// let object1 = pool.get(); -/// let object2 = pool.get_async().await; -/// pool.add(3); -/// let inner1 = object1.unwrap().release(); // won't be returned back -/// } -/// ``` +/// A pool of caller-supplied objects, returned automatically when wrappers drop. /// -/// Create with custom config: -/// ``` -/// let config = auto_pool::config::AutoPoolConfig { -/// wait_duration: std::time::Duration::from_millis(5), -/// ..Default::default() -/// }; -/// let pool = auto_pool::pool::AutoPool::new_with_config(config, [1, 2]); -/// let item = pool.get(); -/// ``` +/// See the [crate documentation](crate) for synchronous and asynchronous examples. +/// Checked-out objects borrow this pool; no replacement objects are allocated. pub struct AutoPool { config: AutoPoolConfig, storage: Mutex>, condvar: Condvar, + #[cfg(feature = "async")] + available: event_listener::Event, } impl AutoPool { + /// Create a pool with unlimited waiting and LIFO selection. pub fn new(items: impl IntoIterator) -> Self { Self::new_with_config(AutoPoolConfig::default(), items) } + /// Create a pool with the supplied checkout policy and initial objects. pub fn new_with_config(config: AutoPoolConfig, items: impl IntoIterator) -> Self { let objects = items.into_iter().collect(); Self { config, storage: Mutex::new(objects), condvar: Condvar::new(), + #[cfg(feature = "async")] + available: event_listener::Event::new(), } } - /// Take an object from the pool. + /// Take an object, returning `None` when the configured overall budget expires. + /// Zero tries immediately. Unlimited waits can block forever on an empty pool. + /// Scheduling and mutex reacquisition may delay completion past the deadline. pub fn get(&'_ self) -> Option> { self.get_with_timeout(self.config.wait_duration) } - /// Async version - tries to get object, sleep if fails until timeout + /// Wait asynchronously for an object, using the configured overall timeout. + /// + /// Exhaustion suspends the future without blocking a thread. The storage mutex + /// is held briefly for checkout; no mutex guard is held across an await. + /// Dropping a pending future cancels its wait without removing an object. + /// This works on any executor; finite timeouts use smol's timer. #[cfg(feature = "async")] pub async fn get_async(&'_ self) -> Option> { if self.config.wait_duration.is_zero() { return self.get(); } + let deadline = Instant::now().checked_add(self.config.wait_duration); + loop { + // Available items need neither a listener allocation nor a timer. + if let Some(object) = self.extract_object(self.storage.lock()) { + return Some(object); + } + if deadline.is_some_and(|deadline| Instant::now() >= deadline) { + return None; + } - let start_time = std::time::Instant::now(); - while std::time::Instant::now() - start_time < self.config.wait_duration { - if let Some(obj) = self.get_with_timeout(self.config.lock_duration) { - return Some(obj); + // Register before rechecking so a concurrent return cannot be missed. + let listener = self.available.listen(); + if let Some(object) = self.extract_object(self.storage.lock()) { + return Some(object); + } + if let Some(deadline) = deadline { + let notified = smol::future::race( + async { + listener.await; + true + }, + async { + smol::Timer::at(deadline).await; + false + }, + ) + .await; + if !notified { + return None; + } + } else { + listener.await; } - smol::Timer::after(self.config.sleep_duration).await; } - None } - /// Is used to return item back - /// Also allows to add new item to the pool + /// Add or return an object and wake waiting consumers. pub fn add(&self, item: T) { self.storage.lock().push(item); self.condvar.notify_one(); + #[cfg(feature = "async")] + // Separate returns must wake separate waiters. A cancelled notified + // listener forwards its notification to another listener on drop. + self.available.notify_additional(1); } - /// Get the number of available items + /// Return the number of available objects, excluding checked-out objects. pub fn size(&self) -> usize { self.storage.lock().len() } - /// Shrink the pool to fit current number of items + /// Shrink storage to the number of currently available objects. + /// This holds the storage mutex while reallocating. pub fn shrink_to_fit(&self) { self.storage.lock().shrink_to_fit(); } fn get_with_timeout(&'_ self, timeout: Duration) -> Option> { - let mut locked_storage = self.storage.lock(); + let deadline = Instant::now().checked_add(timeout); + let mut locked_storage = if timeout.is_zero() { + self.storage.try_lock()? + } else if let Some(deadline) = deadline { + self.storage.try_lock_until(deadline)? + } else { + self.storage.lock() + }; while locked_storage.is_empty() { - let wait_res = self.condvar.wait_for(&mut locked_storage, timeout); - if wait_res.timed_out() { - return None; + if let Some(deadline) = deadline { + if Instant::now() >= deadline || self.condvar.wait_until(&mut locked_storage, deadline).timed_out() { + return None; + } + } else { + self.condvar.wait(&mut locked_storage); } } self.extract_object(locked_storage) @@ -103,11 +134,63 @@ impl AutoPool { 1 => locked_storage.pop(), items_cnt => { let index = rand::rng().next_u64() as usize % items_cnt; - locked_storage.swap(index, items_cnt - 1); - locked_storage.pop() + Some(locked_storage.swap_remove(index)) } }, }; inner.map(|inner| PoolObject::new(inner, self)) } } + +#[cfg(test)] +mod timeout_tests { + use super::*; + use std::time::Instant; + + #[test] + fn test_notifications_do_not_restart_sync_budget() { + let pool = AutoPool::::new_with_config( + AutoPoolConfig { + wait_duration: Duration::from_millis(40), + ..Default::default() + }, + [], + ); + std::thread::scope(|scope| { + scope.spawn(|| { + for _ in 0..25 { + std::thread::sleep(Duration::from_millis(10)); + pool.condvar.notify_all(); + } + }); + let start = Instant::now(); + assert!(pool.get().is_none()); + assert!(start.elapsed() >= Duration::from_millis(40)); + assert!(start.elapsed() < Duration::from_millis(200)); + }); + } + + #[test] + fn test_sync_budget_includes_mutex_acquisition() { + let pool = AutoPool::::new_with_config( + AutoPoolConfig { + wait_duration: Duration::from_millis(20), + ..Default::default() + }, + [1], + ); + std::thread::scope(|scope| { + let (tx, rx) = std::sync::mpsc::channel(); + let pool = &pool; + scope.spawn(move || { + let _guard = pool.storage.lock(); + tx.send(()).unwrap(); + std::thread::sleep(Duration::from_millis(200)); + }); + rx.recv().unwrap(); + let start = Instant::now(); + assert!(pool.get().is_none()); + assert!(start.elapsed() < Duration::from_millis(100)); + }); + } +} diff --git a/auto_pool/src/test.rs b/auto_pool/src/test.rs index 1b09cdd..64801d0 100644 --- a/auto_pool/src/test.rs +++ b/auto_pool/src/test.rs @@ -1,6 +1,5 @@ use crate::config::{AutoPoolConfig, PickStrategy}; use crate::pool::AutoPool; -use std::collections::HashMap; use std::ops::Deref; #[test] @@ -94,15 +93,84 @@ fn test_pick_strategy_random() { ..Default::default() }; let pool = AutoPool::new_with_config(config, [1, 2, 3]); - let mut match_counter = HashMap::new(); - for _ in 0..3000 { - let obj1 = pool.get(); - let value = *obj1.unwrap(); - match_counter.entry(value).and_modify(|v| *v += 1).or_insert(1); + let mut values = Vec::new(); + for _ in 0..3 { + values.push(pool.get().unwrap().release()); } + values.sort_unstable(); + assert_eq!(values, [1, 2, 3]); + assert!(pool.get().is_none()); +} + +#[cfg(feature = "async")] +#[test] +fn test_async_first_poll_does_not_wait_for_an_item() { + let pool = AutoPool::::new_with_config( + AutoPoolConfig { + wait_duration: std::time::Duration::from_millis(20), + lock_duration: std::time::Duration::from_millis(200), + ..Default::default() + }, + [], + ); + let mut get = Box::pin(pool.get_async()); + let start = std::time::Instant::now(); + assert!(smol::block_on(smol::future::poll_once(&mut get)).is_none()); + assert!(start.elapsed() < std::time::Duration::from_millis(100)); +} + +#[cfg(feature = "async")] +#[test] +fn test_async_add_wakes_without_polling_delay() { + let pool = AutoPool::new_with_config( + AutoPoolConfig { + wait_duration: std::time::Duration::from_secs(1), + lock_duration: std::time::Duration::ZERO, + sleep_duration: std::time::Duration::from_secs(60), + ..Default::default() + }, + [], + ); + let mut get = Box::pin(pool.get_async()); + assert!(smol::block_on(smol::future::poll_once(&mut get)).is_none()); + pool.add(42); + let result = smol::block_on(smol::future::poll_once(&mut get)); + assert_eq!(result.flatten().map(|item| item.release()), Some(42)); +} - // not guaranteed to pass, but should be close - assert!(match_counter[&1] > 200); - assert!(match_counter[&2] > 200); - assert!(match_counter[&3] > 200); +#[test] +fn test_return_preserves_mutation_and_release_removes_item() { + let pool = AutoPool::new([String::from("hello")]); + { + let mut item = pool.get().unwrap(); + item.push_str(" world"); + assert_eq!(pool.size(), 0); + } + assert_eq!(pool.size(), 1); + assert_eq!(pool.get().unwrap().release(), "hello world"); + assert_eq!(pool.size(), 0); +} + +#[test] +fn test_zero_timeout_checks_available_items() { + let pool = AutoPool::new_with_config( + AutoPoolConfig { + wait_duration: std::time::Duration::ZERO, + ..Default::default() + }, + [7], + ); + assert_eq!(pool.get().unwrap().release(), 7); + assert!(pool.get().is_none()); +} + +#[test] +fn test_default_timeout_waits_for_return() { + let pool = AutoPool::new([7]); + let item = pool.get().unwrap(); + std::thread::scope(|scope| { + let waiter = scope.spawn(|| pool.get().unwrap().release()); + drop(item); + assert_eq!(waiter.join().unwrap(), 7); + }); }