rrddmma 0.7.3

A Rust library for RDMA.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146

//! Shared Receive Queues.

use std::io::{self, Error as IoError};
use std::ptr::NonNull;
use std::sync::Arc;
use std::{fmt, ptr};

use crate::bindings::*;
use crate::rdma::{context::Context, cq::Cq, mr::*, pd::Pd, qp::build_sgl};
use crate::utils::interop::*;

/// Wrapper for `*mut ibv_srq`.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
#[repr(transparent)]
pub(crate) struct IbvSrq(Option<NonNull<ibv_srq>>);

impl IbvSrq {
    /// Destroy the SRQ.
    ///
    /// # Safety
    ///
    /// - An SRQ must not be destroyed more than once.
    /// - Destroyed SRQs must not be used anymore.
    pub unsafe fn destroy(self) -> io::Result<()> {
        // SAFETY: FFI.
        let ret = ibv_destroy_srq(self.as_ptr());
        from_c_ret(ret)
    }
}

impl_ibv_wrapper_traits!(ibv_srq, IbvSrq);

/// Ownership holder of the SRQ.
struct SrqInner {
    pd: Pd,
    srq: IbvSrq,
    num: u32,
}

impl Drop for SrqInner {
    fn drop(&mut self) {
        // SAFETY: call only once, and no UAF since I will be dropped.
        unsafe { self.srq.destroy() }.expect("cannot destroy SRQ on drop");
    }
}

/// Shared receive queue.
///
/// Currently only supports DCT.
#[derive(Clone)]
pub struct Srq {
    /// Cached SRQ pointer.
    srq: IbvSrq,

    /// SRQ body.
    inner: Arc<SrqInner>,
}

impl fmt::Debug for Srq {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.write_fmt(format_args!("Srq<{:p}>", self.as_raw()))
    }
}

impl Srq {
    /// Create a shared receive queue on the given RDMA protection domain.
    pub fn new(pd: &Pd, cq: Option<&Cq>, max_wr: u32, max_sge: u32) -> io::Result<Self> {
        fn make_srq(pd: &Pd, _cq: Option<&Cq>, max_wr: u32, max_sge: u32) -> io::Result<IbvSrq> {
            let mut init_attr = ibv_srq_init_attr {
                srq_context: ptr::null_mut(),
                attr: ibv_srq_attr {
                    max_wr,
                    max_sge,
                    srq_limit: 0,
                },
            };

            // SAFETY: FFI.
            let srq = unsafe { ibv_create_srq(pd.as_raw(), &mut init_attr) };
            let srq = NonNull::new(srq).ok_or_else(IoError::last_os_error)?;
            Ok(IbvSrq::from(srq))
        }

        let srq = make_srq(pd, cq, max_wr, max_sge)?;

        // Query srq_num.
        let mut num = 0;
        // SAFETY: FFI.
        let ret = unsafe { ibv_get_srq_num(srq.as_ptr(), &mut num) };
        from_c_ret(ret)?;

        let srq = Srq {
            inner: Arc::new(SrqInner {
                pd: pd.clone(),
                srq,
                num,
            }),
            srq,
        };
        Ok(srq)
    }

    /// Get the underlying `ibv_srq` pointer.
    #[inline]
    pub fn as_raw(&self) -> *mut ibv_srq {
        self.srq.as_ptr()
    }

    /// Get the protection domain of the SRQ.
    pub fn pd(&self) -> &Pd {
        &self.inner.pd
    }

    /// Get the RDMA device context of the SRQ.
    pub fn context(&self) -> &Context {
        self.inner.pd.context()
    }

    /// Get the SRQ number.
    pub fn srq_num(&self) -> u32 {
        self.inner.num
    }

    /// Post a receive work request to the SRQ.
    pub fn recv(&self, local: &[MrSlice], wr_id: u64) -> io::Result<()> {
        assert!(local.len() == 1);

        let mut sgl = build_sgl(local);
        let mut wr = ibv_recv_wr {
            wr_id,
            next: ptr::null_mut(),
            sg_list: if local.is_empty() {
                ptr::null_mut()
            } else {
                sgl.as_mut_ptr()
            },
            num_sge: local.len() as i32,
        };
        let ret = {
            let mut bad_wr = ptr::null_mut();
            // SAFETY: FFI.
            unsafe { ibv_post_srq_recv(self.as_raw(), &mut wr, &mut bad_wr) }
        };
        from_c_ret(ret)
    }
}