1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
|
use crate::sync::atomic::{
AtomicI32,
Ordering::{Acquire, Relaxed, Release},
};
use crate::sys::futex::{futex_wait, futex_wake, futex_wake_all};
use crate::time::Duration;
pub type MovableMutex = Mutex;
pub type MovableCondvar = Condvar;
pub struct Mutex {
/// 0: unlocked
/// 1: locked, no other threads waiting
/// 2: locked, and other threads waiting (contended)
futex: AtomicI32,
}
impl Mutex {
pub const fn new() -> Self {
Self { futex: AtomicI32::new(0) }
}
#[inline]
pub unsafe fn init(&mut self) {}
#[inline]
pub unsafe fn destroy(&self) {}
#[inline]
pub unsafe fn try_lock(&self) -> bool {
self.futex.compare_exchange(0, 1, Acquire, Relaxed).is_ok()
}
#[inline]
pub unsafe fn lock(&self) {
if self.futex.compare_exchange(0, 1, Acquire, Relaxed).is_err() {
self.lock_contended();
}
}
fn lock_contended(&self) {
loop {
// Put the lock in contended state, if it wasn't already.
if self.futex.swap(2, Acquire) == 0 {
// It was unlocked, so we just locked it.
return;
}
// Wait for the futex to change state.
futex_wait(&self.futex, 2, None);
}
}
#[inline]
pub unsafe fn unlock(&self) {
if self.futex.swap(0, Release) == 2 {
// We only wake up one thread. When that thread locks the mutex, it
// will mark the mutex as contended (2) (see lock_contended above),
// which makes sure that any other waiting threads will also be
// woken up eventually.
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: AtomicI32,
}
impl Condvar {
#[inline]
pub const fn new() -> Self {
Self { futex: AtomicI32::new(0) }
}
#[inline]
pub unsafe fn init(&mut self) {}
#[inline]
pub unsafe fn destroy(&self) {}
// All the memory orderings here are `Relaxed`,
// because synchronization is done by unlocking and locking the mutex.
#[inline]
pub unsafe fn notify_one(&self) {
self.futex.fetch_add(1, Relaxed);
futex_wake(&self.futex);
}
#[inline]
pub unsafe fn notify_all(&self) {
self.futex.fetch_add(1, Relaxed);
futex_wake_all(&self.futex);
}
#[inline]
pub unsafe fn wait(&self, mutex: &Mutex) {
self.wait_optional_timeout(mutex, None);
}
#[inline]
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 {
// Check 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
}
}
|
