rdma 0.3.0

Low-level RDMA API
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

use crate::bindings as C;
use crate::cq::{self, CompletionQueue};
use crate::ctx::Context;
use crate::error::{create_resource, custom_error};
use crate::weakset::WeakSet;

use std::os::raw::c_void;
use std::os::unix::prelude::{AsRawFd, RawFd};
use std::ptr::NonNull;
use std::sync::{Arc, Weak};
use std::{io, ptr};

use parking_lot::Mutex;

#[derive(Clone)]
pub struct CompChannel(Arc<Owner>);

impl CompChannel {
    pub(crate) fn ffi_ptr(&self) -> *mut C::ibv_comp_channel {
        self.0.ffi_ptr()
    }

    #[inline]
    pub fn create(ctx: &Context) -> io::Result<Self> {
        // SAFETY: ffi
        let owner = unsafe {
            let cc = create_resource(
                || C::ibv_create_comp_channel(ctx.ffi_ptr()),
                || "failed to create completion channel",
            )?;

            Arc::new(Owner {
                cc,
                cq_ref: Mutex::new(WeakSet::new()),
                _ctx: ctx.clone(),
            })
        };
        Ok(Self(owner))
    }

    #[inline]
    pub fn wait_cq_event(&self) -> io::Result<CompletionQueue> {
        let cc = self.ffi_ptr();
        let mut cq: *mut C::ibv_cq = ptr::null_mut();
        let mut cq_context: *mut c_void = ptr::null_mut();
        // SAFETY: ffi
        unsafe {
            let ret = C::ibv_get_cq_event(cc, &mut cq, &mut cq_context);
            if ret != 0 {
                return Err(custom_error("failed to get completion event"));
            }
            debug_assert_eq!((*cq).cq_context, cq_context);
        }
        // SAFETY:
        // 1. the cq is associated with the cc
        // 2. the cc is holding a weak reference to the cq
        // 3. here may panic because the cq may have been destroyed
        unsafe { Ok(CompletionQueue::from_cq_context(cq_context)) }
    }

    pub(crate) fn add_cq_ref(&self, cq: Weak<cq::Owner>) {
        self.0.cq_ref.lock().insert(cq);
    }

    pub(crate) fn del_cq_ref(&self, cq: &cq::Owner) -> bool {
        self.0.cq_ref.lock().remove(cq)
    }
}

impl AsRawFd for CompChannel {
    #[inline]
    fn as_raw_fd(&self) -> RawFd {
        let cc = self.ffi_ptr();
        // SAFETY: reading a immutable field of a concurrent ffi type
        unsafe { (*cc).fd }
    }
}

struct Owner {
    cc: NonNull<C::ibv_comp_channel>,

    cq_ref: Mutex<WeakSet<cq::Owner>>,
    _ctx: Context,
}

/// SAFETY: owned type
unsafe impl Send for Owner {}
/// SAFETY: owned type
unsafe impl Sync for Owner {}

impl Owner {
    pub(crate) fn ffi_ptr(&self) -> *mut C::ibv_comp_channel {
        self.cc.as_ptr()
    }
}

impl Drop for Owner {
    fn drop(&mut self) {
        // SAFETY: ffi
        unsafe {
            let cc = self.ffi_ptr();
            let ret = C::ibv_destroy_comp_channel(cc);
            assert_eq!(ret, 0);
        }
    }
}