saluki_core/pooling/
on_demand.rs

1use std::{
2    future::{ready, Ready},
3    sync::Arc,
4};
5
6use saluki_common::resource_tracking::ResourceGroupToken;
7
8use super::{Clearable, ObjectPool, PoolMetrics, Poolable, ReclaimStrategy};
9
10/// An object pool that allocates objects on demand.
11///
12/// This pool implementation is meant to satisfy the interface of [`ObjectPool`] without performing any actual pooling.
13pub struct OnDemandObjectPool<T: Poolable> {
14    strategy: Arc<OnDemandStrategy<T>>,
15}
16
17impl<T> OnDemandObjectPool<T>
18where
19    T: Poolable + 'static,
20    T::Data: Default,
21{
22    /// Creates a new `OnDemandObjectPool`.
23    pub fn new<S>(pool_name: S) -> Self
24    where
25        S: AsRef<str>,
26    {
27        Self::with_builder(pool_name, T::Data::default)
28    }
29}
30
31impl<T> OnDemandObjectPool<T>
32where
33    T: Poolable + 'static,
34{
35    /// Creates a new `OnDemandObjectPool` with the given item builder.
36    ///
37    /// `builder` is called to construct each item.
38    pub fn with_builder<S, B>(pool_name: S, builder: B) -> Self
39    where
40        S: AsRef<str>,
41        B: Fn() -> T::Data + Send + Sync + 'static,
42    {
43        let strategy = Arc::new(OnDemandStrategy::with_builder(pool_name, builder));
44
45        Self { strategy }
46    }
47}
48
49impl<T: Poolable> Clone for OnDemandObjectPool<T> {
50    fn clone(&self) -> Self {
51        Self {
52            strategy: self.strategy.clone(),
53        }
54    }
55}
56
57impl<T> ObjectPool for OnDemandObjectPool<T>
58where
59    T: Poolable + Send + Unpin + 'static,
60{
61    type Item = T;
62    type AcquireFuture = Ready<T>;
63
64    fn acquire(&self) -> Self::AcquireFuture {
65        let strategy = Arc::clone(&self.strategy);
66        let item = strategy.build();
67        ready(T::from_data(strategy, item))
68    }
69}
70
71struct OnDemandStrategy<T: Poolable> {
72    builder: Box<dyn Fn() -> T::Data + Send + Sync>,
73    resource_group: ResourceGroupToken,
74    metrics: PoolMetrics,
75}
76
77impl<T: Poolable> OnDemandStrategy<T> {
78    fn with_builder<S, B>(pool_name: S, builder: B) -> Self
79    where
80        S: AsRef<str>,
81        B: Fn() -> T::Data + Send + Sync + 'static,
82    {
83        let builder = Box::new(builder);
84
85        let metrics = PoolMetrics::new(pool_name.as_ref());
86        metrics.capacity().set(usize::MAX as f64);
87
88        Self {
89            builder,
90            resource_group: ResourceGroupToken::current(),
91            metrics,
92        }
93    }
94
95    fn build(&self) -> T::Data {
96        self.metrics.created().increment(1);
97        self.metrics.in_use().increment(1.0);
98
99        let _ = self.resource_group.enter();
100        (self.builder)()
101    }
102}
103
104impl<T: Poolable> ReclaimStrategy<T> for OnDemandStrategy<T> {
105    fn reclaim(&self, mut data: T::Data) {
106        data.clear();
107        drop(data);
108
109        self.metrics.released().increment(1);
110        self.metrics.in_use().decrement(1.0);
111    }
112}
113
114#[cfg(test)]
115mod tests {
116    use saluki_metrics::test::TestRecorder;
117    use tokio_test::{assert_ready, task::spawn};
118
119    use super::*;
120    use crate::pooled;
121
122    pooled! {
123        struct PooledValue {
124            value: u32,
125        }
126
127        clear => |this| this.value = 0
128    }
129
130    impl std::fmt::Debug for PooledValue {
131        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
132            f.debug_struct("PooledValue").finish_non_exhaustive()
133        }
134    }
135
136    #[test]
137    fn allocates_a_fresh_item_on_every_acquire_without_blocking() {
138        let recorder = TestRecorder::default();
139        let _guard = metrics::set_default_local_recorder(&recorder);
140
141        let pool = OnDemandObjectPool::<PooledValue>::new("test");
142
143        // Documented contract: this pool performs no actual pooling. Every acquire builds a brand-new
144        // item and completes immediately, even while a previously-acquired item is still outstanding
145        // (a fixed-size pool would instead block on the second acquire here).
146        let mut first_acquire = spawn(pool.acquire());
147        let first = assert_ready!(first_acquire.poll());
148        let mut second_acquire = spawn(pool.acquire());
149        let second = assert_ready!(second_acquire.poll());
150
151        // Both acquisitions allocated on demand, so two items were created and two are in use.
152        assert_eq!(
153            recorder.counter((PoolMetrics::created_name(), &[("pool_name", "test")])),
154            Some(2)
155        );
156        assert_eq!(
157            recorder.gauge((PoolMetrics::in_use_name(), &[("pool_name", "test")])),
158            Some(2.0)
159        );
160
161        // Releasing items drops them (nothing is retained) and drives the in-use gauge back to zero.
162        drop(first);
163        drop(second);
164        assert_eq!(
165            recorder.counter((PoolMetrics::released_name(), &[("pool_name", "test")])),
166            Some(2)
167        );
168        assert_eq!(
169            recorder.gauge((PoolMetrics::in_use_name(), &[("pool_name", "test")])),
170            Some(0.0)
171        );
172    }
173
174    #[test]
175    fn reports_effectively_unbounded_capacity() {
176        // The pool never enforces a ceiling, which it advertises via the capacity gauge as `usize::MAX`.
177        let recorder = TestRecorder::default();
178        let _guard = metrics::set_default_local_recorder(&recorder);
179
180        let _pool = OnDemandObjectPool::<PooledValue>::new("test");
181        assert_eq!(
182            recorder.gauge((PoolMetrics::capacity_name(), &[("pool_name", "test")])),
183            Some(usize::MAX as f64)
184        );
185    }
186}
saluki_core/pooling/on_demand.rs

saluki_core/pooling/
on_demand.rs
