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
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
|
#![warn(rust_2018_idioms)]
use std::sync::Arc;
use std::task::Poll;
use futures::future::FutureExt;
use futures::stream;
use futures::stream::StreamExt;
use tokio::sync::{Barrier, RwLock};
use tokio_test::task::spawn;
use tokio_test::{assert_pending, assert_ready};
#[test]
fn into_inner() {
let rwlock = RwLock::new(42);
assert_eq!(rwlock.into_inner(), 42);
}
// multiple reads should be Ready
#[test]
fn read_shared() {
let rwlock = RwLock::new(100);
let mut t1 = spawn(rwlock.read());
let _g1 = assert_ready!(t1.poll());
let mut t2 = spawn(rwlock.read());
assert_ready!(t2.poll());
}
// When there is an active shared owner, exclusive access should not be possible
#[test]
fn write_shared_pending() {
let rwlock = RwLock::new(100);
let mut t1 = spawn(rwlock.read());
let _g1 = assert_ready!(t1.poll());
let mut t2 = spawn(rwlock.write());
assert_pending!(t2.poll());
}
// When there is an active exclusive owner, subsequent exclusive access should not be possible
#[test]
fn read_exclusive_pending() {
let rwlock = RwLock::new(100);
let mut t1 = spawn(rwlock.write());
let _g1 = assert_ready!(t1.poll());
let mut t2 = spawn(rwlock.read());
assert_pending!(t2.poll());
}
// If the max shared access is reached and subsequent shared access is pending
// should be made available when one of the shared accesses is dropped
#[test]
fn exhaust_reading() {
let rwlock = RwLock::with_max_readers(100, 1024);
let mut reads = Vec::new();
loop {
let mut t = spawn(rwlock.read());
match t.poll() {
Poll::Ready(guard) => reads.push(guard),
Poll::Pending => break,
}
}
let mut t1 = spawn(rwlock.read());
assert_pending!(t1.poll());
let g2 = reads.pop().unwrap();
drop(g2);
assert!(t1.is_woken());
assert_ready!(t1.poll());
}
// When there is an active exclusive owner, subsequent exclusive access should not be possible
#[test]
fn write_exclusive_pending() {
let rwlock = RwLock::new(100);
let mut t1 = spawn(rwlock.write());
let _g1 = assert_ready!(t1.poll());
let mut t2 = spawn(rwlock.write());
assert_pending!(t2.poll());
}
// When there is an active shared owner, exclusive access should be possible after shared is dropped
#[test]
fn write_shared_drop() {
let rwlock = RwLock::new(100);
let mut t1 = spawn(rwlock.read());
let g1 = assert_ready!(t1.poll());
let mut t2 = spawn(rwlock.write());
assert_pending!(t2.poll());
drop(g1);
assert!(t2.is_woken());
assert_ready!(t2.poll());
}
// when there is an active shared owner, and exclusive access is triggered,
// subsequent shared access should not be possible as write gathers all the available semaphore permits
#[test]
fn write_read_shared_pending() {
let rwlock = RwLock::new(100);
let mut t1 = spawn(rwlock.read());
let _g1 = assert_ready!(t1.poll());
let mut t2 = spawn(rwlock.read());
assert_ready!(t2.poll());
let mut t3 = spawn(rwlock.write());
assert_pending!(t3.poll());
let mut t4 = spawn(rwlock.read());
assert_pending!(t4.poll());
}
// when there is an active shared owner, and exclusive access is triggered,
// reading should be possible after pending exclusive access is dropped
#[test]
fn write_read_shared_drop_pending() {
let rwlock = RwLock::new(100);
let mut t1 = spawn(rwlock.read());
let _g1 = assert_ready!(t1.poll());
let mut t2 = spawn(rwlock.write());
assert_pending!(t2.poll());
let mut t3 = spawn(rwlock.read());
assert_pending!(t3.poll());
drop(t2);
assert!(t3.is_woken());
assert_ready!(t3.poll());
}
// Acquire an RwLock nonexclusively by a single task
#[tokio::test]
async fn read_uncontested() {
let rwlock = RwLock::new(100);
let result = *rwlock.read().await;
assert_eq!(result, 100);
}
// Acquire an uncontested RwLock in exclusive mode
#[tokio::test]
async fn write_uncontested() {
let rwlock = RwLock::new(100);
let mut result = rwlock.write().await;
*result += 50;
assert_eq!(*result, 150);
}
// RwLocks should be acquired in the order that their Futures are waited upon.
#[tokio::test]
async fn write_order() {
let rwlock = RwLock::<Vec<u32>>::new(vec![]);
let fut2 = rwlock.write().map(|mut guard| guard.push(2));
let fut1 = rwlock.write().map(|mut guard| guard.push(1));
fut1.await;
fut2.await;
let g = rwlock.read().await;
assert_eq!(*g, vec![1, 2]);
}
// A single RwLock is contested by tasks in multiple threads
#[tokio::test(flavor = "multi_thread", worker_threads = 8)]
async fn multithreaded() {
let barrier = Arc::new(Barrier::new(5));
let rwlock = Arc::new(RwLock::<u32>::new(0));
let rwclone1 = rwlock.clone();
let rwclone2 = rwlock.clone();
let rwclone3 = rwlock.clone();
let rwclone4 = rwlock.clone();
let b1 = barrier.clone();
tokio::spawn(async move {
stream::iter(0..1000)
.for_each(move |_| {
let rwlock = rwclone1.clone();
async move {
let mut guard = rwlock.write().await;
*guard += 2;
}
})
.await;
b1.wait().await;
});
let b2 = barrier.clone();
tokio::spawn(async move {
stream::iter(0..1000)
.for_each(move |_| {
let rwlock = rwclone2.clone();
async move {
let mut guard = rwlock.write().await;
*guard += 3;
}
})
.await;
b2.wait().await;
});
let b3 = barrier.clone();
tokio::spawn(async move {
stream::iter(0..1000)
.for_each(move |_| {
let rwlock = rwclone3.clone();
async move {
let mut guard = rwlock.write().await;
*guard += 5;
}
})
.await;
b3.wait().await;
});
let b4 = barrier.clone();
tokio::spawn(async move {
stream::iter(0..1000)
.for_each(move |_| {
let rwlock = rwclone4.clone();
async move {
let mut guard = rwlock.write().await;
*guard += 7;
}
})
.await;
b4.wait().await;
});
barrier.wait().await;
let g = rwlock.read().await;
assert_eq!(*g, 17_000);
}
#[tokio::test]
async fn try_write() {
let lock = RwLock::new(0);
let read_guard = lock.read().await;
assert!(lock.try_write().is_err());
drop(read_guard);
assert!(lock.try_write().is_ok());
}
#[test]
fn try_read_try_write() {
let lock: RwLock<usize> = RwLock::new(15);
{
let rg1 = lock.try_read().unwrap();
assert_eq!(*rg1, 15);
assert!(lock.try_write().is_err());
let rg2 = lock.try_read().unwrap();
assert_eq!(*rg2, 15)
}
{
let mut wg = lock.try_write().unwrap();
*wg = 1515;
assert!(lock.try_read().is_err())
}
assert_eq!(*lock.try_read().unwrap(), 1515);
}
|
