relibc/sync/
cond.rs

1// Used design from https://www.remlab.net/op/futex-condvar.shtml
2
3use crate::{
4    error::Errno,
5    header::{
6        errno::{EINVAL, ETIMEDOUT},
7        pthread::*,
8        time::{CLOCK_MONOTONIC, CLOCK_REALTIME, timespec, timespec_realtime_to_monotonic},
9    },
10    platform::types::clockid_t,
11};
12
13use core::sync::atomic::{AtomicU32 as AtomicUint, Ordering};
14
15#[derive(Clone, Copy)]
16pub struct CondAttr {
17    pub clock: clockid_t,
18    pub pshared: i32,
19}
20
21impl Default for CondAttr {
22    fn default() -> Self {
23        Self {
24            // defaults according to POSIX
25            clock: CLOCK_REALTIME,            // for timedwait
26            pshared: PTHREAD_PROCESS_PRIVATE, // TODO
27        }
28    }
29}
30
31pub struct Cond {
32    cur: AtomicUint,
33    prev: AtomicUint,
34}
35
36type Result<T, E = Errno> = core::result::Result<T, E>;
37
38impl Default for Cond {
39    fn default() -> Self {
40        Self::new()
41    }
42}
43
44impl Cond {
45    pub fn new() -> Self {
46        Self {
47            cur: AtomicUint::new(0),
48            prev: AtomicUint::new(0),
49        }
50    }
51    fn wake(&self, count: i32) -> Result<(), Errno> {
52        // This is formally correct as long as we don't have more than u32::MAX threads.
53        let prev = self.prev.load(Ordering::Relaxed);
54        self.cur.store(prev.wrapping_add(1), Ordering::Relaxed);
55
56        crate::sync::futex_wake(&self.cur, count);
57        Ok(())
58    }
59    pub fn broadcast(&self) -> Result<(), Errno> {
60        self.wake(i32::MAX)
61    }
62    pub fn signal(&self) -> Result<(), Errno> {
63        self.broadcast()
64        //self.wake(1)
65    }
66    pub fn clockwait(
67        &self,
68        mutex: &RlctMutex,
69        timeout: &timespec,
70        clock_id: clockid_t,
71    ) -> Result<(), Errno> {
72        let relative = match clock_id {
73            // FUTEX expect monotonic clock
74            CLOCK_MONOTONIC => timeout.clone(),
75            CLOCK_REALTIME => timespec_realtime_to_monotonic(timeout)?,
76            _ => return Err(Errno(EINVAL)),
77        };
78
79        self.wait_inner(mutex, Some(&relative))
80    }
81    pub fn timedwait(&self, mutex: &RlctMutex, timeout: &timespec) -> Result<(), Errno> {
82        // TODO: The clock can be other than CLOCK_REALTIME depends on CondAttr
83        self.clockwait(mutex, timeout, CLOCK_REALTIME)
84    }
85    fn wait_inner(&self, mutex: &RlctMutex, timeout: Option<&timespec>) -> Result<(), Errno> {
86        self.wait_inner_generic(|| mutex.unlock(), || mutex.lock(), timeout)
87    }
88    pub fn wait_inner_typedmutex<'lock, T>(
89        &self,
90        guard: crate::sync::MutexGuard<'lock, T>,
91    ) -> crate::sync::MutexGuard<'lock, T> {
92        let mut newguard = None;
93        let lock = guard.mutex;
94        self.wait_inner_generic(
95            move || {
96                drop(guard);
97                Ok(())
98            },
99            || {
100                newguard = Some(lock.lock());
101                Ok(())
102            },
103            None,
104        )
105        .unwrap();
106        newguard.unwrap()
107    }
108    // TODO: FUTEX_REQUEUE
109    fn wait_inner_generic(
110        &self,
111        unlock: impl FnOnce() -> Result<()>,
112        lock: impl FnOnce() -> Result<()>,
113        deadline: Option<&timespec>,
114    ) -> Result<(), Errno> {
115        // TODO: Error checking for certain types (i.e. robust and errorcheck) of mutexes, e.g. if the
116        // mutex is not locked.
117        let current = self.cur.load(Ordering::Relaxed);
118        self.prev.store(current, Ordering::Relaxed);
119
120        unlock()?;
121        let futex_r = crate::sync::futex_wait(&self.cur, current, deadline);
122        lock()?;
123
124        match futex_r {
125            super::FutexWaitResult::Waited => Ok(()),
126            super::FutexWaitResult::Stale => Ok(()),
127            super::FutexWaitResult::TimedOut => Err(Errno(ETIMEDOUT)),
128        }
129    }
130    pub fn wait(&self, mutex: &RlctMutex) -> Result<(), Errno> {
131        self.wait_inner(mutex, None)
132    }
133}
relibc/sync/cond.rs

relibc/sync/
cond.rs
