Line data Source code
1 : use std::{collections::VecDeque, sync::Arc};
2 :
3 : use parking_lot::{Mutex, MutexGuard};
4 :
5 : use crate::executor::{self, PollSome, Waker};
6 :
7 : /// FIFO channel with blocking send and receive. Can be cloned and shared between threads.
8 : /// Blocking functions should be used only from threads that are managed by the executor.
9 : pub struct Chan<T> {
10 : shared: Arc<State<T>>,
11 : }
12 :
13 : impl<T> Clone for Chan<T> {
14 4822100 : fn clone(&self) -> Self {
15 4822100 : Chan {
16 4822100 : shared: self.shared.clone(),
17 4822100 : }
18 4822100 : }
19 : }
20 :
21 : impl<T> Default for Chan<T> {
22 0 : fn default() -> Self {
23 0 : Self::new()
24 0 : }
25 : }
26 :
27 : impl<T> Chan<T> {
28 629681 : pub fn new() -> Chan<T> {
29 629681 : Chan {
30 629681 : shared: Arc::new(State {
31 629681 : queue: Mutex::new(VecDeque::new()),
32 629681 : waker: Waker::new(),
33 629681 : }),
34 629681 : }
35 629681 : }
36 :
37 : /// Get a message from the front of the queue, block if the queue is empty.
38 : /// If not called from the executor thread, it can block forever.
39 4096 : pub fn recv(&self) -> T {
40 4096 : self.shared.recv()
41 4096 : }
42 :
43 : /// Panic if the queue is empty.
44 515984 : pub fn must_recv(&self) -> T {
45 515984 : self.shared
46 515984 : .try_recv()
47 515984 : .expect("message should've been ready")
48 515984 : }
49 :
50 : /// Get a message from the front of the queue, return None if the queue is empty.
51 : /// Never blocks.
52 446053 : pub fn try_recv(&self) -> Option<T> {
53 446053 : self.shared.try_recv()
54 446053 : }
55 :
56 : /// Send a message to the back of the queue.
57 944397 : pub fn send(&self, t: T) {
58 944397 : self.shared.send(t);
59 944397 : }
60 : }
61 :
62 : struct State<T> {
63 : queue: Mutex<VecDeque<T>>,
64 : waker: Waker,
65 : }
66 :
67 : impl<T> State<T> {
68 944397 : fn send(&self, t: T) {
69 944397 : self.queue.lock().push_back(t);
70 944397 : self.waker.wake_all();
71 944397 : }
72 :
73 962037 : fn try_recv(&self) -> Option<T> {
74 962037 : let mut q = self.queue.lock();
75 962037 : q.pop_front()
76 962037 : }
77 :
78 4096 : fn recv(&self) -> T {
79 4096 : // interrupt the receiver to prevent consuming everything at once
80 4096 : executor::yield_me(0);
81 4096 :
82 4096 : let mut queue = self.queue.lock();
83 4096 : if let Some(t) = queue.pop_front() {
84 0 : return t;
85 4096 : }
86 : loop {
87 10099 : self.waker.wake_me_later();
88 10099 : if let Some(t) = queue.pop_front() {
89 3618 : return t;
90 6003 : }
91 6003 : MutexGuard::unlocked(&mut queue, || {
92 6003 : executor::yield_me(-1);
93 6003 : });
94 : }
95 3618 : }
96 : }
97 :
98 : impl<T> PollSome for Chan<T> {
99 : /// Schedules a wakeup for the current thread.
100 6012022 : fn wake_me(&self) {
101 6012022 : self.shared.waker.wake_me_later();
102 6012022 : }
103 :
104 : /// Checks if chan has any pending messages.
105 4841805 : fn has_some(&self) -> bool {
106 4841805 : !self.shared.queue.lock().is_empty()
107 4841805 : }
108 : }
|