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