Move condvar impl to a mostly-safe module.

src/sync/cond.rs

@@ -0,0 +1,57 @@
+use crate::header::pthread::*;
+use crate::header::bits_pthread::*;
+use crate::header::time::timespec;
+use crate::pthread::Errno;
+
+use core::sync::atomic::{AtomicU32 as AtomicUint, Ordering};
+
+pub struct Cond {
+    cur: AtomicUint,
+    prev: AtomicUint,
+}
+impl Cond {
+    pub fn new() -> Self {
+        Self{
+            cur: AtomicUint::new(0),
+            prev: AtomicUint::new(0),
+        }
+    }
+    fn wake(&self, count: i32) -> Result<(), Errno> {
+        // This is formally correct as long as we don't have more than u32::MAX threads.
+        let prev = self.prev.load(Ordering::Relaxed);
+        self.cur.store(prev.wrapping_add(1), Ordering::Relaxed);
+
+        crate::sync::futex_wake(&self.cur, count);
+        Ok(())
+    }
+    pub fn broadcast(&self) -> Result<(), Errno> {
+        self.wake(i32::MAX)
+    }
+    pub fn signal(&self) -> Result<(), Errno> {
+        self.wake(1)
+    }
+    // TODO: Safe version using RlctMutexGuard?
+    pub unsafe fn timedwait(&self, mutex_ptr: *mut pthread_mutex_t, timeout: Option<&timespec>) -> Result<(), Errno> {
+        // TODO: Error checking for certain types (i.e. robust and errorcheck) of mutexes, e.g. if the
+        // mutex is not locked.
+        let current = self.cur.load(Ordering::Relaxed);
+        self.prev.store(current, Ordering::Relaxed); // TODO: ordering?
+
+        pthread_mutex_unlock(mutex_ptr);
+
+        match timeout {
+            Some(timeout) => {
+                crate::sync::futex_wait(&self.cur, current, timespec::subtract(*timeout, crate::sync::rttime()).as_ref());
+                pthread_mutex_timedlock(mutex_ptr, timespec::subtract(*timeout, crate::sync::rttime()).as_ref().map_or(core::ptr::null(), |r| r as *const timespec));
+            }
+            None => {
+                crate::sync::futex_wait(&self.cur, current, None);
+                pthread_mutex_lock(mutex_ptr);
+            }
+        }
+        Ok(())
+    }
+    pub unsafe fn wait(&self, mutex_ptr: *mut pthread_mutex_t) -> Result<(), Errno> {
+        self.timedwait(mutex_ptr, None)
+    }
+}

src/sync/mod.rs

@@ -1,4 +1,5 @@
 pub mod barrier;
+pub mod cond;
 pub mod mutex;
 pub mod once;
 pub mod semaphore;
@@ -15,6 +16,7 @@ use crate::{
     platform::{types::*, Pal, Sys},
 };
 use core::{
+    mem::MaybeUninit,
     ops::Deref,
     sync::atomic::{self, AtomicI32, AtomicU32, AtomicI32 as AtomicInt},
 };
@@ -76,6 +78,18 @@ pub fn futex_wake(atomic: &impl FutexAtomicTy, n: i32) -> usize {
 pub fn futex_wait<T: FutexAtomicTy>(atomic: &T, value: T::Ty, timeout_opt: Option<&timespec>) -> bool {
     unsafe { futex_wait_ptr(atomic.ptr(), value, timeout_opt) }
 }
+
+pub fn rttime() -> timespec {
+    unsafe {
+        let mut time = MaybeUninit::uninit();
+
+        // TODO: Handle error
+        Sys::clock_gettime(crate::header::time::CLOCK_REALTIME, time.as_mut_ptr());
+
+        time.assume_init()
+    }
+}
+
 pub fn wait_until_generic<F1, F2>(word: &AtomicInt, attempt: F1, mark_long: F2, long: c_int)
 where
     F1: Fn(&AtomicInt) -> AttemptStatus,