Async system functions and first-class async support?

[hanabi1224]

In the async_compute example, users need to block_on the result of poll_once, which is inconvenient and may get big performance penalty if they use multiple block_on(s) instead a single one, the performance gap can be consistently verified with the benchmark program in this PR

Proposed API change: #2691

It's ideal if a single block_on can be performed inside ECS by allowing async system functions, and I think using a single block_on for all async system functions on every frame update is the best way (in terms of performance).

Regarding the performance gap, a bevy Task is not a system thread (coordinated by OS) but actually a fiber or green thread or coroutine, you name it, which should ideally be coordinated with a single scheduler(block_on) to get the best performance.

And in terms of first-class async support, I would consider it's missing from (current) bevy, as long as it requires users to write ad-hoc block_on(s).

As @TheRawMeatball has pointed out, tasks in this example are scheduled with ECS scheduler and then I think there should not be any poll_once or block_on in user code at all. Another spawn method in TaskPool that returns TaskWrapper below which contains the Option result will work, in a poll_once and block_on free way, and my test shows it has the best performance as there's no extra overhead and the API is much cleaner.

#[derive(Clone)]
struct TaskWrapper<T> {
    pub result: Arc<RwLock<Option<T>>>,
}

impl<T> TaskWrapper<T> {
    pub fn new() -> Self {
        Self {
            result: Arc::new(RwLock::new(None)),
        }
    }

    pub async fn register(&mut self, t: impl std::future::Future<Output = T>) {
        let ret = t.await;
        let mut lock = self.result.write().unwrap();
        *lock = Some(ret);
    }
}

fn spawn_tasks_no_poll_once(mut commands: Commands, thread_pool: Res<AsyncComputeTaskPool>) {
    for step in 0..N_STEPS {
        for _i in 0..N_TASKS {
            let wrapper = TaskWrapper::<()>::new();
            let mut wrapper_clone = wrapper.clone();
            let task = thread_pool.spawn(async move {
                wrapper_clone
                    .register(async move {
                        let start_time = Instant::now();
                        let duration = Duration::from_secs_f32(TASK_DURATION_SEC * (step as f32));
                        while Instant::now() - start_time < duration {
                            futures_timer::Delay::new(Duration::from_secs_f32(0.1)).await;
                        }
                    })
                    .await;
            });
            commands.spawn().insert(wrapper.clone()).insert(task);
        }
    }
}

fn handle_tasks_no_poll_once(
    mut commands: Commands,
    transform_tasks: Query<(Entity, &TaskWrapper<()>)>,
    mut app_exit_events: EventWriter<AppExit>,
    time: Res<Time>,
    frame_counter: Res<FrameCounter>,
) {
    let mut n_tasks = 0;
    for (entity, task) in transform_tasks.iter() {
        n_tasks += 1;
        let lock = task.result.read().unwrap();
        if lock.is_some() {
            commands
                .entity(entity)
                .remove::<TaskWrapper<()>>()
                .remove::<Task<()>>();
        }
    }
    if n_tasks == 0 {
        print_statistics("handle_tasks_no_poll_once", &frame_counter, &time);
        app_exit_events.send(AppExit);
    }
}

[alice-i-cecile]

This is a great analysis. @Ratysz and @TheRawMeatball do you have thoughts on how this might fit into the scheduling API?

[DJMcNab]

The implementation of async systems you are advocating for (where they are automatically blocked on in bevy's own code then dropped within the same frame) would be a massive footgun.

I think the correct way to do this is something like:

fn handle_tasks(
    mut commands: Commands,
    mut transform_tasks: Query<(Entity, &mut Task<Transform>)>,
    box_mesh_handle: Res<BoxMeshHandle>,
    box_material_handle: Res<BoxMaterialHandle>,
) {
let task = async move {
    for (entity, mut task) in transform_tasks.iter_mut() {
        if let Some(transform) = future::poll_once(&mut *task).await {
            // Add our new PbrBundle of components to our tagged entity
            commands.entity(entity).insert_bundle(PbrBundle {
                mesh: box_mesh_handle.0.clone(),
                material: box_material_handle.0.clone(),
                transform,
                ..Default::default()
            });

            // Task is complete, so remove task component from entity
            commands.entity(entity).remove::<Task<Transform>>();
        }
    }
}
futures::block_on(task);
}

[Ratysz]

I don't think we need to do anything on Bevy side for this: there is no one-size-fits-all way of applying task results because this is entirely user code, and the amount of boilerplate a built-in general solution would remove is next to nothing.

If I were to develop that example from scratch, I would try something like a local FuturesUnordered in handle_tasks: steal entity-attached tasks into it, no matter their completion status, poll it, apply results as they come with Query::get_mut().

[alice-i-cecile]

Perhaps the solution is to extend the examples and documentation to show more sophisticated patterns?

[hanabi1224]

because this is entirely user code

@Ratysz That's true for the logic inside a user system function but I don't think the scheduler/block_on selection part is. Users can use tokio as a scheduler and that might cause errors, I think it's ECS' responsibility to choose the correct scheduler instead of exposing it.
cc @DJMcNab

BTW, I've got another PR to enable alternative async schedulers like tokio, is that something feasible?

[Ratysz]

why bevy chose to split these scenarios

I'm not sure I understand what you mean here.

Architecture of any sensible game is inherently synchronous, due to it running in real time on real hardware and, most commonly, independent of IO. Thus Bevy, being an engine to make games sensibly, is inherently synchronous as well.

Meanwhile, Tokio is for making heavily IO-driven apps, sensible architecture for which is asynchronous.

Async is not magic; it is, at it's core, syntax sugar, and our particular case does not benefit from it. I'm not saying that it's impossible to improve the "long async calculation" story in Bevy, but the proposed approach really doesn't feel like it would be that.

[hanabi1224]

Architecture of any sensible game is inherently synchronous

Totally agree with that, hence most of the usage of async can be implied as IO bounded(although it's cpu-bounded in the async-compute example), my point is that, if bevy would ever allow usage of async, it should provide a systematic way to coordinate those futures, rather than letting users write ad-hoc code to block_on the futures in their systems. IMHO, the sync part and async part does not collide with each other, I'm not advocating bevy should change the main game loop flow to async based, but if bevy systems allow async (e.g. for networking), there should be a better API to coordinate all the async part, inside ECS, and it might be good idea to refer to tokio ONLY for this async part.

The code you suggest works but it seems that bevy does not provide first-class support of async by asking users to write block_on, and maybe I should rephrase my question to 'Is first-class async support in bevy's roadmap?'

[hanabi1224]

@Ratysz as @TheRawMeatball has pointed out, tasks are actually scheduled by a ECS scheduler so actually there's no need to call block_on or poll_once at all as they help nothing but add extra overhead, I've update the PR (also updated the post with code) with a poll_once and block_on free solution and it turns out to have the best performance.

[Ratysz]

... I assumed you had some reason to do it this way that I'm not aware of.

[hanabi1224]

@Ratysz It's very confusing to let users call block_on, as if they're responsible for scheduling the task, which turns out to be not true in ECS case, also it adds non-trivial performance overhead, after all these useful discussions, here is my proposed API change: #2691

[hanabi1224]

would be a massive footgun

@DJMcNab Could you give a concrete example? My take is otherwise, like ppl may use some other scheduler / async frameworks and cause
confusions and unexpected errors with current exposed APIs, like #2532 block_on and delay should be exposed under bevy::task namespace instead and would be even better if users don't need to call block_on explicitly (with async system functions).

[DJMcNab]

I mean, imagine the reasonable seeming 'solution' to your problem:

async fn handle_tasks(
    mut commands: Commands,
    mut transform_tasks: Query<(Entity, &mut Task<Transform>)>,
    box_mesh_handle: Res<BoxMeshHandle>,
    box_material_handle: Res<BoxMaterialHandle>,
) {
    let futures =  transform_tasks.iter_mut().map(|(e, task)|  async move {
               let transform = (&mut *task).await;
               (e, transform)
    }).collect::<FuturesUnordered>();
    while let Some((entity, transform) = futures.next().await;
    {
            // Add our new PbrBundle of components to our tagged entity
            commands.entity(entity).insert_bundle(PbrBundle {
                mesh: box_mesh_handle.0.clone(),
                material: box_material_handle.0.clone(),
                transform,
                ..Default::default()
            });
            commands.entity(entity).remove::<Task<Transform>>();
        }
    }
}

This would be a fine async function by itself (since it correct doesn't care about the orders of the tasks, etc.)

But in your scheme, where we pretend to have async systems but just make it automatically block, this will naturally block progress in the first frame until all of the tasks are done.

[hanabi1224]

The problem divides into 2 parts, one is whether ECS plans to provide first-class async support, which I consider missing if it requires users to write ad-hoc block_on(s), and the other is how to implement it properly if the answer to the first one is yes

[TheRawMeatball]

Something important that doesn't seem to have been mentioned is the block_on calls in the async compute example aren't actually blocking at all - the block_on(poll_once(&mut *task)) is semantically equilavent to a try_recv you'd call on a channel - it returns Some if it's complete, and None if it's not. The execution is driven elsewhere. We could add a method on Task so users don't need to touch block on directly, but nothing fundamentally changes.

Something else that's very important is, references to anything that lives in the world, be it a Query or a Res can never live across an await point, since for the frame execution to continue, commands which need exclusive world access will need to be ran. As such, the api you suggest is effectively impossible. Async systems have been experimented with before, in #1393, where this restriction was worked around by having an accessor api that scoped access to world data in a synchronous function that is ran as part of the regular schedule. Because that PR also introduced lots of redundant stuff to the simple usage of tasks, the accessor api is now moved to #1900, though that PR is unmaintained atm due to not this api not being as useful as I thought it'd be.

[hanabi1224]

block_on calls in the async compute example aren't actually blocking at all

I understand it's not actually blocking, however, to convert an async function/future into a sync code block, there has to be some engine/scheduler (futures_lite / tokio / async-io) involved, and as long as that happens, it's almost always better to use a single scheduler, as my test shows, doing multiple block_on(s) against non-blocking PollOnce can actually get non-trivial perf penalty, although it does not seem to be so as PollOnce.await returns Option.

[TheRawMeatball]

There is indeed a single scheduler. The future is executed by other threads in the task pool you spawned the future into. This block on + poll once just checks if it's done.

[hanabi1224]

There is indeed a single scheduler.

In the example case, I doubt it. In the setup system, futures are created as pending, and then be waited on in the other system with block_on, I don't see how ECS provides a single scheduler in this case, and if that's true and user's block_on(s) only do simple status check, how to explain the big performance gap seen in the benchmark program?

I searched the code base but could not find any effective block_on(s) in engine code for this case, I think currently it just fully delegate task/green thread/coroutines handling to users, and that's why I was saying first-class async support is missing from current bevy

[TheRawMeatball]

When the future is added to the task pool by using spawn, the returned Task is only a handle that can be used to receive it's result. The actual execution is handled by a group of threads running these futures using async_executor.

bevy/crates/bevy_tasks/src/task_pool.rs

Line 139 in 98a08cf

future::block_on(shutdown_future).unwrap_err();

This block on is the one that actually executes all the tasks - the futures are managed by, and a part of the async executor Executor, and these tasks are simultaneously run by the multiple threads which are spawned here.

[hanabi1224]

@TheRawMeatball You're right, thanks for pointing it out! And in this case, since the task is scheduled on spawn by ECS, there's no reason to call poll_once or block_on in user code at all as that would help nothing but add extra overhead. I updated the PR with another solution that uses neither poll_once nor block_on in user code and it gets the best performance number.
Result:

[handle_tasks_no_poll_once]n_frames executed: 206, avg fps: 34.8(target:120), duration: 5.920s
[handle_tasks]        n_frames executed: 72, avg fps: 12.4(target:120), duration: 5.784s
[handle_tasks_par]    n_frames executed: 123, avg fps: 21.6(target:120), duration: 5.697s
[handle_tasks_par_2]  n_frames executed: 88, avg fps: 15.3(target:120), duration: 5.751s

System code:

#[derive(Clone)]
struct TaskWrapper<T> {
    pub result: Arc<RwLock<Option<T>>>,
}

impl<T> TaskWrapper<T> {
    pub fn new() -> Self {
        Self {
            result: Arc::new(RwLock::new(None)),
        }
    }

    pub async fn register(&mut self, t: impl std::future::Future<Output = T>) {
        let ret = t.await;
        let mut lock = self.result.write().unwrap();
        *lock = Some(ret);
    }
}

fn spawn_tasks_no_poll_once(mut commands: Commands, thread_pool: Res<AsyncComputeTaskPool>) {
    for step in 0..N_STEPS {
        for _i in 0..N_TASKS {
            let wrapper = TaskWrapper::<()>::new();
            let mut wrapper_clone = wrapper.clone();
            let task = thread_pool.spawn(async move {
                wrapper_clone
                    .register(async move {
                        let start_time = Instant::now();
                        let duration = Duration::from_secs_f32(TASK_DURATION_SEC * (step as f32));
                        while Instant::now() - start_time < duration {
                            futures_timer::Delay::new(Duration::from_secs_f32(0.1)).await;
                        }
                    })
                    .await;
            });
            commands.spawn().insert(wrapper.clone()).insert(task);
        }
    }
}

fn handle_tasks_no_poll_once(
    mut commands: Commands,
    transform_tasks: Query<(Entity, &TaskWrapper<()>)>,
    mut app_exit_events: EventWriter<AppExit>,
    time: Res<Time>,
    frame_counter: Res<FrameCounter>,
) {
    let mut n_tasks = 0;
    for (entity, task) in transform_tasks.iter() {
        n_tasks += 1;
        let lock = task.result.read().unwrap();
        if lock.is_some() {
            commands
                .entity(entity)
                .remove::<TaskWrapper<()>>()
                .remove::<Task<()>>();
        }
    }
    if n_tasks == 0 {
        print_statistics("handle_tasks_no_poll_once", &frame_counter, &time);
        app_exit_events.send(AppExit);
    }
}

[Stock84-dev]

Here is another example. We have an async system that runs at the same time as it's sync counterpart would. So that we can use ecs directly. Inside the game loop we poll once on every update. We are basically making a system that polls a future with captured references at correct time.
The only problem is that references cannot live across an await point. Some other system could add a component to an entity which could require reallocation and know reference points to garbage.

But how can we change a reference that is captured by a future? We capture a reference to a reference. Before polling a future we must update all references to the correct location.

But now we have another problem. User could copy inner reference and hold it across an await. For that we use guards. We make them !Sync and require that the future produced by a system is Sync. This will error at compile time if the guard is held across an await point.

We could use a proc macro to hide boilerplate of updating references. For an optimization we could store a reference to a structure of references that is generated by a proc macro. And remove Arc.

Pros

• Async system that can access ecs
• Relatively easy to use
• Arc free, lock free, copy free

Cons

• Requires guard API
• Requires async versions of all SystemParam types, eg. AsyncRes, AsyncQuery. And special states that are held across an await (eg. enumeration index of a query iterator).
• Execution speed is tied to Bevy's scheduler

// How it could look like
#[async_system]
async fn async_system(name: AsyncRes<String>) {
    println!("before");
    {
        let guard = name.get();
        dbg!(guard.deref());
    }
    MyFuture(false).await;
    println!("after");
    let guard = name.get();
    dbg!(guard.deref());
}

Working demo

use std::{
    future::Future,
    marker::PhantomData,
    pin::Pin,
    sync::Arc,
    task::{Context, Poll},
};

use bevy_ecs::prelude::*;
use futures_lite::future;

fn wrapper_system<'w>(
    mut query: Query<(Entity, Option<&mut BevyFuture<()>>)>,
    mut res: ResMut<'w, Vec<String>>,
    mut commands: Commands,
    mut polled: Local<u8>,
) {
    match query.get_single_mut() {
        Ok((id, Some(mut fut))) => {
            let captured_reference = &*fut.captured.0;
            if *polled == 1 {
                println!("changing");
                res.push(String::from("This string has been changed and moved"));
                unsafe {
                    *(captured_reference as *const _ as *mut &String) = &res[1];
                }
            }
            match future::block_on(future::poll_once(&mut fut.future)) {
                None => {}
                Some(_) => {
                    commands.entity(id).despawn();
                }
            }
            *polled += 1;
        }
        _ => {
            println!("spawning another future");
            let reference = &res[0];
            commands.spawn().insert(desugared_async_system(reference));
        }
    }
}

fn desugared_async_system<'a>(item: &'a String) -> BevyFuture<()> {
    let mut item = AsyncRef::new(item);
    BevyFuture {
        captured: item.clone(),
        future: Box::pin(async move {
            println!("before");
            {
                let guard = item.get();
                dbg!(guard.deref());
            }
            MyFuture(false).await;
            println!("after");
            let guard = item.get();
            dbg!(guard.deref());
        }),
    }
}

#[derive(Component)]
struct BevyFuture<F> {
    captured: AsyncRef<String>,
    future: Pin<Box<dyn Future<Output = F> + Send + Sync>>,
}

struct AsyncRef<T: 'static>(Arc<&'static T>);

impl<T> Clone for AsyncRef<T> {
    fn clone(&self) -> Self {
        Self(self.0.clone())
    }
}

struct Guard<'a, T> {
    reference: &'a T,
    _t: PhantomData<*const ()>,
}

impl<'a, T> Guard<'a, T> {
    pub fn deref(&'a self) -> &'a T {
        self.reference
    }
}

impl<T: 'static> AsyncRef<T> {
    pub fn new(value: &T) -> Self {
        // change to 'static lifetime
        Self(Arc::new(unsafe { std::mem::transmute(value) }))
    }

    pub fn get<'a>(&'a mut self) -> Guard<'a, T> {
        let a = *self.0;
        Guard {
            reference: *self.0,
            _t: Default::default(),
        }
    }
}

struct MyFuture(bool);
impl Future for MyFuture {
    type Output = ();

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        if self.0 {
            return Poll::Ready(());
        }
        self.0 = true;
        Poll::Pending
    }
}

#[cfg(test)]
mod tests {
    use bevy::prelude::App;

    use crate::wrapper_system;

    #[test]
    fn test() {
        let mut app = App::empty();
        app.add_default_stages()
            .insert_resource(vec!["Hello world".to_string()]);
        app.add_system(wrapper_system);
        app.update();
        app.update();
        app.update();
        app.update();
        app.update();
        app.update();
    }
}

[alice-i-cecile]

Oh interesting! I'd be interested in seeing a pull-request of this so we can investigate it in depth.

[james7132]

I actually tried a version of this to attempt to allow synchronization systems for rendering systems to be non-blocking on the thread they're running on.

Here's a crude MS pain diagram of what I was aiming to do.

You can see the attempt here: https://github.com/bevyengine/bevy/compare/main...james7132:async-systems?expand=1

Came off of a discussion on Discord starting here: https://discord.com/channels/691052431525675048/749335865876021248/978094783148998716

The main issue with async systems is that you cannot elide lifetimes. They must be explicitly written out. This absolutely obliterates any ergonomics Bevy is known for. For example:

 async fn recv_channel_system<'w, 's, 'a, 'b>(
     mut query: Query<'w, 's, &'a MyChannel>,
     mut round: ResMut<'b, Round>,
 ) {
     for channel in query.iter() {
         println!("My message: {}", channel.recv().await);
     }
}

[alice-i-cecile]

Explicit lifetimes in async systems are not too bad IMO.

[Stock84-dev]

Actually, there is a big problem with my approach. If another future inside an async system captures a reference, then there is no way to update it. Guards would disallow that. But that defeats the point because we would need to clone things.

fn desugared_async_system<'a>(item: &'a String) -> BevyFuture<()> {
    let mut item = AsyncRef::new(item);
    BevyFuture {
        captured: item.clone(),
        future: Box::pin(async move {
            let must_be_cloned;
            {
                let guard = item.get();
                must_be_cloned = guard.deref().clone();
            }
            MyFuture(&must_be_cloned, false).await;
        }),
    }
}

[gilescope]

Certainly with wasm it's not that easy to do async currently given it returns a fake task. Would be amazing to have any kind of out of the box first class async system support whatever the initial limitations. Could it be done as a plugin and then iteratred on out-of-tree?

[jkb0o]

While there is not about first-class async functions, but could be related.
There is https://github.com/jkb0o/pecs, Promise implementation for Bevy:

    commands.add(
         Promise::start(asyn!(state, time: Res<Time> => {
             info!("Wait a second..");
             let started_at = time.elapsed_seconds();
             state.with(started_at).asyn().timeout(1.0)
         }))
         .then(asyn!(state, _ => {
             info!("How large is is the Bevy main web page?");
             state.asyn().http().get("https://bevyengine.org")
         }))
         .then(asyn!(state, result => {
             match result {
                 Ok(response) => info!("It is {} bytes!", response.bytes.len()),
                 Err(err) => info!("Ahhh... something goes wrong: {err}")
             }
             state.pass()
         }))
         .then(asyn!(state, _, time: Res<Time>, mut exit: EventWriter<AppExit> => {
             let duration = time.elapsed_seconds() - state.value;
             info!("It tooks {duration:0.2}s to do this job.");
             info!("Exiting now");
             exit.send(AppExit);
             state.pass()
         })),
     );

[frewsxcv]

I also have https://github.com/frewsxcv/bevy_jobs which can be used to spawn async jobs in the background

[jmjoy]

Has anyone used Bevy to make an online game in practice? Networking should rely on Rust’s async ecosystem, right? Of course, using the Rust standard library’s sync APIs is also possible.