1 #![warn(rust_2018_idioms)]
2 #![cfg(feature = "sync")]
3
4 #[cfg(all(target_family = "wasm", not(target_os = "wasi")))]
5 use wasm_bindgen_test::wasm_bindgen_test as test;
6 #[cfg(all(target_family = "wasm", not(target_os = "wasi")))]
7 use wasm_bindgen_test::wasm_bindgen_test as maybe_tokio_test;
8
9 #[cfg(not(all(target_family = "wasm", not(target_os = "wasi"))))]
10 use tokio::test as maybe_tokio_test;
11
12 use std::task::Poll;
13
14 use futures::future::FutureExt;
15
16 use tokio::sync::{RwLock, RwLockWriteGuard};
17 use tokio_test::task::spawn;
18 use tokio_test::{assert_pending, assert_ready};
19
20 #[test]
into_inner()21 fn into_inner() {
22 let rwlock = RwLock::new(42);
23 assert_eq!(rwlock.into_inner(), 42);
24 }
25
26 // multiple reads should be Ready
27 #[test]
read_shared()28 fn read_shared() {
29 let rwlock = RwLock::new(100);
30
31 let mut t1 = spawn(rwlock.read());
32 let _g1 = assert_ready!(t1.poll());
33 let mut t2 = spawn(rwlock.read());
34 let _g2 = assert_ready!(t2.poll());
35 }
36
37 // When there is an active shared owner, exclusive access should not be possible
38 #[test]
write_shared_pending()39 fn write_shared_pending() {
40 let rwlock = RwLock::new(100);
41 let mut t1 = spawn(rwlock.read());
42
43 let _g1 = assert_ready!(t1.poll());
44 let mut t2 = spawn(rwlock.write());
45 assert_pending!(t2.poll());
46 }
47
48 // When there is an active exclusive owner, subsequent exclusive access should not be possible
49 #[test]
read_exclusive_pending()50 fn read_exclusive_pending() {
51 let rwlock = RwLock::new(100);
52 let mut t1 = spawn(rwlock.write());
53
54 let _g1 = assert_ready!(t1.poll());
55 let mut t2 = spawn(rwlock.read());
56 assert_pending!(t2.poll());
57 }
58
59 // If the max shared access is reached and subsequent shared access is pending
60 // should be made available when one of the shared accesses is dropped
61 #[test]
exhaust_reading()62 fn exhaust_reading() {
63 let rwlock = RwLock::with_max_readers(100, 1024);
64 let mut reads = Vec::new();
65 loop {
66 let mut t = spawn(rwlock.read());
67 match t.poll() {
68 Poll::Ready(guard) => reads.push(guard),
69 Poll::Pending => break,
70 }
71 }
72
73 let mut t1 = spawn(rwlock.read());
74 assert_pending!(t1.poll());
75 let g2 = reads.pop().unwrap();
76 drop(g2);
77 assert!(t1.is_woken());
78 let _g1 = assert_ready!(t1.poll());
79 }
80
81 // When there is an active exclusive owner, subsequent exclusive access should not be possible
82 #[test]
write_exclusive_pending()83 fn write_exclusive_pending() {
84 let rwlock = RwLock::new(100);
85 let mut t1 = spawn(rwlock.write());
86
87 let _g1 = assert_ready!(t1.poll());
88 let mut t2 = spawn(rwlock.write());
89 assert_pending!(t2.poll());
90 }
91
92 // When there is an active shared owner, exclusive access should be possible after shared is dropped
93 #[test]
write_shared_drop()94 fn write_shared_drop() {
95 let rwlock = RwLock::new(100);
96 let mut t1 = spawn(rwlock.read());
97
98 let g1 = assert_ready!(t1.poll());
99 let mut t2 = spawn(rwlock.write());
100 assert_pending!(t2.poll());
101 drop(g1);
102 assert!(t2.is_woken());
103 let _g2 = assert_ready!(t2.poll());
104 }
105
106 // when there is an active shared owner, and exclusive access is triggered,
107 // subsequent shared access should not be possible as write gathers all the available semaphore permits
108 #[test]
write_read_shared_pending()109 fn write_read_shared_pending() {
110 let rwlock = RwLock::new(100);
111 let mut t1 = spawn(rwlock.read());
112 let _g1 = assert_ready!(t1.poll());
113
114 let mut t2 = spawn(rwlock.read());
115 let _g2 = assert_ready!(t2.poll());
116
117 let mut t3 = spawn(rwlock.write());
118 assert_pending!(t3.poll());
119
120 let mut t4 = spawn(rwlock.read());
121 assert_pending!(t4.poll());
122 }
123
124 // when there is an active shared owner, and exclusive access is triggered,
125 // reading should be possible after pending exclusive access is dropped
126 #[test]
write_read_shared_drop_pending()127 fn write_read_shared_drop_pending() {
128 let rwlock = RwLock::new(100);
129 let mut t1 = spawn(rwlock.read());
130 let _g1 = assert_ready!(t1.poll());
131
132 let mut t2 = spawn(rwlock.write());
133 assert_pending!(t2.poll());
134
135 let mut t3 = spawn(rwlock.read());
136 assert_pending!(t3.poll());
137 drop(t2);
138
139 assert!(t3.is_woken());
140 let _t3 = assert_ready!(t3.poll());
141 }
142
143 // Acquire an RwLock nonexclusively by a single task
144 #[maybe_tokio_test]
read_uncontested()145 async fn read_uncontested() {
146 let rwlock = RwLock::new(100);
147 let result = *rwlock.read().await;
148
149 assert_eq!(result, 100);
150 }
151
152 // Acquire an uncontested RwLock in exclusive mode
153 #[maybe_tokio_test]
write_uncontested()154 async fn write_uncontested() {
155 let rwlock = RwLock::new(100);
156 let mut result = rwlock.write().await;
157 *result += 50;
158 assert_eq!(*result, 150);
159 }
160
161 // RwLocks should be acquired in the order that their Futures are waited upon.
162 #[maybe_tokio_test]
write_order()163 async fn write_order() {
164 let rwlock = RwLock::<Vec<u32>>::new(vec![]);
165 let fut2 = rwlock.write().map(|mut guard| guard.push(2));
166 let fut1 = rwlock.write().map(|mut guard| guard.push(1));
167 fut1.await;
168 fut2.await;
169
170 let g = rwlock.read().await;
171 assert_eq!(*g, vec![1, 2]);
172 }
173
174 // A single RwLock is contested by tasks in multiple threads
175 #[cfg(all(feature = "full", not(target_os = "wasi")))] // Wasi doesn't support threads
176 #[cfg_attr(miri, ignore)] // Too slow on miri.
177 #[tokio::test(flavor = "multi_thread", worker_threads = 8)]
multithreaded()178 async fn multithreaded() {
179 use futures::stream::{self, StreamExt};
180 use std::sync::Arc;
181 use tokio::sync::Barrier;
182
183 let barrier = Arc::new(Barrier::new(5));
184 let rwlock = Arc::new(RwLock::<u32>::new(0));
185 let rwclone1 = rwlock.clone();
186 let rwclone2 = rwlock.clone();
187 let rwclone3 = rwlock.clone();
188 let rwclone4 = rwlock.clone();
189
190 let b1 = barrier.clone();
191 tokio::spawn(async move {
192 stream::iter(0..1000)
193 .for_each(move |_| {
194 let rwlock = rwclone1.clone();
195 async move {
196 let mut guard = rwlock.write().await;
197 *guard += 2;
198 }
199 })
200 .await;
201 b1.wait().await;
202 });
203
204 let b2 = barrier.clone();
205 tokio::spawn(async move {
206 stream::iter(0..1000)
207 .for_each(move |_| {
208 let rwlock = rwclone2.clone();
209 async move {
210 let mut guard = rwlock.write().await;
211 *guard += 3;
212 }
213 })
214 .await;
215 b2.wait().await;
216 });
217
218 let b3 = barrier.clone();
219 tokio::spawn(async move {
220 stream::iter(0..1000)
221 .for_each(move |_| {
222 let rwlock = rwclone3.clone();
223 async move {
224 let mut guard = rwlock.write().await;
225 *guard += 5;
226 }
227 })
228 .await;
229 b3.wait().await;
230 });
231
232 let b4 = barrier.clone();
233 tokio::spawn(async move {
234 stream::iter(0..1000)
235 .for_each(move |_| {
236 let rwlock = rwclone4.clone();
237 async move {
238 let mut guard = rwlock.write().await;
239 *guard += 7;
240 }
241 })
242 .await;
243 b4.wait().await;
244 });
245
246 barrier.wait().await;
247 let g = rwlock.read().await;
248 assert_eq!(*g, 17_000);
249 }
250
251 #[maybe_tokio_test]
try_write()252 async fn try_write() {
253 let lock = RwLock::new(0);
254 let read_guard = lock.read().await;
255 assert!(lock.try_write().is_err());
256 drop(read_guard);
257 assert!(lock.try_write().is_ok());
258 }
259
260 #[test]
try_read_try_write()261 fn try_read_try_write() {
262 let lock: RwLock<usize> = RwLock::new(15);
263
264 {
265 let rg1 = lock.try_read().unwrap();
266 assert_eq!(*rg1, 15);
267
268 assert!(lock.try_write().is_err());
269
270 let rg2 = lock.try_read().unwrap();
271 assert_eq!(*rg2, 15)
272 }
273
274 {
275 let mut wg = lock.try_write().unwrap();
276 *wg = 1515;
277
278 assert!(lock.try_read().is_err())
279 }
280
281 assert_eq!(*lock.try_read().unwrap(), 1515);
282 }
283
284 #[maybe_tokio_test]
downgrade_map()285 async fn downgrade_map() {
286 let lock = RwLock::new(0);
287 let write_guard = lock.write().await;
288 let mut read_t = spawn(lock.read());
289
290 // We can't create a read when a write exists
291 assert_pending!(read_t.poll());
292
293 // During the call to `f`, `read_t` doesn't have access yet.
294 let read_guard1 = RwLockWriteGuard::downgrade_map(write_guard, |v| {
295 assert_pending!(read_t.poll());
296 v
297 });
298
299 // After the downgrade, `read_t` got the lock
300 let read_guard2 = assert_ready!(read_t.poll());
301
302 // Ensure they're equal, as we return the original value
303 assert_eq!(&*read_guard1 as *const _, &*read_guard2 as *const _);
304 }
305
306 #[maybe_tokio_test]
try_downgrade_map()307 async fn try_downgrade_map() {
308 let lock = RwLock::new(0);
309 let write_guard = lock.write().await;
310 let mut read_t = spawn(lock.read());
311
312 // We can't create a read when a write exists
313 assert_pending!(read_t.poll());
314
315 // During the call to `f`, `read_t` doesn't have access yet.
316 let write_guard = RwLockWriteGuard::try_downgrade_map(write_guard, |_| {
317 assert_pending!(read_t.poll());
318 None::<&()>
319 })
320 .expect_err("downgrade didn't fail");
321
322 // After `f` returns `None`, `read_t` doesn't have access
323 assert_pending!(read_t.poll());
324
325 // After `f` returns `Some`, `read_t` does have access
326 let read_guard1 = RwLockWriteGuard::try_downgrade_map(write_guard, |v| Some(v))
327 .expect("downgrade didn't succeed");
328 let read_guard2 = assert_ready!(read_t.poll());
329
330 // Ensure they're equal, as we return the original value
331 assert_eq!(&*read_guard1 as *const _, &*read_guard2 as *const _);
332 }
333