lf-queue 0.1.0

A lock-free multi-producer multi-consumer unbounded queue.
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

//! Holds an item of the [`Queue`].
//!
//! The [`Slot`]'s state uses below bit flags to report its progress:
//!
//! ```txt
//! INITIAL  0b00000000
//! FILLED   0b00000001 -> Bit flag added after as successful write of the item into the slot.
//! READING  0b00000010 -> Bit flag added from the time a thread starts to read the value.
//! DRAINING 0b00000100 -> Added when a draining of the slot has been scheduled.
//! ```
//!
//! The state of the [`Slot`] can only move forward, resulting in bit flags being added in below
//! order:
//!
//! ```txt
//! INITIAL             0b00000000
//! INITIAL -> FILLED   0b00000001
//! FILLED  -> READING  0b00000011
//! READING -> DRAINING 0b00000111
//! ```
//!
//! [`Node`]: crate::node::Node
//! [`NODE_CAPACITY`]: crate::node::NODE_CAPACITY
//! [`Queue`]: crate::queue::Queue

use crate::variant::cell::UnsafeCell;
use crate::variant::sync::atomic::AtomicUsize;
use crate::variant::thread;

use std::mem::MaybeUninit;
use std::sync::atomic::Ordering;

/// Holds an item of the [`Queue`].
///
/// [`Queue`]: crate::queue::Queue
#[derive(Debug)]
pub(crate) struct Slot<T> {
    /// Holds an item pushed to the [`Queue`].
    ///
    /// [`Queue`]: crate::queue::Queue
    pub(crate) item: UnsafeCell<MaybeUninit<T>>,

    /// Reports the state of the [`Slot`].
    pub(crate) state: AtomicUsize,
}

impl<T> Slot<T> {
    /// When creating a new [`Node`], [`NODE_CAPACITY`] unititialized
    /// [`Slot`] are added to the [`Node`] container. Using a constant
    /// help us reducing the cost of this operation.
    ///
    /// [`Node`]: crate::node::Node
    /// [`NODE_CAPACITY`]: crate::node::NODE_CAPACITY
    #[cfg(not(loom))]
    pub(crate) const UNINIT: Slot<T> = Self {
        item: UnsafeCell::new(MaybeUninit::uninit()),
        state: AtomicUsize::new(0),
    };

    // Loom model checking can't work with constants as it needs to keep
    // track of the initialized items. In loom, `AtomicUsize::new` is therefore
    // not a `const fn`. `Slot::UNINIT` can't be use in loom context as a
    // constant item can't be initialized with calls to non constant expressions.
    #[cfg(loom)]
    pub(crate) fn new() -> Self {
        Self {
            item: UnsafeCell::new(MaybeUninit::uninit()),
            state: AtomicUsize::new(0),
        }
    }

    /// Waits until the state has a [`FILLED`] state.
    pub(crate) fn wait_filled(&self) {
        while self.state.load(Ordering::Acquire) & FILLED == 0 {
            thread::yield_now()
        }
    }
}

// As we can't use compile time constants and function to create a new
// Node container in loom, we implement the Default trait to simplify
// the creation of a new Node container with uninitialized Slots.
#[cfg(loom)]
impl<T> Default for Slot<T> {
    fn default() -> Self {
        Self::new()
    }
}

/// Bit flag added when the [`Slot`] holds an item.
pub(crate) const FILLED: usize = 1;

/// Bit flag added when the [`Slot`] item is being used by a thread.
pub(crate) const READING: usize = 2;

/// Bit flag added when the [`Slot`] is scheduled for deletion.
pub(crate) const DRAINING: usize = 4;