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
use std::{
collections::VecDeque,
sync::{
Arc,
atomic::{AtomicBool, AtomicUsize, Ordering},
},
};
pub use crate::native::sync::mpsc::{RecvError, RecvTimeoutError, SendError, TryRecvError};
use crate::{
sync::{Condvar, Mutex},
time::Instant,
};
struct Chan<T> {
/// Cleared on `Receiver` drop so a later `send` reports `SendError`.
receiver_alive: AtomicBool,
/// Live `Sender` count. Reaching 0 means "disconnected": a blocked
/// `recv` returns [`RecvError`], `try_recv` returns `Disconnected`.
senders: AtomicUsize,
cv: Condvar,
queue: Mutex<VecDeque<T>>,
}
pub struct Sender<T>(Arc<Chan<T>>);
pub struct Receiver<T>(Arc<Chan<T>>);
/// Create a new unbounded channel.
#[must_use]
pub fn channel<T>() -> (Sender<T>, Receiver<T>) {
let chan = Arc::new(Chan {
queue: Mutex::default(),
cv: Condvar::default(),
senders: AtomicUsize::new(1),
receiver_alive: AtomicBool::new(true),
});
(Sender(Arc::clone(&chan)), Receiver(chan))
}
impl<T> Sender<T> {
/// Send a value synchronously.
///
/// # Errors
///
/// Returns [`SendError`] if the receiver has been dropped.
pub fn send(&self, value: T) -> Result<(), SendError<T>> {
if !self.0.receiver_alive.load(Ordering::Acquire) {
return Err(SendError(value));
}
self.0.queue.lock().push_back(value);
// A receiver registers its condvar waiter UNDER the queue lock and
// releases the lock only as it parks, so this notify (which we issue
// after releasing the lock above) can never land before the waiter
// is registered — no lost wakeup.
self.0.cv.notify_one();
Ok(())
}
}
impl<T> Clone for Sender<T> {
fn clone(&self) -> Self {
self.0.senders.fetch_add(1, Ordering::AcqRel);
Self(Arc::clone(&self.0))
}
}
impl<T> Drop for Sender<T> {
fn drop(&mut self) {
if self.0.senders.fetch_sub(1, Ordering::AcqRel) == 1 {
// Last sender gone. Take the queue lock so this serializes with a
// receiver registering its waiter under the same lock, then wake
// it to observe the disconnect (no lost wakeup against the
// unlocked `senders` predicate).
let guard = self.0.queue.lock();
self.0.cv.notify_all();
drop(guard);
}
}
}
impl<T> Receiver<T> {
/// Iterate over received values, blocking until all senders disconnect.
pub fn iter(&self) -> impl Iterator<Item = T> + '_ {
std::iter::from_fn(move || self.recv().ok())
}
/// Block until a value arrives.
///
/// # Errors
///
/// Returns [`RecvError`] if all senders have been dropped.
pub fn recv(&self) -> Result<T, RecvError> {
let mut q = self.0.queue.lock();
loop {
if let Some(v) = q.pop_front() {
return Ok(v);
}
if self.0.senders.load(Ordering::Acquire) == 0 {
return Err(RecvError);
}
q = self.0.cv.wait(q);
}
}
/// Block until a value arrives or `deadline` elapses.
///
/// # Errors
///
/// Returns [`RecvTimeoutError::Timeout`] when no value arrives before
/// `deadline`, or [`RecvTimeoutError::Disconnected`] if all senders are
/// dropped.
pub fn recv_timeout(&self, deadline: Instant) -> Result<T, RecvTimeoutError> {
let mut q = self.0.queue.lock();
loop {
if let Some(v) = q.pop_front() {
return Ok(v);
}
if self.0.senders.load(Ordering::Acquire) == 0 {
return Err(RecvTimeoutError::Disconnected);
}
if Instant::now() >= deadline {
return Err(RecvTimeoutError::Timeout);
}
q = self.0.cv.wait_timeout(q, deadline);
}
}
/// Iterate over currently-available values without blocking.
pub fn try_iter(&self) -> impl Iterator<Item = T> + '_ {
std::iter::from_fn(move || self.try_recv().ok())
}
/// Try to receive without blocking.
///
/// # Errors
///
/// Returns [`TryRecvError`] if no value is available or senders are dropped.
pub fn try_recv(&self) -> Result<T, TryRecvError> {
let mut q = self.0.queue.lock();
match q.pop_front() {
Some(v) => Ok(v),
None if self.0.senders.load(Ordering::Acquire) == 0 => Err(TryRecvError::Disconnected),
None => Err(TryRecvError::Empty),
}
}
}
impl<T> Drop for Receiver<T> {
fn drop(&mut self) {
self.0.receiver_alive.store(false, Ordering::Release);
}
}