oxicuda_memory/pool.rs
1//! Stream-ordered memory pool for efficient async allocation.
2//!
3//! Requires CUDA 11.2+ driver. Gated behind the `pool` feature.
4//!
5//! Stream-ordered memory pools allow allocation and deallocation to be
6//! ordered relative to other operations on a CUDA stream, enabling the
7//! driver to reuse memory more aggressively and avoid synchronisation
8//! barriers that would otherwise be needed for conventional
9//! `cuMemAlloc` / `cuMemFree` calls.
10//!
11//! # Implementation note
12//!
13//! This implementation provides a practical fallback pool that reuses freed
14//! allocations by size and uses `cuMemAlloc_v2` / `cuMemFree_v2` under the
15//! hood. It keeps the same API surface as a stream-ordered pool, but does
16//! not yet expose native CUDA mempool handles.
17//!
18//! Reuse is stream-ordered via a recycle event: dropping a [`PooledBuffer`]
19//! records a `CU_EVENT_DISABLE_TIMING` event on the stream it was allocated
20//! from rather than immediately handing the pointer back out, so a second
21//! concurrent allocation can only reuse the pointer once all GPU work
22//! enqueued before the drop has actually completed. Each [`MemoryPool`] also
23//! retains its device's primary context for its lifetime and binds every
24//! allocation/free/event operation to that context, so the pool is safe to
25//! share across threads without mixing pointers from different devices.
26//!
27//! # API
28//!
29//! ```rust,ignore
30//! let pool = MemoryPool::new(device)?;
31//! let buf = PooledBuffer::<f32>::alloc_async(&pool, 1024, &stream)?;
32//! // … use buf in kernels on `stream` …
33//! // `buf` is dropped here: a recycle event is enqueued on `stream`, and the
34//! // pointer becomes reusable by a later `alloc_async` only once that event
35//! // (i.e. all prior work on `stream`) has completed.
36//! ```
37
38#![cfg(feature = "pool")]
39
40use std::collections::HashMap;
41use std::marker::PhantomData;
42use std::sync::atomic::{AtomicUsize, Ordering};
43use std::sync::{Arc, Mutex};
44
45use oxicuda_driver::error::{CudaError, CudaResult, check};
46use oxicuda_driver::ffi::{
47 CU_EVENT_DISABLE_TIMING, CUDA_ERROR_NOT_READY, CUcontext, CUdeviceptr, CUevent,
48 CUmemAllocationHandleType, CUmemAllocationType, CUmemLocation, CUmemLocationType,
49 CUmemPoolProps, CUmemoryPool, CUstream,
50};
51use oxicuda_driver::loader::{DriverApi, try_driver};
52use oxicuda_driver::stream::Stream;
53use tracing::warn;
54
55// ---------------------------------------------------------------------------
56// MemoryPool
57// ---------------------------------------------------------------------------
58
59/// A stream-ordered memory pool (CUDA 11.2+).
60///
61/// Memory pools allow the driver to reuse freed allocations without
62/// returning them to the OS, reducing allocation latency and avoiding
63/// the implicit synchronisation of `cuMemFree`.
64///
65/// # Status
66///
67/// `MemoryPool` is a software pool layered on top of `cuMemAlloc_v2`.
68/// For a thin wrapper over the *native* CUDA stream-ordered memory pool
69/// API (`cuMemPoolCreate`, `cuMemPoolDestroy`, `cuMemAllocFromPoolAsync`,
70/// `cuMemFreeAsync`), use [`NativeMemoryPool`].
71///
72/// Statistics for a memory pool's allocation behaviour.
73///
74/// These statistics track the total bytes allocated, peak usage,
75/// allocation count, and free count for a given pool.
76#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
77pub struct PoolStats {
78 /// Total bytes currently allocated from the pool.
79 pub allocated_bytes: usize,
80 /// Peak bytes allocated at any point during the pool's lifetime.
81 pub peak_bytes: usize,
82 /// Total number of allocations performed.
83 pub allocation_count: u64,
84 /// Total number of frees performed.
85 pub free_count: u64,
86}
87
88#[derive(Debug)]
89struct MemoryPoolInner {
90 handle: u64,
91 device_ordinal: i32,
92 /// The device's primary context, retained for the lifetime of the pool.
93 ///
94 /// All allocations, frees, and event operations issued by this pool are
95 /// performed with this context made current on the calling thread (see
96 /// [`with_device_context`](Self::with_device_context)), so that a pool
97 /// shared across threads always targets `device_ordinal` regardless of
98 /// whatever context happens to already be current on a given thread.
99 ctx: CUcontext,
100 threshold_bytes: AtomicUsize,
101 cached_bytes: AtomicUsize,
102 stats: Mutex<PoolStats>,
103 /// Free-list bins keyed by allocation size. Each entry pairs a device
104 /// pointer with the (possibly null) recycle-safety event that was
105 /// recorded on the freeing `PooledBuffer`'s stream; the pointer must not
106 /// be handed out again until that event has completed (see
107 /// [`try_pop_reuse`](Self::try_pop_reuse)).
108 free_bins: Mutex<HashMap<usize, Vec<(CUdeviceptr, CUevent)>>>,
109}
110
111impl MemoryPoolInner {
112 /// Runs `f` with this pool's device context current on the calling
113 /// thread, restoring the caller's previous context before returning.
114 ///
115 /// This is the mechanism that binds every driver call issued by the pool
116 /// to `device_ordinal`'s context, so a `MemoryPool` shared across threads
117 /// (or used from a thread whose current context targets a different
118 /// device) never mixes pointers from different devices in one free-bin
119 /// map.
120 fn with_device_context<R>(
121 &self,
122 f: impl FnOnce(&'static DriverApi) -> CudaResult<R>,
123 ) -> CudaResult<R> {
124 let api = try_driver()?;
125 let mut prev = CUcontext::default();
126 check(unsafe { (api.cu_ctx_get_current)(&mut prev) })?;
127 check(unsafe { (api.cu_ctx_set_current)(self.ctx) })?;
128
129 let result = f(api);
130
131 let restore_rc = check(unsafe { (api.cu_ctx_set_current)(prev) });
132 match result {
133 Ok(value) => restore_rc.map(|()| value),
134 Err(e) => {
135 if let Err(restore_err) = restore_rc {
136 warn!(
137 "failed to restore previous CUDA context after pool operation \
138 (original error: {e}): {restore_err}"
139 );
140 }
141 Err(e)
142 }
143 }
144 }
145
146 fn allocate_fresh(&self, bytes: usize) -> CudaResult<CUdeviceptr> {
147 self.with_device_context(|api| {
148 let mut ptr: CUdeviceptr = 0;
149 let rc = unsafe { (api.cu_mem_alloc_v2)(&mut ptr, bytes) };
150 oxicuda_driver::check(rc)?;
151 Ok(ptr)
152 })
153 }
154
155 fn free_ptr(&self, ptr: CUdeviceptr) -> CudaResult<()> {
156 self.with_device_context(|api| {
157 let rc = unsafe { (api.cu_mem_free_v2)(ptr) };
158 oxicuda_driver::check(rc)
159 })
160 }
161
162 /// Creates a `CU_EVENT_DISABLE_TIMING` event and records it on `stream`.
163 ///
164 /// Returns the raw event handle; the caller owns it and must eventually
165 /// pass it to [`destroy_event`](Self::destroy_event).
166 fn record_recycle_event(&self, stream: CUstream) -> CudaResult<CUevent> {
167 self.with_device_context(|api| {
168 let mut event = CUevent::default();
169 check(unsafe { (api.cu_event_create)(&mut event, CU_EVENT_DISABLE_TIMING) })?;
170 if let Err(e) = check(unsafe { (api.cu_event_record)(event, stream) }) {
171 let _ = unsafe { (api.cu_event_destroy_v2)(event) };
172 return Err(e);
173 }
174 Ok(event)
175 })
176 }
177
178 /// Returns `true` if `event` is null (nothing to wait for) or has
179 /// completed; `false` if it is still pending on its stream.
180 fn event_ready(&self, event: CUevent) -> bool {
181 if event.is_null() {
182 return true;
183 }
184 self.with_device_context(|api| {
185 let rc = unsafe { (api.cu_event_query)(event) };
186 if rc == 0 {
187 Ok(true)
188 } else if rc == CUDA_ERROR_NOT_READY {
189 Ok(false)
190 } else {
191 Err(CudaError::from_raw(rc))
192 }
193 })
194 // A query error (rather than "not ready") is treated conservatively
195 // as "not yet safe to reuse" so the pointer is never handed out
196 // early; the entry simply remains in the free bin.
197 .unwrap_or(false)
198 }
199
200 /// Blocks until `event` has completed. A null event is a no-op.
201 fn synchronize_event(&self, event: CUevent) -> CudaResult<()> {
202 if event.is_null() {
203 return Ok(());
204 }
205 self.with_device_context(|api| check(unsafe { (api.cu_event_synchronize)(event) }))
206 }
207
208 /// Destroys `event`, logging (but not propagating) any driver error.
209 /// A null event is a no-op.
210 fn destroy_event(&self, event: CUevent) {
211 if event.is_null() {
212 return;
213 }
214 let result =
215 self.with_device_context(|api| check(unsafe { (api.cu_event_destroy_v2)(event) }));
216 if let Err(e) = result {
217 warn!("cuEventDestroy_v2 failed for pooled-buffer recycle event: {e}");
218 }
219 }
220
221 fn try_pop_reuse(&self, bytes: usize) -> CudaResult<Option<CUdeviceptr>> {
222 let popped = {
223 let mut bins = self.free_bins.lock().map_err(|_| CudaError::Unknown(0))?;
224 let Some(vec) = bins.get_mut(&bytes) else {
225 return Ok(None);
226 };
227 // Only hand out an entry whose recycle event has completed —
228 // i.e. all GPU work enqueued before the corresponding `Drop`
229 // has actually finished — so a second concurrent user can never
230 // receive a pointer that is still in flight. Entries that are
231 // not yet ready are left in the bin for a later call.
232 let ready_idx = vec.iter().position(|(_, event)| self.event_ready(*event));
233 ready_idx.map(|idx| vec.swap_remove(idx))
234 };
235
236 let Some((ptr, event)) = popped else {
237 return Ok(None);
238 };
239 self.destroy_event(event);
240 self.cached_bytes.fetch_sub(bytes, Ordering::Relaxed);
241 Ok(Some(ptr))
242 }
243
244 fn stash_freed(&self, ptr: CUdeviceptr, bytes: usize, event: CUevent) -> CudaResult<()> {
245 let mut bins = self.free_bins.lock().map_err(|_| CudaError::Unknown(0))?;
246 bins.entry(bytes).or_default().push((ptr, event));
247 self.cached_bytes.fetch_add(bytes, Ordering::Relaxed);
248 Ok(())
249 }
250
251 fn release_cached_until(&self, keep_bytes: usize) -> CudaResult<()> {
252 loop {
253 let cached = self.cached_bytes.load(Ordering::Relaxed);
254 if cached <= keep_bytes {
255 return Ok(());
256 }
257
258 let popped = {
259 let mut bins = self.free_bins.lock().map_err(|_| CudaError::Unknown(0))?;
260 let mut candidate: Option<(usize, CUdeviceptr, CUevent)> = None;
261 for (size, vec) in bins.iter_mut() {
262 if let Some((ptr, event)) = vec.pop() {
263 candidate = Some((*size, ptr, event));
264 break;
265 }
266 }
267 candidate
268 };
269
270 let Some((size, ptr, event)) = popped else {
271 return Ok(());
272 };
273 // A cached pointer may still have GPU work in flight against it;
274 // block until its recorded recycle event completes before
275 // handing the memory back to `cuMemFree_v2`.
276 self.synchronize_event(event)?;
277 self.destroy_event(event);
278 self.free_ptr(ptr)?;
279 self.cached_bytes.fetch_sub(size, Ordering::Relaxed);
280 }
281 }
282
283 fn update_alloc_stats(&self, bytes: usize) {
284 if let Ok(mut stats) = self.stats.lock() {
285 stats.allocated_bytes = stats.allocated_bytes.saturating_add(bytes);
286 stats.allocation_count = stats.allocation_count.saturating_add(1);
287 if stats.allocated_bytes > stats.peak_bytes {
288 stats.peak_bytes = stats.allocated_bytes;
289 }
290 }
291 }
292
293 fn update_free_stats(&self, bytes: usize) {
294 if let Ok(mut stats) = self.stats.lock() {
295 stats.allocated_bytes = stats.allocated_bytes.saturating_sub(bytes);
296 stats.free_count = stats.free_count.saturating_add(1);
297 }
298 }
299}
300
301impl Drop for MemoryPoolInner {
302 fn drop(&mut self) {
303 let Ok(mut bins) = self.free_bins.lock() else {
304 return;
305 };
306 let mut to_free: Vec<(CUdeviceptr, CUevent)> = Vec::new();
307 for vec in bins.values_mut() {
308 to_free.append(vec);
309 }
310 drop(bins);
311
312 for (ptr, event) in to_free {
313 // Cached pointers may still have GPU work in flight; wait for
314 // their recycle event before freeing the underlying memory.
315 if let Err(e) = self.synchronize_event(event) {
316 warn!("cuEventSynchronize failed while draining pool on drop: {e}");
317 }
318 self.destroy_event(event);
319 if let Err(e) = self.free_ptr(ptr) {
320 warn!("failed to free pooled pointer {ptr:#x} during drop: {e}");
321 }
322 }
323
324 if !self.ctx.is_null() {
325 if let Ok(api) = try_driver() {
326 let rc = unsafe { (api.cu_device_primary_ctx_release_v2)(self.device_ordinal) };
327 if rc != 0 {
328 warn!(
329 cuda_error = rc,
330 device_ordinal = self.device_ordinal,
331 "cuDevicePrimaryCtxRelease_v2 failed while dropping MemoryPool"
332 );
333 }
334 }
335 }
336 }
337}
338
339/// A stream-ordered memory pool (CUDA 11.2+).
340pub struct MemoryPool {
341 inner: Arc<MemoryPoolInner>,
342}
343
344impl MemoryPool {
345 /// Creates a new memory pool on the given device.
346 ///
347 /// This retains the device's primary context for the lifetime of the
348 /// pool (released when the pool is dropped), and binds every subsequent
349 /// allocation and free issued through this pool to that context, so the
350 /// pool is safe to share across threads without mixing pointers from
351 /// different devices.
352 ///
353 /// # Errors
354 ///
355 /// * [`CudaError::InvalidDevice`] if `device_ordinal` is negative.
356 /// * [`CudaError::NotInitialized`] if no CUDA driver is available.
357 /// * Other [`CudaError`] variants if `cuDevicePrimaryCtxRetain` fails
358 /// (e.g. an out-of-range device ordinal).
359 pub fn new(device_ordinal: i32) -> CudaResult<Self> {
360 if device_ordinal < 0 {
361 return Err(CudaError::InvalidDevice);
362 }
363 let api = try_driver()?;
364 let mut ctx = CUcontext::default();
365 check(unsafe { (api.cu_device_primary_ctx_retain)(&mut ctx, device_ordinal) })?;
366 Ok(Self {
367 inner: Arc::new(MemoryPoolInner {
368 handle: 0,
369 device_ordinal,
370 ctx,
371 threshold_bytes: AtomicUsize::new(0),
372 cached_bytes: AtomicUsize::new(0),
373 stats: Mutex::new(PoolStats::default()),
374 free_bins: Mutex::new(HashMap::new()),
375 }),
376 })
377 }
378
379 /// Returns the raw pool handle.
380 ///
381 /// # Status
382 ///
383 /// Returns `0` until the pool is properly initialised.
384 #[inline]
385 pub fn raw_handle(&self) -> u64 {
386 self.inner.handle
387 }
388
389 /// Returns the device ordinal this pool targets.
390 #[inline]
391 pub fn device_ordinal(&self) -> i32 {
392 self.inner.device_ordinal
393 }
394
395 /// Returns current pool statistics.
396 ///
397 /// The statistics track allocation behaviour over the pool's lifetime.
398 #[inline]
399 pub fn stats(&self) -> PoolStats {
400 self.inner.stats.lock().map(|s| *s).unwrap_or_default()
401 }
402
403 /// Trims the pool, releasing unused memory back to the OS.
404 ///
405 /// Attempts to release memory such that the pool retains at most
406 /// `min_bytes` of unused memory.
407 ///
408 /// # Errors
409 ///
410 pub fn trim(&mut self, min_bytes: usize) -> CudaResult<()> {
411 self.inner.release_cached_until(min_bytes)
412 }
413
414 /// Sets the threshold at which the pool will automatically release
415 /// memory back to the OS.
416 ///
417 /// When the pool's unused memory exceeds `bytes`, subsequent frees
418 /// will trigger automatic trimming.
419 ///
420 /// # Errors
421 ///
422 pub fn set_threshold(&mut self, bytes: usize) -> CudaResult<()> {
423 self.inner.threshold_bytes.store(bytes, Ordering::Relaxed);
424 self.inner.release_cached_until(bytes)
425 }
426}
427
428// ---------------------------------------------------------------------------
429// PooledBuffer<T>
430// ---------------------------------------------------------------------------
431
432/// A device buffer allocated from a [`MemoryPool`].
433///
434/// Unlike [`DeviceBuffer`](crate::DeviceBuffer), a `PooledBuffer` is freed
435/// asynchronously — the free operation is enqueued on the stream rather
436/// than blocking the CPU. This enables overlap of allocation, computation,
437/// and deallocation across multiple stream operations.
438///
439/// # Stream-ordering
440///
441/// Dropping a `PooledBuffer` does not immediately reuse its device pointer.
442/// Instead, a `CU_EVENT_DISABLE_TIMING` event is recorded on the stream the
443/// buffer was allocated on (the `stream` argument to
444/// [`alloc_async`](Self::alloc_async)); the pointer is only handed back out
445/// to a later [`alloc_async`](Self::alloc_async) call once that event has completed, i.e. once
446/// all work enqueued on the stream before the drop has actually finished
447/// executing on the device. A `PooledBuffer` must therefore not outlive its
448/// stream. If the recycle event cannot be created or recorded, `Drop` falls
449/// back to a blocking `cuStreamSynchronize` on the owning stream before
450/// freeing the pointer directly, so correctness is preserved at the cost of
451/// losing the pool's overlap benefits for that allocation.
452///
453/// # Status
454///
455/// This type allocates from an in-process memory pool and returns buffers to
456/// that pool on drop.
457pub struct PooledBuffer<T: Copy> {
458 /// Raw device pointer to the pooled allocation.
459 ptr: CUdeviceptr,
460 /// Number of `T` elements.
461 len: usize,
462 /// Number of bytes in this allocation.
463 bytes: usize,
464 /// Owning pool.
465 pool: Arc<MemoryPoolInner>,
466 /// The stream this allocation was requested on. The recycle-safety
467 /// event enqueued in `Drop` is recorded against this stream.
468 stream: CUstream,
469 /// Marker for the element type.
470 _phantom: PhantomData<T>,
471}
472
473impl<T: Copy> PooledBuffer<T> {
474 /// Asynchronously allocates a buffer of `n` elements from the given pool.
475 ///
476 /// The allocation is ordered relative to other operations on `stream`;
477 /// `stream` is also recorded against so that [`Drop`] can establish that
478 /// all work using this buffer has completed before its pointer is
479 /// recycled to a later caller (see the type-level docs).
480 ///
481 /// # Errors
482 ///
483 /// * [`CudaError::InvalidValue`] if `n` is zero or `n * size_of::<T>()`
484 /// overflows.
485 /// * Other [`CudaError`] variants if the underlying allocation fails.
486 pub fn alloc_async(pool: &MemoryPool, n: usize, stream: &Stream) -> CudaResult<Self> {
487 if n == 0 {
488 return Err(CudaError::InvalidValue);
489 }
490 let bytes = n
491 .checked_mul(std::mem::size_of::<T>())
492 .ok_or(CudaError::InvalidValue)?;
493 let ptr = if let Some(reused) = pool.inner.try_pop_reuse(bytes)? {
494 reused
495 } else {
496 pool.inner.allocate_fresh(bytes)?
497 };
498 pool.inner.update_alloc_stats(bytes);
499
500 Ok(Self {
501 ptr,
502 len: n,
503 bytes,
504 pool: Arc::clone(&pool.inner),
505 stream: stream.raw(),
506 _phantom: PhantomData,
507 })
508 }
509
510 /// Returns the number of `T` elements in this buffer.
511 #[inline]
512 pub fn len(&self) -> usize {
513 self.len
514 }
515
516 /// Returns `true` if the buffer contains zero elements.
517 #[inline]
518 pub fn is_empty(&self) -> bool {
519 self.len == 0
520 }
521
522 /// Returns the total size of the allocation in bytes.
523 #[inline]
524 pub fn byte_size(&self) -> usize {
525 self.bytes
526 }
527
528 /// Returns the raw [`CUdeviceptr`] handle.
529 #[inline]
530 pub fn as_device_ptr(&self) -> CUdeviceptr {
531 self.ptr
532 }
533}
534
535impl<T: Copy> Drop for PooledBuffer<T> {
536 fn drop(&mut self) {
537 if self.ptr == 0 {
538 return;
539 }
540
541 let event = match self.pool.record_recycle_event(self.stream) {
542 Ok(event) => event,
543 Err(e) => {
544 // Without an event we cannot prove the pointer is safe to
545 // reuse without blocking, so synchronise the owning stream
546 // directly and free the pointer rather than risk handing it
547 // to a second concurrent user while GPU work may still be
548 // in flight.
549 warn!(
550 "failed to record pooled-buffer recycle event ({e}); falling back to a \
551 blocking stream synchronize before freeing directly"
552 );
553 if let Ok(api) = try_driver() {
554 let rc = unsafe { (api.cu_stream_synchronize)(self.stream) };
555 if rc != 0 {
556 warn!(cuda_error = rc, "fallback cuStreamSynchronize failed");
557 }
558 }
559 if let Err(free_err) = self.pool.free_ptr(self.ptr) {
560 warn!("direct free of pooled pointer failed: {free_err}");
561 }
562 self.pool.update_free_stats(self.bytes);
563 self.ptr = 0;
564 return;
565 }
566 };
567
568 if let Err(e) = self.pool.stash_freed(self.ptr, self.bytes, event) {
569 warn!("failed to return pooled pointer to free list: {e}; freeing directly");
570 self.pool.destroy_event(event);
571 if let Err(free_err) = self.pool.free_ptr(self.ptr) {
572 warn!("direct free of pooled pointer failed: {free_err}");
573 }
574 self.pool.update_free_stats(self.bytes);
575 self.ptr = 0;
576 return;
577 }
578
579 self.pool.update_free_stats(self.bytes);
580 let threshold = self.pool.threshold_bytes.load(Ordering::Relaxed);
581 if let Err(e) = self.pool.release_cached_until(threshold) {
582 warn!("pool threshold trim failed: {e}");
583 }
584 self.ptr = 0;
585 }
586}
587
588// ---------------------------------------------------------------------------
589// NativeMemoryPool — thin wrapper over the CUDA stream-ordered pool API
590// ---------------------------------------------------------------------------
591
592/// Configuration for a [`NativeMemoryPool`].
593#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
594pub struct NativeMemoryPoolProps {
595 /// Device ordinal that physically backs the pool.
596 pub device_ordinal: i32,
597 /// Maximum aggregate size (bytes) the pool may hold. `0` = unlimited.
598 pub max_size_bytes: usize,
599}
600
601/// Thin wrapper around the CUDA driver's stream-ordered memory pool
602/// (`cuMemPoolCreate` / `cuMemPoolDestroy`).
603///
604/// Allocations are issued via [`NativeMemoryPool::alloc_async`] which
605/// invokes `cuMemAllocFromPoolAsync`; frees are issued via
606/// [`NativeMemoryPool::free_async`] which invokes `cuMemFreeAsync`.
607///
608/// # Stream-ordering
609///
610/// The CUDA stream-ordered pool API requires the caller to ensure all
611/// outstanding work on the stream has completed before destroying the
612/// pool. The [`Drop`] implementation calls `cuMemPoolDestroy` and
613/// silently swallows any error to honour the standard Drop convention.
614/// Call [`NativeMemoryPool::destroy`] explicitly to surface destruction
615/// errors.
616///
617/// # Status
618///
619/// On systems without a CUDA driver (e.g. macOS), [`NativeMemoryPool::new`]
620/// fails with [`CudaError::NotInitialized`]. On older drivers that lack
621/// the pool entry points it fails with [`CudaError::NotSupported`].
622pub struct NativeMemoryPool {
623 raw: CUmemoryPool,
624 device_ordinal: i32,
625}
626
627// SAFETY: `CUmemoryPool` is an opaque driver handle. The CUDA driver is
628// thread-safe; multiple threads may issue stream-ordered allocations from
629// the same pool concurrently.
630unsafe impl Send for NativeMemoryPool {}
631unsafe impl Sync for NativeMemoryPool {}
632
633impl NativeMemoryPool {
634 /// Creates a new native memory pool on the device described by `props`.
635 ///
636 /// # Errors
637 ///
638 /// * [`CudaError::InvalidValue`] if `device_ordinal` is negative.
639 /// * [`CudaError::NotInitialized`] if no CUDA driver is available.
640 /// * [`CudaError::NotSupported`] if the driver does not export
641 /// `cuMemPoolCreate`.
642 /// * Other [`CudaError`] variants on driver failure.
643 pub fn new(props: NativeMemoryPoolProps) -> CudaResult<Self> {
644 if props.device_ordinal < 0 {
645 return Err(CudaError::InvalidDevice);
646 }
647
648 let api = try_driver()?;
649 let f = api.cu_mem_pool_create.ok_or(CudaError::NotSupported)?;
650
651 let pool_props = CUmemPoolProps {
652 alloc_type: CUmemAllocationType::Pinned as u32,
653 handle_types: CUmemAllocationHandleType::None as u32,
654 location: CUmemLocation {
655 loc_type: CUmemLocationType::Device as u32,
656 id: props.device_ordinal,
657 },
658 max_size: props.max_size_bytes,
659 ..CUmemPoolProps::default()
660 };
661
662 let mut raw = CUmemoryPool::default();
663 check(unsafe { f(&mut raw, &pool_props) })?;
664
665 Ok(Self {
666 raw,
667 device_ordinal: props.device_ordinal,
668 })
669 }
670
671 /// Returns the raw [`CUmemoryPool`] handle.
672 #[inline]
673 pub fn raw(&self) -> CUmemoryPool {
674 self.raw
675 }
676
677 /// Returns the device ordinal that backs this pool.
678 #[inline]
679 pub fn device_ordinal(&self) -> i32 {
680 self.device_ordinal
681 }
682
683 /// Asynchronously allocates `bytes` of memory from the pool, ordered
684 /// against `stream`.
685 ///
686 /// # Errors
687 ///
688 /// * [`CudaError::InvalidValue`] if `bytes` is zero.
689 /// * [`CudaError::NotInitialized`] if no CUDA driver is available.
690 /// * [`CudaError::NotSupported`] if the driver does not export
691 /// `cuMemAllocFromPoolAsync`.
692 /// * Other [`CudaError`] variants on driver failure.
693 pub fn alloc_async(&self, bytes: usize, stream: &Stream) -> CudaResult<CUdeviceptr> {
694 if bytes == 0 {
695 return Err(CudaError::InvalidValue);
696 }
697 let api = try_driver()?;
698 let f = api
699 .cu_mem_alloc_from_pool_async
700 .ok_or(CudaError::NotSupported)?;
701 let mut ptr: CUdeviceptr = 0;
702 check(unsafe { f(&mut ptr, bytes, self.raw, stream.raw()) })?;
703 Ok(ptr)
704 }
705
706 /// Asynchronously frees a pointer previously returned by
707 /// [`alloc_async`](Self::alloc_async), ordered against `stream`.
708 ///
709 /// # Errors
710 ///
711 /// * [`CudaError::NotInitialized`] if no CUDA driver is available.
712 /// * [`CudaError::NotSupported`] if the driver does not export
713 /// `cuMemFreeAsync`.
714 /// * Other [`CudaError`] variants on driver failure.
715 pub fn free_async(&self, ptr: CUdeviceptr, stream: &Stream) -> CudaResult<()> {
716 let api = try_driver()?;
717 let f = api.cu_mem_free_async.ok_or(CudaError::NotSupported)?;
718 check(unsafe { f(ptr, stream.raw()) })
719 }
720
721 /// Destroys the pool, returning any driver error to the caller.
722 ///
723 /// The caller is responsible for ensuring all outstanding work on
724 /// streams that allocated from this pool has completed before calling
725 /// `destroy`.
726 ///
727 /// After this call returns, the [`Drop`] implementation will be a
728 /// no-op.
729 ///
730 /// # Errors
731 ///
732 /// * [`CudaError::NotInitialized`] if no CUDA driver is available.
733 /// * [`CudaError::NotSupported`] if the driver does not export
734 /// `cuMemPoolDestroy`.
735 /// * Other [`CudaError`] variants on driver failure.
736 pub fn destroy(mut self) -> CudaResult<()> {
737 self.destroy_inner()
738 }
739
740 fn destroy_inner(&mut self) -> CudaResult<()> {
741 if self.raw.is_null() {
742 return Ok(());
743 }
744 let api = try_driver()?;
745 let f = api.cu_mem_pool_destroy.ok_or(CudaError::NotSupported)?;
746 let result = check(unsafe { f(self.raw) });
747 // Always clear the handle so Drop is a no-op even if destroy fails.
748 self.raw = CUmemoryPool::default();
749 result
750 }
751}
752
753impl Drop for NativeMemoryPool {
754 fn drop(&mut self) {
755 if let Err(e) = self.destroy_inner() {
756 warn!("failed to destroy native memory pool during drop: {e}");
757 }
758 }
759}
760
761// ---------------------------------------------------------------------------
762// Tests
763// ---------------------------------------------------------------------------
764
765#[cfg(test)]
766mod tests {
767 use super::*;
768
769 fn is_driver_unavailable(err: &CudaError) -> bool {
770 matches!(err, CudaError::NotInitialized | CudaError::NotSupported)
771 }
772
773 #[test]
774 fn native_memory_pool_props_default() {
775 let props = NativeMemoryPoolProps::default();
776 assert_eq!(props.device_ordinal, 0);
777 assert_eq!(props.max_size_bytes, 0);
778 }
779
780 #[test]
781 fn native_memory_pool_new_negative_device_fails() {
782 let props = NativeMemoryPoolProps {
783 device_ordinal: -1,
784 max_size_bytes: 0,
785 };
786 let result = NativeMemoryPool::new(props);
787 assert_eq!(result.err(), Some(CudaError::InvalidDevice));
788 }
789
790 /// Without a CUDA driver, `NativeMemoryPool::new` must fail with one of
791 /// the driver-unavailability error kinds rather than panicking.
792 #[test]
793 fn native_memory_pool_new_no_driver_returns_driver_unavailable() {
794 let result = NativeMemoryPool::new(NativeMemoryPoolProps::default());
795 match result {
796 Ok(pool) => {
797 // CUDA available: explicit destroy must succeed too.
798 let destroy = pool.destroy();
799 assert!(destroy.is_ok(), "destroy failed: {destroy:?}");
800 }
801 Err(e) => assert!(
802 is_driver_unavailable(&e),
803 "expected driver-unavailable error, got {e:?}"
804 ),
805 }
806 }
807
808 /// On macOS specifically, every driver-calling method must return
809 /// [`CudaError::NotInitialized`] (no library to load).
810 #[cfg(target_os = "macos")]
811 #[test]
812 fn macos_native_pool_returns_not_initialized() {
813 let result = NativeMemoryPool::new(NativeMemoryPoolProps::default());
814 let err = match result {
815 Err(e) => e,
816 Ok(_) => panic!("expected NotInitialized on macOS, got Ok"),
817 };
818 assert!(
819 matches!(err, CudaError::NotInitialized),
820 "expected NotInitialized, got {err:?}"
821 );
822 }
823
824 // -- MemoryPool / PooledBuffer tests -------------------------------------
825
826 /// The negative-ordinal check must reject before any driver call, so
827 /// this must hold regardless of whether a CUDA driver is present.
828 #[test]
829 fn memory_pool_new_negative_device_fails() {
830 let result = MemoryPool::new(-1);
831 assert_eq!(result.err(), Some(CudaError::InvalidDevice));
832 }
833
834 #[cfg(feature = "gpu-tests")]
835 mod gpu_tests {
836 use super::*;
837 use std::ffi::c_void;
838 use std::sync::Arc;
839
840 /// Bootstraps a real CUDA context on device 0. Returns `None` if no
841 /// driver/GPU is available so callers can skip gracefully.
842 fn real_context() -> Option<Arc<oxicuda_driver::context::Context>> {
843 if oxicuda_driver::init().is_err()
844 || oxicuda_driver::device::Device::count().unwrap_or(0) == 0
845 {
846 return None;
847 }
848 let dev = oxicuda_driver::device::Device::get(0).ok()?;
849 oxicuda_driver::context::Context::new(&dev)
850 .ok()
851 .map(Arc::new)
852 }
853
854 /// `MemoryPool::new` must bind to the requested device (retaining
855 /// its primary context — the F075 fix) and allocations issued
856 /// through the pool must round-trip real data correctly (a
857 /// regression guard that the save/restore-context wrapping in
858 /// `with_device_context` does not corrupt normal HtoD/DtoH traffic).
859 #[test]
860 fn memory_pool_binds_device_and_round_trips_data() {
861 let Some(ctx) = real_context() else {
862 return;
863 };
864 let Ok(pool) = MemoryPool::new(0) else {
865 return;
866 };
867 assert_eq!(pool.device_ordinal(), 0);
868 let Ok(stream) = Stream::new(&ctx) else {
869 return;
870 };
871
872 let host_in: Vec<f32> = (0..256).map(|i| i as f32).collect();
873 let api = try_driver().expect("driver must be present under gpu-tests");
874
875 let buf =
876 PooledBuffer::<f32>::alloc_async(&pool, 256, &stream).expect("alloc_async failed");
877 assert_eq!(buf.len(), 256);
878 assert_eq!(buf.byte_size(), 256 * std::mem::size_of::<f32>());
879
880 let rc = unsafe {
881 (api.cu_memcpy_htod_v2)(
882 buf.as_device_ptr(),
883 host_in.as_ptr().cast::<c_void>(),
884 buf.byte_size(),
885 )
886 };
887 check(rc).expect("HtoD copy failed");
888
889 let mut host_out = vec![0.0f32; 256];
890 let rc = unsafe {
891 (api.cu_memcpy_dtoh_v2)(
892 host_out.as_mut_ptr().cast::<c_void>(),
893 buf.as_device_ptr(),
894 buf.byte_size(),
895 )
896 };
897 check(rc).expect("DtoH copy failed");
898 assert_eq!(host_out, host_in);
899
900 drop(buf);
901 let stats = pool.stats();
902 assert_eq!(stats.allocation_count, 1);
903 assert_eq!(stats.free_count, 1);
904 }
905
906 /// Drops a `PooledBuffer` and immediately requests a new allocation
907 /// of the same size, then writes and reads back fresh data. This
908 /// exercises the event-gated recycle path (F010): whether or not
909 /// the pointer is physically reused, the second buffer's contents
910 /// must reflect exactly what was written to it, never stale data
911 /// from a still-in-flight free.
912 #[test]
913 fn memory_pool_reuse_after_drop_preserves_data_integrity() {
914 let Some(ctx) = real_context() else {
915 return;
916 };
917 let Ok(pool) = MemoryPool::new(0) else {
918 return;
919 };
920 let Ok(stream) = Stream::new(&ctx) else {
921 return;
922 };
923 let api = try_driver().expect("driver must be present under gpu-tests");
924
925 let first_pattern: Vec<u32> = vec![0xAAAA_AAAA; 128];
926 {
927 let buf = PooledBuffer::<u32>::alloc_async(&pool, 128, &stream)
928 .expect("first alloc_async failed");
929 let rc = unsafe {
930 (api.cu_memcpy_htod_v2)(
931 buf.as_device_ptr(),
932 first_pattern.as_ptr().cast::<c_void>(),
933 buf.byte_size(),
934 )
935 };
936 check(rc).expect("first HtoD copy failed");
937 // `buf` drops here, enqueuing a recycle event on `stream`.
938 }
939
940 let second_pattern: Vec<u32> = (0..128u32).collect();
941 let buf2 = PooledBuffer::<u32>::alloc_async(&pool, 128, &stream)
942 .expect("second alloc_async failed");
943 let rc = unsafe {
944 (api.cu_memcpy_htod_v2)(
945 buf2.as_device_ptr(),
946 second_pattern.as_ptr().cast::<c_void>(),
947 buf2.byte_size(),
948 )
949 };
950 check(rc).expect("second HtoD copy failed");
951
952 let mut readback = vec![0u32; 128];
953 let rc = unsafe {
954 (api.cu_memcpy_dtoh_v2)(
955 readback.as_mut_ptr().cast::<c_void>(),
956 buf2.as_device_ptr(),
957 buf2.byte_size(),
958 )
959 };
960 check(rc).expect("DtoH copy failed");
961 assert_eq!(readback, second_pattern);
962
963 let stats = pool.stats();
964 assert_eq!(stats.allocation_count, 2);
965 assert_eq!(stats.free_count, 1);
966 }
967 }
968}