Skip to main content

monarch_rdma/
local_memory.rs

1/*
2 * Copyright (c) Meta Platforms, Inc. and affiliates.
3 * All rights reserved.
4 *
5 * This source code is licensed under the BSD-style license found in the
6 * LICENSE file in the root directory of this source tree.
7 */
8
9//! Local memory abstractions for RDMA operations.
10//!
11//! [`KeepaliveLocalMemory`] wraps a raw pointer with a [`Keepalive`]
12//! guard and dispatches reads/writes to CPU or CUDA paths.
13
14use std::fmt::Debug;
15use std::sync::Arc;
16use std::sync::Condvar;
17use std::sync::Mutex;
18use std::sync::OnceLock;
19
20use crate::backend::ibverbs::memory_region::IbvMemoryRegionView;
21
22/// Returns `true` when `addr` is a CUDA device pointer.
23///
24/// Probes the CUDA driver via `cuPointerGetAttribute`; returns `false`
25/// when CUDA is unavailable or the pointer is not device memory.
26pub fn is_device_ptr(addr: usize) -> bool {
27    // SAFETY: FFI call that queries pointer metadata without accessing
28    // the pointed-to memory.
29    unsafe {
30        let mut mem_type: u32 = 0;
31        let err = rdmaxcel_sys::rdmaxcel_cuPointerGetAttribute(
32            &mut mem_type as *mut _ as *mut std::ffi::c_void,
33            rdmaxcel_sys::CU_POINTER_ATTRIBUTE_MEMORY_TYPE,
34            addr as rdmaxcel_sys::CUdeviceptr,
35        );
36        err == rdmaxcel_sys::CUDA_SUCCESS && mem_type == rdmaxcel_sys::CU_MEMORYTYPE_DEVICE
37    }
38}
39
40/// RAII guard that restores the previous CUDA context on drop and, if a
41/// primary context was retained, releases it.
42pub(crate) struct CudaCtxGuard {
43    prev: rdmaxcel_sys::CUcontext,
44    /// Set when the fallback path called `cuDevicePrimaryCtxRetain`.
45    retained_device: Option<rdmaxcel_sys::CUdevice>,
46}
47
48impl Drop for CudaCtxGuard {
49    fn drop(&mut self) {
50        unsafe {
51            rdmaxcel_sys::rdmaxcel_cuCtxSetCurrent(self.prev);
52            if let Some(device) = self.retained_device {
53                rdmaxcel_sys::rdmaxcel_cuDevicePrimaryCtxRelease(device);
54            }
55        }
56    }
57}
58
59/// Make the CUDA context that owns `addr` current on the calling
60/// thread, returning a guard that restores the previous context on
61/// drop.
62///
63/// First tries `CU_POINTER_ATTRIBUTE_CONTEXT` to get the exact context
64/// the allocation belongs to.  When that returns null (runtime-API or
65/// memory-pool allocations such as PyTorch's caching allocator), falls
66/// back to the device's primary context via
67/// `CU_POINTER_ATTRIBUTE_DEVICE_ORDINAL` + `cuDevicePrimaryCtxRetain`.
68///
69/// # Safety
70///
71/// `addr` must be a valid CUDA device pointer.
72pub(crate) unsafe fn set_ctx_for_ptr(addr: usize) -> Result<CudaCtxGuard, anyhow::Error> {
73    let mut prev: rdmaxcel_sys::CUcontext = std::ptr::null_mut();
74    unsafe {
75        rdmaxcel_sys::rdmaxcel_cuCtxGetCurrent(&mut prev);
76    }
77
78    let mut ctx: rdmaxcel_sys::CUcontext = std::ptr::null_mut();
79    let rc = unsafe {
80        rdmaxcel_sys::rdmaxcel_cuPointerGetAttribute(
81            &mut ctx as *mut _ as *mut std::ffi::c_void,
82            rdmaxcel_sys::CU_POINTER_ATTRIBUTE_CONTEXT,
83            addr as rdmaxcel_sys::CUdeviceptr,
84        )
85    };
86
87    // Null context: allocation came from the runtime API or a memory
88    // pool.  Fall back to the owning device's primary context.
89    let mut retained_device = None;
90    if rc != rdmaxcel_sys::CUDA_SUCCESS || ctx.is_null() {
91        let mut ordinal: i32 = -1;
92        let rc = unsafe {
93            rdmaxcel_sys::rdmaxcel_cuPointerGetAttribute(
94                &mut ordinal as *mut _ as *mut std::ffi::c_void,
95                rdmaxcel_sys::CU_POINTER_ATTRIBUTE_DEVICE_ORDINAL,
96                addr as rdmaxcel_sys::CUdeviceptr,
97            )
98        };
99        anyhow::ensure!(
100            rc == rdmaxcel_sys::CUDA_SUCCESS,
101            "cuPointerGetAttribute(DEVICE_ORDINAL) failed with error code {rc}"
102        );
103
104        let mut device: rdmaxcel_sys::CUdevice = 0;
105        let rc = unsafe { rdmaxcel_sys::rdmaxcel_cuDeviceGet(&mut device, ordinal) };
106        anyhow::ensure!(
107            rc == rdmaxcel_sys::CUDA_SUCCESS,
108            "cuDeviceGet({ordinal}) failed with error code {rc}"
109        );
110
111        let rc = unsafe { rdmaxcel_sys::rdmaxcel_cuDevicePrimaryCtxRetain(&mut ctx, device) };
112        anyhow::ensure!(
113            rc == rdmaxcel_sys::CUDA_SUCCESS,
114            "cuDevicePrimaryCtxRetain failed with error code {rc}"
115        );
116        retained_device = Some(device);
117    }
118
119    let rc = unsafe { rdmaxcel_sys::rdmaxcel_cuCtxSetCurrent(ctx) };
120    anyhow::ensure!(
121        rc == rdmaxcel_sys::CUDA_SUCCESS,
122        "cuCtxSetCurrent failed with error code {rc}"
123    );
124
125    Ok(CudaCtxGuard {
126        prev,
127        retained_device,
128    })
129}
130
131/// Verify that an access at `offset` with `len` bytes fits within `size`.
132fn check_bounds(offset: usize, len: usize, size: usize) -> Result<(), anyhow::Error> {
133    anyhow::ensure!(
134        offset.checked_add(len).is_some_and(|end| end <= size),
135        "access at offset {offset} with length {len} exceeds region size {size}"
136    );
137    Ok(())
138}
139
140/// Copy `dst.len()` bytes from host memory at `addr + offset` into `dst`.
141///
142/// # Safety
143///
144/// The caller must ensure that `addr` points to a valid host allocation of
145/// at least `offset + dst.len()` bytes.
146unsafe fn read_cpu(addr: usize, offset: usize, dst: &mut [u8]) {
147    unsafe {
148        std::ptr::copy_nonoverlapping((addr + offset) as *const u8, dst.as_mut_ptr(), dst.len());
149    }
150}
151
152/// Copy `src.len()` bytes from `src` into host memory at `addr + offset`.
153///
154/// # Safety
155///
156/// The caller must ensure that `addr` points to a valid host allocation of
157/// at least `offset + src.len()` bytes.
158unsafe fn write_cpu(addr: usize, offset: usize, src: &[u8]) {
159    unsafe {
160        std::ptr::copy_nonoverlapping(src.as_ptr(), (addr + offset) as *mut u8, src.len());
161    }
162}
163
164/// Copy `dst.len()` bytes from device memory at `addr + offset` into `dst`.
165///
166/// # Safety
167///
168/// The caller must ensure that `addr` is a valid CUDA device pointer to an
169/// allocation of at least `offset + dst.len()` bytes.
170unsafe fn read_gpu(addr: usize, offset: usize, dst: &mut [u8]) -> Result<(), anyhow::Error> {
171    let _guard = unsafe { set_ctx_for_ptr(addr)? };
172    let rc = unsafe {
173        rdmaxcel_sys::rdmaxcel_cuMemcpyDtoH_v2(
174            dst.as_mut_ptr() as *mut std::ffi::c_void,
175            (addr + offset) as rdmaxcel_sys::CUdeviceptr,
176            dst.len(),
177        )
178    };
179    anyhow::ensure!(
180        rc == rdmaxcel_sys::CUDA_SUCCESS,
181        "cuMemcpyDtoH failed with error code {rc}"
182    );
183    Ok(())
184}
185
186/// Copy `src.len()` bytes from `src` into device memory at `addr + offset`.
187///
188/// # Safety
189///
190/// The caller must ensure that `addr` is a valid CUDA device pointer to an
191/// allocation of at least `offset + src.len()` bytes.
192unsafe fn write_gpu(addr: usize, offset: usize, src: &[u8]) -> Result<(), anyhow::Error> {
193    let _guard = unsafe { set_ctx_for_ptr(addr)? };
194    let rc = unsafe {
195        rdmaxcel_sys::rdmaxcel_cuMemcpyHtoD_v2(
196            (addr + offset) as rdmaxcel_sys::CUdeviceptr,
197            src.as_ptr() as *const std::ffi::c_void,
198            src.len(),
199        )
200    };
201    anyhow::ensure!(
202        rc == rdmaxcel_sys::CUDA_SUCCESS,
203        "cuMemcpyHtoD failed with error code {rc}"
204    );
205    Ok(())
206}
207
208/// Three-mode access lock used by [`KeepaliveLocalMemory`] to coordinate
209/// concurrent reads, exclusive writes, and parallel "disjoint" writes
210/// (writers that the caller has promised target disjoint ranges).
211///
212/// - [`AccessLock::read`] returns when no exclusive writer and no
213///   disjoint writer is active. Multiple readers are permitted to hold
214///   the lock at the same time.
215/// - [`AccessLock::disjoint_write`] returns when no reader and no
216///   exclusive writer is active. Multiple disjoint writers are
217///   permitted to hold the lock at the same time.
218/// - [`AccessLock::exclusive`] returns only when no one else holds the
219///   lock.
220///
221/// Read mode and disjoint-write mode are mutually exclusive, which is
222/// what gives readers a torn-free view of memory in the presence of
223/// disjoint parallel writers.
224#[derive(Debug, Default)]
225struct AccessLock {
226    state: Mutex<AccessState>,
227    cond: Condvar,
228}
229
230#[derive(Debug, Default)]
231enum AccessState {
232    #[default]
233    Idle,
234    Reading(usize),
235    DisjointWriting(usize),
236    Exclusive,
237}
238
239impl AccessLock {
240    fn new() -> Self {
241        Self::default()
242    }
243
244    fn read(&self) -> AccessReadGuard<'_> {
245        let mut state = self.state.lock().expect("AccessLock poisoned");
246        loop {
247            match &mut *state {
248                AccessState::Idle => {
249                    *state = AccessState::Reading(1);
250                    return AccessReadGuard(self);
251                }
252                AccessState::Reading(n) => {
253                    *n += 1;
254                    return AccessReadGuard(self);
255                }
256                AccessState::DisjointWriting(_) | AccessState::Exclusive => {
257                    state = self.cond.wait(state).expect("AccessLock poisoned");
258                }
259            }
260        }
261    }
262
263    fn disjoint_write(&self) -> AccessDisjointWriteGuard<'_> {
264        let mut state = self.state.lock().expect("AccessLock poisoned");
265        loop {
266            match &mut *state {
267                AccessState::Idle => {
268                    *state = AccessState::DisjointWriting(1);
269                    return AccessDisjointWriteGuard(self);
270                }
271                AccessState::DisjointWriting(n) => {
272                    *n += 1;
273                    return AccessDisjointWriteGuard(self);
274                }
275                AccessState::Reading(_) | AccessState::Exclusive => {
276                    state = self.cond.wait(state).expect("AccessLock poisoned");
277                }
278            }
279        }
280    }
281
282    fn exclusive(&self) -> AccessExclusiveGuard<'_> {
283        let mut state = self.state.lock().expect("AccessLock poisoned");
284        loop {
285            if matches!(*state, AccessState::Idle) {
286                *state = AccessState::Exclusive;
287                return AccessExclusiveGuard(self);
288            }
289            state = self.cond.wait(state).expect("AccessLock poisoned");
290        }
291    }
292}
293
294struct AccessReadGuard<'a>(&'a AccessLock);
295impl Drop for AccessReadGuard<'_> {
296    fn drop(&mut self) {
297        let mut state = self.0.state.lock().expect("AccessLock poisoned");
298        match &mut *state {
299            AccessState::Reading(1) => {
300                *state = AccessState::Idle;
301                self.0.cond.notify_all();
302            }
303            AccessState::Reading(n) => *n -= 1,
304            other => unreachable!("AccessReadGuard dropped in non-Reading state: {other:?}"),
305        }
306    }
307}
308
309struct AccessDisjointWriteGuard<'a>(&'a AccessLock);
310impl Drop for AccessDisjointWriteGuard<'_> {
311    fn drop(&mut self) {
312        let mut state = self.0.state.lock().expect("AccessLock poisoned");
313        match &mut *state {
314            AccessState::DisjointWriting(1) => {
315                *state = AccessState::Idle;
316                self.0.cond.notify_all();
317            }
318            AccessState::DisjointWriting(n) => *n -= 1,
319            other => unreachable!(
320                "AccessDisjointWriteGuard dropped in non-DisjointWriting state: {other:?}"
321            ),
322        }
323    }
324}
325
326struct AccessExclusiveGuard<'a>(&'a AccessLock);
327impl Drop for AccessExclusiveGuard<'_> {
328    fn drop(&mut self) {
329        let mut state = self.0.state.lock().expect("AccessLock poisoned");
330        debug_assert!(matches!(*state, AccessState::Exclusive));
331        *state = AccessState::Idle;
332        self.0.cond.notify_all();
333    }
334}
335
336/// Trait for values that keep a backing memory allocation alive and
337/// know its address and size.
338///
339/// As long as a value implementing this trait exists, the memory region
340/// it describes is guaranteed to remain valid.
341pub trait Keepalive: Send + Sync {
342    /// Start address of the memory region this keepalive pins.
343    fn addr(&self) -> usize;
344
345    /// Size in bytes of the memory region this keepalive pins.
346    fn size(&self) -> usize;
347
348    /// Produce a [`WeakKeepalive`] pointing at the same underlying
349    /// resource. Defaults to `None` for impls with no weak form.
350    fn downgrade(&self) -> Option<Arc<dyn WeakKeepalive>> {
351        None
352    }
353}
354
355/// Counterpart to [`Keepalive`]: a non-pinning reference to the same
356/// underlying resource that can be re-promoted to a [`Keepalive`] as
357/// long as the resource is still alive.
358pub trait WeakKeepalive: Send + Sync {
359    /// Re-acquire a strong [`Keepalive`] for the underlying resource,
360    /// or `None` if the referent has gone away.
361    fn upgrade(&self) -> Option<Arc<dyn Keepalive>>;
362}
363
364impl Keepalive for Box<[u8]> {
365    fn addr(&self) -> usize {
366        self.as_ptr() as usize
367    }
368
369    fn size(&self) -> usize {
370        self.len()
371    }
372}
373
374/// Backing state of a [`KeepaliveLocalMemory`].
375///
376/// Holds the addressing/bandwidth metadata, the access-coordination
377/// lock, and a single-slot home for an [`IbvMemoryRegionView`]
378/// registered against this region. Cloning shares the slot and the
379/// access lock by `Arc`, so every handle derived from the same
380/// allocation observes the same registered MR and the same
381/// reader/writer coordination.
382///
383/// All access goes through methods on [`KeepaliveLocalMemory`];
384/// nothing outside the module pokes at these fields directly.
385#[derive(Clone)]
386pub(crate) struct LocalMemoryInner {
387    addr: usize,
388    size: usize,
389    /// Bandwidth (bytes/s) for direct host-thread pointer access, or `None`
390    /// if the memory is not host-accessible.
391    direct_access_host_bandwidth: Option<u64>,
392    /// Bandwidth (bytes/s) for direct device-thread pointer access, or
393    /// `None` if the memory is not device-accessible.
394    direct_access_device_bandwidth: Option<u64>,
395    /// Per-allocation slot for the [`IbvMemoryRegionView`] registered
396    /// against this region. Populated lazily by
397    /// `IbvManagerActor::resolve_local_mr` on first use.
398    mr_slot: Arc<OnceLock<IbvMemoryRegionView>>,
399    /// Coordinates concurrent reads, exclusive writes, and parallel
400    /// disjoint writes against this region.
401    access: Arc<AccessLock>,
402}
403
404impl LocalMemoryInner {
405    fn new(addr: usize, size: usize) -> Self {
406        // TODO(slurye): Using placeholder values for now. Fill in with real values.
407        let (host_bw, device_bw) = if is_device_ptr(addr) {
408            (None, Some(1))
409        } else {
410            (Some(1), None)
411        };
412        Self {
413            addr,
414            size,
415            direct_access_host_bandwidth: host_bw,
416            direct_access_device_bandwidth: device_bw,
417            mr_slot: Arc::new(OnceLock::new()),
418            access: Arc::new(AccessLock::new()),
419        }
420    }
421}
422
423/// Local memory handle that keeps its backing allocation alive via an
424/// [`Arc<dyn Keepalive>`].
425///
426/// Detects at construction time whether the address is a CUDA device
427/// pointer and dispatches `read_at`/`write_at` accordingly.
428///
429/// All three access methods are `unsafe`: the [`Keepalive`] only
430/// guarantees the allocation stays mapped, not that this handle has
431/// unique ownership. The internal [`AccessLock`] coordinates concurrent
432/// callers that share the same clone of this handle (readers run in
433/// parallel, exclusive writers run alone, disjoint writers run in
434/// parallel with one another but exclude readers and exclusive
435/// writers), but callers must additionally rule out concurrent access
436/// through other views of the same allocation.
437///
438/// The `direct_access_host_bandwidth` and `direct_access_device_bandwidth`
439/// fields indicate the speed of reading the memory via pointer dereference
440/// on a host or device thread, respectively. A value of `None` means the
441/// memory is not directly accessible from that context.
442#[derive(Clone)]
443pub struct KeepaliveLocalMemory {
444    inner: LocalMemoryInner,
445    _keepalive: Arc<dyn Keepalive>,
446}
447
448impl Debug for KeepaliveLocalMemory {
449    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
450        f.debug_struct("KeepaliveLocalMemory")
451            .field("addr", &self.inner.addr)
452            .field("size", &self.inner.size)
453            .field(
454                "direct_access_host_bandwidth",
455                &self.inner.direct_access_host_bandwidth,
456            )
457            .field(
458                "direct_access_device_bandwidth",
459                &self.inner.direct_access_device_bandwidth,
460            )
461            .finish_non_exhaustive()
462    }
463}
464
465impl KeepaliveLocalMemory {
466    /// Create a new handle. Derives `addr` and `size` from the
467    /// `keepalive` via [`Keepalive::addr`] /
468    /// [`Keepalive::size`], then probes the CUDA driver to
469    /// determine whether the address is a device pointer and sets the
470    /// bandwidth fields accordingly.
471    pub fn new(keepalive: Arc<dyn Keepalive>) -> Self {
472        let addr = keepalive.addr();
473        let size = keepalive.size();
474        Self {
475            inner: LocalMemoryInner::new(addr, size),
476            _keepalive: keepalive,
477        }
478    }
479
480    /// Starting virtual address of the memory region.
481    pub fn addr(&self) -> usize {
482        self.inner.addr
483    }
484
485    /// Size of the memory region in bytes.
486    pub fn size(&self) -> usize {
487        self.inner.size
488    }
489
490    /// Shared slot for the [`IbvMemoryRegionView`] registered against
491    /// this region. Populated lazily by
492    /// [`IbvManagerActor::resolve_local_mr`] on first use; the slot
493    /// is cloned `Arc` so every handle derived from the same
494    /// allocation sees the same registered MR.
495    pub fn mr_slot(&self) -> &Arc<OnceLock<IbvMemoryRegionView>> {
496        &self.inner.mr_slot
497    }
498
499    /// Copy `dst.len()` bytes from this memory region starting at `offset`
500    /// into `dst`.
501    ///
502    /// Mutually exclusive with both `write_at` and `write_at_disjoint`
503    /// *across clones of this handle*: the [`AccessLock`] guarantees a
504    /// reader and any writer (exclusive or disjoint) that share the
505    /// same lock never observe each other's partial state. Multiple
506    /// concurrent `read_at` calls on shared clones are permitted and
507    /// run in parallel.
508    ///
509    /// # Safety
510    ///
511    /// The [`Keepalive`] guarantees the allocation stays mapped, but it
512    /// does *not* imply unique ownership: another component may hold its
513    /// own view of the same allocation and read or write it concurrently
514    /// outside this handle's [`AccessLock`]. The caller must ensure that
515    /// no such external access produces a torn read of
516    /// `offset..offset + dst.len()` for the duration of this call.
517    pub unsafe fn read_at(&self, offset: usize, dst: &mut [u8]) -> Result<(), anyhow::Error> {
518        let _guard = self.inner.access.read();
519        check_bounds(offset, dst.len(), self.inner.size)?;
520        // SAFETY: the `_keepalive` field keeps the allocation live, the
521        // read guard above excludes concurrent exclusive and disjoint
522        // writers that share this lock, `check_bounds` verified the access
523        // is in range, and the caller upholds the no-external-writer
524        // obligation documented on this method.
525        unsafe {
526            if self.inner.direct_access_host_bandwidth.is_some() {
527                read_cpu(self.inner.addr, offset, dst);
528                Ok(())
529            } else {
530                read_gpu(self.inner.addr, offset, dst)
531            }
532        }
533    }
534
535    /// Copy `src.len()` bytes from `src` into this memory region starting
536    /// at `offset`.
537    ///
538    /// Mutually exclusive with every other read and write against this
539    /// region *across clones of this handle*: the [`AccessLock`] blocks
540    /// concurrent readers and writers that share the same lock. Use
541    /// [`KeepaliveLocalMemory::write_at_disjoint`] when multiple writers
542    /// can be proven to target disjoint byte ranges.
543    ///
544    /// # Safety
545    ///
546    /// See [`KeepaliveLocalMemory::read_at`]. The [`Keepalive`] guarantee
547    /// covers liveness only; the caller must ensure no concurrent
548    /// external reader or writer observes an overlapping byte range.
549    pub unsafe fn write_at(&self, offset: usize, src: &[u8]) -> Result<(), anyhow::Error> {
550        let _guard = self.inner.access.exclusive();
551        check_bounds(offset, src.len(), self.inner.size)?;
552        // SAFETY: the `_keepalive` field keeps the allocation live, the
553        // exclusive guard above excludes every other reader and writer
554        // that shares this lock, `check_bounds` verified the access is
555        // in range, and the caller upholds the no-external-access
556        // obligation documented on this method.
557        unsafe {
558            if self.inner.direct_access_host_bandwidth.is_some() {
559                write_cpu(self.inner.addr, offset, src);
560                Ok(())
561            } else {
562                write_gpu(self.inner.addr, offset, src)
563            }
564        }
565    }
566
567    /// Like [`KeepaliveLocalMemory::write_at`], but allows other
568    /// concurrent `write_at_disjoint` calls (across clones of this
569    /// handle) to proceed in parallel. Still mutually exclusive with
570    /// `read_at` and `write_at` through the [`AccessLock`].
571    ///
572    /// # Safety
573    ///
574    /// In addition to the obligations of
575    /// [`KeepaliveLocalMemory::write_at`] (no external concurrent
576    /// reader or writer of the same byte range), the caller must
577    /// ensure that no other concurrent call to this method targets a
578    /// byte range that overlaps `offset..offset + src.len()`. Disjoint
579    /// byte ranges across concurrent disjoint callers are sound.
580    pub unsafe fn write_at_disjoint(&self, offset: usize, src: &[u8]) -> Result<(), anyhow::Error> {
581        let _guard = self.inner.access.disjoint_write();
582        check_bounds(offset, src.len(), self.inner.size)?;
583        // SAFETY: the `_keepalive` field keeps the allocation live, the
584        // disjoint-write guard above excludes concurrent readers and
585        // exclusive writers that share this lock, `check_bounds`
586        // verified the access is in range, and the caller upholds both
587        // safety obligations documented on this method (no external access,
588        // no overlap with other concurrent disjoint writers).
589        unsafe {
590            if self.inner.direct_access_host_bandwidth.is_some() {
591                write_cpu(self.inner.addr, offset, src);
592                Ok(())
593            } else {
594                write_gpu(self.inner.addr, offset, src)
595            }
596        }
597    }
598
599    /// Pair off a [`WeakLocalMemory`] that shares this handle's
600    /// [`LocalMemoryInner`] (and therefore the same MR slot and
601    /// access lock). Returns `None` when the underlying [`Keepalive`]
602    /// does not provide a weak form.
603    pub fn downgrade(&self) -> Option<WeakLocalMemory> {
604        let weak_keepalive = self._keepalive.downgrade()?;
605        Some(WeakLocalMemory {
606            inner: self.inner.clone(),
607            weak_keepalive,
608        })
609    }
610}
611
612/// Non-pinning counterpart of [`KeepaliveLocalMemory`].
613///
614/// Holds the shared [`LocalMemoryInner`] (so a re-promoted strong
615/// handle sees the same MR slot and access lock) plus a
616/// [`WeakKeepalive`] that can be upgraded to a fresh
617/// [`Arc<dyn Keepalive>`] as long as the referent is still alive.
618#[derive(Clone)]
619pub struct WeakLocalMemory {
620    inner: LocalMemoryInner,
621    weak_keepalive: Arc<dyn WeakKeepalive>,
622}
623
624impl WeakLocalMemory {
625    /// Starting virtual address of the memory region.
626    pub fn addr(&self) -> usize {
627        self.inner.addr
628    }
629
630    /// Size of the memory region in bytes.
631    pub fn size(&self) -> usize {
632        self.inner.size
633    }
634
635    /// Materialize a strong [`KeepaliveLocalMemory`] sharing this
636    /// handle's [`LocalMemoryInner`]. Returns `None` if the
637    /// referent has gone away **or** if its currently-computed
638    /// `(addr, size)` no longer matches the values stored on this
639    /// handle — the latter guarding against the live referent
640    /// describing a different memory region than the one this weak
641    /// handle was paired with at downgrade time.
642    pub fn upgrade(&self) -> Option<KeepaliveLocalMemory> {
643        let keepalive = self.weak_keepalive.upgrade()?;
644        let new_addr = keepalive.addr();
645        let new_size = keepalive.size();
646        if new_addr != self.inner.addr || new_size != self.inner.size {
647            tracing::warn!(
648                expected_addr = self.inner.addr,
649                actual_addr = new_addr,
650                expected_size = self.inner.size,
651                actual_size = new_size,
652                "WeakLocalMemory upgrade rejected: backing keepalive's (addr, size) changed since downgrade",
653            );
654            return None;
655        }
656        Some(KeepaliveLocalMemory {
657            inner: self.inner.clone(),
658            _keepalive: keepalive,
659        })
660    }
661}
662
663#[cfg(test)]
664mod tests {
665    use super::*;
666
667    // -- KeepaliveLocalMemory (host) --
668
669    fn host_keepalive_mem(data: Box<[u8]>) -> KeepaliveLocalMemory {
670        KeepaliveLocalMemory::new(Arc::new(data))
671    }
672
673    #[test]
674    fn keepalive_host_read_at() {
675        let mem = host_keepalive_mem(Box::from([1, 2, 3, 4, 5]));
676        let mut buf = [0u8; 3];
677        // SAFETY: `mem` is the sole handle to the allocation, no other
678        // thread or component holds a view of it.
679        unsafe { mem.read_at(1, &mut buf) }.unwrap();
680        assert_eq!(buf, [2, 3, 4]);
681    }
682
683    #[test]
684    fn keepalive_host_write_then_read() {
685        let mem = host_keepalive_mem(vec![0; 5].into_boxed_slice());
686        // SAFETY: `mem` is the sole handle to the allocation, no other
687        // thread or component holds a view of it.
688        unsafe { mem.write_at(1, &[7, 8, 9]) }.unwrap();
689        let mut buf = [0u8; 5];
690        // SAFETY: same as above.
691        unsafe { mem.read_at(0, &mut buf) }.unwrap();
692        assert_eq!(buf, [0, 7, 8, 9, 0]);
693    }
694
695    #[test]
696    fn keepalive_host_out_of_bounds() {
697        let mem = host_keepalive_mem(vec![0; 3].into_boxed_slice());
698        let mut buf = [0u8; 3];
699        // SAFETY: `mem` is the sole handle to the allocation; the
700        // bounds check fires before any pointer dereference.
701        assert!(unsafe { mem.read_at(1, &mut buf) }.is_err());
702        // SAFETY: same as above.
703        assert!(unsafe { mem.write_at(1, &[7, 8, 9]) }.is_err());
704    }
705}