Skip to main content

tenferro_internal_cpu_kernels/
buffer_pool.rs

1//! Typed host buffer pooling for reusable tensor allocations.
2//!
3//! # Examples
4//!
5//! ```rust
6//! use tenferro_internal_cpu_kernels::buffer_pool::{BufferPool, PoolScalar};
7//!
8//! let mut pool = BufferPool::new();
9//! let mut buf = pool.acquire_zeroed::<f64>(4);
10//! buf.fill(1.0);
11//! <f64 as PoolScalar>::pool_release(&mut pool, buf);
12//! assert_eq!(pool.len(), 1);
13//! ```
14
15use std::collections::BTreeMap;
16use std::env;
17use std::ffi::OsString;
18use std::fmt;
19use std::mem::{size_of, ManuallyDrop, MaybeUninit};
20use std::sync::OnceLock;
21
22use num_complex::{Complex32, Complex64};
23
24use crate::CacheStats;
25
26/// Non-Copy proof of a tracked pool checkout.
27#[derive(Debug)]
28pub(crate) enum UninitCheckoutToken {
29    /// The allocation was freshly allocated.
30    Fresh { actual_capacity: usize },
31    /// Retained storage was removed, with its actual capacity.
32    Reused { actual_capacity: usize },
33}
34
35/// Environment variable overriding the CPU buffer-pool retention cap in bytes.
36///
37/// The value is parsed as an unsigned integer. Invalid values fall back to
38/// [`DEFAULT_MAX_RETAINED_CAPACITY_BYTES`].
39pub const BUFFER_POOL_MAX_RETAINED_BYTES_ENV: &str = "TENFERRO_BUFFER_POOL_MAX_RETAINED_BYTES";
40
41/// Default retained CPU buffer capacity per backend.
42///
43/// The cap keeps long-running workloads from accumulating obsolete buffer
44/// sizes as tensor shapes grow while still preserving reuse for hot working
45/// sets.
46pub const DEFAULT_MAX_RETAINED_CAPACITY_BYTES: usize = 100 * 1024 * 1024;
47
48static DEFAULT_MAX_RETAINED_CAPACITY_FROM_ENV: OnceLock<usize> = OnceLock::new();
49
50/// Snapshot of typed host buffers retained by a [`BufferPool`].
51///
52/// `buffers` counts retained `Vec` allocations, while `capacity_bytes` counts
53/// their total element capacity in bytes. Allocators may keep freed memory in
54/// process-local arenas after a pool is cleared, so this reports memory that is
55/// still live in the pool rather than operating-system RSS.
56#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
57pub struct BufferPoolStats {
58    /// Number of retained vector allocations.
59    pub buffers: usize,
60    /// Total retained vector capacity in bytes.
61    pub capacity_bytes: usize,
62}
63
64/// Typed buffer pool keyed by element capacity and separated by scalar type.
65///
66/// Each supported dtype has an independent best-fit pool. Acquired buffers are
67/// returned without zero-initialization so kernels can avoid redundant writes
68/// when they fully overwrite the output. Use [`PoolScalar::pool_acquire_zeroed`]
69/// when the caller may read the buffer before writing every element.
70///
71/// # Examples
72///
73/// ```rust
74/// use tenferro_internal_cpu_kernels::buffer_pool::{BufferPool, PoolScalar};
75///
76/// let mut pool = BufferPool::new();
77/// let buf = pool.acquire_zeroed::<f32>(8);
78/// <f32 as PoolScalar>::pool_release(&mut pool, buf);
79/// assert_eq!(pool.len(), 1);
80/// ```
81pub struct BufferPool {
82    f64_pool: BTreeMap<usize, Vec<Vec<f64>>>,
83    f32_pool: BTreeMap<usize, Vec<Vec<f32>>>,
84    i32_pool: BTreeMap<usize, Vec<Vec<i32>>>,
85    i64_pool: BTreeMap<usize, Vec<Vec<i64>>>,
86    bool_pool: BTreeMap<usize, Vec<Vec<bool>>>,
87    c64_pool: BTreeMap<usize, Vec<Vec<Complex64>>>,
88    c32_pool: BTreeMap<usize, Vec<Vec<Complex32>>>,
89    f64_in_flight: BTreeMap<usize, usize>,
90    f32_in_flight: BTreeMap<usize, usize>,
91    i32_in_flight: BTreeMap<usize, usize>,
92    i64_in_flight: BTreeMap<usize, usize>,
93    bool_in_flight: BTreeMap<usize, usize>,
94    c64_in_flight: BTreeMap<usize, usize>,
95    c32_in_flight: BTreeMap<usize, usize>,
96    retained_capacity_bytes: usize,
97    max_retained_capacity_bytes: usize,
98}
99
100impl fmt::Debug for BufferPool {
101    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
102        f.debug_struct("BufferPool")
103            .field("stats", &self.stats())
104            .field(
105                "max_retained_capacity_bytes",
106                &self.max_retained_capacity_bytes,
107            )
108            .finish_non_exhaustive()
109    }
110}
111
112/// Scalar types supported by [`BufferPool`].
113///
114/// The trait is sealed to the scalar dtypes that tenferro currently pools for
115/// CPU execution.
116///
117/// # Examples
118///
119/// ```rust
120/// use tenferro_internal_cpu_kernels::buffer_pool::{BufferPool, PoolScalar};
121///
122/// let mut pool = BufferPool::new();
123/// let mut buf = pool.acquire_zeroed::<f64>(2);
124/// buf.copy_from_slice(&[3.0, 4.0]);
125/// <f64 as PoolScalar>::pool_release(&mut pool, buf);
126/// ```
127pub trait PoolScalar:
128    Copy + Sized + Send + Sync + tenferro_tensor::TensorScalar + private::Sealed
129{
130    /// Zero value used to initialize acquired buffers.
131    fn pool_zero() -> Self;
132
133    /// Acquire a buffer with length `len` and every element set to zero.
134    ///
135    /// This is the safe path for callers that may read the buffer before every
136    /// element is overwritten. Full-overwrite kernels should use
137    /// [`crate::PooledUninitOutput`] or an operation-specific uninitialized
138    /// destination guard.
139    ///
140    /// # Examples
141    ///
142    /// ```rust
143    /// use tenferro_internal_cpu_kernels::buffer_pool::{BufferPool, PoolScalar};
144    ///
145    /// let mut pool = BufferPool::new();
146    /// let buf = <f64 as PoolScalar>::pool_acquire_zeroed(&mut pool, 2);
147    /// assert_eq!(buf, vec![0.0, 0.0]);
148    /// ```
149    fn pool_acquire_zeroed(pool: &mut BufferPool, len: usize) -> Vec<Self>;
150
151    /// Return a buffer to the typed pool for later reuse.
152    ///
153    /// Zero-capacity buffers are ignored.
154    ///
155    /// # Examples
156    ///
157    /// ```rust
158    /// use tenferro_internal_cpu_kernels::buffer_pool::{BufferPool, PoolScalar};
159    ///
160    /// let mut pool = BufferPool::new();
161    /// let buf = vec![1.0_f32; 4];
162    /// <f32 as PoolScalar>::pool_release(&mut pool, buf);
163    /// assert_eq!(pool.len(), 1);
164    /// ```
165    fn pool_release(pool: &mut BufferPool, buf: Vec<Self>);
166}
167
168pub(crate) mod private {
169    use std::mem::MaybeUninit;
170
171    // INVARIANT: this sealed crate-private trait is the only implementation
172    // boundary for tracked guard tokens; it is not part of the public API.
173    #[allow(private_interfaces)]
174    pub trait Sealed {
175        /// Checks out uninitialized storage with an exact cleanup token.
176        ///
177        /// # Errors
178        /// Returns `Error::Validation` if retained-capacity byte accounting
179        /// overflows, or `Error::BackendSource` if fresh allocation fails.
180        fn pool_acquire_uninit_tracked(
181            pool: &mut super::BufferPool,
182            len: usize,
183        ) -> crate::Result<(Vec<MaybeUninit<Self>>, super::UninitCheckoutToken)>
184        where
185            Self: Sized;
186        fn pool_discard_uninit(
187            pool: &mut super::BufferPool,
188            data: Vec<MaybeUninit<Self>>,
189            checkout: super::UninitCheckoutToken,
190        ) where
191            Self: Sized;
192    }
193}
194
195fn take_best_fit<T>(pool: &mut BTreeMap<usize, Vec<Vec<T>>>, len: usize) -> Option<Vec<T>> {
196    let key = *pool.range(len..).next()?.0;
197    let buf = {
198        let vecs = pool.get_mut(&key)?;
199        vecs.pop()
200    };
201    if pool.get(&key).is_some_and(Vec::is_empty) {
202        pool.remove(&key);
203    }
204    buf
205}
206
207fn pool_len<T>(pool: &BTreeMap<usize, Vec<Vec<T>>>) -> usize {
208    pool.values().map(Vec::len).sum()
209}
210
211fn evict_one_from_pool<T>(pool: &mut BTreeMap<usize, Vec<Vec<T>>>) -> Option<usize> {
212    let key = *pool.keys().next()?;
213    let vecs = pool.get_mut(&key)?;
214    let _ = vecs.pop()?;
215    if vecs.is_empty() {
216        pool.remove(&key);
217    }
218    Some(key.saturating_mul(size_of::<T>()))
219}
220
221#[derive(Clone, Copy)]
222enum TypedPoolKind {
223    F64,
224    F32,
225    I32,
226    I64,
227    Bool,
228    C64,
229    C32,
230}
231
232fn smallest_pool_candidate<T>(
233    pool: &BTreeMap<usize, Vec<Vec<T>>>,
234    kind: TypedPoolKind,
235) -> Option<(usize, TypedPoolKind)> {
236    pool.keys()
237        .next()
238        .map(|&capacity| (capacity.saturating_mul(size_of::<T>()), kind))
239}
240
241fn increment_in_flight(in_flight: &mut BTreeMap<usize, usize>, cap: usize) {
242    if cap > 0 {
243        *in_flight.entry(cap).or_default() += 1;
244    }
245}
246
247fn decrement_in_flight(in_flight: &mut BTreeMap<usize, usize>, cap: usize) {
248    if cap == 0 {
249        return;
250    }
251    let Some(count) = in_flight.get_mut(&cap) else {
252        return;
253    };
254    *count -= 1;
255    if *count == 0 {
256        in_flight.remove(&cap);
257    }
258}
259
260fn replenish_in_flight_for<T>(
261    pool: &mut BTreeMap<usize, Vec<Vec<T>>>,
262    in_flight: &mut BTreeMap<usize, usize>,
263    retained_capacity_bytes: &mut usize,
264) {
265    for (&cap, &count) in in_flight.iter() {
266        for _ in 0..count {
267            let mut replacement = Vec::new();
268            if replacement.try_reserve_exact(cap).is_err() {
269                continue;
270            }
271            let actual_cap = replacement.capacity();
272            *retained_capacity_bytes =
273                retained_capacity_bytes.saturating_add(actual_cap.saturating_mul(size_of::<T>()));
274            pool.entry(actual_cap).or_default().push(replacement);
275        }
276    }
277    in_flight.clear();
278}
279
280macro_rules! impl_pool_scalar {
281    ($ty:ty, $field:ident, $in_flight:ident, $zero:expr) => {
282        // INVARIANT: this implementation is reachable only through the sealed
283        // crate-private helper and cannot be named by sibling crates.
284        #[allow(private_interfaces)]
285        impl private::Sealed for $ty {
286            fn pool_acquire_uninit_tracked(
287                pool: &mut BufferPool,
288                len: usize,
289            ) -> crate::Result<(Vec<MaybeUninit<Self>>, UninitCheckoutToken)> {
290                match take_best_fit(&mut pool.$field, len) {
291                    Some(buf) => {
292                        let cap = buf.capacity();
293                        let bytes = cap.checked_mul(size_of::<Self>()).ok_or_else(|| {
294                            crate::Error::invalid_argument(
295                                "pooled_uninit_output",
296                                "length",
297                                "pool capacity byte length overflow",
298                            )
299                        })?;
300                        pool.retained_capacity_bytes -= bytes;
301                        increment_in_flight(&mut pool.$in_flight, cap);
302                        let mut buf = ManuallyDrop::new(buf);
303                        // SAFETY: MaybeUninit<Self> has the same layout as Self and len <= cap.
304                        let buf = unsafe { Vec::from_raw_parts(buf.as_mut_ptr().cast(), len, cap) };
305                        Ok((
306                            buf,
307                            UninitCheckoutToken::Reused {
308                                actual_capacity: cap,
309                            },
310                        ))
311                    }
312                    None => {
313                        let mut buf = Vec::new();
314                        buf.try_reserve_exact(len).map_err(|err| {
315                            crate::Error::backend_source("pooled_uninit_output", err)
316                        })?;
317                        // SAFETY: every bit pattern is valid for MaybeUninit.
318                        unsafe { buf.set_len(len) };
319                        let actual_capacity = buf.capacity();
320                        Ok((buf, UninitCheckoutToken::Fresh { actual_capacity }))
321                    }
322                }
323            }
324
325            fn pool_discard_uninit(
326                pool: &mut BufferPool,
327                data: Vec<MaybeUninit<Self>>,
328                checkout: UninitCheckoutToken,
329            ) {
330                drop(data);
331                if let UninitCheckoutToken::Reused { actual_capacity } = checkout {
332                    decrement_in_flight(&mut pool.$in_flight, actual_capacity);
333                }
334            }
335        }
336
337        impl PoolScalar for $ty {
338            fn pool_zero() -> Self {
339                $zero
340            }
341
342            fn pool_acquire_zeroed(pool: &mut BufferPool, len: usize) -> Vec<Self> {
343                match take_best_fit(&mut pool.$field, len) {
344                    Some(mut buf) => {
345                        pool.retained_capacity_bytes = pool
346                            .retained_capacity_bytes
347                            .saturating_sub(buf.capacity().saturating_mul(size_of::<Self>()));
348                        increment_in_flight(&mut pool.$in_flight, buf.capacity());
349                        buf.resize(len, Self::pool_zero());
350                        buf.fill(Self::pool_zero());
351                        buf
352                    }
353                    None => vec![Self::pool_zero(); len],
354                }
355            }
356
357            fn pool_release(pool: &mut BufferPool, buf: Vec<Self>) {
358                let cap = buf.capacity();
359                if cap > 0 {
360                    decrement_in_flight(&mut pool.$in_flight, cap);
361                    pool.retained_capacity_bytes = pool
362                        .retained_capacity_bytes
363                        .saturating_add(cap.saturating_mul(size_of::<Self>()));
364                    pool.$field.entry(cap).or_default().push(buf);
365                    pool.enforce_retention_limit();
366                }
367            }
368        }
369    };
370}
371
372impl_pool_scalar!(f64, f64_pool, f64_in_flight, 0.0);
373impl_pool_scalar!(f32, f32_pool, f32_in_flight, 0.0);
374impl_pool_scalar!(i32, i32_pool, i32_in_flight, 0);
375impl_pool_scalar!(i64, i64_pool, i64_in_flight, 0);
376impl_pool_scalar!(bool, bool_pool, bool_in_flight, false);
377impl_pool_scalar!(Complex64, c64_pool, c64_in_flight, Complex64::new(0.0, 0.0));
378impl_pool_scalar!(Complex32, c32_pool, c32_in_flight, Complex32::new(0.0, 0.0));
379
380impl BufferPool {
381    #[cfg(test)]
382    pub(crate) fn in_flight_is_empty(&self) -> bool {
383        self.f64_in_flight.is_empty()
384            && self.f32_in_flight.is_empty()
385            && self.i32_in_flight.is_empty()
386            && self.i64_in_flight.is_empty()
387            && self.bool_in_flight.is_empty()
388            && self.c64_in_flight.is_empty()
389            && self.c32_in_flight.is_empty()
390    }
391    /// Create an empty typed buffer pool.
392    ///
393    /// # Examples
394    ///
395    /// ```rust
396    /// use tenferro_internal_cpu_kernels::buffer_pool::BufferPool;
397    ///
398    /// let pool = BufferPool::new();
399    /// assert!(pool.is_empty());
400    /// ```
401    pub fn new() -> Self {
402        Self::with_max_retained_capacity_bytes(default_max_retained_capacity_bytes())
403    }
404
405    /// Create an empty typed buffer pool with a specific retention cap.
406    ///
407    /// A cap of zero disables retention. Use [`BufferPool::unbounded`] only for
408    /// diagnostics or workloads that are externally memory-limited.
409    ///
410    /// # Examples
411    ///
412    /// ```rust
413    /// use tenferro_internal_cpu_kernels::buffer_pool::BufferPool;
414    ///
415    /// let pool = BufferPool::with_max_retained_capacity_bytes(1024);
416    /// assert_eq!(pool.max_retained_capacity_bytes(), 1024);
417    /// ```
418    pub fn with_max_retained_capacity_bytes(max_retained_capacity_bytes: usize) -> Self {
419        Self {
420            f64_pool: BTreeMap::new(),
421            f32_pool: BTreeMap::new(),
422            i32_pool: BTreeMap::new(),
423            i64_pool: BTreeMap::new(),
424            bool_pool: BTreeMap::new(),
425            c64_pool: BTreeMap::new(),
426            c32_pool: BTreeMap::new(),
427            f64_in_flight: BTreeMap::new(),
428            f32_in_flight: BTreeMap::new(),
429            i32_in_flight: BTreeMap::new(),
430            i64_in_flight: BTreeMap::new(),
431            bool_in_flight: BTreeMap::new(),
432            c64_in_flight: BTreeMap::new(),
433            c32_in_flight: BTreeMap::new(),
434            retained_capacity_bytes: 0,
435            max_retained_capacity_bytes,
436        }
437    }
438
439    /// Create an empty typed buffer pool without a retention cap.
440    ///
441    /// This preserves the historical behavior and is mainly useful for
442    /// diagnostics or controlled benchmarks.
443    ///
444    /// # Examples
445    ///
446    /// ```rust
447    /// use tenferro_internal_cpu_kernels::buffer_pool::BufferPool;
448    ///
449    /// let pool = BufferPool::unbounded();
450    /// assert_eq!(pool.max_retained_capacity_bytes(), usize::MAX);
451    /// ```
452    pub fn unbounded() -> Self {
453        Self::with_max_retained_capacity_bytes(usize::MAX)
454    }
455
456    /// Maximum retained typed host-buffer capacity in bytes.
457    ///
458    /// # Examples
459    ///
460    /// ```rust
461    /// use tenferro_internal_cpu_kernels::buffer_pool::BufferPool;
462    ///
463    /// let pool = BufferPool::with_max_retained_capacity_bytes(4096);
464    /// assert_eq!(pool.max_retained_capacity_bytes(), 4096);
465    /// ```
466    pub fn max_retained_capacity_bytes(&self) -> usize {
467        self.max_retained_capacity_bytes
468    }
469
470    /// Update the maximum retained typed host-buffer capacity in bytes.
471    ///
472    /// Shrinking below the currently retained capacity immediately evicts
473    /// retained buffers until the new cap is satisfied. A cap of zero disables
474    /// retention.
475    ///
476    /// # Examples
477    ///
478    /// ```
479    /// use tenferro_internal_cpu_kernels::buffer_pool::{BufferPool, PoolScalar};
480    ///
481    /// let mut pool = BufferPool::with_max_retained_capacity_bytes(1024);
482    /// <f64 as PoolScalar>::pool_release(&mut pool, Vec::with_capacity(128));
483    /// pool.set_max_retained_capacity_bytes(0);
484    /// assert_eq!(pool.max_retained_capacity_bytes(), 0);
485    /// assert!(pool.is_empty());
486    /// ```
487    pub fn set_max_retained_capacity_bytes(&mut self, max_retained_capacity_bytes: usize) {
488        self.max_retained_capacity_bytes = max_retained_capacity_bytes;
489        self.enforce_retention_limit();
490    }
491
492    /// Number of retained buffers across all typed pools.
493    ///
494    /// # Examples
495    ///
496    /// ```rust
497    /// use tenferro_internal_cpu_kernels::buffer_pool::{BufferPool, PoolScalar};
498    ///
499    /// let mut pool = BufferPool::new();
500    /// <f64 as PoolScalar>::pool_release(&mut pool, vec![0.0; 2]);
501    /// assert_eq!(pool.len(), 1);
502    /// ```
503    pub fn len(&self) -> usize {
504        self.stats().buffers
505    }
506
507    /// Total retained typed host-buffer capacity in bytes.
508    ///
509    /// This counts capacity that is still live in the pool. The operating
510    /// system RSS may remain high after clearing the pool because the process
511    /// allocator can keep freed pages for future allocations.
512    ///
513    /// # Examples
514    ///
515    /// ```rust
516    /// use tenferro_internal_cpu_kernels::buffer_pool::{BufferPool, PoolScalar};
517    ///
518    /// let mut pool = BufferPool::new();
519    /// <f64 as PoolScalar>::pool_release(&mut pool, Vec::with_capacity(2));
520    /// assert_eq!(pool.retained_capacity_bytes(), 16);
521    /// ```
522    pub fn retained_capacity_bytes(&self) -> usize {
523        self.stats().capacity_bytes
524    }
525
526    /// Snapshot retained-buffer count and capacity.
527    ///
528    /// # Examples
529    ///
530    /// ```rust
531    /// use tenferro_internal_cpu_kernels::buffer_pool::{BufferPool, PoolScalar};
532    ///
533    /// let mut pool = BufferPool::new();
534    /// <f32 as PoolScalar>::pool_release(&mut pool, Vec::with_capacity(4));
535    /// let stats = pool.stats();
536    /// assert_eq!(stats.buffers, 1);
537    /// assert_eq!(stats.capacity_bytes, 16);
538    /// ```
539    pub fn stats(&self) -> BufferPoolStats {
540        BufferPoolStats {
541            buffers: pool_len(&self.f64_pool)
542                + pool_len(&self.f32_pool)
543                + pool_len(&self.i32_pool)
544                + pool_len(&self.i64_pool)
545                + pool_len(&self.bool_pool)
546                + pool_len(&self.c64_pool)
547                + pool_len(&self.c32_pool),
548            capacity_bytes: self.retained_capacity_bytes,
549        }
550    }
551
552    /// Return cache-style stats for the buffers retained by this pool.
553    ///
554    /// `entries` is the number of retained buffers, and `retained_bytes` is the
555    /// total retained vector capacity in bytes.
556    ///
557    /// # Examples
558    ///
559    /// ```
560    /// use tenferro_internal_cpu_kernels::buffer_pool::{BufferPool, PoolScalar};
561    ///
562    /// let mut pool = BufferPool::new();
563    /// <f32 as PoolScalar>::pool_release(&mut pool, Vec::with_capacity(4));
564    /// let stats = pool.cache_stats();
565    /// assert_eq!(stats.entries, 1);
566    /// assert_eq!(stats.retained_bytes, 16);
567    /// ```
568    pub fn cache_stats(&self) -> CacheStats {
569        let stats = self.stats();
570        CacheStats {
571            entries: stats.buffers,
572            retained_bytes: stats.capacity_bytes,
573            hits: 0,
574            misses: 0,
575            evictions: 0,
576            clears: 0,
577        }
578    }
579
580    /// Acquire a typed vector with length 0 and at least `cap` capacity.
581    ///
582    /// Returned buffers come from the typed pool when possible and are ready
583    /// for push-based population.
584    ///
585    /// # Examples
586    ///
587    /// ```rust
588    /// use tenferro_internal_cpu_kernels::buffer_pool::BufferPool;
589    ///
590    /// let mut pool = BufferPool::new();
591    /// let mut buf = pool.acquire_with_capacity::<f64>(4);
592    /// buf.extend_from_slice(&[1.0, 2.0]);
593    /// assert_eq!(buf.len(), 2);
594    /// assert!(buf.capacity() >= 4);
595    /// ```
596    pub fn acquire_with_capacity<T: PoolScalar>(&mut self, cap: usize) -> Vec<T> {
597        if cap == 0 {
598            return Vec::new();
599        }
600
601        let (data, _checkout) = <T as private::Sealed>::pool_acquire_uninit_tracked(self, cap)
602            .expect("validated typed pool capacity must be acquirable");
603        let mut data = ManuallyDrop::new(data);
604        let ptr = data.as_mut_ptr().cast::<T>();
605        let capacity = data.capacity();
606        // SAFETY: `MaybeUninit<T>` and `T` have identical layouts, and the
607        // returned vector has length zero, so no element is read before push.
608        unsafe { Vec::from_raw_parts(ptr, 0, capacity) }
609    }
610
611    /// Acquire a typed vector with length `len` initialized to zero.
612    ///
613    /// Use this only when the caller may read elements before overwriting the
614    /// entire buffer. Full-overwrite kernels should use
615    /// [`crate::PooledUninitOutput`] to avoid the initialization cost.
616    ///
617    /// # Examples
618    ///
619    /// ```rust
620    /// use tenferro_internal_cpu_kernels::buffer_pool::BufferPool;
621    ///
622    /// let mut pool = BufferPool::new();
623    /// let buf = pool.acquire_zeroed::<f32>(3);
624    /// assert_eq!(buf, vec![0.0, 0.0, 0.0]);
625    /// ```
626    pub fn acquire_zeroed<T: PoolScalar>(&mut self, len: usize) -> Vec<T> {
627        T::pool_acquire_zeroed(self, len)
628    }
629
630    /// Whether all typed pools are empty.
631    ///
632    /// # Examples
633    ///
634    /// ```rust
635    /// use tenferro_internal_cpu_kernels::buffer_pool::BufferPool;
636    ///
637    /// let pool = BufferPool::new();
638    /// assert!(pool.is_empty());
639    /// ```
640    pub fn is_empty(&self) -> bool {
641        self.f64_pool.is_empty()
642            && self.f32_pool.is_empty()
643            && self.i32_pool.is_empty()
644            && self.i64_pool.is_empty()
645            && self.bool_pool.is_empty()
646            && self.c64_pool.is_empty()
647            && self.c32_pool.is_empty()
648    }
649
650    /// Drop all retained buffers from the pool.
651    ///
652    /// This releases the vectors owned by the pool. The process allocator may
653    /// still keep freed pages mapped for reuse, so operating-system RSS is not
654    /// guaranteed to fall immediately.
655    ///
656    /// # Examples
657    ///
658    /// ```rust
659    /// use tenferro_internal_cpu_kernels::buffer_pool::{BufferPool, PoolScalar};
660    ///
661    /// let mut pool = BufferPool::new();
662    /// <f64 as PoolScalar>::pool_release(&mut pool, Vec::with_capacity(8));
663    /// pool.clear();
664    /// assert!(pool.is_empty());
665    /// ```
666    pub fn clear(&mut self) {
667        self.f64_pool.clear();
668        self.f32_pool.clear();
669        self.i32_pool.clear();
670        self.i64_pool.clear();
671        self.bool_pool.clear();
672        self.c64_pool.clear();
673        self.c32_pool.clear();
674        self.clear_in_flight_retained();
675        self.retained_capacity_bytes = 0;
676    }
677
678    #[doc(hidden)]
679    pub fn clear_in_flight_retained(&mut self) {
680        self.f64_in_flight.clear();
681        self.f32_in_flight.clear();
682        self.i32_in_flight.clear();
683        self.i64_in_flight.clear();
684        self.bool_in_flight.clear();
685        self.c64_in_flight.clear();
686        self.c32_in_flight.clear();
687    }
688
689    #[doc(hidden)]
690    pub fn replenish_in_flight_retained(&mut self) {
691        replenish_in_flight_for(
692            &mut self.f64_pool,
693            &mut self.f64_in_flight,
694            &mut self.retained_capacity_bytes,
695        );
696        replenish_in_flight_for(
697            &mut self.f32_pool,
698            &mut self.f32_in_flight,
699            &mut self.retained_capacity_bytes,
700        );
701        replenish_in_flight_for(
702            &mut self.i32_pool,
703            &mut self.i32_in_flight,
704            &mut self.retained_capacity_bytes,
705        );
706        replenish_in_flight_for(
707            &mut self.i64_pool,
708            &mut self.i64_in_flight,
709            &mut self.retained_capacity_bytes,
710        );
711        replenish_in_flight_for(
712            &mut self.bool_pool,
713            &mut self.bool_in_flight,
714            &mut self.retained_capacity_bytes,
715        );
716        replenish_in_flight_for(
717            &mut self.c64_pool,
718            &mut self.c64_in_flight,
719            &mut self.retained_capacity_bytes,
720        );
721        replenish_in_flight_for(
722            &mut self.c32_pool,
723            &mut self.c32_in_flight,
724            &mut self.retained_capacity_bytes,
725        );
726        self.enforce_retention_limit();
727    }
728
729    fn enforce_retention_limit(&mut self) {
730        while self.retained_capacity_bytes > self.max_retained_capacity_bytes {
731            let Some(evicted_bytes) = self.evict_smallest_retained_buffer() else {
732                self.retained_capacity_bytes = 0;
733                return;
734            };
735            if evicted_bytes == 0 {
736                if self.is_empty() {
737                    self.retained_capacity_bytes = 0;
738                    return;
739                }
740                continue;
741            }
742            self.retained_capacity_bytes =
743                self.retained_capacity_bytes.saturating_sub(evicted_bytes);
744        }
745    }
746
747    fn evict_smallest_retained_buffer(&mut self) -> Option<usize> {
748        let candidates = [
749            smallest_pool_candidate(&self.f64_pool, TypedPoolKind::F64),
750            smallest_pool_candidate(&self.f32_pool, TypedPoolKind::F32),
751            smallest_pool_candidate(&self.i32_pool, TypedPoolKind::I32),
752            smallest_pool_candidate(&self.i64_pool, TypedPoolKind::I64),
753            smallest_pool_candidate(&self.bool_pool, TypedPoolKind::Bool),
754            smallest_pool_candidate(&self.c64_pool, TypedPoolKind::C64),
755            smallest_pool_candidate(&self.c32_pool, TypedPoolKind::C32),
756        ];
757        let (_, kind) = candidates
758            .into_iter()
759            .flatten()
760            .min_by_key(|(bytes, _)| *bytes)?;
761        match kind {
762            TypedPoolKind::F64 => evict_one_from_pool(&mut self.f64_pool),
763            TypedPoolKind::F32 => evict_one_from_pool(&mut self.f32_pool),
764            TypedPoolKind::I32 => evict_one_from_pool(&mut self.i32_pool),
765            TypedPoolKind::I64 => evict_one_from_pool(&mut self.i64_pool),
766            TypedPoolKind::Bool => evict_one_from_pool(&mut self.bool_pool),
767            TypedPoolKind::C64 => evict_one_from_pool(&mut self.c64_pool),
768            TypedPoolKind::C32 => evict_one_from_pool(&mut self.c32_pool),
769        }
770    }
771}
772
773fn default_max_retained_capacity_bytes() -> usize {
774    *DEFAULT_MAX_RETAINED_CAPACITY_FROM_ENV.get_or_init(|| {
775        parse_default_max_retained_capacity_bytes(env::var_os(BUFFER_POOL_MAX_RETAINED_BYTES_ENV))
776    })
777}
778
779fn parse_default_max_retained_capacity_bytes(value: Option<OsString>) -> usize {
780    value
781        .and_then(|value| value.into_string().ok())
782        .and_then(|value| value.parse().ok())
783        .unwrap_or(DEFAULT_MAX_RETAINED_CAPACITY_BYTES)
784}
785
786impl Default for BufferPool {
787    fn default() -> Self {
788        Self::new()
789    }
790}
791
792#[cfg(test)]
793mod tests;