mpmc: add NBLFQ implementation

Adds the NBLFQ lock-free queue implementation.

Author: 0xllx0

Cargo.toml

@@ -59,6 +59,9 @@ mpmc_large = []
 # Enable crossbeam ArrayQueue MPMC implementation.
 mpmc_crossbeam = ["dep:crossbeam-utils"]
 
+# Enable NBLFQ MPMC implementation.
+mpmc_nblfq = []
+
 # Implement some alloc Vec interoperability
 alloc = []
 
@@ -84,6 +87,7 @@ stable_deref_trait = { version = "1", default-features = false }
 [dev-dependencies]
 critical-section = { version = "1.1", features = ["std"] }
 static_assertions = "1.1.0"
+thread-priority = "3.0"
 
 [target.'cfg(loom)'.dependencies]
 loom = "0.7.2"

src/c_string.rs

@@ -382,7 +382,7 @@ mod tests {
         let empty = CString::<1>::new();
 
         assert_eq!(empty.as_c_str(), c"");
-        assert_eq!(empty.as_bytes(), &[]);
+        assert!(empty.as_bytes().is_empty());
         assert_eq!(empty.to_str(), Ok(""));
     }
 

src/history_buf.rs

@@ -670,7 +670,7 @@ mod tests {
         assert!(x.is_full());
 
         let x: HistoryBuf<u8, 4> = HistoryBuf::new();
-        assert_eq!(x.as_slice(), []);
+        assert!(x.as_slice().is_empty());
         assert!(!x.is_full());
     }
 
@@ -694,7 +694,7 @@ mod tests {
     fn clear() {
         let mut x: HistoryBuf<u8, 4> = HistoryBuf::new_with(1);
         x.clear();
-        assert_eq!(x.as_slice(), []);
+        assert!(x.as_slice().is_empty());
 
         let mut x: HistoryBuf<u8, 4> = HistoryBuf::new();
         x.clear_with(1);
@@ -782,7 +782,7 @@ mod tests {
     fn as_slice() {
         let mut x: HistoryBuf<u8, 4> = HistoryBuf::new();
 
-        assert_eq!(x.as_slice(), []);
+        assert!(x.as_slice().is_empty());
 
         x.extend([1, 2, 3, 4, 5].iter());
 

src/mpmc.rs

@@ -99,20 +99,27 @@ type UintSize = usize;
 type UintSize = u8;
 
 #[cfg(feature = "mpmc_large")]
+#[allow(unused)]
 type IntSize = isize;
 #[cfg(not(feature = "mpmc_large"))]
+#[allow(unused)]
 type IntSize = i8;
 
-#[cfg(not(feature = "mpmc_crossbeam"))]
+#[cfg(all(not(feature = "mpmc_nblfq"), not(feature = "mpmc_crossbeam")))]
 mod original;
-#[cfg(not(feature = "mpmc_crossbeam"))]
+#[cfg(all(not(feature = "mpmc_nblfq"), not(feature = "mpmc_crossbeam")))]
 pub use original::*;
 
 #[cfg(feature = "mpmc_crossbeam")]
 mod crossbeam_array_queue;
 #[cfg(feature = "mpmc_crossbeam")]
 pub use crossbeam_array_queue::*;
 
+#[cfg(all(feature = "mpmc_nblfq", not(feature = "mpmc_crossbeam")))]
+mod nblfq;
+#[cfg(all(feature = "mpmc_nblfq", not(feature = "mpmc_crossbeam")))]
+pub use nblfq::*;
+
 /// A [`Queue`] with dynamic capacity.
 ///
 /// [`Queue`] coerces to `QueueView`. `QueueView` is `!Sized`, meaning it can only ever be used by reference.
@@ -125,6 +132,8 @@ mod tests {
 
     use super::Queue;
 
+    const N: usize = 4;
+
     // Ensure a `Queue` containing `!Send` values stays `!Send` itself.
     assert_not_impl_any!(Queue<*const (), 4>: Send);
 
@@ -196,7 +205,153 @@ mod tests {
         // Queue is full, this should not block forever.
         q.enqueue(0x55).unwrap_err();
     }
+
+    #[test]
+    fn test_enqueue_contention_rt() {
+        use thread_priority::*;
+
+        let q0 = std::sync::Arc::new(Queue::<u8, N>::new());
+
+        for i in 0..N {
+            q0.enqueue(i as u8).expect("new enqueue");
+        }
+
+        let model_thread = |q0: std::sync::Arc<Queue<u8, N>>| {
+            for k in 0..N {
+                match q0.dequeue() {
+                    Some(_i) => (),
+                    None if q0.is_empty() => (),
+                    None => panic!(
+                        "enqueue: Dequeue failed on iteration: {k}, empty queue?: {}, queue len: {}",
+                        q0.is_empty(),
+                        q0.len()
+                    ),
+                };
+
+                q0.enqueue(k as u8).unwrap();
+            }
+        };
+
+        let q1 = q0.clone();
+        let h1 = ThreadBuilder::default()
+            .name("h1")
+            .priority(ThreadPriority::Max)
+            .policy(ThreadSchedulePolicy::Realtime(
+                RealtimeThreadSchedulePolicy::Fifo,
+            ))
+            .spawn(move |_| model_thread(q1))
+            .unwrap();
+
+        let q2 = q0.clone();
+        let h2 = ThreadBuilder::default()
+            .name("h2")
+            .priority(ThreadPriority::Max)
+            .policy(ThreadSchedulePolicy::Realtime(
+                RealtimeThreadSchedulePolicy::Fifo,
+            ))
+            .spawn(move |_| model_thread(q2))
+            .unwrap();
+
+        h1.join().unwrap();
+        h2.join().unwrap();
+    }
+
+    #[test]
+    fn test_dequeue_contention_rt() {
+        use thread_priority::*;
+
+        let q0 = std::sync::Arc::new(Queue::<u8, N>::new());
+
+        let model_thread = |q0: std::sync::Arc<Queue<u8, N>>| {
+            for k in 0..N {
+                q0.enqueue(k as u8).unwrap();
+                match q0.dequeue() {
+                    Some(_i) => (),
+                    None if q0.is_empty() => (),
+                    None => {
+                        panic!(
+                            "dequeue: Dequeue failed on iteration: {k}, queue is empty?: {}, queue len: {}",
+                            q0.is_empty(),
+                            q0.len()
+                        );
+                    }
+                }
+            }
+        };
+
+        let q1 = q0.clone();
+        let h1 = ThreadBuilder::default()
+            .name("h1")
+            .priority(ThreadPriority::Max)
+            .policy(ThreadSchedulePolicy::Realtime(
+                RealtimeThreadSchedulePolicy::Fifo,
+            ))
+            .spawn(move |_| model_thread(q1))
+            .unwrap();
+
+        let q2 = q0.clone();
+        let h2 = ThreadBuilder::default()
+            .name("h2")
+            .priority(ThreadPriority::Max)
+            .policy(ThreadSchedulePolicy::Realtime(
+                RealtimeThreadSchedulePolicy::Fifo,
+            ))
+            .spawn(move |_| model_thread(q2))
+            .unwrap();
+
+        h1.join().unwrap();
+        h2.join().unwrap();
+    }
+
+    #[test]
+    fn test_issue_583_enqueue_rt() {
+        use thread_priority::*;
+
+        fn to_vec<T>(q: &Queue<T, N>) -> Vec<T> {
+            // inaccurate
+            let mut ret = vec![];
+            while let Some(v) = q.dequeue() {
+                ret.push(v);
+            }
+            ret
+        }
+
+        let q0 = std::sync::Arc::new(Queue::<u8, N>::new());
+
+        let model_thread = move |q0: std::sync::Arc<Queue<u8, N>>| {
+            for k in 0..1_000_000 {
+                if let Some(v) = q0.dequeue() {
+                    q0.enqueue(v)
+                        .unwrap_or_else(|v| panic!("{}: q0 -> q0: {}, {:?}", k, v, to_vec(&q0)));
+                }
+            }
+        };
+
+        let q1 = q0.clone();
+        let h1 = ThreadBuilder::default()
+            .name("h1")
+            .priority(ThreadPriority::Max)
+            .policy(ThreadSchedulePolicy::Realtime(
+                RealtimeThreadSchedulePolicy::Fifo,
+            ))
+            .spawn(move |_| model_thread(q1))
+            .unwrap();
+
+        let q2 = q0.clone();
+        let h2 = ThreadBuilder::default()
+            .name("h2")
+            .priority(ThreadPriority::Max)
+            .policy(ThreadSchedulePolicy::Realtime(
+                RealtimeThreadSchedulePolicy::Fifo,
+            ))
+            .spawn(move |_| model_thread(q2))
+            .unwrap();
+
+        h1.join().unwrap();
+        h2.join().unwrap();
+    }
 }
+
 #[cfg(all(loom, test))]
 mod tests_loom {
     use super::*;

src/mpmc/crossbeam_array_queue.rs

@@ -636,7 +636,7 @@ impl<T, const N: usize> Iterator for IntoIter<T, N> {
                 // Set to `{ lap: lap.wrapping_add(1), index: 0 }`.
                 lap.wrapping_add(value.one_lap)
             };
-            value.head.store(new, Ordering::AcqRel);
+            value.head.store(new, Ordering::Release);
             Some(val)
         } else {
             None