-
Notifications
You must be signed in to change notification settings - Fork 126
/
messages_queue.rs
85 lines (75 loc) · 2.66 KB
/
messages_queue.rs
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
use std::collections::VecDeque;
use std::time::{Duration, Instant};
use std::sync::{Arc, Mutex, Condvar};
enum Control<T> {
Elem(T),
Unblock,
}
pub struct MessagesQueue<T> where T: Send {
queue: Mutex<VecDeque<Control<T>>>,
condvar: Condvar,
}
impl<T> MessagesQueue<T> where T: Send {
pub fn with_capacity(capacity: usize) -> Arc<MessagesQueue<T>> {
Arc::new(MessagesQueue {
queue: Mutex::new(VecDeque::with_capacity(capacity)),
condvar: Condvar::new(),
})
}
/// Pushes an element to the queue.
pub fn push(&self, value: T) {
let mut queue = self.queue.lock().unwrap();
queue.push_back(Control::Elem(value));
self.condvar.notify_one();
}
/// Unblock one thread stuck in pop loop.
pub fn unblock(&self) {
let mut queue = self.queue.lock().unwrap();
queue.push_back(Control::Unblock);
self.condvar.notify_one();
}
/// Pops an element. Blocks until one is available.
/// Returns None in case unblock() was issued.
pub fn pop(&self) -> Option<T> {
let mut queue = self.queue.lock().unwrap();
loop {
match queue.pop_front() {
Some(Control::Elem(value)) => return Some(value),
Some(Control::Unblock) => return None,
None => (),
}
queue = self.condvar.wait(queue).unwrap();
}
}
/// Tries to pop an element without blocking.
pub fn try_pop(&self) -> Option<T> {
let mut queue = self.queue.lock().unwrap();
match queue.pop_front() {
Some(Control::Elem(value)) => Some(value),
Some(Control::Unblock) | None => None,
}
}
/// Tries to pop an element without blocking
/// more than the specified timeout duration
/// or unblock() was issued
pub fn pop_timeout(&self, timeout: Duration) -> Option<T> {
let mut queue = self.queue.lock().unwrap();
let mut duration = timeout;
loop {
match queue.pop_front() {
Some(Control::Elem(value)) => return Some(value),
Some(Control::Unblock) => return None,
None => (),
}
let now = Instant::now();
let (_queue, result) = self.condvar.wait_timeout(queue, timeout).unwrap();
queue = _queue;
let sleep_time = now.elapsed();
duration = if duration > sleep_time { duration - sleep_time } else { Duration::from_millis(0) };
if result.timed_out() ||
(duration.as_secs() == 0 && duration.subsec_nanos() < 1000000) {
return None;
}
}
}
}