MpMcQueue can fail to push even when it's not full under contention

[nodakai]

https://play.rust-lang.org/?version=stable&mode=debug&edition=2024&gist=0a9dde3dac37be94314c5719726060c7

#[test]
fn test() {
    fn to_vec<T>(q: &QueueView<T>) -> Vec<T> {
        // inaccurate
        let mut ret = vec![];
        while let Some(v) = q.dequeue() {
            ret.push(v);
        }
        ret
    }
    const N: usize = 4;

    let q0 = Queue::<u8, N>::new();
    for i in 0..N {
        q0.enqueue(i as u8).expect("new enqueue");
    }
    eprintln!("start!");

    std::thread::scope(|sc| {
        for _ in 0..2 {
            sc.spawn(|| {
                for k in 0..1000_000 {
                    if let Some(v) = q0.dequeue() {
                        q0.enqueue(v).unwrap_or_else(|v| {
                            panic!("{}: q0 -> q0: {}, {:?}", k, v, to_vec(&q0))
                        });
                    }
                }
            });
        }
    });
}

Confirmed an implementation based on a mutex, and another based on ring-channel didn't panic

[sosthene-nitrokey]

I think I see the case that can fail. Let's name the 2 threads: A and B, B is the one that will panic:

1. The queue is full, enqueue_pos = i + N and dequeue_pos = i
2. A calls dequeue
3. A gets to the compare_exchange of the dequeue operation (reserving its own dequeuing to item i)
4. B calls dequeue dequeuing completely item i + 1.
5. B calls enqueue, tries to enqueue on the slot of i + N, which still contains item i which has not been dequeued yet (thread A has been parked by the scheduler and is still in the dequeue operation)

This is consistent with what I observe in the tests regarding the position of the dequeue operation that precedes the failure to enqueue.

The good news is that no invariant ever gets invalidated. The bad news is that I have no idea how to fix this without completely changing the implementation.
It might make sense to use an implementation that is better documented, like crossbeam's ArrayQueue. Though it might not make the same performance tradeoffs (it certainly seems to be using a lot more code so will lead to a larger binary size).

This queue is designed for the use cases where items can be dropped when the consumer side is being too slow

[sosthene-nitrokey]

Actually the same can happen when dequeueing:

#[test]
fn issue_583_dequeue() {
    const N: usize = 4;

    let q0 = Queue::<u8, N>::new();
    eprintln!("start!");
    std::thread::scope(|sc| {
        for _ in 0..2 {
            sc.spawn(|| {
                for k in 0..1000_000 {
                    q0.enqueue(k as u8).unwrap();
                    if q0.dequeue().is_none() {
                        panic!("{k}");
                    }
                }
            });
        }
    });
}

[sosthene-nitrokey]

I found sort of a fix.

It works to fix the issue in the given example but in practice I don't think it's suitable. Even with std threads it's likely to require a lot of iterations (on my machine 10 000 loops) are required so that the tests stops failing spuriously.
Also, heapless will often be used in the context of a real-time scheduler, so no matter how long the failing operation waits it the lower-priority operation that has been preempted will not be scheduled again.

Finally, for some reasons I don't fully understand this fails when the positions wrap around, so this makes it almost useless when mpmc_large is not enabled.

Patch that fixes the issue in some cases

diff --git a/src/mpmc.rs b/src/mpmc.rs
index f676c7c9..44dc2713 100644
--- a/src/mpmc.rs
+++ b/src/mpmc.rs
@@ -208,7 +208,14 @@ impl<T, S: Storage> QueueInner<T, S> {
 
     /// Returns the item in the front of the queue, or `None` if the queue is empty.
     pub fn dequeue(&self) -> Option<T> {
-        unsafe { dequeue(S::as_ptr(self.buffer.get()), &self.dequeue_pos, self.mask()) }
+        unsafe {
+            dequeue(
+                S::as_ptr(self.buffer.get()),
+                &self.dequeue_pos,
+                &self.enqueue_pos,
+                self.mask(),
+            )
+        }
     }
 
     /// Adds an `item` to the end of the queue.
@@ -218,6 +225,7 @@ impl<T, S: Storage> QueueInner<T, S> {
         unsafe {
             enqueue(
                 S::as_ptr(self.buffer.get()),
+                &self.dequeue_pos,
                 &self.enqueue_pos,
                 self.mask(),
                 item,
@@ -255,18 +263,25 @@ impl<T> Cell<T> {
     }
 }
 
+const CONTENTION_RETRY_COUNT: usize = 10000;
+
 unsafe fn dequeue<T>(
     buffer: *mut Cell<T>,
     dequeue_pos: &AtomicTargetSize,
+    enqueue_pos: &AtomicTargetSize,
     mask: UintSize,
 ) -> Option<T> {
     let mut pos = dequeue_pos.load(Ordering::Relaxed);
 
     let mut cell;
+    let mut seq;
+    let mut dif;
+    let mut contention_retry_count = 0;
+
     loop {
         cell = buffer.add(usize::from(pos & mask));
-        let seq = (*cell).sequence.load(Ordering::Acquire);
-        let dif = (seq as IntSize).wrapping_sub((pos.wrapping_add(1)) as IntSize);
+        seq = (*cell).sequence.load(Ordering::Acquire);
+        dif = (seq as IntSize).wrapping_sub((pos.wrapping_add(1)) as IntSize);
 
         match dif.cmp(&0) {
             core::cmp::Ordering::Equal => {
@@ -283,6 +298,16 @@ unsafe fn dequeue<T>(
                 }
             }
             core::cmp::Ordering::Less => {
+                if pos != enqueue_pos.load(Ordering::Relaxed)
+                    && contention_retry_count < CONTENTION_RETRY_COUNT
+                {
+                    // In this case according to the positions the queue is not empty
+                    // This suggests that there is a enqueue operations that is in progress in some other task
+                    // Therefore we can wait a bit hoping that the other task can finish its `enqueue` operation complete
+                    core::hint::spin_loop();
+                    contention_retry_count += 1;
+                    continue;
+                }
                 return None;
             }
             core::cmp::Ordering::Greater => {
@@ -300,6 +325,7 @@ unsafe fn dequeue<T>(
 
 unsafe fn enqueue<T>(
     buffer: *mut Cell<T>,
+    dequeue_pos: &AtomicTargetSize,
     enqueue_pos: &AtomicTargetSize,
     mask: UintSize,
     item: T,
@@ -307,6 +333,7 @@ unsafe fn enqueue<T>(
     let mut pos = enqueue_pos.load(Ordering::Relaxed);
 
     let mut cell;
+    let mut contention_retry_count = 0;
     loop {
         cell = buffer.add(usize::from(pos & mask));
         let seq = (*cell).sequence.load(Ordering::Acquire);
@@ -325,8 +352,19 @@ unsafe fn enqueue<T>(
                 {
                     break;
                 }
+                pos = enqueue_pos.load(Ordering::Relaxed);
             }
             core::cmp::Ordering::Less => {
+                if dequeue_pos.load(Ordering::Relaxed).wrapping_add(mask + 1) != pos
+                    && contention_retry_count < CONTENTION_RETRY_COUNT
+                {
+                    // In this case according to the positions the queue is not full
+                    // This suggests that there is a dequeue operation that is in progress in some other task
+                    // Therefore we can wait a bit hoping that the other task can finish its `dequeue` operation completes
+                    core::hint::spin_loop();
+                    contention_retry_count += 1;
+                    continue;
+                }
                 return Err(item);
             }
             core::cmp::Ordering::Greater => {
@@ -344,9 +382,10 @@ unsafe fn enqueue<T>(
 
 #[cfg(test)]
 mod tests {
+
     use static_assertions::assert_not_impl_any;
 
-    use super::Queue;
+    use super::*;
 
     // Ensure a `Queue` containing `!Send` values stays `!Send` itself.
     assert_not_impl_any!(Queue<*const (), 4>: Send);
@@ -419,4 +458,59 @@ mod tests {
         // Queue is full, this should not block forever.
         q.enqueue(0x55).unwrap_err();
     }
+
+    #[cfg_attr(not(feature = "mpmc_large"), ignore)]
+    #[test]
+    fn issue_583() {
+        fn to_vec<T>(q: &QueueView<T>) -> Vec<T> {
+            // inaccurate
+            let mut ret = vec![];
+            while let Some(v) = q.dequeue() {
+                ret.push(v);
+            }
+            ret
+        }
+        const N: usize = 4;
+
+        let q0 = Queue::<u8, N>::new();
+        for i in 0..N {
+            q0.enqueue(i as u8).expect("new enqueue");
+        }
+        eprintln!("start!");
+
+        std::thread::scope(|sc| {
+            for _ in 0..2 {
+                sc.spawn(|| {
+                    for k in 0..1000_000 {
+                        if let Some(v) = q0.dequeue() {
+                            q0.enqueue(v).unwrap_or_else(|v| {
+                                panic!("{}: q0 -> q0: {}, {:?}", k, v, to_vec(&q0))
+                            });
+                        }
+                    }
+                });
+            }
+        });
+    }
+
+    #[cfg_attr(not(feature = "mpmc_large"), ignore)]
+    #[test]
+    fn issue_583_dequeue() {
+        const N: usize = 4;
+
+        let q0 = Queue::<u8, N>::new();
+        eprintln!("start!");
+        std::thread::scope(|sc| {
+            for _ in 0..2 {
+                sc.spawn(|| {
+                    for k in 0..1000_000 {
+                        q0.enqueue(k as u8).unwrap();
+                        if q0.dequeue().is_none() {
+                            panic!("{k}");
+                        }
+                    }
+                });
+            }
+        });
+    }
 }

I don't think it makes sense to fix this. @nodakai what is your use case for this?

[jannic]

The good news is that no invariant ever gets invalidated.

I think (but may be wrong!) that the current behavior doesn't provide the lock-free guarantee. If I understand https://en.wikipedia.org/wiki/Non-blocking_algorithm correctly, a lock-free algorithm must be able to make progress even if some thread is suspended indefinitely. So the "thread A has been parked by the scheduler and is still in the dequeue operation" situation must not prevent other threads from making progress.

[jannic]

The doc comment says that this is an implementation of http://www.1024cores.net/home/lock-free-algorithms/queues/bounded-mpmc-queue, and that page says "It's not lockfree in the official meaning, just implemented by means of atomic RMW operations w/o mutexes."

So maybe the right solution is to change the documentation to not claim the queue is lock-free? (And maybe mention that if execution is suspended during a dequeue operation, the queue may stay "full" until the thread continues?)

[sgued]

Updating the documentation to document this case would be good.

It'd be nice if we could find another queue design that doesn't have this limitation

[nodakai]

My use case is a memory pool. Something like

struct Pool<T, const N: usize> {
  storage: [T; N],
  avail: MpMcQueue<*mut T, N>,
}

For this use case, returning an object to the pool must not fail.

Recently I'm using ArrayQueue for the avail part, it's working fine so far.

[0xllx0]

It'd be nice if we could find another queue design that doesn't have this limitation

This was my thinking, too. Finding a viable MPMC lock-free algorithm seems like the ideal solution. The change to the docs is still a good idea while implementing whatever lock-free algorithm is settled on.

After a quick search, I found some promising candidates:

• https://inria.hal.science/hal-04851700v1
• https://github.com/erez-strauss/lockfree_mpmc_queue
• https://arxiv.org/abs/1908.04511

I'd be happy to help in this effort / work on the implementation.