fix(port_std): replace std::sync::Mutex with a remote-park-friendly implementation on Windows

The test suite was intermittently failing on Windows. The execution
stalled in some cases. When I attached a debugger, I observed that two
threads (only one of which I speculate was suspended) were attempting to
acquire a lock, but there was no thread holding the lock. This suggests
`std::sync::Mutex` can deadlock if one of the waiters is suspended by
a remote park operation.

Author: yvt

src/r3_port_std/src/threading_unix.rs

@@ -14,7 +14,10 @@ use std::{
     thread,
 };
 
-pub use self::thread::ThreadId;
+pub use std::{
+    sync::{Mutex, MutexGuard},
+    thread::ThreadId,
+};
 
 thread_local! {
     static EXIT_JMP_BUF: Cell<Option<JmpBuf>> = Cell::new(None);

src/r3_port_std/src/threading_windows.rs

@@ -2,7 +2,7 @@ use std::{
     mem::MaybeUninit,
     sync::{
         atomic::{AtomicIsize, Ordering},
-        mpsc, Arc, Mutex,
+        mpsc, Arc,
     },
     thread,
 };
@@ -17,7 +17,7 @@ pub unsafe fn exit_thread() -> ! {
     unreachable!();
 }
 
-pub use self::thread::ThreadId;
+pub use std::thread::ThreadId;
 
 /// [`std::thread::JoinHandle`] with extra functionalities.
 #[derive(Debug)]
@@ -208,3 +208,95 @@ fn panic_last_error() -> ! {
         errhandlingapi::GetLastError()
     });
 }
+
+// `std::sync::Mutex` isn't remote-park-friendly
+pub use mutex::{Mutex, MutexGuard};
+
+mod mutex {
+    use std::{
+        cell::UnsafeCell,
+        fmt,
+        sync::atomic::{AtomicBool, Ordering},
+    };
+    use winapi::um::{synchapi, winbase::INFINITE};
+
+    /// Remote-park-friendly [`std::sync::Mutex`].
+    pub struct Mutex<T: ?Sized> {
+        locked: AtomicBool,
+        data: UnsafeCell<T>,
+    }
+
+    unsafe impl<T: ?Sized + Send> Sync for Mutex<T> {}
+
+    pub struct MutexGuard<'a, T: ?Sized> {
+        data: &'a mut T,
+        locked: &'a AtomicBool,
+    }
+
+    impl<T> Mutex<T> {
+        #[inline]
+        pub const fn new(x: T) -> Self {
+            Self {
+                data: UnsafeCell::new(x),
+                locked: AtomicBool::new(false),
+            }
+        }
+    }
+
+    impl<T: ?Sized> Mutex<T> {
+        #[inline]
+        pub fn lock(&self) -> Result<MutexGuard<'_, T>, ()> {
+            while self
+                .locked
+                .compare_exchange(false, true, Ordering::Acquire, Ordering::Relaxed)
+                .is_err()
+            {
+                unsafe {
+                    synchapi::WaitOnAddress(
+                        self.locked.as_mut_ptr().cast(),  // location to watch
+                        (&true) as *const bool as *mut _, // undesired value
+                        std::mem::size_of::<bool>(),      // value size
+                        INFINITE,                         // timeout
+                    );
+                }
+            }
+
+            // Poisoning is not supported by this `Mutex`
+
+            Ok(MutexGuard {
+                data: unsafe { &mut *self.data.get() },
+                locked: &self.locked,
+            })
+        }
+    }
+
+    impl<T: ?Sized + fmt::Debug> fmt::Debug for Mutex<T> {
+        fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+            f.write_str("Mutex")
+        }
+    }
+
+    impl<T: ?Sized> Drop for MutexGuard<'_, T> {
+        #[inline]
+        fn drop(&mut self) {
+            self.locked.store(false, Ordering::Release);
+            unsafe { synchapi::WakeByAddressSingle(self.locked.as_mut_ptr().cast()) };
+        }
+    }
+
+    impl<T: ?Sized> std::ops::Deref for MutexGuard<'_, T> {
+        type Target = T;
+
+        #[inline]
+        fn deref(&self) -> &Self::Target {
+            self.data
+        }
+    }
+
+    impl<T: ?Sized> std::ops::DerefMut for MutexGuard<'_, T> {
+        #[inline]
+        fn deref_mut(&mut self) -> &mut Self::Target {
+            self.data
+        }
+    }
+}

src/r3_port_std/src/ums.rs

@@ -3,7 +3,7 @@ use once_cell::sync::OnceCell;
 use slab::Slab;
 use std::{
     panic::{catch_unwind, AssertUnwindSafe},
-    sync::{mpsc, Arc, Mutex, MutexGuard},
+    sync::{mpsc, Arc},
     thread::Result,
 };
 
@@ -18,7 +18,7 @@ mod tests;
 /// `Sched: `[`Scheduler`].
 #[derive(Debug)]
 pub struct ThreadGroup<Sched: ?Sized> {
-    state: Arc<Mutex<State<Sched>>>,
+    state: Arc<threading::Mutex<State<Sched>>>,
 }
 
 impl<Sched: ?Sized> Clone for ThreadGroup<Sched> {
@@ -37,8 +37,8 @@ pub struct ThreadGroupJoinHandle {
 
 /// RAII guard returned by [`ThreadGroup::lock`].
 pub struct ThreadGroupLockGuard<'a, Sched: ?Sized> {
-    state_ref: &'a Arc<Mutex<State<Sched>>>,
-    guard: MutexGuard<'a, State<Sched>>,
+    state_ref: &'a Arc<threading::Mutex<State<Sched>>>,
+    guard: threading::MutexGuard<'a, State<Sched>>,
 }
 
 /// Identifies a thread in [`ThreadGroup`].
@@ -82,7 +82,7 @@ struct ThreadLocalBlock {
     /// The current thread ID.
     thread_id: ThreadId,
     /// The thread group the current worker thread belongs to.
-    state: Arc<Mutex<State<dyn Scheduler>>>,
+    state: Arc<threading::Mutex<State<dyn Scheduler>>>,
 }
 
 impl<Sched: Scheduler> ThreadGroup<Sched> {
@@ -91,7 +91,7 @@ impl<Sched: Scheduler> ThreadGroup<Sched> {
     pub fn new(sched: Sched) -> (Self, ThreadGroupJoinHandle) {
         let (send, recv) = mpsc::channel();
 
-        let state = Arc::new(Mutex::new(State {
+        let state = Arc::new(threading::Mutex::new(State {
             threads: Slab::new(),
             num_threads: 0,
             cur_thread_id: None,
@@ -264,7 +264,7 @@ impl State<dyn Scheduler> {
 ///
 /// Panics if the current thread is not a worker thread of some [`ThreadGroup`].
 pub fn yield_now() {
-    let thread_group: Arc<Mutex<State<dyn Scheduler>>> = TLB
+    let thread_group: Arc<threading::Mutex<State<dyn Scheduler>>> = TLB
         .with(|cell| cell.get().map(|tlb| Arc::clone(&tlb.state)))
         .expect("current thread does not belong to a thread group");
 
@@ -303,7 +303,7 @@ pub unsafe fn exit_thread() -> ! {
 
 /// Mark the specified thread as exited.
 fn finalize_thread(
-    thread_group: Arc<Mutex<State<dyn Scheduler>>>,
+    thread_group: Arc<threading::Mutex<State<dyn Scheduler>>>,
     thread_id: ThreadId,
     result: Result<()>,
 ) {