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;