relibc/sync/
pthread_mutex.rs

1use core::{
2    cell::Cell,
3    sync::atomic::{AtomicU32 as AtomicUint, Ordering},
4};
5
6use crate::{
7    error::Errno,
8    header::{errno::*, pthread::*, time::timespec},
9};
10
11use crate::platform::{Pal, Sys, types::c_int};
12
13use super::FutexWaitResult;
14
15pub struct RlctMutex {
16    // Actual locking word.
17    inner: AtomicUint,
18    recursive_count: AtomicUint,
19
20    ty: Ty,
21    robust: bool,
22}
23
24const STATE_UNLOCKED: u32 = 0;
25const WAITING_BIT: u32 = 1 << 31;
26const INDEX_MASK: u32 = !WAITING_BIT;
27
28// TODO: Lower limit is probably better.
29const RECURSIVE_COUNT_MAX_INCLUSIVE: u32 = u32::MAX;
30// TODO: How many spins should we do before it becomes more time-economical to enter kernel mode
31// via futexes?
32const SPIN_COUNT: usize = 0;
33
34impl RlctMutex {
35    pub(crate) fn new(attr: &RlctMutexAttr) -> Result<Self, Errno> {
36        let RlctMutexAttr {
37            prioceiling,
38            protocol,
39            pshared: _,
40            robust,
41            ty,
42        } = *attr;
43
44        Ok(Self {
45            inner: AtomicUint::new(STATE_UNLOCKED),
46            recursive_count: AtomicUint::new(0),
47            robust: match robust {
48                PTHREAD_MUTEX_STALLED => false,
49                PTHREAD_MUTEX_ROBUST => true,
50
51                _ => return Err(Errno(EINVAL)),
52            },
53            ty: match ty {
54                PTHREAD_MUTEX_DEFAULT => Ty::Def,
55                PTHREAD_MUTEX_ERRORCHECK => Ty::Errck,
56                PTHREAD_MUTEX_RECURSIVE => Ty::Recursive,
57                PTHREAD_MUTEX_NORMAL => Ty::Normal,
58
59                _ => return Err(Errno(EINVAL)),
60            },
61        })
62    }
63    pub fn prioceiling(&self) -> Result<c_int, Errno> {
64        todo_skip!(0, "pthread_getprioceiling: not implemented");
65        Ok(0)
66    }
67    pub fn replace_prioceiling(&self, _: c_int) -> Result<c_int, Errno> {
68        todo_skip!(0, "pthread_setprioceiling: not implemented");
69        Ok(0)
70    }
71    pub fn make_consistent(&self) -> Result<(), Errno> {
72        todo_skip!(0, "pthread robust mutexes: not implemented");
73        Ok(())
74    }
75    fn lock_inner(&self, deadline: Option<&timespec>) -> Result<(), Errno> {
76        let this_thread = os_tid_invalid_after_fork();
77
78        //let mut spins_left = SPIN_COUNT;
79
80        loop {
81            let result = self.inner.compare_exchange_weak(
82                STATE_UNLOCKED,
83                this_thread,
84                Ordering::Acquire,
85                Ordering::Relaxed,
86            );
87
88            match result {
89                // CAS succeeded
90                Ok(_) => {
91                    if self.ty == Ty::Recursive {
92                        self.increment_recursive_count()?;
93                    }
94                    return Ok(());
95                }
96                // CAS failed, but the mutex was recursive and we already own the lock.
97                Err(thread) if thread & INDEX_MASK == this_thread && self.ty == Ty::Recursive => {
98                    self.increment_recursive_count()?;
99                    return Ok(());
100                }
101                // CAS failed, but the mutex was error-checking and we already own the lock.
102                Err(thread) if thread & INDEX_MASK == this_thread && self.ty == Ty::Errck => {
103                    return Err(Errno(EAGAIN));
104                }
105                // CAS spuriously failed, simply retry the CAS. TODO: Use core::hint::spin_loop()?
106                Err(thread) if thread & INDEX_MASK == 0 => {
107                    continue;
108                }
109                // CAS failed because some other thread owned the lock. We must now wait.
110                Err(thread) => {
111                    /*if spins_left > 0 {
112                        // TODO: Faster to spin trying to load the flag, compared to CAS?
113                        spins_left -= 1;
114                        core::hint::spin_loop();
115                        continue;
116                    }
117
118                    spins_left = SPIN_COUNT;
119
120                    let inner = self.inner.fetch_or(WAITING_BIT, Ordering::Relaxed);
121
122                    if inner == STATE_UNLOCKED {
123                        continue;
124                    }*/
125
126                    // If the mutex is not robust, simply futex_wait until unblocked.
127                    //crate::sync::futex_wait(&self.inner, inner | WAITING_BIT, None);
128                    if crate::sync::futex_wait(&self.inner, thread, deadline)
129                        == FutexWaitResult::TimedOut
130                    {
131                        return Err(Errno(ETIMEDOUT));
132                    }
133                }
134            }
135        }
136    }
137    pub fn lock(&self) -> Result<(), Errno> {
138        self.lock_inner(None)
139    }
140    pub fn lock_with_timeout(&self, deadline: &timespec) -> Result<(), Errno> {
141        self.lock_inner(Some(deadline))
142    }
143    fn increment_recursive_count(&self) -> Result<(), Errno> {
144        // We don't have to worry about asynchronous signals here, since pthread_mutex_trylock
145        // is not async-signal-safe.
146        //
147        // TODO: Maybe just use Cell? Send/Sync doesn't matter much anyway, and will be
148        // protected by the lock itself anyway.
149
150        let prev_recursive_count = self.recursive_count.load(Ordering::Relaxed);
151
152        if prev_recursive_count == RECURSIVE_COUNT_MAX_INCLUSIVE {
153            return Err(Errno(EAGAIN));
154        }
155
156        self.recursive_count
157            .store(prev_recursive_count + 1, Ordering::Relaxed);
158
159        Ok(())
160    }
161    pub fn try_lock(&self) -> Result<(), Errno> {
162        let this_thread = os_tid_invalid_after_fork();
163
164        // TODO: If recursive, omitting CAS may be faster if it is already owned by this thread.
165        let result = self.inner.compare_exchange(
166            STATE_UNLOCKED,
167            this_thread,
168            Ordering::Acquire,
169            Ordering::Relaxed,
170        );
171
172        if self.ty == Ty::Recursive {
173            match result {
174                Err(index) if index & INDEX_MASK != this_thread => return Err(Errno(EBUSY)),
175                _ => (),
176            }
177
178            self.increment_recursive_count()?;
179
180            return Ok(());
181        }
182
183        match result {
184            Ok(_) => Ok(()),
185            Err(index) if index & INDEX_MASK == this_thread && self.ty == Ty::Errck => {
186                Err(Errno(EDEADLK))
187            }
188            Err(_) => Err(Errno(EBUSY)),
189        }
190    }
191    // Safe because we are not protecting any data.
192    pub fn unlock(&self) -> Result<(), Errno> {
193        if self.robust || matches!(self.ty, Ty::Recursive | Ty::Errck) {
194            if self.inner.load(Ordering::Relaxed) & INDEX_MASK != os_tid_invalid_after_fork() {
195                return Err(Errno(EPERM));
196            }
197
198            // TODO: Is this fence correct?
199            core::sync::atomic::fence(Ordering::Acquire);
200        }
201
202        if self.ty == Ty::Recursive {
203            let next = self.recursive_count.load(Ordering::Relaxed) - 1;
204            self.recursive_count.store(next, Ordering::Relaxed);
205
206            if next > 0 {
207                return Ok(());
208            }
209        }
210
211        self.inner.store(STATE_UNLOCKED, Ordering::Release);
212        crate::sync::futex_wake(&self.inner, i32::MAX);
213        /*let was_waiting = self.inner.swap(STATE_UNLOCKED, Ordering::Release) & WAITING_BIT != 0;
214
215        if was_waiting {
216            let _ = crate::sync::futex_wake(&self.inner, 1);
217        }*/
218
219        Ok(())
220    }
221}
222
223#[repr(u8)]
224#[derive(PartialEq)]
225enum Ty {
226    // The only difference between PTHREAD_MUTEX_NORMAL and PTHREAD_MUTEX_DEFAULT appears to be
227    // that "normal" mutexes deadlock if locked multiple times on the same thread, whereas
228    // "default" mutexes are UB in that case. So we can treat them as being the same type.
229    Normal,
230    Def,
231
232    Errck,
233    Recursive,
234}
235
236// Children after fork can only call async-signal-safe functions until they exec.
237#[thread_local]
238static CACHED_OS_TID_INVALID_AFTER_FORK: Cell<u32> = Cell::new(0);
239
240// Assumes TIDs are unique between processes, which I only know is true for Redox.
241fn os_tid_invalid_after_fork() -> u32 {
242    // TODO: Coordinate better if using shared == PTHREAD_PROCESS_SHARED, with up to 2^32 separate
243    // threads within possibly distinct processes, using the mutex. OS thread IDs on Redox are
244    // pointer-sized, but relibc and POSIX uses int everywhere.
245
246    let value = CACHED_OS_TID_INVALID_AFTER_FORK.get();
247
248    if value == 0 {
249        let tid = Sys::gettid();
250
251        assert_ne!(tid, -1, "failed to obtain current thread ID");
252
253        CACHED_OS_TID_INVALID_AFTER_FORK.set(tid as u32);
254
255        tid as u32
256    } else {
257        value
258    }
259}
relibc/sync/pthread_mutex.rs

relibc/sync/
pthread_mutex.rs
