Implement thread parking, RwLock, Once fallbacks

- Bring back "Generic" thread parker
- Add checks for Win8 and WinXP APIs to decide which thread parker impl to use
- Use `queue::RwLock` for rust9x. Now that the thread parker is available, we get this high-quality `RwLock` implementation on all API levels, without any extra fallback impls!
- Same for `Once`

Author: seritools

library/std/src/sys/pal/windows/c.rs

@@ -5,7 +5,9 @@
 #![unstable(issue = "none", feature = "windows_c")]
 #![allow(clippy::style)]
 
-use core::ffi::{c_uint, c_ulong, c_ushort, c_void};
+#[allow(unused_imports)]
+use core::ffi::c_void;
+use core::ffi::{c_uint, c_ulong, c_ushort};
 use core::{mem, ptr};
 
 mod windows_sys;
@@ -150,7 +152,7 @@ compat_fn_with_fallback! {
     }
 }
 
-#[cfg(not(target_vendor = "win7"))]
+#[cfg(not(any(target_vendor = "win7", target_vendor = "rust9x")))]
 // Use raw-dylib to import synchronization functions to workaround issues with the older mingw import library.
 #[cfg_attr(
     target_arch = "x86",
@@ -187,11 +189,31 @@ compat_fn_optional! {
     pub fn WakeByAddressSingle(address: *const c_void);
 }
 
-#[cfg(any(target_vendor = "win7", target_vendor = "uwp"))]
+#[cfg(target_vendor = "rust9x")]
+compat_fn_with_fallback! {
+    pub static SYNCH: &CStr = c"api-ms-win-core-synch-l1-2-0" => { load: true, unicows: false };
+
+    pub fn WaitOnAddress(
+        address: *const c_void,
+        compareaddress: *const c_void,
+        addresssize: usize,
+        dwmilliseconds: u32
+    ) -> BOOL {
+        rtabort!("unimplemented")
+    }
+    pub fn WakeByAddressSingle(address: *const c_void) {
+        rtabort!("unimplemented")
+    }
+    pub fn WakeByAddressAll(address: *const c_void) {
+        rtabort!("unimplemented")
+    }
+}
+
+#[cfg(any(target_vendor = "win7", target_vendor = "rust9x", target_vendor = "uwp"))]
 compat_fn_with_fallback! {
     pub static NTDLL: &CStr = c"ntdll" => { load: true, unicows: false };
 
-    #[cfg(target_vendor = "win7")]
+    #[cfg(any(target_vendor = "win7", target_vendor = "rust9x"))]
     pub fn NtCreateKeyedEvent(
         KeyedEventHandle: *mut HANDLE,
         DesiredAccess: u32,
@@ -200,7 +222,7 @@ compat_fn_with_fallback! {
     ) -> NTSTATUS {
         panic!("keyed events not available")
     }
-    #[cfg(target_vendor = "win7")]
+    #[cfg(any(target_vendor = "win7", target_vendor = "rust9x"))]
     pub fn NtReleaseKeyedEvent(
         EventHandle: HANDLE,
         Key: *const c_void,
@@ -209,7 +231,7 @@ compat_fn_with_fallback! {
     ) -> NTSTATUS {
         panic!("keyed events not available")
     }
-    #[cfg(target_vendor = "win7")]
+    #[cfg(any(target_vendor = "win7", target_vendor = "rust9x"))]
     pub fn NtWaitForKeyedEvent(
         EventHandle: HANDLE,
         Key: *const c_void,

library/std/src/sys/pal/windows/compat.rs

@@ -26,6 +26,9 @@ use crate::sys::c;
 #[cfg(target_vendor = "rust9x")]
 pub(crate) mod checks;
 
+#[cfg(target_vendor = "rust9x")]
+pub(crate) mod thread_parking;
+
 // This uses a static initializer to preload some imported functions.
 // The CRT (C runtime) executes static initializers before `main`
 // is called (for binaries) and before `DllMain` is called (for DLLs).

library/std/src/sys/pal/windows/compat/thread_parking.rs

@@ -0,0 +1,40 @@
+static mut THREAD_PARKING_IMPL: Option<ThreadParkingImpl> = None;
+
+#[derive(Clone, Copy, PartialEq, Eq)]
+pub(crate) enum ThreadParkingImpl {
+    // WaitOnAddress-based, Windows 8+
+    Futex,
+    // NtCreatedKeyedEvent-based, Windows XP+
+    KeyedEvent,
+    // Generic thread parker based on Mutex+Condvar
+    Generic,
+}
+
+// This CANNOT be called during global init because it has to `LoadLibrary` to figure out which impl
+// is available.
+pub(crate) fn thread_parking_impl() -> ThreadParkingImpl {
+    if let Some(implementation) = unsafe { THREAD_PARKING_IMPL } {
+        return implementation;
+    }
+
+    let implementation = {
+        if crate::sys::c::WaitOnAddress::available().is_some()
+            && crate::sys::c::WakeByAddressSingle::available().is_some()
+        {
+            ThreadParkingImpl::Futex
+        } else if crate::sys::c::NtCreateKeyedEvent::available().is_some()
+            && crate::sys::c::NtReleaseKeyedEvent::available().is_some()
+            && crate::sys::c::NtWaitForKeyedEvent::available().is_some()
+        {
+            ThreadParkingImpl::KeyedEvent
+        } else {
+            ThreadParkingImpl::Generic
+        }
+    };
+
+    unsafe {
+        THREAD_PARKING_IMPL = Some(implementation);
+    }
+
+    implementation
+}

library/std/src/sys/sync/once/mod.rs

@@ -9,7 +9,7 @@
 
 cfg_if::cfg_if! {
     if #[cfg(any(
-        all(target_os = "windows", not(target_vendor="win7")),
+        all(target_os = "windows", not(any(target_vendor = "win7", target_vendor = "rust9x"))),
         target_os = "linux",
         target_os = "android",
         all(target_arch = "wasm32", target_feature = "atomics"),

library/std/src/sys/sync/rwlock/mod.rs

@@ -1,6 +1,6 @@
 cfg_if::cfg_if! {
     if #[cfg(any(
-        all(target_os = "windows", not(target_vendor = "win7")),
+        all(target_os = "windows", not(any(target_vendor = "win7", target_vendor = "rust9x"))),
         target_os = "linux",
         target_os = "android",
         target_os = "freebsd",
@@ -14,7 +14,7 @@ cfg_if::cfg_if! {
         pub use futex::RwLock;
     } else if #[cfg(any(
         target_family = "unix",
-        all(target_os = "windows", target_vendor = "win7"),
+        all(target_os = "windows", any(target_vendor = "win7", target_vendor = "rust9x")),
         all(target_vendor = "fortanix", target_env = "sgx"),
         target_os = "xous",
     ))] {

library/std/src/sys/sync/thread_parking/generic.rs

@@ -0,0 +1,125 @@
+//! Parker implementation based on a Mutex and Condvar.
+
+use crate::pin::Pin;
+use crate::sync::atomic::AtomicUsize;
+use crate::sync::atomic::Ordering::SeqCst;
+use crate::sync::{Condvar, Mutex};
+use crate::time::Duration;
+
+const EMPTY: usize = 0;
+const PARKED: usize = 1;
+const NOTIFIED: usize = 2;
+
+pub struct Parker {
+    state: AtomicUsize,
+    lock: Mutex<()>,
+    cvar: Condvar,
+}
+
+impl Parker {
+    /// Construct the generic parker. The UNIX parker implementation
+    /// requires this to happen in-place.
+    pub unsafe fn new_in_place(parker: *mut Parker) {
+        parker.write(Parker {
+            state: AtomicUsize::new(EMPTY),
+            lock: Mutex::new(()),
+            cvar: Condvar::new(),
+        });
+    }
+
+    // This implementation doesn't require `unsafe` and `Pin`, but other implementations do.
+    pub unsafe fn park(self: Pin<&Self>) {
+        // If we were previously notified then we consume this notification and
+        // return quickly.
+        if self.state.compare_exchange(NOTIFIED, EMPTY, SeqCst, SeqCst).is_ok() {
+            return;
+        }
+
+        // Otherwise we need to coordinate going to sleep
+        let mut m = self.lock.lock().unwrap();
+        match self.state.compare_exchange(EMPTY, PARKED, SeqCst, SeqCst) {
+            Ok(_) => {}
+            Err(NOTIFIED) => {
+                // We must read here, even though we know it will be `NOTIFIED`.
+                // This is because `unpark` may have been called again since we read
+                // `NOTIFIED` in the `compare_exchange` above. We must perform an
+                // acquire operation that synchronizes with that `unpark` to observe
+                // any writes it made before the call to unpark. To do that we must
+                // read from the write it made to `state`.
+                let old = self.state.swap(EMPTY, SeqCst);
+                assert_eq!(old, NOTIFIED, "park state changed unexpectedly");
+                return;
+            } // should consume this notification, so prohibit spurious wakeups in next park.
+            Err(_) => panic!("inconsistent park state"),
+        }
+        loop {
+            m = self.cvar.wait(m).unwrap();
+            match self.state.compare_exchange(NOTIFIED, EMPTY, SeqCst, SeqCst) {
+                Ok(_) => return, // got a notification
+                Err(_) => {}     // spurious wakeup, go back to sleep
+            }
+        }
+    }
+
+    // This implementation doesn't require `unsafe` and `Pin`, but other implementations do.
+    pub unsafe fn park_timeout(self: Pin<&Self>, dur: Duration) {
+        // Like `park` above we have a fast path for an already-notified thread, and
+        // afterwards we start coordinating for a sleep.
+        // return quickly.
+        if self.state.compare_exchange(NOTIFIED, EMPTY, SeqCst, SeqCst).is_ok() {
+            return;
+        }
+        let m = self.lock.lock().unwrap();
+        match self.state.compare_exchange(EMPTY, PARKED, SeqCst, SeqCst) {
+            Ok(_) => {}
+            Err(NOTIFIED) => {
+                // We must read again here, see `park`.
+                let old = self.state.swap(EMPTY, SeqCst);
+                assert_eq!(old, NOTIFIED, "park state changed unexpectedly");
+                return;
+            } // should consume this notification, so prohibit spurious wakeups in next park.
+            Err(_) => panic!("inconsistent park_timeout state"),
+        }
+
+        // Wait with a timeout, and if we spuriously wake up or otherwise wake up
+        // from a notification we just want to unconditionally set the state back to
+        // empty, either consuming a notification or un-flagging ourselves as
+        // parked.
+        let (_m, _result) = self.cvar.wait_timeout(m, dur).unwrap();
+        match self.state.swap(EMPTY, SeqCst) {
+            NOTIFIED => {} // got a notification, hurray!
+            PARKED => {}   // no notification, alas
+            n => panic!("inconsistent park_timeout state: {n}"),
+        }
+    }
+
+    // This implementation doesn't require `Pin`, but other implementations do.
+    pub fn unpark(self: Pin<&Self>) {
+        // To ensure the unparked thread will observe any writes we made
+        // before this call, we must perform a release operation that `park`
+        // can synchronize with. To do that we must write `NOTIFIED` even if
+        // `state` is already `NOTIFIED`. That is why this must be a swap
+        // rather than a compare-and-swap that returns if it reads `NOTIFIED`
+        // on failure.
+        match self.state.swap(NOTIFIED, SeqCst) {
+            EMPTY => return,    // no one was waiting
+            NOTIFIED => return, // already unparked
+            PARKED => {}        // gotta go wake someone up
+            _ => panic!("inconsistent state in unpark"),
+        }
+
+        // There is a period between when the parked thread sets `state` to
+        // `PARKED` (or last checked `state` in the case of a spurious wake
+        // up) and when it actually waits on `cvar`. If we were to notify
+        // during this period it would be ignored and then when the parked
+        // thread went to sleep it would never wake up. Fortunately, it has
+        // `lock` locked at this stage so we can acquire `lock` to wait until
+        // it is ready to receive the notification.
+        //
+        // Releasing `lock` before the call to `notify_one` means that when the
+        // parked thread wakes it doesn't get woken only to have to wait for us
+        // to release `lock`.
+        drop(self.lock.lock().unwrap());
+        self.cvar.notify_one()
+    }
+}

library/std/src/sys/sync/thread_parking/mod.rs

@@ -1,6 +1,6 @@
 cfg_if::cfg_if! {
     if #[cfg(any(
-        all(target_os = "windows", not(target_vendor = "win7")),
+        all(target_os = "windows", not(any(target_vendor = "win7", target_vendor = "rust9x"))),
         target_os = "linux",
         target_os = "android",
         all(target_arch = "wasm32", target_feature = "atomics"),
@@ -22,6 +22,12 @@ cfg_if::cfg_if! {
     } else if #[cfg(target_vendor = "win7")] {
         mod windows7;
         pub use windows7::Parker;
+    } else if #[cfg(target_vendor = "rust9x")] {
+        mod futex;
+        mod windowsxp;
+        mod generic;
+        mod rust9x;
+        pub use rust9x::Parker;
     } else if #[cfg(all(target_vendor = "apple", not(miri)))] {
         // Doesn't work in Miri, see <https://github.com/rust-lang/miri/issues/2589>.
         mod darwin;

library/std/src/sys/sync/thread_parking/rust9x.rs

@@ -0,0 +1,83 @@
+use crate::mem::ManuallyDrop;
+use crate::pin::Pin;
+use crate::sys::compat::thread_parking::{ThreadParkingImpl, thread_parking_impl};
+use crate::time::Duration;
+
+pub union Parker {
+    futex: ManuallyDrop<super::futex::Parker>,
+    keyed_event: ManuallyDrop<super::windowsxp::Parker>,
+    generic: ManuallyDrop<super::generic::Parker>,
+}
+
+impl Parker {
+    pub unsafe fn new_in_place(parker: *mut Parker) {
+        let impl_ = thread_parking_impl();
+
+        match impl_ {
+            ThreadParkingImpl::Futex => unsafe {
+                super::futex::Parker::new_in_place(&raw mut (*(*parker).futex))
+            },
+            ThreadParkingImpl::KeyedEvent => unsafe {
+                super::windowsxp::Parker::new_in_place(&raw mut (*(*parker).keyed_event))
+            },
+            ThreadParkingImpl::Generic => unsafe {
+                super::generic::Parker::new_in_place(&raw mut (*(*parker).generic))
+            },
+        }
+    }
+
+    pub unsafe fn park(self: Pin<&Self>) {
+        let impl_ = thread_parking_impl();
+
+        match impl_ {
+            ThreadParkingImpl::Futex => unsafe { self.map_unchecked(|p| &*p.futex).park() },
+            ThreadParkingImpl::KeyedEvent => unsafe {
+                self.map_unchecked(|p| &*p.keyed_event).park()
+            },
+            ThreadParkingImpl::Generic => unsafe { self.map_unchecked(|p| &*p.generic).park() },
+        }
+    }
+
+    pub unsafe fn park_timeout(self: Pin<&Self>, timeout: Duration) {
+        let impl_ = thread_parking_impl();
+
+        match impl_ {
+            ThreadParkingImpl::Futex => unsafe {
+                self.map_unchecked(|p| &*p.futex).park_timeout(timeout)
+            },
+            ThreadParkingImpl::KeyedEvent => unsafe {
+                self.map_unchecked(|p| &*p.keyed_event).park_timeout(timeout)
+            },
+            ThreadParkingImpl::Generic => unsafe {
+                self.map_unchecked(|p| &*p.generic).park_timeout(timeout)
+            },
+        }
+    }
+
+    pub fn unpark(self: Pin<&Self>) {
+        let impl_ = thread_parking_impl();
+
+        match impl_ {
+            ThreadParkingImpl::Futex => unsafe {
+                self.map_unchecked(|p| &*p.futex).unpark();
+            },
+            ThreadParkingImpl::KeyedEvent => unsafe {
+                self.map_unchecked(|p| &*p.keyed_event).unpark()
+            },
+            ThreadParkingImpl::Generic => unsafe { self.map_unchecked(|p| &*p.generic).unpark() },
+        }
+    }
+}
+
+impl Drop for Parker {
+    fn drop(&mut self) {
+        let impl_ = thread_parking_impl();
+
+        match impl_ {
+            ThreadParkingImpl::Futex | ThreadParkingImpl::KeyedEvent => {
+                // these don't require any cleanup
+            }
+            ThreadParkingImpl::Generic => unsafe { ManuallyDrop::drop(&mut self.generic) },
+        }
+    }
+}