Misc improvements, move barrier to safe module.

src/sync/barrier.rs

@@ -0,0 +1,86 @@
+use core::cmp;
+use core::num::NonZeroU32;
+use core::sync::atomic::{AtomicU32 as AtomicUint, Ordering};
+
+pub struct Barrier {
+    waited_count: AtomicUint,
+    notified_count: AtomicUint,
+    original_count: NonZeroU32,
+}
+
+pub enum WaitResult {
+    Waited,
+    NotifiedAll,
+}
+
+impl Barrier {
+    pub fn new(count: NonZeroU32) -> Self {
+        Self {
+            waited_count: AtomicUint::new(0),
+            notified_count: AtomicUint::new(0),
+            original_count: count,
+        }
+    }
+    pub fn wait(&self) -> WaitResult {
+        // The barrier wait operation can be divided into two parts: (1) incrementing the wait count where
+        // N-1 waiters wait and one notifies the rest, and (2) notifying all threads that have been
+        // waiting.
+
+        let original_count = self.original_count.get();
+
+        loop {
+            let new = self.waited_count.fetch_add(1, Ordering::Acquire) + 1;
+
+            match Ord::cmp(&new, &original_count) {
+                cmp::Ordering::Less => {
+                    // new < original_count, i.e. we were one of the threads that incremented the counter,
+                    // but need to continue waiting for the last waiter to notify the others.
+
+                    loop {
+                        let count = self.waited_count.load(Ordering::Acquire);
+
+                        if count >= original_count { break }
+
+                        let _ = crate::sync::futex_wait(&self.waited_count, count, None);
+                    }
+
+                    // When the required number of threads have called pthread_barrier_wait so waited_count
+                    // >= original_count (should never be able to exceed that value), we can safely reset
+                    // the counter to zero.
+
+                    if self.notified_count.fetch_add(1, Ordering::Relaxed) + 1 >= original_count {
+                        self.waited_count.store(0, Ordering::Relaxed);
+                    }
+
+                    return WaitResult::Waited;
+                }
+                cmp::Ordering::Equal => {
+                    // new == original_count, i.e. we were the one thread doing the last increment, and we
+                    // will be responsible for waking up all other waiters.
+
+                    crate::sync::futex_wake(&self.waited_count, i32::MAX);
+
+                    if self.notified_count.fetch_add(1, Ordering::Relaxed) + 1 >= original_count {
+                        self.waited_count.store(0, Ordering::Relaxed);
+                    }
+
+                    return WaitResult::NotifiedAll;
+                }
+                // FIXME: Starvation?
+                cmp::Ordering::Greater => {
+                    let mut cached = new;
+                    while cached >= original_count {
+                        // new > original_count, i.e. we are waiting on a barrier that is already finished, but
+                        // which has not yet awoken all its waiters and re-initialized the self. The
+                        // simplest way to handle this is to wait for waited_count to return to zero, and
+                        // start over.
+
+                        crate::sync::futex_wait(&self.waited_count, cached, None);
+
+                        cached = self.waited_count.load(Ordering::Acquire);
+                    }
+                }
+            }
+        }
+    }
+}

src/sync/mod.rs

@@ -1,3 +1,4 @@
+pub mod barrier;
 pub mod mutex;
 pub mod once;
 pub mod semaphore;
@@ -15,7 +16,7 @@ use crate::{
 };
 use core::{
     ops::Deref,
-    sync::atomic::{self, AtomicI32 as AtomicInt},
+    sync::atomic::{self, AtomicI32, AtomicU32, AtomicI32 as AtomicInt},
 };
 
 const FUTEX_WAIT: c_int = 0;
@@ -28,19 +29,52 @@ pub enum AttemptStatus {
     Other,
 }
 
-pub unsafe fn futex_wake_ptr(ptr: *mut i32, n: i32) -> usize {
+pub trait FutexTy {
+    fn conv(self) -> i32;
+}
+pub trait FutexAtomicTy {
+    type Ty: FutexTy;
+
+    fn as_mut_ptr(&self) -> *mut Self::Ty;
+}
+impl FutexTy for u32 {
+    fn conv(self) -> i32 {
+        self as i32
+    }
+}
+impl FutexTy for i32 {
+    fn conv(self) -> i32 {
+        self
+    }
+}
+impl FutexAtomicTy for AtomicU32 {
+    type Ty = u32;
+
+    fn as_mut_ptr(&self) -> *mut u32 {
+        AtomicU32::as_mut_ptr(self)
+    }
+}
+impl FutexAtomicTy for AtomicI32 {
+    type Ty = i32;
+
+    fn as_mut_ptr(&self) -> *mut i32 {
+        AtomicI32::as_mut_ptr(self)
+    }
+}
+
+pub unsafe fn futex_wake_ptr(ptr: *mut impl FutexTy, n: i32) -> usize {
     // TODO: unwrap_unchecked?
-    Sys::futex(ptr, FUTEX_WAKE, n, 0).unwrap() as usize
+    Sys::futex(ptr.cast(), FUTEX_WAKE, n, 0).unwrap() as usize
 }
-pub unsafe fn futex_wait_ptr(ptr: *mut i32, value: i32, timeout_opt: Option<&timespec>) -> bool {
+pub unsafe fn futex_wait_ptr<T: FutexTy>(ptr: *mut T, value: T, timeout_opt: Option<&timespec>) -> bool {
     // TODO: unwrap_unchecked?
-    Sys::futex(ptr, FUTEX_WAIT, value, timeout_opt.map_or(0, |t| t as *const _ as usize)) == Ok(0)
+    Sys::futex(ptr.cast(), FUTEX_WAIT, value.conv(), timeout_opt.map_or(0, |t| t as *const _ as usize)) == Ok(0)
 }
-pub fn futex_wake(atomic: &AtomicInt, n: i32) -> usize {
-    unsafe { futex_wake_ptr(atomic.as_ptr(), n) }
+pub fn futex_wake(atomic: &impl FutexAtomicTy, n: i32) -> usize {
+    unsafe { futex_wake_ptr(atomic.as_mut_ptr(), n) }
 }
-pub fn futex_wait(atomic: &AtomicInt, value: i32, timeout_opt: Option<&timespec>) -> bool {
-    unsafe { futex_wait_ptr(atomic.as_ptr(), value, timeout_opt) }
+pub fn futex_wait<T: FutexAtomicTy>(atomic: &T, value: T::Ty, timeout_opt: Option<&timespec>) -> bool {
+    unsafe { futex_wait_ptr(atomic.as_mut_ptr(), value, timeout_opt) }
 }
 pub fn wait_until_generic<F1, F2>(word: &AtomicInt, attempt: F1, mark_long: F2, long: c_int)
 where