Skip to main content

tenferro_cpu/
buffer_pool.rs

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