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
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
use crate::spinlock::SpinLock;
use crate::remedy;
use crate::remedy::Remedy;
use std::sync::{Condvar, Mutex};
use std::time::Duration;
pub trait Monitor<S: ?Sized> {
fn enter<F: FnMut(&mut S) -> Directive>(&self, f: F);
/// Invokes the given closure exactly once, supplying the encapsulated state for alteration
/// or observation.
///
/// When there is no need to wait for or notify other threads, this method is preferred
/// over [`enter`](Self::enter), as it takes a stronger form of closure that is evaluated once.
///
/// # Examples
/// ```
/// use anode::monitor::{Monitor, SpeculativeMonitor};
/// struct State {
/// foo: u64
/// }
/// let monitor = SpeculativeMonitor::new(State { foo: 42 });
/// let mut foo = None;
/// monitor.alter(|state| {
/// foo = Some(state.foo);
/// state.foo *= 1;
/// });
/// assert_eq!(Some(42), foo);
/// ```
fn alter<F: FnOnce(&mut S)>(&self, f: F);
/// Performs some computation over the encapsulated state. It may be as simple as
/// extracting a value.
///
/// # Examples
/// ```
/// use anode::monitor::{Monitor, SpeculativeMonitor};
/// struct State {
/// foo: u64,
/// bar: u64,
/// }
/// let monitor = SpeculativeMonitor::new(State { foo: 42, bar: 24 });
/// let foo = monitor.compute(|state| state.foo + state.bar);
/// assert_eq!(66, foo);
/// ```
fn compute<T, F: FnOnce(&S) -> T>(&self, f: F) -> T {
let mut val = None;
self.alter(|state| {
val = Some(f(state));
});
val.unwrap() // guaranteed to be initialised
}
}
pub enum Directive {
Return,
Wait(Duration),
NotifyOne,
NotifyAll
}
struct Tracker<S: ?Sized> {
waiting: u32,
data: S,
}
pub struct SpeculativeMonitor<S: ?Sized> {
mutex: Mutex<()>,
cond: Condvar,
tracker: SpinLock<Tracker<S>>,
}
impl<S> SpeculativeMonitor<S> {
#[inline(always)]
pub fn new(s: S) -> Self {
Self {
tracker: SpinLock::new(Tracker {
data: s,
waiting: 0,
}),
mutex: Mutex::new(()),
cond: Default::default(),
}
}
}
impl<S: ?Sized> SpeculativeMonitor<S> {
pub fn num_waiting(&self) -> u32 {
self.tracker.lock().waiting
}
}
impl<S: ?Sized> Monitor<S> for SpeculativeMonitor<S> {
#[inline(always)]
fn enter<F: FnMut(&mut S) -> Directive>(&self, mut f: F) {
let mut mutex_guard = None;
let mut woken = false;
loop {
let mut spin_guard = self.tracker.lock();
if woken {
woken = false;
spin_guard.waiting -= 1;
}
let data = &mut spin_guard.data;
let directive = f(data);
match directive {
Directive::Return => {
return;
}
Directive::Wait(duration) => {
match mutex_guard.take() {
None => {
// println!("init lock");
drop(spin_guard);
mutex_guard = Some(self.mutex.lock().remedy());
}
Some(guard) => {
spin_guard.waiting += 1;
drop(spin_guard);
let (guard, timed_out) =
remedy::cond_wait_remedy(&self.cond, guard, duration);
if timed_out {
// println!("timed out");
let mut spin_guard = self.tracker.lock();
spin_guard.waiting -= 1;
return;
} else {
// println!("keep going");
mutex_guard = Some(guard);
woken = true;
}
}
}
}
Directive::NotifyOne | Directive::NotifyAll => {
if spin_guard.waiting > 0 {
drop(spin_guard);
match mutex_guard.take() {
None => {
// println!("init lock");
mutex_guard = Some(self.mutex.lock().remedy());
}
Some(guard) => {
drop(guard);
match directive {
Directive::NotifyOne => {
self.cond.notify_one();
}
Directive::NotifyAll => {
self.cond.notify_all();
}
_ => unreachable!()
}
return;
}
}
} else {
return;
}
}
}
}
}
#[inline(always)]
fn alter<F: FnOnce(&mut S)>(&self, f: F) {
let mut spin_guard = self.tracker.lock();
f(&mut spin_guard.data);
}
}
#[cfg(test)]
mod tests;