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;