diff --git a/LICENSE b/LICENSE index f0925947..86747f2d 100644 --- a/LICENSE +++ b/LICENSE @@ -426,10 +426,6 @@ is followed by its upstream source path or paths: fastpool/src/unbounded.rs asyncband/src/pool/unbounded.rs fastpool/src/unbounded.rs - tests-integration/tests/pool_recycle_cancelled_test.rs - fastpool/tests/recycle_cancelled_tests.rs - tests-integration/tests/pool_replenish_test.rs - fastpool/tests/replenish_tests.rs The exact upstream revision is: diff --git a/asyncband/src/broadcast/mpmc/bounded/tests.rs b/asyncband/src/broadcast/mpmc/bounded/tests.rs index 594936ce..a6256e2d 100644 --- a/asyncband/src/broadcast/mpmc/bounded/tests.rs +++ b/asyncband/src/broadcast/mpmc/bounded/tests.rs @@ -16,7 +16,7 @@ // under the License. // These run under Miri via `cargo x miri`, so they stay single-threaded and small. Behavior -// reachable from the public API is covered in `tests-integration/broadcast_mpmc_bounded_test.rs`. +// reachable from the public API is covered in `tests-integration/tests/broadcast_test/bounded.rs`. use std::task::Waker; diff --git a/licenserc.toml b/licenserc.toml index 4ec6f841..7b6c05d9 100644 --- a/licenserc.toml +++ b/licenserc.toml @@ -37,8 +37,6 @@ excludes = [ "asyncband/src/rwlock/owned_write_guard.rs", "asyncband/src/rwlock/read_guard.rs", "asyncband/src/rwlock/write_guard.rs", - "tests-integration/tests/pool_recycle_cancelled_test.rs", - "tests-integration/tests/pool_replenish_test.rs", ] includes = [ "**/*.md", diff --git a/tests-integration/tests/broadcast_mpmc_bounded_test.rs b/tests-integration/tests/broadcast_test/bounded.rs similarity index 100% rename from tests-integration/tests/broadcast_mpmc_bounded_test.rs rename to tests-integration/tests/broadcast_test/bounded.rs diff --git a/tests-integration/tests/broadcast_test/main.rs b/tests-integration/tests/broadcast_test/main.rs new file mode 100644 index 00000000..e0ac8347 --- /dev/null +++ b/tests-integration/tests/broadcast_test/main.rs @@ -0,0 +1,19 @@ +// 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. + +mod bounded; +mod unbounded; diff --git a/tests-integration/tests/broadcast_mpmc_unbounded_test.rs b/tests-integration/tests/broadcast_test/unbounded.rs similarity index 100% rename from tests-integration/tests/broadcast_mpmc_unbounded_test.rs rename to tests-integration/tests/broadcast_test/unbounded.rs diff --git a/tests-integration/tests/auto_reset_event_test.rs b/tests-integration/tests/event_test/auto_reset.rs similarity index 100% rename from tests-integration/tests/auto_reset_event_test.rs rename to tests-integration/tests/event_test/auto_reset.rs diff --git a/tests-integration/tests/event_test/main.rs b/tests-integration/tests/event_test/main.rs new file mode 100644 index 00000000..60e0b8c7 --- /dev/null +++ b/tests-integration/tests/event_test/main.rs @@ -0,0 +1,19 @@ +// 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. + +mod auto_reset; +mod manual_reset; diff --git a/tests-integration/tests/event_test.rs b/tests-integration/tests/event_test/manual_reset.rs similarity index 100% rename from tests-integration/tests/event_test.rs rename to tests-integration/tests/event_test/manual_reset.rs diff --git a/tests-integration/tests/pool_recycle_cancelled_test.rs b/tests-integration/tests/pool_recycle_cancelled_test.rs deleted file mode 100644 index 144beaa5..00000000 --- a/tests-integration/tests/pool_recycle_cancelled_test.rs +++ /dev/null @@ -1,245 +0,0 @@ -// Copyright 2025 FastLabs Developers -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. -// -// This file contains code ported from Fastpool 1.1.1. -// The incorporated code has been modified for use in Apache Asyncband. -// Upstream source: -// https://github.com/fast/fastpool/blob/e4c65f1ed38395abc58d68eda8bd09925c13028e/fastpool/tests/recycle_cancelled_tests.rs - -use std::future::Future; -use std::future::poll_fn; -use std::pin::pin; -use std::sync::Arc; -use std::sync::atomic::AtomicBool; -use std::sync::atomic::AtomicUsize; -use std::sync::atomic::Ordering; -use std::task::Poll; - -use asyncband::pool::ManageObject; -use asyncband::pool::ObjectStatus; -use asyncband::pool::RecycleCancelledStrategy; - -#[derive(Default)] -struct Controls { - created: AtomicUsize, - recycle_ready: AtomicBool, - reject_recycle: AtomicBool, -} - -struct ControlledRecycleManager { - controls: Arc, -} - -impl ManageObject for ControlledRecycleManager { - type Object = usize; - type Error = (); - - async fn create(&self) -> Result { - Ok(self.controls.created.fetch_add(1, Ordering::Relaxed)) - } - - async fn is_recyclable( - &self, - _object: &mut Self::Object, - _status: &ObjectStatus, - ) -> Result<(), Self::Error> { - poll_fn(|_| { - if !self.controls.recycle_ready.load(Ordering::Acquire) { - Poll::Pending - } else if self.controls.reject_recycle.load(Ordering::Relaxed) { - Poll::Ready(Err(())) - } else { - Poll::Ready(Ok(())) - } - }) - .await - } -} - -fn manager() -> (ControlledRecycleManager, Arc) { - let controls = Arc::new(Controls::default()); - ( - ControlledRecycleManager { - controls: controls.clone(), - }, - controls, - ) -} - -fn poll_and_cancel(future: impl Future) { - let mut future = pin!(future); - assert!(tests_integration::poll_once(future.as_mut()).is_pending()); -} - -mod bounded_tests { - use asyncband::pool::bounded::Pool; - use asyncband::pool::bounded::PoolConfig; - - use super::*; - - #[tokio::test] - async fn cancellation_detaches_by_default() { - let (manager, controls) = manager(); - let pool = Pool::new(PoolConfig::new(1), manager); - - let object = pool.get().await.unwrap(); - assert_eq!(*object, 0); - drop(object); - - poll_and_cancel(pool.get()); - assert_eq!(pool.status().current_size, 0); - assert_eq!(pool.status().idle_count, 0); - - let object = pool.get().await.unwrap(); - assert_eq!(*object, 1); - assert_eq!(controls.created.load(Ordering::Relaxed), 2); - } - - #[tokio::test] - async fn cancellation_can_restore_the_idle_object() { - let (manager, controls) = manager(); - let config = PoolConfig::new(1) - .with_recycle_cancelled_strategy(RecycleCancelledStrategy::ReturnToPool); - let pool = Pool::new(config, manager); - - let object = pool.get().await.unwrap(); - drop(object); - let mut last_used_before = None; - pool.retain(|_, status| { - last_used_before = Some(status.last_used()); - true - }); - - poll_and_cancel(pool.get()); - assert_eq!(pool.status().current_size, 1); - assert_eq!(pool.status().idle_count, 1); - - let mut last_used_after = None; - pool.retain(|_, status| { - last_used_after = Some(status.last_used()); - true - }); - assert_eq!(last_used_after, last_used_before); - - controls.recycle_ready.store(true, Ordering::Release); - let object = pool.get().await.unwrap(); - assert_eq!(*object, 0); - assert_eq!(controls.created.load(Ordering::Relaxed), 1); - } - - #[tokio::test] - async fn repeated_cancellation_does_not_shrink_a_restoring_pool() { - let (manager, _) = manager(); - let config = PoolConfig::new(3) - .with_recycle_cancelled_strategy(RecycleCancelledStrategy::ReturnToPool); - let pool = Pool::new(config, manager); - - let objects = [ - pool.get().await.unwrap(), - pool.get().await.unwrap(), - pool.get().await.unwrap(), - ]; - drop(objects); - - for _ in 0..5 { - poll_and_cancel(pool.get()); - } - assert_eq!(pool.status().current_size, 3); - assert_eq!(pool.status().idle_count, 3); - } - - #[tokio::test] - async fn rejected_recycle_detaches_even_when_cancellation_would_restore() { - let (manager, controls) = manager(); - let config = PoolConfig::new(1) - .with_recycle_cancelled_strategy(RecycleCancelledStrategy::ReturnToPool); - let pool = Pool::new(config, manager); - - let object = pool.get().await.unwrap(); - drop(object); - controls.reject_recycle.store(true, Ordering::Relaxed); - controls.recycle_ready.store(true, Ordering::Release); - - let object = pool.get().await.unwrap(); - assert_eq!(*object, 1); - assert_eq!(controls.created.load(Ordering::Relaxed), 2); - assert_eq!(pool.status().current_size, 1); - } -} - -mod unbounded_tests { - use asyncband::pool::unbounded::Pool; - use asyncband::pool::unbounded::PoolConfig; - - use super::*; - - #[tokio::test] - async fn cancellation_detaches_by_default() { - let (manager, controls) = manager(); - let pool = Pool::new(PoolConfig::default(), manager); - - let object = pool.get().await.unwrap(); - assert_eq!(*object, 0); - drop(object); - - poll_and_cancel(pool.get()); - assert_eq!(pool.status().current_size, 0); - assert_eq!(pool.status().idle_count, 0); - - let object = pool.get().await.unwrap(); - assert_eq!(*object, 1); - assert_eq!(controls.created.load(Ordering::Relaxed), 2); - } - - #[tokio::test] - async fn cancellation_can_restore_the_idle_object() { - let (manager, controls) = manager(); - let config = PoolConfig::new() - .with_recycle_cancelled_strategy(RecycleCancelledStrategy::ReturnToPool); - let pool = Pool::new(config, manager); - - let object = pool.get().await.unwrap(); - drop(object); - poll_and_cancel(pool.get()); - - assert_eq!(pool.status().current_size, 1); - assert_eq!(pool.status().idle_count, 1); - controls.recycle_ready.store(true, Ordering::Release); - - let object = pool.get().await.unwrap(); - assert_eq!(*object, 0); - assert_eq!(controls.created.load(Ordering::Relaxed), 1); - } - - #[tokio::test] - async fn repeated_cancellation_does_not_shrink_a_restoring_pool() { - let (manager, _) = manager(); - let config = PoolConfig::new() - .with_recycle_cancelled_strategy(RecycleCancelledStrategy::ReturnToPool); - let pool = Pool::new(config, manager); - - let objects = [ - pool.get().await.unwrap(), - pool.get().await.unwrap(), - pool.get().await.unwrap(), - ]; - drop(objects); - - for _ in 0..5 { - poll_and_cancel(pool.get()); - } - assert_eq!(pool.status().current_size, 3); - assert_eq!(pool.status().idle_count, 3); - } -} diff --git a/tests-integration/tests/pool_replenish_test.rs b/tests-integration/tests/pool_replenish_test.rs deleted file mode 100644 index 5dbca192..00000000 --- a/tests-integration/tests/pool_replenish_test.rs +++ /dev/null @@ -1,308 +0,0 @@ -// Copyright 2025 FastLabs Developers -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. -// -// This file contains code ported from Fastpool 1.1.1. -// The incorporated code has been modified for use in Apache Asyncband. -// Upstream source: -// https://github.com/fast/fastpool/blob/e4c65f1ed38395abc58d68eda8bd09925c13028e/fastpool/tests/replenish_tests.rs - -use std::convert::Infallible; -use std::future::poll_fn; -use std::sync::Arc; -use std::sync::atomic::AtomicBool; -use std::sync::atomic::AtomicUsize; -use std::sync::atomic::Ordering; -use std::task::Poll; - -use asyncband::pool::ManageObject; -use asyncband::pool::ObjectStatus; -use asyncband::pool::bounded::Pool; -use asyncband::pool::bounded::PoolConfig; - -#[tokio::test] -async fn test_replenish_to() { - #[derive(Default)] - struct Manager; - - impl ManageObject for Manager { - type Object = (); - type Error = Infallible; - - async fn create(&self) -> Result { - 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::new(PoolConfig::new(MAX_SIZE), Manager) - } - - for i in 0..5 { - let pool = make_default(); - let n = pool.replenish_to(i).await.unwrap(); - assert_eq!(n, i.min(MAX_SIZE)); - } - - // stage one idle object - { - let pool = make_default(); - pool.get().await.unwrap(); - let n = pool.replenish_to(2).await.unwrap(); - 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_to(2).await.unwrap(); - assert_eq!(n, 0); - } -} - -#[derive(Debug, PartialEq, Eq)] -struct CreateError; - -struct FailingManager { - calls: Arc, -} - -impl ManageObject for FailingManager { - type Object = usize; - type Error = CreateError; - - async fn create(&self) -> Result { - let call = self.calls.fetch_add(1, Ordering::Relaxed); - if call == 1 { - Err(CreateError) - } else { - Ok(call) - } - } - - async fn is_recyclable( - &self, - _object: &mut Self::Object, - _status: &ObjectStatus, - ) -> Result<(), Self::Error> { - Ok(()) - } -} - -#[tokio::test] -async fn replenish_to_reports_create_errors_and_keeps_prior_objects() { - let pool = Pool::new( - PoolConfig::new(2), - FailingManager { - calls: Arc::new(AtomicUsize::new(0)), - }, - ); - - assert_eq!(pool.replenish_to(2).await, Err(CreateError)); - assert_eq!(pool.status().current_size, 1); - assert_eq!(pool.status().idle_count, 1); - - assert_eq!(pool.replenish_to(2).await, Ok(1)); - assert_eq!(pool.status().current_size, 2); - assert_eq!(pool.status().idle_count, 2); -} - -struct ControlledManager { - calls: Arc, - allow_create: Arc, -} - -impl ManageObject for ControlledManager { - type Object = usize; - type Error = Infallible; - - async fn create(&self) -> Result { - let call = self.calls.fetch_add(1, Ordering::Relaxed); - if call != 0 { - poll_fn(|_| { - if self.allow_create.load(Ordering::Acquire) { - Poll::Ready(()) - } else { - Poll::Pending - } - }) - .await; - } - Ok(call) - } - - async fn is_recyclable( - &self, - _object: &mut Self::Object, - _status: &ObjectStatus, - ) -> Result<(), Self::Error> { - Ok(()) - } -} - -#[tokio::test] -async fn concurrent_replenish_to_calls_respect_capacity() { - let calls = Arc::new(AtomicUsize::new(0)); - let allow_create = Arc::new(AtomicBool::new(false)); - let pool = Pool::new( - PoolConfig::new(2), - ControlledManager { - calls: calls.clone(), - allow_create: allow_create.clone(), - }, - ); - - assert_eq!(pool.replenish_to(1).await, Ok(1)); - - let mut first = Box::pin(pool.replenish_to(2)); - assert!(tests_integration::poll_once(first.as_mut()).is_pending()); - - let mut second = Box::pin(pool.replenish_to(2)); - assert_eq!( - tests_integration::poll_once(second.as_mut()), - Poll::Ready(Ok(0)) - ); - - allow_create.store(true, Ordering::Release); - assert_eq!( - tests_integration::poll_once(first.as_mut()), - Poll::Ready(Ok(1)) - ); - assert_eq!(calls.load(Ordering::Relaxed), 2); - assert_eq!(pool.status().current_size, 2); - assert_eq!(pool.status().idle_count, 2); -} - -#[tokio::test] -async fn concurrent_get_and_replenish_to_respect_capacity() { - let calls = Arc::new(AtomicUsize::new(0)); - let allow_create = Arc::new(AtomicBool::new(false)); - let pool = Pool::new( - PoolConfig::new(2), - ControlledManager { - calls: calls.clone(), - allow_create: allow_create.clone(), - }, - ); - - let first = pool.get().await.unwrap(); - - let mut replenish = Box::pin(pool.replenish_to(2)); - assert!(tests_integration::poll_once(replenish.as_mut()).is_pending()); - - let mut get = Box::pin(pool.get()); - assert!(tests_integration::poll_once(get.as_mut()).is_pending()); - assert_eq!(pool.status().current_size, 1); - - allow_create.store(true, Ordering::Release); - assert_eq!( - tests_integration::poll_once(replenish.as_mut()), - Poll::Ready(Ok(1)) - ); - - let second = match tests_integration::poll_once(get.as_mut()) { - Poll::Ready(Ok(object)) => object, - _ => panic!("get should consume the replenished object"), - }; - assert_eq!(calls.load(Ordering::Relaxed), 2); - assert_eq!(pool.status().current_size, 2); - assert_eq!(pool.status().idle_count, 0); - - drop((first, second)); - assert_eq!(pool.status().idle_count, 2); -} - -struct BlockingManager { - allow_create: Arc, -} - -impl ManageObject for BlockingManager { - type Object = (); - type Error = Infallible; - - async fn create(&self) -> Result { - poll_fn(|_| { - if self.allow_create.load(Ordering::Acquire) { - Poll::Ready(()) - } else { - Poll::Pending - } - }) - .await; - Ok(()) - } - - async fn is_recyclable( - &self, - _object: &mut Self::Object, - _status: &ObjectStatus, - ) -> Result<(), Self::Error> { - Ok(()) - } -} - -#[tokio::test] -async fn replenish_to_respects_max_size_with_active_and_idle_objects() { - let pool = Pool::new( - PoolConfig::new(2), - BlockingManager { - allow_create: Arc::new(AtomicBool::new(true)), - }, - ); - - assert_eq!(pool.replenish_to(2).await, Ok(2)); - let active = pool.get().await.unwrap(); - assert_eq!(pool.status().current_size, 2); - assert_eq!(pool.status().idle_count, 1); - - assert_eq!(pool.replenish_to(usize::MAX).await, Ok(0)); - assert_eq!(pool.status().current_size, 2); - assert_eq!(pool.status().idle_count, 1); - - drop(active); - assert_eq!(pool.status().idle_count, 2); -} - -#[tokio::test] -async fn cancelling_replenish_to_releases_reserved_capacity() { - let allow_create = Arc::new(AtomicBool::new(false)); - let pool = Pool::new( - PoolConfig::new(1), - BlockingManager { - allow_create: allow_create.clone(), - }, - ); - - let mut replenish = Box::pin(pool.replenish_to(1)); - assert!(tests_integration::poll_once(replenish.as_mut()).is_pending()); - drop(replenish); - - allow_create.store(true, Ordering::Release); - let mut get = Box::pin(pool.get()); - assert!(tests_integration::poll_once(get.as_mut()).is_ready()); - assert_eq!(pool.status().idle_count, 1); -} diff --git a/tests-integration/tests/pool_acquire_cancelled_test.rs b/tests-integration/tests/pool_test/acquire.rs similarity index 66% rename from tests-integration/tests/pool_acquire_cancelled_test.rs rename to tests-integration/tests/pool_test/acquire.rs index 2f34dda0..39276acc 100644 --- a/tests-integration/tests/pool_acquire_cancelled_test.rs +++ b/tests-integration/tests/pool_test/acquire.rs @@ -16,61 +16,30 @@ // under the License. use std::future::Future; -use std::future::poll_fn; use std::pin::pin; use std::sync::Arc; -use std::sync::atomic::AtomicBool; -use std::sync::atomic::AtomicUsize; -use std::sync::atomic::Ordering; use std::task::Context; use std::task::Poll; use std::task::Wake; use std::task::Waker; -use asyncband::pool::ManageObject; -use asyncband::pool::ObjectStatus; use asyncband::pool::bounded::Pool; use asyncband::pool::bounded::PoolConfig; use tests_integration::WakeCounter; -struct Manager { - create_calls: Arc, - create_ready: Arc, -} - -impl ManageObject for Manager { - type Object = usize; - type Error = (); +use super::support::Manager; +use super::support::ManagerError; - async fn create(&self) -> Result { - let id = self.create_calls.fetch_add(1, Ordering::Relaxed); - // Tests explicitly poll again after changing this gate; no executor drives it. - poll_fn(|_| { - if self.create_ready.load(Ordering::Acquire) { - Poll::Ready(Ok(id)) - } else { - Poll::Pending - } - }) - .await - } - - async fn is_recyclable( - &self, - _object: &mut Self::Object, - _status: &ObjectStatus, - ) -> Result<(), Self::Error> { - Ok(()) - } -} - -fn ready(future: impl Future>) -> T { +fn ready(future: impl Future>) -> T { let wakes = Arc::new(WakeCounter::default()); wakes.wake_by_ref(); ready_after_wake(future, &wakes) } -fn ready_after_wake(future: impl Future>, wakes: &Arc) -> T { +fn ready_after_wake( + future: impl Future>, + wakes: &Arc, +) -> T { let mut future = pin!(future); let waker = Waker::from(wakes.clone()); // Allow cooperative yields, but never poll away a missing notification or spin forever. @@ -78,7 +47,7 @@ fn ready_after_wake(future: impl Future>, wakes: &Arc< assert!(wakes.take() > 0, "missing wake"); match future.as_mut().poll(&mut Context::from_waker(&waker)) { Poll::Ready(Ok(value)) => return value, - Poll::Ready(Err(())) => panic!("operation should succeed"), + Poll::Ready(Err(ManagerError)) => panic!("operation should succeed"), Poll::Pending => {} } } @@ -86,14 +55,8 @@ fn ready_after_wake(future: impl Future>, wakes: &Arc< } fn cancel_waiter(return_before_cancel: bool) { - let create_calls = Arc::new(AtomicUsize::new(0)); - let pool = Pool::new( - PoolConfig::new(1), - Manager { - create_calls: create_calls.clone(), - create_ready: Arc::new(AtomicBool::new(true)), - }, - ); + let manager = Manager::default(); + let pool = Pool::new(PoolConfig::new(1), manager.clone()); let held = ready(pool.get()); let mut first_wakes = Arc::new(WakeCounter::default()); let mut next_wakes = Arc::new(WakeCounter::default()); @@ -112,7 +75,7 @@ fn cancel_waiter(return_before_cancel: bool) { .poll(&mut Context::from_waker(&next_waker)) .is_pending() ); - assert_eq!(create_calls.load(Ordering::Relaxed), 1); + assert_eq!(manager.created(), 1); if return_before_cancel { drop(held); @@ -132,13 +95,13 @@ fn cancel_waiter(return_before_cancel: bool) { // A notification from returning the object is just as valid as one from cancellation. let object = ready_after_wake(next.as_mut(), &next_wakes); assert_eq!(*object, 0); - assert_eq!(create_calls.load(Ordering::Relaxed), 1); + assert_eq!(manager.created(), 1); assert_eq!(pool.status().current_size, 1); assert_eq!(pool.status().idle_count, 0); let mut extra = Box::pin(pool.get()); assert!(tests_integration::poll_once(extra.as_mut()).is_pending()); - assert_eq!(create_calls.load(Ordering::Relaxed), 1); + assert_eq!(manager.created(), 1); drop(extra); drop(object); assert_eq!(pool.status().idle_count, 1); @@ -157,18 +120,12 @@ fn cancelling_notified_get_preserves_follower_progress() { #[test] fn cancelling_create_restores_capacity_for_waiting_get() { - let create_calls = Arc::new(AtomicUsize::new(0)); - let create_ready = Arc::new(AtomicBool::new(false)); - let pool = Pool::new( - PoolConfig::new(1), - Manager { - create_calls: create_calls.clone(), - create_ready: create_ready.clone(), - }, - ); + let manager = Manager::default(); + let creation = manager.pause_create(); + let pool = Pool::new(PoolConfig::new(1), manager.clone()); let mut creating = Box::pin(pool.get()); assert!(tests_integration::poll_once(creating.as_mut()).is_pending()); - assert_eq!(create_calls.load(Ordering::Relaxed), 1); + assert_eq!(manager.created(), 1); assert_eq!(pool.status().current_size, 0); let wakes = Arc::new(WakeCounter::default()); @@ -179,24 +136,24 @@ fn cancelling_create_restores_capacity_for_waiting_get() { .poll(&mut Context::from_waker(&waker)) .is_pending() ); - assert_eq!(create_calls.load(Ordering::Relaxed), 1); + assert_eq!(manager.created(), 1); drop(creating); assert!(wakes.count() > 0); assert_eq!(pool.status().current_size, 0); assert_eq!(pool.status().idle_count, 0); - create_ready.store(true, Ordering::Release); + assert!(creation.is_closed()); let object = ready_after_wake(next.as_mut(), &wakes); assert_eq!(*object, 1); - assert_eq!(create_calls.load(Ordering::Relaxed), 2); + assert_eq!(manager.created(), 2); assert_eq!(pool.status().current_size, 1); assert_eq!(pool.status().idle_count, 0); let mut extra = Box::pin(pool.get()); assert!(tests_integration::poll_once(extra.as_mut()).is_pending()); - assert_eq!(create_calls.load(Ordering::Relaxed), 2); + assert_eq!(manager.created(), 2); drop(extra); drop(object); assert_eq!(pool.status().idle_count, 1); assert_eq!(*ready(pool.get()), 1); - assert_eq!(create_calls.load(Ordering::Relaxed), 2); + assert_eq!(manager.created(), 2); } diff --git a/tests-integration/tests/pool_behavior_test.rs b/tests-integration/tests/pool_test/behavior.rs similarity index 71% rename from tests-integration/tests/pool_behavior_test.rs rename to tests-integration/tests/pool_test/behavior.rs index ad2417ca..9d68269f 100644 --- a/tests-integration/tests/pool_behavior_test.rs +++ b/tests-integration/tests/pool_test/behavior.rs @@ -19,65 +19,23 @@ use std::cell::Cell; use std::convert::Infallible; use std::panic::AssertUnwindSafe; use std::panic::catch_unwind; -use std::sync::Arc; -use std::sync::atomic::AtomicUsize; -use std::sync::atomic::Ordering; use std::time::Instant; -use asyncband::pool::ManageObject; -use asyncband::pool::ObjectStatus; use asyncband::pool::QueueStrategy; use asyncband::pool::bounded; use asyncband::pool::unbounded; -struct CountingManager { - next: Arc, - detached: Arc, -} - -impl ManageObject for CountingManager { - type Object = usize; - type Error = Infallible; - - async fn create(&self) -> Result { - Ok(self.next.fetch_add(1, Ordering::Relaxed)) - } - - async fn is_recyclable( - &self, - _object: &mut Self::Object, - _status: &ObjectStatus, - ) -> Result<(), Self::Error> { - Ok(()) - } - - fn on_detached(&self, object: &mut Self::Object) { - self.detached.fetch_add(1, Ordering::Relaxed); - *object += 1000; - } -} +use super::support::Manager; #[test] #[should_panic(expected = "bounded pool max_size must be greater than zero")] fn bounded_pool_rejects_zero_capacity() { - bounded::Pool::new( - bounded::PoolConfig::new(0), - CountingManager { - next: Arc::new(AtomicUsize::new(0)), - detached: Arc::new(AtomicUsize::new(0)), - }, - ); + bounded::Pool::new(bounded::PoolConfig::new(0), Manager::default()); } #[test] fn bounded_construction_allocates_idle_storage_lazily() { - let pool = bounded::Pool::new( - bounded::PoolConfig::new(usize::MAX), - CountingManager { - next: Arc::new(AtomicUsize::new(0)), - detached: Arc::new(AtomicUsize::new(0)), - }, - ); + let pool = bounded::Pool::new(bounded::PoolConfig::new(usize::MAX), Manager::default()); assert_eq!(pool.status().max_size, usize::MAX); assert_eq!(pool.status().current_size, 0); @@ -86,13 +44,7 @@ fn bounded_construction_allocates_idle_storage_lazily() { #[tokio::test] async fn bounded_last_used_tracks_the_end_of_a_checkout() { - let pool = bounded::Pool::new( - bounded::PoolConfig::new(1), - CountingManager { - next: Arc::new(AtomicUsize::new(0)), - detached: Arc::new(AtomicUsize::new(0)), - }, - ); + let pool = bounded::Pool::new(bounded::PoolConfig::new(1), Manager::default()); let object = pool.get().await.unwrap(); let before_return = Instant::now(); @@ -105,14 +57,8 @@ async fn bounded_last_used_tracks_the_end_of_a_checkout() { #[tokio::test] async fn retain_invokes_detachment_hook_once_per_removed_object() { - let detached = Arc::new(AtomicUsize::new(0)); - let pool = bounded::Pool::new( - bounded::PoolConfig::new(4), - CountingManager { - next: Arc::new(AtomicUsize::new(0)), - detached: detached.clone(), - }, - ); + let manager = Manager::default(); + let pool = bounded::Pool::new(bounded::PoolConfig::new(4), manager.clone()); let mut objects = vec![]; for _ in 0..4 { @@ -125,7 +71,7 @@ async fn retain_invokes_detachment_hook_once_per_removed_object() { assert_eq!(result.retained, 2); assert_eq!(result.removed, [1001, 1003]); - assert_eq!(detached.load(Ordering::Relaxed), 2); + assert_eq!(manager.detached(), [1, 3]); assert_eq!(pool.status().current_size, 2); assert_eq!(pool.status().idle_count, 2); } diff --git a/tests-integration/tests/pool_test/main.rs b/tests-integration/tests/pool_test/main.rs new file mode 100644 index 00000000..683f0f1a --- /dev/null +++ b/tests-integration/tests/pool_test/main.rs @@ -0,0 +1,22 @@ +// 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. + +mod acquire; +mod behavior; +mod recycle; +mod replenish; +mod support; diff --git a/tests-integration/tests/pool_test/recycle.rs b/tests-integration/tests/pool_test/recycle.rs new file mode 100644 index 00000000..03ff53ef --- /dev/null +++ b/tests-integration/tests/pool_test/recycle.rs @@ -0,0 +1,193 @@ +// 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 asyncband::pool::RecycleCancelledStrategy; +use tests_integration::WakeCounter; +use tests_integration::expect_ready; +use tests_integration::poll_once; +use tests_integration::poll_with; + +use super::support::Manager; +use super::support::ManagerError; +use super::support::ready; + +// Both pool variants promise the same object lifecycle during recycle cancellation. +macro_rules! recycle_contract { + () => { + #[test] + fn cancelled_validation_applies_the_configured_ownership_policy() { + for strategy in [None, Some(RecycleCancelledStrategy::ReturnToPool)] { + let manager = Manager::default(); + let pool = pool(1, manager.clone(), strategy); + let original = ready(pool.get()).unwrap(); + let id = *original; + drop(original); + + let mut returned_at = None; + pool.retain(|_, status| { + returned_at = Some(status.last_used()); + true + }); + + // Repeating the cancellation checks that restored objects stay reusable. + let attempts = if strategy.is_some() { 4 } else { 1 }; + for _ in 0..attempts { + let validation = manager.pause_recycle(); + let mut checkout = Box::pin(pool.get()); + assert!(poll_once(checkout.as_mut()).is_pending()); + assert_eq!(pool.status().idle_count, 0); + drop(checkout); + assert!(validation.is_closed()); + + let retained = usize::from(strategy.is_some()); + assert_eq!(pool.status().current_size, retained); + assert_eq!(pool.status().idle_count, retained); + assert_eq!(manager.created(), 1); + } + + if strategy.is_some() { + assert!(manager.detached().is_empty()); + pool.retain(|object, status| { + assert_eq!(*object, id); + assert_eq!(Some(status.last_used()), returned_at); + assert_eq!(status.recycle_count(), 0); + true + }); + } else { + assert_eq!(manager.detached(), [id]); + } + let replacement = ready(pool.get()).unwrap(); + if strategy.is_some() { + assert_eq!(*replacement, id); + assert_eq!(manager.created(), 1); + assert_eq!(replacement.status().recycle_count(), 1); + } else { + assert_ne!(*replacement, id); + assert_eq!(manager.created(), 2); + } + assert_eq!(pool.status().current_size, 1); + assert_eq!(pool.status().idle_count, 0); + } + } + + #[test] + fn recycle_completion_wakes_checkout_and_rejection_replaces_the_object() { + for reject in [false, true] { + let manager = Manager::default(); + let pool = pool( + 1, + manager.clone(), + Some(RecycleCancelledStrategy::ReturnToPool), + ); + drop(ready(pool.get()).unwrap()); + let validation = manager.pause_recycle(); + let (waker, wakes) = WakeCounter::new(); + let mut checkout = Box::pin(pool.get()); + assert!(poll_with(checkout.as_mut(), &waker).is_pending()); + validation + .send(if reject { Err(ManagerError) } else { Ok(()) }) + .unwrap(); + assert!(wakes.count() > 0); + + let object = expect_ready(poll_with(checkout.as_mut(), &waker)).unwrap(); + assert_eq!(*object, usize::from(reject)); + assert_eq!(manager.created(), 1 + usize::from(reject)); + assert_eq!(manager.detached(), if reject { vec![0] } else { vec![] }); + drop(object); + assert_eq!(pool.status().current_size, 1); + assert_eq!(pool.status().idle_count, 1); + } + } + + #[test] + fn cancelling_one_validation_leaves_other_idle_objects_available() { + let manager = Manager::default(); + let pool = pool( + 3, + manager.clone(), + Some(RecycleCancelledStrategy::ReturnToPool), + ); + let objects: Vec<_> = (0..3).map(|_| ready(pool.get()).unwrap()).collect(); + let mut ids: Vec<_> = objects.iter().map(|object| **object).collect(); + drop(objects); + + for _ in 0..6 { + let validation = manager.pause_recycle(); + let mut checkout = Box::pin(pool.get()); + assert!(poll_once(checkout.as_mut()).is_pending()); + assert_eq!(pool.status().idle_count, 2); + drop(checkout); + assert!(validation.is_closed()); + assert_eq!(pool.status().idle_count, 3); + } + + let objects: Vec<_> = (0..3).map(|_| ready(pool.get()).unwrap()).collect(); + let mut recycled: Vec<_> = objects.iter().map(|object| **object).collect(); + ids.sort_unstable(); + recycled.sort_unstable(); + assert_eq!(recycled, ids); + assert_eq!(manager.created(), 3); + assert!(manager.detached().is_empty()); + } + }; +} + +mod bounded { + use std::sync::Arc; + + use asyncband::pool::bounded::Pool; + use asyncband::pool::bounded::PoolConfig; + + use super::*; + + fn pool( + capacity: usize, + manager: Manager, + strategy: Option, + ) -> Arc> { + let mut config = PoolConfig::new(capacity); + if let Some(strategy) = strategy { + config = config.with_recycle_cancelled_strategy(strategy); + } + Pool::new(config, manager) + } + + recycle_contract!(); +} + +mod unbounded { + use std::sync::Arc; + + use asyncband::pool::unbounded::Pool; + use asyncband::pool::unbounded::PoolConfig; + + use super::*; + + fn pool( + _capacity: usize, + manager: Manager, + strategy: Option, + ) -> Arc> { + let mut config = PoolConfig::new(); + if let Some(strategy) = strategy { + config = config.with_recycle_cancelled_strategy(strategy); + } + Pool::new(config, manager) + } + + recycle_contract!(); +} diff --git a/tests-integration/tests/pool_test/replenish.rs b/tests-integration/tests/pool_test/replenish.rs new file mode 100644 index 00000000..b0a351dc --- /dev/null +++ b/tests-integration/tests/pool_test/replenish.rs @@ -0,0 +1,152 @@ +// 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::task::Poll; + +use asyncband::pool::bounded::Pool; +use asyncband::pool::bounded::PoolConfig; +use tests_integration::WakeCounter; +use tests_integration::expect_ready; +use tests_integration::poll_once; +use tests_integration::poll_with; + +use super::support::Manager; +use super::support::ManagerError; +use super::support::ready; + +#[test] +fn targets_account_for_idle_objects_and_checked_out_capacity() { + for capacity in [1, 4] { + for checked_out in 0..=capacity { + for idle in 0..=capacity - checked_out { + for target in [0, 1, capacity, usize::MAX] { + let manager = Manager::default(); + let pool = Pool::new(PoolConfig::new(capacity), manager.clone()); + let active: Vec<_> = (0..checked_out) + .map(|_| ready(pool.get()).unwrap()) + .collect(); + assert_eq!(ready(pool.replenish_to(idle)), Ok(idle)); + let expected_idle = target.min(capacity - checked_out).max(idle); + assert_eq!(ready(pool.replenish_to(target)), Ok(expected_idle - idle)); + assert_eq!(pool.status().idle_count, expected_idle); + assert_eq!(pool.status().current_size, checked_out + expected_idle); + assert_eq!(manager.created(), checked_out + expected_idle); + drop(active); + assert_eq!(pool.status().idle_count, manager.created()); + } + } + } + } +} + +#[test] +fn creation_failure_keeps_completed_work_and_allows_a_retry() { + let manager = Manager::default(); + manager.pause_create().send(Ok(())).unwrap(); + manager.pause_create().send(Err(ManagerError)).unwrap(); + let pool = Pool::new(PoolConfig::new(3), manager.clone()); + + assert_eq!(ready(pool.replenish_to(3)), Err(ManagerError)); + assert_eq!(pool.status().current_size, 1); + assert_eq!(pool.status().idle_count, 1); + assert_eq!(manager.created(), 2); + + assert_eq!(ready(pool.replenish_to(3)), Ok(2)); + let mut ids: Vec<_> = (0..3) + .map(|_| ready(pool.get()).unwrap().detach()) + .collect(); + ids.sort_unstable(); + assert_eq!(ids, [1000, 1002, 1003]); +} + +#[test] +fn in_flight_creation_reserves_capacity_against_other_replenishers() { + let manager = Manager::default(); + let pool = Pool::new(PoolConfig::new(3), manager.clone()); + assert_eq!(ready(pool.replenish_to(1)), Ok(1)); + + let creation = manager.pause_create(); + let mut first = Box::pin(pool.replenish_to(3)); + assert!(poll_once(first.as_mut()).is_pending()); + assert_eq!(manager.created(), 2); + assert_eq!(ready(pool.replenish_to(3)), Ok(0)); + assert_eq!(manager.created(), 2); + + creation.send(Ok(())).unwrap(); + assert_eq!(poll_once(first.as_mut()), Poll::Ready(Ok(2))); + assert_eq!(pool.status().current_size, 3); + assert_eq!(pool.status().idle_count, 3); +} + +#[test] +fn replenished_object_wakes_a_checkout_waiting_for_capacity() { + let manager = Manager::default(); + let pool = Pool::new(PoolConfig::new(2), manager.clone()); + let held = ready(pool.get()).unwrap(); + let creation = manager.pause_create(); + let mut replenish = Box::pin(pool.replenish_to(2)); + assert!(poll_once(replenish.as_mut()).is_pending()); + + let (waker, wakes) = WakeCounter::new(); + let mut checkout = Box::pin(pool.get()); + assert!(poll_with(checkout.as_mut(), &waker).is_pending()); + assert_eq!(manager.created(), 2); + creation.send(Ok(())).unwrap(); + assert_eq!(poll_once(replenish.as_mut()), Poll::Ready(Ok(1))); + assert!(wakes.count() > 0); + let acquired = expect_ready(poll_with(checkout.as_mut(), &waker)).unwrap(); + assert_ne!(*held, *acquired); + assert_eq!(manager.created(), 2); + assert_eq!(pool.status().idle_count, 0); + drop((held, acquired)); + assert_eq!(pool.status().idle_count, 2); +} + +#[test] +fn cancellation_keeps_finished_objects_and_releases_unfilled_reservations() { + for completed in [0, 1] { + let manager = Manager::default(); + for _ in 0..completed { + manager.pause_create().send(Ok(())).unwrap(); + } + let creation = manager.pause_create(); + let pool = Pool::new(PoolConfig::new(3), manager.clone()); + let mut replenish = Box::pin(pool.replenish_to(3)); + assert!(poll_once(replenish.as_mut()).is_pending()); + assert_eq!(pool.status().current_size, completed); + assert_eq!(pool.status().idle_count, completed); + + let mut active: Vec<_> = (0..completed).map(|_| ready(pool.get()).unwrap()).collect(); + let (waker, wakes) = WakeCounter::new(); + let mut checkout = Box::pin(pool.get()); + assert!(poll_with(checkout.as_mut(), &waker).is_pending()); + drop(replenish); + assert!(creation.is_closed()); + assert!(wakes.count() > 0); + let acquired = expect_ready(poll_with(checkout.as_mut(), &waker)).unwrap(); + active.push(acquired); + for _ in active.len()..3 { + active.push(ready(pool.get()).unwrap()); + } + assert_eq!(pool.status().current_size, 3); + assert_eq!(pool.status().idle_count, 0); + assert!(manager.detached().is_empty()); + assert!(poll_once(Box::pin(pool.get()).as_mut()).is_pending()); + drop(active); + assert_eq!(pool.status().idle_count, 3); + } +} diff --git a/tests-integration/tests/pool_test/support.rs b/tests-integration/tests/pool_test/support.rs new file mode 100644 index 00000000..f4a23596 --- /dev/null +++ b/tests-integration/tests/pool_test/support.rs @@ -0,0 +1,100 @@ +// 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 std::future::Future; +use std::pin::pin; +use std::sync::Arc; +use std::sync::Mutex; +use std::sync::atomic::AtomicUsize; +use std::sync::atomic::Ordering; + +use asyncband::pool::ManageObject; +use asyncband::pool::ObjectStatus; +use tests_integration::expect_ready; +use tests_integration::poll_once; +use tokio::sync::oneshot; + +#[derive(Debug, PartialEq, Eq)] +pub struct ManagerError; + +type Step = oneshot::Receiver>; + +#[derive(Default)] +struct State { + created: AtomicUsize, + creating: Mutex>, + recycling: Mutex>, + detached: Mutex>, +} + +#[derive(Clone, Default)] +pub struct Manager(Arc); + +impl Manager { + pub fn created(&self) -> usize { + self.0.created.load(Ordering::Relaxed) + } + + pub fn detached(&self) -> Vec { + self.0.detached.lock().unwrap().clone() + } + + pub fn pause_create(&self) -> oneshot::Sender> { + let (sender, receiver) = oneshot::channel(); + self.0.creating.lock().unwrap().push_back(receiver); + sender + } + + pub fn pause_recycle(&self) -> oneshot::Sender> { + let (sender, receiver) = oneshot::channel(); + self.0.recycling.lock().unwrap().push_back(receiver); + sender + } +} + +impl ManageObject for Manager { + type Object = usize; + type Error = ManagerError; + + async fn create(&self) -> Result { + let id = self.0.created.fetch_add(1, Ordering::Relaxed); + let step = self.0.creating.lock().unwrap().pop_front(); + if let Some(step) = step { + step.await.expect("creation controller dropped")?; + } + Ok(id) + } + + async fn is_recyclable(&self, _: &mut usize, _: &ObjectStatus) -> Result<(), ManagerError> { + let step = self.0.recycling.lock().unwrap().pop_front(); + if let Some(step) = step { + step.await.expect("recycle controller dropped")?; + } + Ok(()) + } + + fn on_detached(&self, object: &mut usize) { + self.0.detached.lock().unwrap().push(*object); + // Make hook execution visible in the values returned by retain/detach as well. + *object += 1000; + } +} + +pub fn ready(future: F) -> F::Output { + expect_ready(poll_once(pin!(future))) +}