Auto merge of #9 - japaric:contention, r=japaric

add contention test

This needs to be fixed before the next release

Interestingly I see that the test passes when compiled without optimizations but fails when
optimized. Perhaps there's some UB in here.

cc @pftbest

ci/script.sh

@@ -11,8 +11,9 @@ main() {
             cargo test --target $TARGET
             cargo test --target $TARGET --release
 
-            export TSAN_OPTIONS="suppressions=$(pwd)/blacklist.txt"
             export RUSTFLAGS="-Z sanitizer=thread"
+            export RUST_TEST_THREADS=1
+            export TSAN_OPTIONS="suppressions=$(pwd)/blacklist.txt"
 
             cargo test --test tsan --target $TARGET
             cargo test --test tsan --target $TARGET --release

src/ring_buffer/mod.rs

@@ -29,6 +29,10 @@ impl AtomicUsize {
         unsafe { &mut *self.v.get() }
     }
 
+    pub fn load_acquire(&self) -> usize {
+        unsafe { intrinsics::atomic_load_acq(self.v.get()) }
+    }
+
     pub fn load_relaxed(&self) -> usize {
         unsafe { intrinsics::atomic_load_relaxed(self.v.get()) }
     }
@@ -129,6 +133,11 @@ where
         }
     }
 
+    /// Returns `true` if the ring buffer has a length of 0
+    pub fn is_empty(&self) -> bool {
+        self.len() == 0
+    }
+
     /// Iterates from the front of the queue to the back
     pub fn iter(&self) -> Iter<T, A> {
         Iter {

src/ring_buffer/spsc.rs

@@ -45,7 +45,7 @@ where
         let n = rb.capacity() + 1;
         let buffer: &[T] = unsafe { rb.buffer.as_ref() };
 
-        let tail = rb.tail.load_relaxed();
+        let tail = rb.tail.load_acquire();
         let head = rb.head.load_relaxed();
         if head != tail {
             let item = unsafe { ptr::read(buffer.get_unchecked(head)) };

tests/tsan.rs

@@ -73,3 +73,48 @@ fn scoped() {
 
     rb.dequeue().unwrap();
 }
+
+#[test]
+fn contention() {
+    const N: usize = 1024;
+
+    let mut rb: RingBuffer<u8, [u8; N]> = RingBuffer::new();
+
+    {
+        let (mut p, mut c) = rb.split();
+
+        Pool::new(2).scoped(move |scope| {
+            scope.execute(move || {
+                let mut sum: u32 = 0;
+
+                for i in 0..N {
+                    let i = i as u8;
+                    sum = sum.wrapping_add(i as u32);
+                    while let Err(_) = p.enqueue(i) {}
+                }
+
+                println!("producer: {}", sum);
+            });
+
+            scope.execute(move || {
+                let mut sum: u32 = 0;
+
+                for _ in 0..N {
+                    loop {
+                        match c.dequeue() {
+                            Some(v) => {
+                                sum = sum.wrapping_add(v as u32);
+                                break;
+                            }
+                            _ => {}
+                        }
+                    }
+                }
+
+                println!("consumer: {}", sum);
+            });
+        });
+    }
+
+    assert!(rb.is_empty());
+}