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
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
use crate::prelude::*;
use async_channel::Receiver;
use async_channel::TryRecvError;
pub struct AsyncRunner;
/// Tick global task pools to progress local tasks.
/// This is required because `spawn_local` tasks can only be polled by the
/// thread that owns the LocalExecutor.
#[inline]
fn tick_task_pools() {
#[cfg(not(target_arch = "wasm32"))]
bevy::tasks::tick_global_task_pools_on_main_thread();
}
#[extend::ext]
pub impl App {
/// Run the app asynchronously, particularly useful for cases like
/// wasm where `App::run` just succeeds immediately
fn run_async(&mut self) -> impl 'static + Future<Output = AppExit> {
AsyncRunner::run(std::mem::take(self))
}
}
impl AsyncRunner {
async fn run(mut app: App) -> AppExit {
app.init_plugin::<AsyncPlugin>();
app.init();
// this is an outer loop that will run when there are no
// in-flight async tasks. We'll just do a 100ms update loop
loop {
// 1. flush async tasks (also runs update)
Self::flush_async_tasks(app.world_mut()).await;
// 2. exit if instructed
if let Some(exit) = app.should_exit() {
return exit;
}
// 3. delay next update
// TODO no idea how long the correct duration is here,
// does it depend on use-case?
time_ext::sleep_millis(1).await;
}
}
/// Run an loop at regular updates until all tasks have completed or
/// an AppExit is triggered. Note that some tasks like http/socket listeners
/// will never complete in which case this will never return.
async fn flush_async_tasks(world: &mut World) -> Option<AppExit> {
// yield required for wasm to spawn tasks
async_ext::yield_now().await;
loop {
// 1. update first to process command queues
world.update_local();
// 2. tick local tasks in multi-threaded mode
tick_task_pools();
// 3. exit if AppExit
if let Some(exit) = world.should_exit() {
return Some(exit);
}
// 4. exit if no remaining tasks
if world.resource::<AsyncChannel>().task_count() == 0 {
return None;
}
// 5. short delay
time_ext::sleep_millis(1).await;
}
}
/// Update the world in 1ms increments until recv has a value.
/// Ticks task pools after yielding to ensure spawned local tasks make progress.
/// Runs one final update after receiving the result to process any pending commands.
pub async fn poll_and_update<T>(
mut update: impl FnMut(),
recv: Receiver<T>,
) -> T {
loop {
match recv.try_recv() {
Ok(out) => {
// Run one final update to process any commands the async task
// sent before completing (e.g. resource modifications)
update();
return out;
}
Err(TryRecvError::Empty) => {
// Update to process command queues
update();
// Tick task pools BEFORE yielding to ensure newly spawned
// local tasks are polled in the same tick
tick_task_pools();
// Yield to let the executor poll other tasks.
// On WASM we need to actually sleep to return control to
// the JS event loop, otherwise setTimeout callbacks never fire.
#[cfg(target_arch = "wasm32")]
time_ext::sleep_millis(1).await;
#[cfg(not(target_arch = "wasm32"))]
async_ext::yield_now().await;
// Tick again after yielding to progress any tasks that were
// waiting on this task to yield
tick_task_pools();
}
Err(TryRecvError::Closed) => {
unreachable!("we control the send");
}
}
}
}
}