1 use crate::sync::oneshot;
2 
3 use futures::future::poll_fn;
4 use loom::future::block_on;
5 use loom::thread;
6 use std::task::Poll::{Pending, Ready};
7 
8 #[test]
smoke()9 fn smoke() {
10     loom::model(|| {
11         let (tx, rx) = oneshot::channel();
12 
13         thread::spawn(move || {
14             tx.send(1).unwrap();
15         });
16 
17         let value = block_on(rx).unwrap();
18         assert_eq!(1, value);
19     });
20 }
21 
22 #[test]
changing_rx_task()23 fn changing_rx_task() {
24     loom::model(|| {
25         let (tx, mut rx) = oneshot::channel();
26 
27         thread::spawn(move || {
28             tx.send(1).unwrap();
29         });
30 
31         let rx = thread::spawn(move || {
32             let ready = block_on(poll_fn(|cx| match Pin::new(&mut rx).poll(cx) {
33                 Ready(Ok(value)) => {
34                     assert_eq!(1, value);
35                     Ready(true)
36                 }
37                 Ready(Err(_)) => unimplemented!(),
38                 Pending => Ready(false),
39             }));
40 
41             if ready {
42                 None
43             } else {
44                 Some(rx)
45             }
46         })
47         .join()
48         .unwrap();
49 
50         if let Some(rx) = rx {
51             // Previous task parked, use a new task...
52             let value = block_on(rx).unwrap();
53             assert_eq!(1, value);
54         }
55     });
56 }
57 
58 #[test]
try_recv_close()59 fn try_recv_close() {
60     // reproduces https://github.com/tokio-rs/tokio/issues/4225
61     loom::model(|| {
62         let (tx, mut rx) = oneshot::channel();
63         thread::spawn(move || {
64             let _ = tx.send(());
65         });
66 
67         rx.close();
68         let _ = rx.try_recv();
69     })
70 }
71 
72 #[test]
recv_closed()73 fn recv_closed() {
74     // reproduces https://github.com/tokio-rs/tokio/issues/4225
75     loom::model(|| {
76         let (tx, mut rx) = oneshot::channel();
77 
78         thread::spawn(move || {
79             let _ = tx.send(1);
80         });
81 
82         rx.close();
83         let _ = block_on(rx);
84     });
85 }
86 
87 // TODO: Move this into `oneshot` proper.
88 
89 use std::future::Future;
90 use std::pin::Pin;
91 use std::task::{Context, Poll};
92 
93 struct OnClose<'a> {
94     tx: &'a mut oneshot::Sender<i32>,
95 }
96 
97 impl<'a> OnClose<'a> {
new(tx: &'a mut oneshot::Sender<i32>) -> Self98     fn new(tx: &'a mut oneshot::Sender<i32>) -> Self {
99         OnClose { tx }
100     }
101 }
102 
103 impl Future for OnClose<'_> {
104     type Output = bool;
105 
poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<bool>106     fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<bool> {
107         let fut = self.get_mut().tx.closed();
108         crate::pin!(fut);
109 
110         Ready(fut.poll(cx).is_ready())
111     }
112 }
113 
114 #[test]
changing_tx_task()115 fn changing_tx_task() {
116     loom::model(|| {
117         let (mut tx, rx) = oneshot::channel::<i32>();
118 
119         thread::spawn(move || {
120             drop(rx);
121         });
122 
123         let tx = thread::spawn(move || {
124             let t1 = block_on(OnClose::new(&mut tx));
125 
126             if t1 {
127                 None
128             } else {
129                 Some(tx)
130             }
131         })
132         .join()
133         .unwrap();
134 
135         if let Some(mut tx) = tx {
136             // Previous task parked, use a new task...
137             block_on(OnClose::new(&mut tx));
138         }
139     });
140 }
141