-
Notifications
You must be signed in to change notification settings - Fork 122
/
s3_single_atomic.rs
73 lines (64 loc) · 1.74 KB
/
s3_single_atomic.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
use std::cell::UnsafeCell;
use std::mem::MaybeUninit;
use std::sync::atomic::AtomicU8;
use std::sync::atomic::Ordering::{Acquire, Relaxed, Release};
const EMPTY: u8 = 0;
const WRITING: u8 = 1;
const READY: u8 = 2;
const READING: u8 = 3;
pub struct Channel<T> {
message: UnsafeCell<MaybeUninit<T>>,
state: AtomicU8,
}
unsafe impl<T: Send> Sync for Channel<T> {}
impl<T> Channel<T> {
pub const fn new() -> Self {
Self {
message: UnsafeCell::new(MaybeUninit::uninit()),
state: AtomicU8::new(EMPTY),
}
}
pub fn send(&self, message: T) {
if self.state.compare_exchange(
EMPTY, WRITING, Relaxed, Relaxed
).is_err() {
panic!("can't send more than one message!");
}
unsafe { (*self.message.get()).write(message) };
self.state.store(READY, Release);
}
pub fn is_ready(&self) -> bool {
self.state.load(Relaxed) == READY
}
pub fn receive(&self) -> T {
if self.state.compare_exchange(
READY, READING, Acquire, Relaxed
).is_err() {
panic!("no message available!");
}
unsafe { (*self.message.get()).assume_init_read() }
}
}
impl<T> Drop for Channel<T> {
fn drop(&mut self) {
if *self.state.get_mut() == READY {
unsafe { self.message.get_mut().assume_init_drop() }
}
}
}
#[test]
fn main() {
use std::thread;
let channel = Channel::new();
let t = thread::current();
thread::scope(|s| {
s.spawn(|| {
channel.send("hello world!");
t.unpark();
});
while !channel.is_ready() {
thread::park();
}
assert_eq!(channel.receive(), "hello world!");
});
}