Miri: enable preemption again

ci/miri.sh

@@ -3,37 +3,40 @@ set -euxo pipefail
 IFS=$'\n\t'
 cd "$(dirname "$0")"/..
 
+# We need 'ts' for the per-line timing
+sudo apt-get -y install moreutils
+echo
+
 export RUSTFLAGS="${RUSTFLAGS:-} -Z randomize-layout"
 
-# disable preemption due to https://github.com/rust-lang/rust/issues/55005
-MIRIFLAGS="-Zmiri-strict-provenance -Zmiri-symbolic-alignment-check -Zmiri-disable-isolation -Zmiri-preemption-rate=0" \
+# -Zmiri-ignore-leaks is needed because we use detached threads in tests/docs: https://github.com/rust-lang/miri/issues/1371
+MIRIFLAGS="-Zmiri-strict-provenance -Zmiri-symbolic-alignment-check -Zmiri-disable-isolation -Zmiri-ignore-leaks" \
     cargo miri test \
     -p crossbeam-queue \
-    -p crossbeam-utils
+    -p crossbeam-utils 2>&1 | ts -i '%.s  '
 
 # -Zmiri-ignore-leaks is needed because we use detached threads in tests/docs: https://github.com/rust-lang/miri/issues/1371
-# disable preemption due to https://github.com/rust-lang/rust/issues/55005
-MIRIFLAGS="-Zmiri-strict-provenance -Zmiri-symbolic-alignment-check -Zmiri-disable-isolation -Zmiri-ignore-leaks -Zmiri-preemption-rate=0" \
+MIRIFLAGS="-Zmiri-strict-provenance -Zmiri-symbolic-alignment-check -Zmiri-disable-isolation -Zmiri-ignore-leaks" \
     cargo miri test \
-    -p crossbeam-channel
+    -p crossbeam-channel 2>&1 | ts -i '%.s  '
 
 # -Zmiri-ignore-leaks is needed for https://github.com/crossbeam-rs/crossbeam/issues/579
 # -Zmiri-disable-stacked-borrows is needed for https://github.com/crossbeam-rs/crossbeam/issues/545
-# disable preemption due to https://github.com/rust-lang/rust/issues/55005
-MIRIFLAGS="-Zmiri-check-number-validity -Zmiri-symbolic-alignment-check -Zmiri-disable-isolation -Zmiri-disable-stacked-borrows -Zmiri-ignore-leaks -Zmiri-preemption-rate=0" \
+MIRIFLAGS="-Zmiri-symbolic-alignment-check -Zmiri-disable-isolation -Zmiri-disable-stacked-borrows -Zmiri-ignore-leaks" \
     cargo miri test \
     -p crossbeam-epoch \
-    -p crossbeam-skiplist
+    -p crossbeam-skiplist 2>&1 | ts -i '%.s  '
 
 # -Zmiri-ignore-leaks is needed for https://github.com/crossbeam-rs/crossbeam/issues/579
 # -Zmiri-disable-stacked-borrows is needed for https://github.com/crossbeam-rs/crossbeam/issues/545
-# disable preemption due to https://github.com/rust-lang/rust/issues/55005
-MIRIFLAGS="-Zmiri-check-number-validity -Zmiri-symbolic-alignment-check -Zmiri-disable-stacked-borrows -Zmiri-ignore-leaks -Zmiri-compare-exchange-weak-failure-rate=0.0 -Zmiri-preemption-rate=0" \
+# -Zmiri-compare-exchange-weak-failure-rate=0.0 is needed because some sequential tests (e.g.,
+# doctest of Stealer::steal) incorrectly assume that sequential weak CAS will never fail.
+# -Zmiri-preemption-rate=0 is needed because this code technically has UB and Miri catches that.
+MIRIFLAGS="-Zmiri-symbolic-alignment-check -Zmiri-disable-stacked-borrows -Zmiri-ignore-leaks -Zmiri-compare-exchange-weak-failure-rate=0.0 -Zmiri-preemption-rate=0" \
     cargo miri test \
-    -p crossbeam-deque
+    -p crossbeam-deque 2>&1 | ts -i '%.s  '
 
 # -Zmiri-ignore-leaks is needed for https://github.com/crossbeam-rs/crossbeam/issues/579
-# disable preemption due to https://github.com/rust-lang/rust/issues/55005
-MIRIFLAGS="-Zmiri-check-number-validity -Zmiri-symbolic-alignment-check -Zmiri-ignore-leaks -Zmiri-preemption-rate=0" \
+MIRIFLAGS="-Zmiri-symbolic-alignment-check -Zmiri-ignore-leaks" \
     cargo miri test \
-    -p crossbeam
+    -p crossbeam 2>&1 | ts -i '%.s  '

crossbeam-channel/tests/array.rs

@@ -532,16 +532,12 @@ fn drops() {
             scope.spawn(|_| {
                 for _ in 0..steps {
                     r.recv().unwrap();
-                    #[cfg(miri)]
-                    std::thread::yield_now(); // https://github.com/rust-lang/miri/issues/1388
                 }
             });
 
             scope.spawn(|_| {
                 for _ in 0..steps {
                     s.send(DropCounter).unwrap();
-                    #[cfg(miri)]
-                    std::thread::yield_now(); // https://github.com/rust-lang/miri/issues/1388
                 }
             });
         })

crossbeam-channel/tests/list.rs

@@ -433,8 +433,6 @@ fn drops() {
             scope.spawn(|_| {
                 for _ in 0..steps {
                     r.recv().unwrap();
-                    #[cfg(miri)]
-                    std::thread::yield_now(); // https://github.com/rust-lang/miri/issues/1388
                 }
             });
 

crossbeam-channel/tests/thread_locals.rs

@@ -1,6 +1,6 @@
 //! Tests that make sure accessing thread-locals while exiting the thread doesn't cause panics.
 
-#![cfg(not(miri))] // error: abnormal termination: the evaluated program aborted execution
+#![cfg(not(miri))] // Miri detects that this test is buggy: the destructor of `FOO` uses `std::thread::current()`!
 
 use std::thread;
 use std::time::Duration;

crossbeam-deque/src/deque.rs

@@ -53,6 +53,8 @@ impl<T> Buffer<T> {
     /// Returns a pointer to the task at the specified `index`.
     unsafe fn at(&self, index: isize) -> *mut T {
         // `self.cap` is always a power of two.
+        // We do all the loads at `MaybeUninit` because we might realize, after loading, that we
+        // don't actually have the right to access this memory.
         self.ptr.offset(index & (self.cap - 1) as isize)
     }
 
@@ -62,18 +64,18 @@ impl<T> Buffer<T> {
     /// technically speaking a data race and therefore UB. We should use an atomic store here, but
     /// that would be more expensive and difficult to implement generically for all types `T`.
     /// Hence, as a hack, we use a volatile write instead.
-    unsafe fn write(&self, index: isize, task: T) {
-        ptr::write_volatile(self.at(index), task)
+    unsafe fn write(&self, index: isize, task: MaybeUninit<T>) {
+        ptr::write_volatile(self.at(index) as *mut MaybeUninit<T>, task)
     }
 
     /// Reads a task from the specified `index`.
     ///
     /// This method might be concurrently called with another `write` at the same index, which is
     /// technically speaking a data race and therefore UB. We should use an atomic load here, but
     /// that would be more expensive and difficult to implement generically for all types `T`.
-    /// Hence, as a hack, we use a volatile write instead.
-    unsafe fn read(&self, index: isize) -> T {
-        ptr::read_volatile(self.at(index))
+    /// Hence, as a hack, we use a volatile load instead.
+    unsafe fn read(&self, index: isize) -> MaybeUninit<T> {
+        ptr::read_volatile(self.at(index) as *mut MaybeUninit<T>)
     }
 }
 
@@ -406,7 +408,7 @@ impl<T> Worker<T> {
 
         // Write `task` into the slot.
         unsafe {
-            buffer.write(b, task);
+            buffer.write(b, MaybeUninit::new(task));
         }
 
         atomic::fence(Ordering::Release);
@@ -461,7 +463,7 @@ impl<T> Worker<T> {
                 unsafe {
                     // Read the popped task.
                     let buffer = self.buffer.get();
-                    let task = buffer.read(f);
+                    let task = buffer.read(f).assume_init();
 
                     // Shrink the buffer if `len - 1` is less than one fourth of the capacity.
                     if buffer.cap > MIN_CAP && len <= buffer.cap as isize / 4 {
@@ -509,8 +511,8 @@ impl<T> Worker<T> {
                             )
                             .is_err()
                         {
-                            // Failed. We didn't pop anything.
-                            mem::forget(task.take());
+                            // Failed. We didn't pop anything. Reset to `None`.
+                            task.take();
                         }
 
                         // Restore the back index to the original task.
@@ -524,7 +526,7 @@ impl<T> Worker<T> {
                         }
                     }
 
-                    task
+                    task.map(|t| unsafe { t.assume_init() })
                 }
             }
         }
@@ -661,12 +663,11 @@ impl<T> Stealer<T> {
                 .is_err()
         {
             // We didn't steal this task, forget it.
-            mem::forget(task);
             return Steal::Retry;
         }
 
         // Return the stolen task.
-        Steal::Success(task)
+        Steal::Success(unsafe { task.assume_init() })
     }
 
     /// Steals a batch of tasks and pushes them into another worker.
@@ -821,7 +822,6 @@ impl<T> Stealer<T> {
                             .is_err()
                     {
                         // We didn't steal this task, forget it and break from the loop.
-                        mem::forget(task);
                         batch_size = i;
                         break;
                     }
@@ -975,7 +975,6 @@ impl<T> Stealer<T> {
                         .is_err()
                 {
                     // We didn't steal this task, forget it.
-                    mem::forget(task);
                     return Steal::Retry;
                 }
 
@@ -992,7 +991,6 @@ impl<T> Stealer<T> {
                     .is_err()
                 {
                     // We didn't steal this task, forget it.
-                    mem::forget(task);
                     return Steal::Retry;
                 }
 
@@ -1037,7 +1035,6 @@ impl<T> Stealer<T> {
                             .is_err()
                     {
                         // We didn't steal this task, forget it and break from the loop.
-                        mem::forget(tmp);
                         batch_size = i;
                         break;
                     }
@@ -1077,7 +1074,7 @@ impl<T> Stealer<T> {
         dest.inner.back.store(dest_b, Ordering::Release);
 
         // Return with success.
-        Steal::Success(task)
+        Steal::Success(unsafe { task.assume_init() })
     }
 }
 
@@ -1535,7 +1532,7 @@ impl<T> Injector<T> {
                         // Read the task.
                         let slot = (*block).slots.get_unchecked(offset + i);
                         slot.wait_write();
-                        let task = slot.task.get().read().assume_init();
+                        let task = slot.task.get().read();
 
                         // Write it into the destination queue.
                         dest_buffer.write(dest_b.wrapping_add(i as isize), task);
@@ -1547,7 +1544,7 @@ impl<T> Injector<T> {
                         // Read the task.
                         let slot = (*block).slots.get_unchecked(offset + i);
                         slot.wait_write();
-                        let task = slot.task.get().read().assume_init();
+                        let task = slot.task.get().read();
 
                         // Write it into the destination queue.
                         dest_buffer.write(dest_b.wrapping_add((batch_size - 1 - i) as isize), task);
@@ -1689,7 +1686,7 @@ impl<T> Injector<T> {
             // Read the task.
             let slot = (*block).slots.get_unchecked(offset);
             slot.wait_write();
-            let task = slot.task.get().read().assume_init();
+            let task = slot.task.get().read();
 
             match dest.flavor {
                 Flavor::Fifo => {
@@ -1698,7 +1695,7 @@ impl<T> Injector<T> {
                         // Read the task.
                         let slot = (*block).slots.get_unchecked(offset + i + 1);
                         slot.wait_write();
-                        let task = slot.task.get().read().assume_init();
+                        let task = slot.task.get().read();
 
                         // Write it into the destination queue.
                         dest_buffer.write(dest_b.wrapping_add(i as isize), task);
@@ -1711,7 +1708,7 @@ impl<T> Injector<T> {
                         // Read the task.
                         let slot = (*block).slots.get_unchecked(offset + i + 1);
                         slot.wait_write();
-                        let task = slot.task.get().read().assume_init();
+                        let task = slot.task.get().read();
 
                         // Write it into the destination queue.
                         dest_buffer.write(dest_b.wrapping_add((batch_size - 1 - i) as isize), task);
@@ -1744,7 +1741,7 @@ impl<T> Injector<T> {
                 }
             }
 
-            Steal::Success(task)
+            Steal::Success(task.assume_init())
         }
     }
 

crossbeam-queue/tests/array_queue.rs

@@ -103,8 +103,6 @@ fn len() {
                         assert_eq!(x, i);
                         break;
                     }
-                    #[cfg(miri)]
-                    std::thread::yield_now(); // https://github.com/rust-lang/miri/issues/1388
                 }
                 let len = q.len();
                 assert!(len <= CAP);
@@ -113,10 +111,7 @@ fn len() {
 
         scope.spawn(|_| {
             for i in 0..COUNT {
-                while q.push(i).is_err() {
-                    #[cfg(miri)]
-                    std::thread::yield_now(); // https://github.com/rust-lang/miri/issues/1388
-                }
+                while q.push(i).is_err() {}
                 let len = q.len();
                 assert!(len <= CAP);
             }
@@ -143,19 +138,14 @@ fn spsc() {
                         assert_eq!(x, i);
                         break;
                     }
-                    #[cfg(miri)]
-                    std::thread::yield_now(); // https://github.com/rust-lang/miri/issues/1388
                 }
             }
             assert!(q.pop().is_none());
         });
 
         scope.spawn(|_| {
             for i in 0..COUNT {
-                while q.push(i).is_err() {
-                    #[cfg(miri)]
-                    std::thread::yield_now(); // https://github.com/rust-lang/miri/issues/1388
-                }
+                while q.push(i).is_err() {}
             }
         });
     })
@@ -184,8 +174,6 @@ fn spsc_ring_buffer() {
                     }
                 }
             }
-            #[cfg(miri)]
-            std::thread::yield_now(); // https://github.com/rust-lang/miri/issues/1388
         });
 
         scope.spawn(|_| {
@@ -320,19 +308,14 @@ fn drops() {
         scope(|scope| {
             scope.spawn(|_| {
                 for _ in 0..steps {
-                    while q.pop().is_none() {
-                        #[cfg(miri)]
-                        std::thread::yield_now(); // https://github.com/rust-lang/miri/issues/1388
-                    }
+                    while q.pop().is_none() {}
                 }
             });
 
             scope.spawn(|_| {
                 for _ in 0..steps {
                     while q.push(DropCounter).is_err() {
                         DROPS.fetch_sub(1, Ordering::SeqCst);
-                        #[cfg(miri)]
-                        std::thread::yield_now(); // https://github.com/rust-lang/miri/issues/1388
                     }
                 }
             });

crossbeam-queue/tests/seg_queue.rs

@@ -69,8 +69,6 @@ fn spsc() {
                         assert_eq!(x, i);
                         break;
                     }
-                    #[cfg(miri)]
-                    std::thread::yield_now(); // https://github.com/rust-lang/miri/issues/1388
                 }
             }
             assert!(q.pop().is_none());
@@ -155,10 +153,7 @@ fn drops() {
         scope(|scope| {
             scope.spawn(|_| {
                 for _ in 0..steps {
-                    while q.pop().is_none() {
-                        #[cfg(miri)]
-                        std::thread::yield_now(); // https://github.com/rust-lang/miri/issues/1388
-                    }
+                    while q.pop().is_none() {}
                 }
             });
 

crossbeam-utils/tests/wait_group.rs

@@ -36,6 +36,7 @@ fn wait() {
 }
 
 #[test]
+#[cfg(not(miri))] // this test makes timing assumptions, but Miri is so slow it violates them
 fn wait_and_drop() {
     let wg = WaitGroup::new();
     let (tx, rx) = mpsc::channel();
@@ -51,8 +52,8 @@ fn wait_and_drop() {
         });
     }
 
-    // At this point, all spawned threads should be sleeping, so we shouldn't get anything from the
-    // channel.
+    // At this point, all spawned threads should be in `thread::sleep`, so we shouldn't get anything
+    // from the channel.
     assert!(rx.try_recv().is_err());
 
     wg.wait();