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