Implementing a custom async task pool

[rohel01]

Hello,

I am using Bevy to implement a camera control software where the device main control loop must run at 10Hz.
I implemented image transfers using async tasks that I spawn on the default IoTaskPool

During testing, I noticed a few things:

1. Data transfers were surprisingly slow
2. Transfer speed was tied to the main scheduler configured loop period (ScheduleRunnerPlugin::run_loop)

Looking at Bevy's source code, I believe the reason for 2. is pool executors are ticked as part of the application main loop. This is perfectly reasonable in games where main loops are expected to run between 25Hz to 240Hz. But, unfortunately for me, this does not work well with the slow loop of my use case. This explains 1.

So, I am considering implementing my own task pool plugin which I would tick separately. However, I found this warning in Bevy's code:

/// A function used by `bevy_app` to tick the global tasks pools on the main thread.
/// This will run a maximum of 100 local tasks per executor per call to this function.
///
/// # Warning
///
/// This function *must* be called on the main thread, or the task pools will not be updated appropriately.
#[cfg(not(all(target_arch = "wasm32", feature = "web")))]
pub fn tick_global_task_pools_on_main_thread() {

Can someone elaborate of why this needed and if it applies to my use case?

[SkiFire13]

How are you spawning your tasks? AFAIK that function is needed only for making progress on local tasks, i.e. those spawned using spawn_local.

[rohel01]

I am using spawn

    IoTaskPool::get()
        .spawn(async move {
            // My async task code
        })
        .detach();

[SkiFire13]

Just to be sure, are you perhaps blocking the threads of the task pool with some blocking code? What happens if you try to increase the number of IO threads in the TaskPoolPlugin configuration?

[hymm]

If you're not using DefaultPlugins, you might need to configure the task pools or add the TaskPoolPlugin. https://docs.rs/bevy/latest/bevy/prelude/struct.TaskPoolPlugin.html

[rohel01]

Yes, I am manually adding the TaskPoolPlugin to my app

The task pool is working. My tasks are making progress, but I observed the rate at which they are making progress is somehow tied to application main loop cycle period.

[vshulcz]

Before writing a custom pool, check whether your build has bevy's multi_threaded feature. If it doesn't, both of your observations follow directly, and a hand-rolled pool would run into the same thing.

Without that feature bevy_tasks compiles single_threaded_task_pool.rs, where spawn is not what the name suggests:

pub fn spawn<T>(&self, future: ...) -> Task<T> {
    LOCAL_EXECUTOR.with(|executor| {
        let task = executor.spawn(future);
        // Loop until all tasks are done
        while executor.try_tick() {}
        task
    })
}

pub fn spawn_local<T>(&self, future: ...) -> Task<T> {
    self.spawn(future)
}

Everything lands on a thread-local executor, and after the spawning call returns it only advances when somebody ticks that executor. The only thing doing that is the function you quoted, called once per main loop iteration:

for _ in 0..100 {
    compute_local_executor.try_tick();
    async_local_executor.try_tick();
    io_local_executor.try_tick();
}

So at 10 Hz your transfers get ten tick windows per second, with at most 100 polls per pool in each. That is the coupling you measured.

The feature comes in through default_platform in bevy's Cargo.toml, which is part of the 2d/3d/ui profiles. A headless app built with default-features = false, which is the usual shape for something driven by ScheduleRunnerPlugin, drops it. Quick check:

cargo tree -e features -i bevy_tasks

If multi_threaded isn't listed, add it to your bevy dependency features.

To your actual question about the warning: with the multi-threaded pool, spawn hands the future to the pool's own worker threads and the main loop is not involved at all. tick_global_task_pools_on_main_thread exists only to drive the local executors behind spawn_local, for futures that aren't Send. They live in thread-local storage, hence the "must be called on the main thread". For spawn on a real pool it is irrelevant.

If you turn out to already be on the multi-threaded pool, then the next thing I'd look at is thread count rather than ticking. IoTaskPool defaults to a quarter of your cores, at least 1 and at most 4:

io: TaskPoolThreadAssignmentPolicy {
    min_threads: 1,
    max_threads: 4,
    percent: 0.25,
    ...

Any blocking call inside those futures occupies one of those threads for its whole duration, so a handful of blocking transfers is enough to serialise everything. That would show up as slow transfers, though it wouldn't tie them to run_loop.

[rohel01]

Thank you for this very detailed analysis.

Your first guess is correct, I did not have multi_threaded activated.

My need was to have the main loop single threaded because I have very few systems that do not benefit from parallelization. However, I still want to use the other task pools in a multi-threaded fashion if that makes sense.

[rohel01]

Edit: I believe I can achieve what I described above if I configure the Main schedule to use ExecutorKind::SingleThreaded. Do you confirm?