File: mutex.rs

package info (click to toggle)
rust-futures-locks 0.7.1-1
  • links: PTS, VCS
  • area: main
  • in suites: bookworm, forky, sid, trixie
  • size: 176 kB
  • sloc: makefile: 2
file content (276 lines) | stat: -rw-r--r-- 7,704 bytes parent folder | download
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
271
272
273
274
275
276
//vim: tw=80

use futures::{FutureExt, stream};
#[cfg(feature = "tokio")]
use futures::future::ready;
use futures::stream::StreamExt;
use std::sync::Arc;
#[cfg(feature = "tokio")]
use std::rc::Rc;
use tokio::{self, sync::Barrier};
#[cfg(feature = "tokio")]
use tokio::runtime;
use tokio_test::task::spawn;
use tokio_test::{assert_pending, assert_ready};
use futures_locks::*;

// Create a MutexWeak and then upgrade it to Mutex
#[test]
fn mutex_weak_some() {
    let mutex = Mutex::<u32>::new(0);
    let mutex_weak = Mutex::downgrade(&mutex);

    assert!(mutex_weak.upgrade().is_some())
}

// Create a MutexWeak and drop the mutex so that MutexWeak::upgrade return None
#[test]
fn mutex_weak_none() {
    let mutex = Mutex::<u32>::new(0);
    let mutex_weak = Mutex::downgrade(&mutex);

    drop(mutex);

    assert!(mutex_weak.upgrade().is_none())
}

// Compare Mutexes if it point to the same value
#[test]
fn mutex_eq_ptr_true() {
    let mutex = Mutex::<u32>::new(0);
    let mutex_other = mutex.clone();

    assert!(Mutex::ptr_eq(&mutex, &mutex_other));
}

// Compare Mutexes if it point to the same value
#[test]
fn mutex_eq_ptr_false() {
    let mutex = Mutex::<u32>::new(0);
    let mutex_other = Mutex::<u32>::new(0);

    assert!(!Mutex::ptr_eq(&mutex, &mutex_other));
}

// When a Mutex gets dropped after gaining ownership but before being polled, it
// should drain its channel and relinquish ownership if a message was found.  If
// not, deadlocks may result.
#[tokio::test]
async fn drop_when_ready() {
    let mutex = Mutex::<u32>::new(0);

    let guard1 = mutex.lock().await;
    let fut2 = mutex.lock();
    drop(guard1);                // ownership transfers to fut2
    drop(fut2);                  // relinquish ownership
    // The mutex should be available again
    let _guard3 = mutex.lock().await;
}

// When a pending Mutex gets dropped after being polled() but before gaining
// ownership, ownership should pass on to the next waiter.
#[test]
fn drop_before_ready() {
    let mutex = Mutex::<u32>::new(0);

    let mut fut1 = spawn(mutex.lock());
    let guard1 = assert_ready!(fut1.poll()); // fut1 immediately gets ownership

    let mut fut2 = spawn(mutex.lock());
    assert_pending!(fut2.poll());            // fut2 is blocked

    let mut fut3 = spawn(mutex.lock());
    assert_pending!(fut3.poll());            // fut3 is blocked, too

    drop(fut2);                  // drop before gaining ownership
    drop(guard1);                // ownership transfers to fut3
    drop(fut1);

    assert!(fut3.is_woken());
    assert_ready!(fut3.poll());
}

// Mutably dereference a uniquely owned Mutex
#[test]
fn get_mut() {
    let mut mutex = Mutex::<u32>::new(42);
    *mutex.get_mut().unwrap() += 1;
    assert_eq!(*mutex.get_mut().unwrap(), 43);
}

// Cloned Mutexes cannot be deferenced
#[test]
fn get_mut_cloned() {
    let mut mutex = Mutex::<u32>::new(42);
    let _clone = mutex.clone();
    assert!(mutex.get_mut().is_none());
}

// Acquire an uncontested Mutex.  poll immediately returns Async::Ready
// #[test]
#[tokio::test]
async fn lock_uncontested() {
    let mutex = Mutex::<u32>::new(0);

    let guard = mutex.lock().await;
    let result = *guard + 5;
    drop(guard);
    assert_eq!(result, 5);
}

// Pend on a Mutex held by another task in the same tokio Reactor.  poll returns
// Async::NotReady.  Later, it gets woken up without involving the OS.
#[test]
fn lock_contested() {
    let mutex = Mutex::<u32>::new(0);

    let mut fut0 = spawn(mutex.lock());
    let guard0 = assert_ready!(fut0.poll()); // fut0 immediately gets ownership

    let mut fut1 = spawn(mutex.lock());
    assert_pending!(fut1.poll());            // fut1 is blocked

    drop(guard0);                            // Ownership transfers to fut1
    assert!(fut1.is_woken());
    assert_ready!(fut1.poll());
}

// A single Mutex is contested by tasks in multiple threads
#[tokio::test]
async fn lock_multithreaded() {
    let mutex = Mutex::<u32>::new(0);
    let mtx_clone0 = mutex.clone();
    let mtx_clone1 = mutex.clone();
    let mtx_clone2 = mutex.clone();
    let mtx_clone3 = mutex.clone();
    let barrier = Arc::new(Barrier::new(5));
    let b0 = barrier.clone();
    let b1 = barrier.clone();
    let b2 = barrier.clone();
    let b3 = barrier.clone();

    tokio::task::spawn(async move {
        stream::iter(0..1000).for_each(move |_| {
            mtx_clone0.lock().map(|mut guard| { *guard += 2 })
        }).await;
        b0.wait().await;
    });
    tokio::task::spawn(async move {
        stream::iter(0..1000).for_each(move |_| {
            mtx_clone1.lock().map(|mut guard| { *guard += 3 })
        }).await;
        b1.wait().await;
    });
    tokio::task::spawn(async move {
        stream::iter(0..1000).for_each(move |_| {
            mtx_clone2.lock().map(|mut guard| { *guard += 5 })
        }).await;
        b2.wait().await;
    });
    tokio::task::spawn(async move {
        stream::iter(0..1000).for_each(move |_| {
            mtx_clone3.lock().map(|mut guard| { *guard += 7 })
        }).await;
        b3.wait().await;
    });

    barrier.wait().await;
    assert_eq!(mutex.try_unwrap().expect("try_unwrap"), 17_000);
}

// Mutexes should be acquired in the order that their Futures are waited upon.
#[tokio::test]
async fn lock_order() {
    let mutex = Mutex::<Vec<u32>>::new(vec![]);
    let fut2 = mutex.lock().map(|mut guard| guard.push(2));
    let fut1 = mutex.lock().map(|mut guard| guard.push(1));

    fut1.then(|_| fut2).await;
    assert_eq!(mutex.try_unwrap().unwrap(), vec![1, 2]);
}

// Acquire an uncontested Mutex with try_lock
#[test]
fn try_lock_uncontested() {
    let mutex = Mutex::<u32>::new(5);

    let guard = mutex.try_lock().unwrap();
    assert_eq!(5, *guard);
}

// Try and fail to acquire a contested Mutex with try_lock
#[test]
fn try_lock_contested() {
    let mutex = Mutex::<u32>::new(0);

    let _guard = mutex.try_lock().unwrap();
    assert!(mutex.try_lock().is_err());
}

#[test]
fn try_unwrap_multiply_referenced() {
    let mtx = Mutex::<u32>::new(0);
    let _mtx2 = mtx.clone();
    assert!(mtx.try_unwrap().is_err());
}

// Returning errors is simpler than in futures-locks 0.5: just return a Result
#[cfg(feature = "tokio")]
#[test]
fn with_err() {
    let mtx = Mutex::<i32>::new(-5);
    let rt = runtime::Builder::new_current_thread().build().unwrap();
    let r = rt.block_on(async {
        mtx.with(|guard| {
            if *guard > 0 {
                ready(Ok(*guard))
            } else {
                ready(Err("Whoops!"))
            }
        }).await
    });
    assert_eq!(r, Err("Whoops!"));
}

#[cfg(feature = "tokio")]
#[test]
fn with_ok() {
    let mtx = Mutex::<i32>::new(5);
    let rt = runtime::Builder::new_current_thread().build().unwrap();
    let r = rt.block_on(async {
        mtx.with(|guard| {
            ready(*guard)
        }).await
    });
    assert_eq!(r, 5);
}

// Mutex::with should work with multithreaded Runtimes as well as
// single-threaded Runtimes.
// https://github.com/asomers/futures-locks/issues/5
#[cfg(feature = "tokio")]
#[test]
fn with_threadpool() {
    let mtx = Mutex::<i32>::new(5);
    let rt = runtime::Builder::new_multi_thread().build().unwrap();
    let r = rt.block_on(async {
        mtx.with(|guard| {
            ready(*guard)
        }).await
    });
    assert_eq!(r, 5);
}

#[cfg(feature = "tokio")]
#[test]
fn with_local_ok() {
    // Note: Rc is not Send
    let mtx = Mutex::<Rc<i32>>::new(Rc::new(5));
    let rt = runtime::Builder::new_current_thread().build().unwrap();
    let r = rt.block_on(async {
        mtx.with_local(|guard| {
            ready(**guard)
        }).await
    });
    assert_eq!(r, 5);
}