test: add Miri tests, fix a stacked borrows violation (#32)

This branch adds support for running Miri tests against `thingbuf`. The
primary reason to use Miri is for testing the wait queue linked list
implementation. Other unsafe code in `thingbuf` is primarily unsafe due
to potential concurrent access issues (which are tested for using
`loom`), rather than doing anything particularly weird with raw pointers
and lifetimes (for which we would want to use Miri).

In order to actually test the implementation using Miri, I wrote a few
tests that exercise basic functionality for the wait queue. 

In the process, I fixed two issues:
- A potential stacked borrows violation in the `enqueue` operation due
to creating a raw pointer to the node being enqueued while there is also
an `&mut` reference to it. This code doesn't currently result in
anything bad happening, as the `&mut` ref doesn't outlive the stack
frame in which the raw ptr is created, but it's a stacked borrows
violation and is technically unsound. This was solved by re-ordering
some of the code so that the raw pointer to the node being enqueued is
not created until we are finished mutating it through the mut ref. 
- Some debug assertions didn't take into account the potential for
`compare_exchange_weak` to fail spuriously with the expected value, and
asserted it was not that value. Apparently miri causes spurious
`compare_exchange_weak` failures (which was surprising to me) and
triggered those assertions. I'm surprised they weren't also triggered
under loom!

There's now a bash script for running Miri locally, and I've added a CI
job to run Miri tests.

Signed-off-by: Eliza Weisman <eliza@buoyant.io>

.github/workflows/miri.yaml

@@ -0,0 +1,30 @@
+on:
+  push:
+    branches:
+      - main
+    paths:
+      - '**.rs'
+  workflow_dispatch:
+  pull_request:
+    paths:
+      - '**.rs'
+
+name: Miri
+jobs:
+  tests:
+    name: Miri tests
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v2
+      - name: Install nightly toolchain
+        uses: actions-rs/toolchain@v1
+        with:
+          profile: minimal
+          toolchain: nightly
+          override: true
+          components: miri
+      - name: Run Miri tests
+      - name: miri
+        run: cargo miri test --lib --no-fail-fast
+        env:
+          MIRIFLAGS: -Zmiri-disable-isolation -Zmiri-tag-raw-pointers

bin/miri

@@ -0,0 +1,10 @@
+#!/usr/bin/env bash
+#
+# Runs `miri` tests
+set -euo pipefail
+set -x
+
+cd "$(dirname "$0")"/..
+
+MIRIFLAGS="-Zmiri-disable-isolation -Zmiri-tag-raw-pointers ${MIRIFLAGS:-}" \
+    cargo +nightly miri test --lib "${@}"

src/mpsc/tests/mpsc_async.rs

@@ -1,7 +1,5 @@
 use super::*;
-use crate::{
-    loom::{self, alloc::Track, future, thread},
-};
+use crate::loom::{self, alloc::Track, future, thread};
 
 #[test]
 #[cfg_attr(ci_skip_slow_models, ignore)]

src/wait/queue.rs

@@ -226,7 +226,7 @@ impl<T: Notify + Unpin> WaitQueue<T> {
                     {
                         Ok(_) => break,
                         Err(actual) => {
-                            debug_assert!(actual == WAKING || actual == CLOSED);
+                            debug_assert!(actual == EMPTY || actual == WAKING || actual == CLOSED);
                             state = actual;
                         }
                     }
@@ -242,7 +242,7 @@ impl<T: Notify + Unpin> WaitQueue<T> {
                         // Consumed the wakeup!
                         Ok(_) => return WaitResult::Notified,
                         Err(actual) => {
-                            debug_assert!(actual == EMPTY || actual == CLOSED);
+                            debug_assert!(actual == WAKING || actual == EMPTY || actual == CLOSED);
                             state = actual;
                         }
                     }
@@ -539,25 +539,25 @@ impl<T> List<T> {
     fn enqueue(&mut self, waiter: Pin<&mut Waiter<T>>) {
         test_println!("List::enqueue({:p})", waiter);
 
-        let node = unsafe { Pin::into_inner_unchecked(waiter) };
-        let ptr = NonNull::from(&*node);
-        debug_assert_ne!(
-            self.head,
-            Some(ptr),
-            "tried to enqueue the same waiter twice!"
-        );
-
-        let head = self.head.replace(ptr);
+        let node = unsafe { waiter.get_unchecked_mut() };
+        let head = self.head.take();
         node.with_node(self, |node| {
             node.next = head;
             node.prev = None;
         });
 
+        let ptr = NonNull::from(node);
+        debug_assert_ne!(
+            self.head,
+            Some(ptr),
+            "tried to enqueue the same waiter twice!"
+        );
         if let Some(mut head) = head {
             unsafe {
                 head.as_mut().set_prev(Some(ptr));
             }
         }
+        self.head = Some(ptr);
 
         if self.tail.is_none() {
             self.tail = Some(ptr);
@@ -632,3 +632,223 @@ impl<T> fmt::Debug for List<T> {
 }
 
 unsafe impl<T: Send> Send for List<T> {}
+
+#[cfg(all(test, not(loom)))]
+mod tests {
+    use super::*;
+    use std::sync::{
+        atomic::{AtomicBool, Ordering},
+        Arc,
+    };
+
+    #[derive(Debug, Clone)]
+    struct MockNotify(Arc<AtomicBool>);
+
+    impl Notify for MockNotify {
+        fn notify(self) {
+            self.0.store(true, Ordering::SeqCst);
+        }
+
+        fn same(&self, &Self(ref other): &Self) -> bool {
+            Arc::ptr_eq(&self.0, other)
+        }
+    }
+
+    impl MockNotify {
+        fn new() -> Self {
+            Self(Arc::new(AtomicBool::new(false)))
+        }
+
+        fn was_notified(&self) -> bool {
+            self.0.load(Ordering::SeqCst)
+        }
+    }
+
+    #[test]
+    fn notify_one() {
+        let q = WaitQueue::new();
+
+        let notify1 = MockNotify::new();
+        let notify2 = MockNotify::new();
+
+        let mut waiter1 = Box::pin(Waiter::new());
+        let mut waiter2 = Box::pin(Waiter::new());
+
+        assert_eq!(q.start_wait(waiter1.as_mut(), &notify1), WaitResult::Wait);
+        assert!(waiter1.is_linked());
+
+        assert_eq!(q.start_wait(waiter2.as_mut(), &notify2), WaitResult::Wait);
+        assert!(waiter2.is_linked());
+
+        assert!(!notify1.was_notified());
+        assert!(!notify2.was_notified());
+
+        assert!(q.notify());
+
+        assert!(notify1.was_notified());
+        assert!(!waiter1.is_linked());
+
+        assert!(!notify2.was_notified());
+        assert!(waiter2.is_linked());
+
+        assert_eq!(
+            q.continue_wait(waiter2.as_mut(), &notify2),
+            WaitResult::Wait
+        );
+
+        assert_eq!(
+            q.continue_wait(waiter1.as_mut(), &notify1),
+            WaitResult::Notified
+        );
+    }
+
+    #[test]
+    fn close() {
+        let q = WaitQueue::new();
+
+        let notify1 = MockNotify::new();
+        let notify2 = MockNotify::new();
+
+        let mut waiter1 = Box::pin(Waiter::new());
+        let mut waiter2 = Box::pin(Waiter::new());
+
+        assert_eq!(q.start_wait(waiter1.as_mut(), &notify1), WaitResult::Wait);
+        assert!(waiter1.is_linked());
+
+        assert_eq!(q.start_wait(waiter2.as_mut(), &notify2), WaitResult::Wait);
+        assert!(waiter2.is_linked());
+
+        assert!(!notify1.was_notified());
+        assert!(!notify2.was_notified());
+
+        q.close();
+
+        assert!(notify1.was_notified());
+        assert!(!waiter1.is_linked());
+
+        assert!(notify2.was_notified());
+        assert!(!waiter2.is_linked());
+
+        assert_eq!(
+            q.continue_wait(waiter2.as_mut(), &notify2),
+            WaitResult::Closed
+        );
+
+        assert_eq!(
+            q.continue_wait(waiter1.as_mut(), &notify1),
+            WaitResult::Closed
+        );
+    }
+
+    #[test]
+    fn remove_from_middle() {
+        let q = WaitQueue::new();
+
+        let notify1 = MockNotify::new();
+        let notify2 = MockNotify::new();
+        let notify3 = MockNotify::new();
+
+        let mut waiter1 = Box::pin(Waiter::new());
+        let mut waiter2 = Box::pin(Waiter::new());
+        let mut waiter3 = Box::pin(Waiter::new());
+
+        assert_eq!(q.start_wait(waiter1.as_mut(), &notify1), WaitResult::Wait);
+        assert!(waiter1.is_linked());
+
+        assert_eq!(q.start_wait(waiter2.as_mut(), &notify2), WaitResult::Wait);
+        assert!(waiter2.is_linked());
+
+        assert_eq!(q.start_wait(waiter3.as_mut(), &notify3), WaitResult::Wait);
+        assert!(waiter2.is_linked());
+
+        assert!(!notify1.was_notified());
+        assert!(!notify2.was_notified());
+        assert!(!notify3.was_notified());
+
+        waiter2.as_mut().remove(&q);
+        assert!(!notify2.was_notified());
+        drop(waiter2);
+
+        assert!(q.notify());
+
+        assert!(notify1.was_notified());
+        assert!(!waiter1.is_linked());
+
+        assert!(!notify3.was_notified());
+        assert!(waiter3.is_linked());
+
+        assert_eq!(
+            q.continue_wait(waiter3.as_mut(), &notify3),
+            WaitResult::Wait
+        );
+
+        assert_eq!(
+            q.continue_wait(waiter1.as_mut(), &notify1),
+            WaitResult::Notified
+        );
+    }
+
+    #[test]
+    fn remove_after_notify() {
+        let q = WaitQueue::new();
+
+        let notify1 = MockNotify::new();
+        let notify2 = MockNotify::new();
+        let notify3 = MockNotify::new();
+
+        let mut waiter1 = Box::pin(Waiter::new());
+        let mut waiter2 = Box::pin(Waiter::new());
+        let mut waiter3 = Box::pin(Waiter::new());
+
+        assert_eq!(q.start_wait(waiter1.as_mut(), &notify1), WaitResult::Wait);
+        assert!(waiter1.is_linked());
+
+        assert_eq!(q.start_wait(waiter2.as_mut(), &notify2), WaitResult::Wait);
+        assert!(waiter2.is_linked());
+
+        assert_eq!(q.start_wait(waiter3.as_mut(), &notify3), WaitResult::Wait);
+        assert!(waiter2.is_linked());
+
+        assert!(!notify1.was_notified());
+        assert!(!notify2.was_notified());
+        assert!(!notify3.was_notified());
+
+        assert!(q.notify());
+
+        assert!(notify1.was_notified());
+        assert!(!waiter1.is_linked());
+
+        assert!(!notify2.was_notified());
+        assert!(waiter2.is_linked());
+
+        assert!(!notify3.was_notified());
+        assert!(waiter3.is_linked());
+
+        assert_eq!(
+            q.continue_wait(waiter3.as_mut(), &notify3),
+            WaitResult::Wait
+        );
+
+        assert_eq!(
+            q.continue_wait(waiter2.as_mut(), &notify2),
+            WaitResult::Wait
+        );
+
+        assert_eq!(
+            q.continue_wait(waiter1.as_mut(), &notify1),
+            WaitResult::Notified
+        );
+
+        waiter2.as_mut().remove(&q);
+        assert!(!notify2.was_notified());
+        drop(waiter2);
+
+        assert!(!notify3.was_notified());
+        assert!(waiter3.is_linked());
+
+        assert_eq!(
+            q.continue_wait(waiter3.as_mut(), &notify3),
+            WaitResult::Wait
+        );
+    }
+}