-
Notifications
You must be signed in to change notification settings - Fork 125
/
s6_blocking.rs
84 lines (74 loc) · 1.99 KB
/
s6_blocking.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
use std::cell::UnsafeCell;
use std::marker::PhantomData;
use std::mem::MaybeUninit;
use std::sync::atomic::AtomicBool;
use std::sync::atomic::Ordering::{Acquire, Release};
use std::thread;
use std::thread::Thread;
pub struct Channel<T> {
message: UnsafeCell<MaybeUninit<T>>,
ready: AtomicBool,
}
unsafe impl<T> Sync for Channel<T> where T: Send {}
pub struct Sender<'a, T> {
channel: &'a Channel<T>,
receiving_thread: Thread, // New!
}
pub struct Receiver<'a, T> {
channel: &'a Channel<T>,
_no_send: PhantomData<*const ()>, // New!
}
impl<T> Channel<T> {
pub const fn new() -> Self {
Self {
message: UnsafeCell::new(MaybeUninit::uninit()),
ready: AtomicBool::new(false),
}
}
pub fn split<'a>(&'a mut self) -> (Sender<'a, T>, Receiver<'a, T>) {
*self = Self::new();
(
Sender {
channel: self,
receiving_thread: thread::current(), // New!
},
Receiver {
channel: self,
_no_send: PhantomData, // New!
}
)
}
}
impl<T> Sender<'_, T> {
pub fn send(self, message: T) {
unsafe { (*self.channel.message.get()).write(message) };
self.channel.ready.store(true, Release);
self.receiving_thread.unpark(); // New!
}
}
impl<T> Receiver<'_, T> {
pub fn receive(self) -> T {
while !self.channel.ready.swap(false, Acquire) {
thread::park();
}
unsafe { (*self.channel.message.get()).assume_init_read() }
}
}
impl<T> Drop for Channel<T> {
fn drop(&mut self) {
if *self.ready.get_mut() {
unsafe { self.message.get_mut().assume_init_drop() }
}
}
}
#[test]
fn main() {
let mut channel = Channel::new();
thread::scope(|s| {
let (sender, receiver) = channel.split();
s.spawn(move || {
sender.send("hello world!");
});
assert_eq!(receiver.receive(), "hello world!");
});
}