Rollup merge of rust-lang#98707 - joboet:fuchsia_locks, r=m-ou-se

std: use futex-based locks on Fuchsia

This switches `Condvar` and `RwLock` to the futex-based implementation currently used on Linux and some BSDs. Additionally, `Mutex` now has its own, priority-inheriting implementation based on the mutex in Fuchsia's `libsync`. It differs from the original in that it panics instead of aborting when reentrant locking is detected.

```@rustbot``` ping fuchsia
r? ```@m-ou-se```

library/std/src/sys/unix/futex.rs

@@ -240,25 +240,34 @@ pub fn futex_wake_all(futex: &AtomicU32) {
 }
 
 #[cfg(target_os = "fuchsia")]
-mod zircon {
-    type zx_time_t = i64;
-    type zx_futex_t = crate::sync::atomic::AtomicU32;
-    type zx_handle_t = u32;
-    type zx_status_t = i32;
+pub mod zircon {
+    pub type zx_futex_t = crate::sync::atomic::AtomicU32;
+    pub type zx_handle_t = u32;
+    pub type zx_status_t = i32;
+    pub type zx_time_t = i64;
 
     pub const ZX_HANDLE_INVALID: zx_handle_t = 0;
-    pub const ZX_ERR_TIMED_OUT: zx_status_t = -21;
+
     pub const ZX_TIME_INFINITE: zx_time_t = zx_time_t::MAX;
 
+    pub const ZX_OK: zx_status_t = 0;
+    pub const ZX_ERR_INVALID_ARGS: zx_status_t = -10;
+    pub const ZX_ERR_BAD_HANDLE: zx_status_t = -11;
+    pub const ZX_ERR_WRONG_TYPE: zx_status_t = -12;
+    pub const ZX_ERR_BAD_STATE: zx_status_t = -20;
+    pub const ZX_ERR_TIMED_OUT: zx_status_t = -21;
+
     extern "C" {
+        pub fn zx_clock_get_monotonic() -> zx_time_t;
         pub fn zx_futex_wait(
             value_ptr: *const zx_futex_t,
             current_value: zx_futex_t,
             new_futex_owner: zx_handle_t,
             deadline: zx_time_t,
         ) -> zx_status_t;
         pub fn zx_futex_wake(value_ptr: *const zx_futex_t, wake_count: u32) -> zx_status_t;
-        pub fn zx_clock_get_monotonic() -> zx_time_t;
+        pub fn zx_futex_wake_single_owner(value_ptr: *const zx_futex_t) -> zx_status_t;
+        pub fn zx_thread_self() -> zx_handle_t;
     }
 }
 
@@ -287,3 +296,8 @@ pub fn futex_wake(futex: &AtomicU32) -> bool {
     unsafe { zircon::zx_futex_wake(futex, 1) };
     false
 }
+
+#[cfg(target_os = "fuchsia")]
+pub fn futex_wake_all(futex: &AtomicU32) {
+    unsafe { zircon::zx_futex_wake(futex, u32::MAX) };
+}

library/std/src/sys/unix/locks/fuchsia_mutex.rs

@@ -0,0 +1,165 @@
+//! A priority inheriting mutex for Fuchsia.
+//!
+//! This is a port of the [mutex in Fuchsia's libsync]. Contrary to the original,
+//! it does not abort the process when reentrant locking is detected, but deadlocks.
+//!
+//! Priority inheritance is achieved by storing the owning thread's handle in an
+//! atomic variable. Fuchsia's futex operations support setting an owner thread
+//! for a futex, which can boost that thread's priority while the futex is waited
+//! upon.
+//!
+//! libsync is licenced under the following BSD-style licence:
+//!
+//! Copyright 2016 The Fuchsia Authors.
+//!
+//! Redistribution and use in source and binary forms, with or without
+//! modification, are permitted provided that the following conditions are
+//! met:
+//!
+//!    * Redistributions of source code must retain the above copyright
+//!      notice, this list of conditions and the following disclaimer.
+//!    * Redistributions in binary form must reproduce the above
+//!      copyright notice, this list of conditions and the following
+//!      disclaimer in the documentation and/or other materials provided
+//!      with the distribution.
+//!
+//! THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+//! "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+//! LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+//! A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+//! OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+//! SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+//! LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+//! DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+//! THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+//! (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+//! OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+//!
+//! [mutex in Fuchsia's libsync]: https://cs.opensource.google/fuchsia/fuchsia/+/main:zircon/system/ulib/sync/mutex.c
+
+use crate::sync::atomic::{
+    AtomicU32,
+    Ordering::{Acquire, Relaxed, Release},
+};
+use crate::sys::futex::zircon::{
+    zx_futex_wait, zx_futex_wake_single_owner, zx_handle_t, zx_thread_self, ZX_ERR_BAD_HANDLE,
+    ZX_ERR_BAD_STATE, ZX_ERR_INVALID_ARGS, ZX_ERR_TIMED_OUT, ZX_ERR_WRONG_TYPE, ZX_OK,
+    ZX_TIME_INFINITE,
+};
+
+// The lowest two bits of a `zx_handle_t` are always set, so the lowest bit is used to mark the
+// mutex as contested by clearing it.
+const CONTESTED_BIT: u32 = 1;
+// This can never be a valid `zx_handle_t`.
+const UNLOCKED: u32 = 0;
+
+pub type MovableMutex = Mutex;
+
+pub struct Mutex {
+    futex: AtomicU32,
+}
+
+#[inline]
+fn to_state(owner: zx_handle_t) -> u32 {
+    owner
+}
+
+#[inline]
+fn to_owner(state: u32) -> zx_handle_t {
+    state | CONTESTED_BIT
+}
+
+#[inline]
+fn is_contested(state: u32) -> bool {
+    state & CONTESTED_BIT == 0
+}
+
+#[inline]
+fn mark_contested(state: u32) -> u32 {
+    state & !CONTESTED_BIT
+}
+
+impl Mutex {
+    #[inline]
+    pub const fn new() -> Mutex {
+        Mutex { futex: AtomicU32::new(UNLOCKED) }
+    }
+
+    #[inline]
+    pub unsafe fn init(&mut self) {}
+
+    #[inline]
+    pub unsafe fn try_lock(&self) -> bool {
+        let thread_self = zx_thread_self();
+        self.futex.compare_exchange(UNLOCKED, to_state(thread_self), Acquire, Relaxed).is_ok()
+    }
+
+    #[inline]
+    pub unsafe fn lock(&self) {
+        let thread_self = zx_thread_self();
+        if let Err(state) =
+            self.futex.compare_exchange(UNLOCKED, to_state(thread_self), Acquire, Relaxed)
+        {
+            self.lock_contested(state, thread_self);
+        }
+    }
+
+    #[cold]
+    fn lock_contested(&self, mut state: u32, thread_self: zx_handle_t) {
+        let owned_state = mark_contested(to_state(thread_self));
+        loop {
+            // Mark the mutex as contested if it is not already.
+            let contested = mark_contested(state);
+            if is_contested(state)
+                || self.futex.compare_exchange(state, contested, Relaxed, Relaxed).is_ok()
+            {
+                // The mutex has been marked as contested, wait for the state to change.
+                unsafe {
+                    match zx_futex_wait(
+                        &self.futex,
+                        AtomicU32::new(contested),
+                        to_owner(state),
+                        ZX_TIME_INFINITE,
+                    ) {
+                        ZX_OK | ZX_ERR_BAD_STATE | ZX_ERR_TIMED_OUT => (),
+                        // Note that if a thread handle is reused after its associated thread
+                        // exits without unlocking the mutex, an arbitrary thread's priority
+                        // could be boosted by the wait, but there is currently no way to
+                        // prevent that.
+                        ZX_ERR_INVALID_ARGS | ZX_ERR_BAD_HANDLE | ZX_ERR_WRONG_TYPE => {
+                            panic!(
+                                "either the current thread is trying to lock a mutex it has
+                                already locked, or the previous owner did not unlock the mutex
+                                before exiting"
+                            )
+                        }
+                        error => panic!("unexpected error in zx_futex_wait: {error}"),
+                    }
+                }
+            }
+
+            // The state has changed or a wakeup occured, try to lock the mutex.
+            match self.futex.compare_exchange(UNLOCKED, owned_state, Acquire, Relaxed) {
+                Ok(_) => return,
+                Err(updated) => state = updated,
+            }
+        }
+    }
+
+    #[inline]
+    pub unsafe fn unlock(&self) {
+        if is_contested(self.futex.swap(UNLOCKED, Release)) {
+            // The woken thread will mark the mutex as contested again,
+            // and return here, waking until there are no waiters left,
+            // in which case this is a noop.
+            self.wake();
+        }
+    }
+
+    #[cold]
+    fn wake(&self) {
+        unsafe {
+            zx_futex_wake_single_owner(&self.futex);
+        }
+    }
+}

library/std/src/sys/unix/locks/futex_condvar.rs

@@ -0,0 +1,58 @@
+use super::Mutex;
+use crate::sync::atomic::{AtomicU32, Ordering::Relaxed};
+use crate::sys::futex::{futex_wait, futex_wake, futex_wake_all};
+use crate::time::Duration;
+
+pub type MovableCondvar = Condvar;
+
+pub struct Condvar {
+    // The value of this atomic is simply incremented on every notification.
+    // This is used by `.wait()` to not miss any notifications after
+    // unlocking the mutex and before waiting for notifications.
+    futex: AtomicU32,
+}
+
+impl Condvar {
+    #[inline]
+    pub const fn new() -> Self {
+        Self { futex: AtomicU32::new(0) }
+    }
+
+    // All the memory orderings here are `Relaxed`,
+    // because synchronization is done by unlocking and locking the mutex.
+
+    pub unsafe fn notify_one(&self) {
+        self.futex.fetch_add(1, Relaxed);
+        futex_wake(&self.futex);
+    }
+
+    pub unsafe fn notify_all(&self) {
+        self.futex.fetch_add(1, Relaxed);
+        futex_wake_all(&self.futex);
+    }
+
+    pub unsafe fn wait(&self, mutex: &Mutex) {
+        self.wait_optional_timeout(mutex, None);
+    }
+
+    pub unsafe fn wait_timeout(&self, mutex: &Mutex, timeout: Duration) -> bool {
+        self.wait_optional_timeout(mutex, Some(timeout))
+    }
+
+    unsafe fn wait_optional_timeout(&self, mutex: &Mutex, timeout: Option<Duration>) -> bool {
+        // Examine the notification counter _before_ we unlock the mutex.
+        let futex_value = self.futex.load(Relaxed);
+
+        // Unlock the mutex before going to sleep.
+        mutex.unlock();
+
+        // Wait, but only if there hasn't been any
+        // notification since we unlocked the mutex.
+        let r = futex_wait(&self.futex, futex_value, timeout);
+
+        // Lock the mutex again.
+        mutex.lock();
+
+        r
+    }
+}

library/std/src/sys/unix/locks/futex.rs → library/std/src/sys/unix/locks/futex_mutex.rs

@@ -2,11 +2,9 @@ use crate::sync::atomic::{
     AtomicU32,
     Ordering::{Acquire, Relaxed, Release},
 };
-use crate::sys::futex::{futex_wait, futex_wake, futex_wake_all};
-use crate::time::Duration;
+use crate::sys::futex::{futex_wait, futex_wake};
 
 pub type MovableMutex = Mutex;
-pub type MovableCondvar = Condvar;
 
 pub struct Mutex {
     /// 0: unlocked
@@ -101,55 +99,3 @@ impl Mutex {
         futex_wake(&self.futex);
     }
 }
-
-pub struct Condvar {
-    // The value of this atomic is simply incremented on every notification.
-    // This is used by `.wait()` to not miss any notifications after
-    // unlocking the mutex and before waiting for notifications.
-    futex: AtomicU32,
-}
-
-impl Condvar {
-    #[inline]
-    pub const fn new() -> Self {
-        Self { futex: AtomicU32::new(0) }
-    }
-
-    // All the memory orderings here are `Relaxed`,
-    // because synchronization is done by unlocking and locking the mutex.
-
-    pub unsafe fn notify_one(&self) {
-        self.futex.fetch_add(1, Relaxed);
-        futex_wake(&self.futex);
-    }
-
-    pub unsafe fn notify_all(&self) {
-        self.futex.fetch_add(1, Relaxed);
-        futex_wake_all(&self.futex);
-    }
-
-    pub unsafe fn wait(&self, mutex: &Mutex) {
-        self.wait_optional_timeout(mutex, None);
-    }
-
-    pub unsafe fn wait_timeout(&self, mutex: &Mutex, timeout: Duration) -> bool {
-        self.wait_optional_timeout(mutex, Some(timeout))
-    }
-
-    unsafe fn wait_optional_timeout(&self, mutex: &Mutex, timeout: Option<Duration>) -> bool {
-        // Examine the notification counter _before_ we unlock the mutex.
-        let futex_value = self.futex.load(Relaxed);
-
-        // Unlock the mutex before going to sleep.
-        mutex.unlock();
-
-        // Wait, but only if there hasn't been any
-        // notification since we unlocked the mutex.
-        let r = futex_wait(&self.futex, futex_value, timeout);
-
-        // Lock the mutex again.
-        mutex.lock();
-
-        r
-    }
-}

library/std/src/sys/unix/locks/mod.rs

@@ -7,10 +7,19 @@ cfg_if::cfg_if! {
         target_os = "openbsd",
         target_os = "dragonfly",
     ))] {
-        mod futex;
+        mod futex_mutex;
         mod futex_rwlock;
-        pub(crate) use futex::{Mutex, MovableMutex, MovableCondvar};
+        mod futex_condvar;
+        pub(crate) use futex_mutex::{Mutex, MovableMutex};
         pub(crate) use futex_rwlock::{RwLock, MovableRwLock};
+        pub(crate) use futex_condvar::MovableCondvar;
+    } else if #[cfg(target_os = "fuchsia")] {
+        mod fuchsia_mutex;
+        mod futex_rwlock;
+        mod futex_condvar;
+        pub(crate) use fuchsia_mutex::{Mutex, MovableMutex};
+        pub(crate) use futex_rwlock::{RwLock, MovableRwLock};
+        pub(crate) use futex_condvar::MovableCondvar;
     } else {
         mod pthread_mutex;
         mod pthread_rwlock;