urcu2 0.1.3

Safe API to liburcu
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

use std::marker::PhantomData;
use std::mem::MaybeUninit;
use std::ops::Deref;

use container_of::container_of;
use urcu_cds_sys::lfq;

use crate::rcu::flavor::RcuFlavor;
use crate::utility::*;

pub struct RawNode<T> {
    handle: lfq::NodeRcu,
    data: T,
}
impl<T> RawNode<T> {
    pub fn new(data: T) -> Box<Self> {
        let mut handle = MaybeUninit::<lfq::NodeRcu>::uninit();

        // SAFETY: We don't need to registered with RCU in any way.
        unsafe { lfq::node_init_rcu(handle.as_mut_ptr()) };

        Box::new(Self {
            // SAFETY: Data has been initialised by `lfq::node_init`.
            handle: unsafe { handle.assume_init() },
            data,
        })
    }

    fn into_handle(self: Box<Self>) -> *mut lfq::NodeRcu {
        let node_ptr = Box::into_raw(self);
        let node = unsafe { node_ptr.as_mut_unchecked() };
        &mut node.handle
    }
}

impl<T> Deref for RawNode<T> {
    type Target = T;

    fn deref(&self) -> &Self::Target {
        &self.data
    }
}

/// #### Safety
///
/// It is safe to send a [`RawNode<T>`] to another thread if `T` is [`Send`].
unsafe impl<T: Send> Send for RawNode<T> {}

/// #### Safety
///
/// It is safe to share a [`RawNode<T>`] between threads if `T` is [`Sync`].
unsafe impl<T: Sync> Sync for RawNode<T> {}

pub struct RawQueue<T, F> {
    handle: lfq::QueueRcu,
    _unsend: PhantomUnsend<(T, F)>,
    _unsync: PhantomUnsync<(T, F)>,
}

impl<T, F> RawQueue<T, F> {
    /// #### Safety
    ///
    /// The caller must call [`RawQueue::init`] once [`RawQueue`] is in a stable memory location.
    pub unsafe fn new() -> Self {
        Self {
            handle: lfq::QueueRcu {
                head: std::ptr::null_mut(),
                tail: std::ptr::null_mut(),
                queue_call_rcu: None,
            },
            _unsend: PhantomData,
            _unsync: PhantomData,
        }
    }

    /// #### Safety
    ///
    /// The caller must ensure [`RawQueue`] is in a stable memory location.
    /// The caller must remove all nodes before dropping this type.
    pub unsafe fn init(&mut self)
    where
        F: RcuFlavor,
    {
        // SAFETY: We don't need to registered with RCU in any way.
        // SAFETY: The unchecked API is used by the C code.
        unsafe { lfq::init_rcu(&mut self.handle, F::unchecked_rcu_api().update_call_rcu) };
    }

    /// #### Safety
    ///
    /// The caller must be inside a RCU critical section.
    pub unsafe fn enqueue(&self, node: Box<RawNode<T>>) {
        let handle = &self.handle as *const lfq::QueueRcu as *mut lfq::QueueRcu;

        // SAFETY: The C call safely mutate the state shared between threads.
        unsafe { lfq::enqueue_rcu(handle, node.into_handle()) }
    }

    // #### Safety
    //
    // The caller must be inside a RCU critical section.
    //
    // The caller must wait a RCU grace period before freeing the node.
    pub unsafe fn dequeue(&self) -> *mut RawNode<T> {
        let handle = &self.handle as *const lfq::QueueRcu as *mut lfq::QueueRcu;

        // SAFETY: The C call safely mutate the state shared between threads.
        let handle = unsafe { lfq::dequeue_rcu(handle) };
        if handle.is_null() {
            std::ptr::null_mut()
        } else {
            container_of!(handle, RawNode<T>, handle)
        }
    }

    // #### Safety
    //
    // The caller must be inside a RCU critical section.
    //
    // The caller must wait a RCU grace period before freeing the nodes.
    pub unsafe fn dequeue_all(&self) -> Vec<*mut RawNode<T>> {
        let mut ptrs = Vec::new();

        loop {
            let ptr = self.dequeue();
            if ptr.is_null() {
                break;
            }

            ptrs.push(ptr);
        }

        ptrs
    }
}

impl<T, F> Drop for RawQueue<T, F> {
    fn drop(&mut self) {
        // SAFETY: The queue creator must empty the queue before dropping.
        let ret = unsafe { lfq::destroy_rcu(&mut self.handle) };

        if ret != 0 {
            log::error!("raw queue was not emptied before dropping");
        }
    }
}