| 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 | |