1 | #![cfg (feature = "macros" )] |
2 | #![allow (clippy::disallowed_names)] |
3 | use std::sync::Arc; |
4 | |
5 | #[cfg (all(target_family = "wasm" , not(target_os = "wasi" )))] |
6 | #[cfg (target_pointer_width = "64" )] |
7 | use wasm_bindgen_test::wasm_bindgen_test as test; |
8 | #[cfg (all(target_family = "wasm" , not(target_os = "wasi" )))] |
9 | use wasm_bindgen_test::wasm_bindgen_test as maybe_tokio_test; |
10 | |
11 | #[cfg (not(all(target_family = "wasm" , not(target_os = "wasi" ))))] |
12 | use tokio::test as maybe_tokio_test ; |
13 | |
14 | use tokio::sync::{oneshot, Semaphore}; |
15 | use tokio_test::{assert_pending, assert_ready, task}; |
16 | |
17 | #[maybe_tokio_test ] |
18 | async fn sync_one_lit_expr_comma() { |
19 | let foo = tokio::join!(async { 1 },); |
20 | |
21 | assert_eq!(foo, (1,)); |
22 | } |
23 | |
24 | #[maybe_tokio_test ] |
25 | async fn sync_one_lit_expr_no_comma() { |
26 | let foo = tokio::join!(async { 1 }); |
27 | |
28 | assert_eq!(foo, (1,)); |
29 | } |
30 | |
31 | #[maybe_tokio_test ] |
32 | async fn sync_two_lit_expr_comma() { |
33 | let foo = tokio::join!(async { 1 }, async { 2 },); |
34 | |
35 | assert_eq!(foo, (1, 2)); |
36 | } |
37 | |
38 | #[maybe_tokio_test ] |
39 | async fn sync_two_lit_expr_no_comma() { |
40 | let foo = tokio::join!(async { 1 }, async { 2 }); |
41 | |
42 | assert_eq!(foo, (1, 2)); |
43 | } |
44 | |
45 | #[maybe_tokio_test ] |
46 | async fn two_await() { |
47 | let (tx1, rx1) = oneshot::channel::<&str>(); |
48 | let (tx2, rx2) = oneshot::channel::<u32>(); |
49 | |
50 | let mut join = task::spawn(async { |
51 | tokio::join!(async { rx1.await.unwrap() }, async { rx2.await.unwrap() }) |
52 | }); |
53 | |
54 | assert_pending!(join.poll()); |
55 | |
56 | tx2.send(123).unwrap(); |
57 | assert!(join.is_woken()); |
58 | assert_pending!(join.poll()); |
59 | |
60 | tx1.send("hello" ).unwrap(); |
61 | assert!(join.is_woken()); |
62 | let res = assert_ready!(join.poll()); |
63 | |
64 | assert_eq!(("hello" , 123), res); |
65 | } |
66 | |
67 | #[test] |
68 | #[cfg (target_pointer_width = "64" )] |
69 | fn join_size() { |
70 | use futures::future; |
71 | use std::mem; |
72 | |
73 | let fut = async { |
74 | let ready = future::ready(0i32); |
75 | tokio::join!(ready) |
76 | }; |
77 | assert_eq!(mem::size_of_val(&fut), 32); |
78 | |
79 | let fut = async { |
80 | let ready1 = future::ready(0i32); |
81 | let ready2 = future::ready(0i32); |
82 | tokio::join!(ready1, ready2) |
83 | }; |
84 | assert_eq!(mem::size_of_val(&fut), 40); |
85 | } |
86 | |
87 | async fn non_cooperative_task(permits: Arc<Semaphore>) -> usize { |
88 | let mut exceeded_budget = 0; |
89 | |
90 | for _ in 0..5 { |
91 | // Another task should run after this task uses its whole budget |
92 | for _ in 0..128 { |
93 | let _permit = permits.clone().acquire_owned().await.unwrap(); |
94 | } |
95 | |
96 | exceeded_budget += 1; |
97 | } |
98 | |
99 | exceeded_budget |
100 | } |
101 | |
102 | async fn poor_little_task(permits: Arc<Semaphore>) -> usize { |
103 | let mut how_many_times_i_got_to_run = 0; |
104 | |
105 | for _ in 0..5 { |
106 | let _permit = permits.clone().acquire_owned().await.unwrap(); |
107 | how_many_times_i_got_to_run += 1; |
108 | } |
109 | |
110 | how_many_times_i_got_to_run |
111 | } |
112 | |
113 | #[tokio::test ] |
114 | async fn join_does_not_allow_tasks_to_starve() { |
115 | let permits = Arc::new(Semaphore::new(1)); |
116 | |
117 | // non_cooperative_task should yield after its budget is exceeded and then poor_little_task should run. |
118 | let (non_cooperative_result, little_task_result) = tokio::join!( |
119 | non_cooperative_task(Arc::clone(&permits)), |
120 | poor_little_task(permits) |
121 | ); |
122 | |
123 | assert_eq!(5, non_cooperative_result); |
124 | assert_eq!(5, little_task_result); |
125 | } |
126 | |
127 | #[tokio::test ] |
128 | async fn a_different_future_is_polled_first_every_time_poll_fn_is_polled() { |
129 | let poll_order = Arc::new(std::sync::Mutex::new(vec![])); |
130 | |
131 | let fut = |x, poll_order: Arc<std::sync::Mutex<Vec<i32>>>| async move { |
132 | for _ in 0..4 { |
133 | { |
134 | let mut guard = poll_order.lock().unwrap(); |
135 | |
136 | guard.push(x); |
137 | } |
138 | |
139 | tokio::task::yield_now().await; |
140 | } |
141 | }; |
142 | |
143 | tokio::join!( |
144 | fut(1, Arc::clone(&poll_order)), |
145 | fut(2, Arc::clone(&poll_order)), |
146 | fut(3, Arc::clone(&poll_order)), |
147 | ); |
148 | |
149 | // Each time the future created by join! is polled, it should start |
150 | // by polling a different future first. |
151 | assert_eq!( |
152 | vec![1, 2, 3, 2, 3, 1, 3, 1, 2, 1, 2, 3], |
153 | *poll_order.lock().unwrap() |
154 | ); |
155 | } |
156 | |
157 | #[tokio::test ] |
158 | #[allow (clippy::unit_cmp)] |
159 | async fn empty_join() { |
160 | assert_eq!(tokio::join!(), ()); |
161 | } |
162 | |