Is Concurrent Processing of JavaScript Async Tasks Feasible in rquickjs?

[mrxsisyphus]

I'm exploring the possibility of batch processing JavaScript async tasks within Rust's asynchronous code in rquickjs.

1. tokio: I'm encountering problems using tokio::spawn because some types from rquickjs do not implement the Send trait, preventing safe usage across threads:

let handle = tokio::spawn(async move {
        task_fn
            .call::<(), Promise>(())?
            .into_future::<Value>()
            .await
});

msg: *mut c_void cannot be sent between threads safely

2. futures::stream: I've considered using futures::stream with buffer_unordered to manage concurrent tasks. While it seems like a potential solution, I'm unsure about its effectiveness and performance in this scenario.

 #[rquickjs::function]
    pub async fn execute_tasks<'js>(
        task_functions: Vec<Function<'js>>,
        max_concurrent: usize,
    ) -> Vec<rquickjs::Result<Value>> {
        let results = stream::iter(task_functions)
            .map(|task_fn| async move {
                task_fn
                    .call::<(), Promise>(())?
                    .into_future::<Value>()
                    .await
            })
            .buffer_unordered(max_concurrent)
            .collect()
            .await;
        results
    }

so Is it feasible to use Rust to perform parallel processing of JavaScript method calls?
Are there any best practices or alternative approaches you would recommend for concurrent handling of async JavaScript tasks in Rust?

Any guidance or examples would be greatly appreciated!

Thank you!

[richarddd]

Use ctx.spawn()

[mrxsisyphus]

@richarddd
Thank you for your quick answer. Could you please further explain this method? And, if possible, provide a usable example?

[DelSkayn]

Tasks can run concurrent via js promises and rust futures. Tasks cannot run in parallel within the same runtime, rquickjs is not thread safe so only a single thread can have access to the runtime at once. If you need parallel execution you will have to create multiple runtimes for each thread.

If you only need concurrent execution you can use ctx::spawn to spawn a future to run within the runtimes schedular. It works pretty much like tokio::spawn except that it will only run futures within the runtime.

[mrxsisyphus]

@DelSkayn
Thank you for your patient response. So, if I use single-threaded Tokio, would it achieve the same effect?
It just feels like ctx.spawn is related to the QJS runtime, so perhaps it might be more efficient?

However, what I’m struggling with now is how to use ctx.spawn to implement the scenario I mentioned above.
After trying, I found that the#[rquickjs::function]macro cannot apply the ctx value. And even if it could be applied in some way, how would it integrate with Tokio for asynchronous waiting and handling of the completion of these ctx.spawn calls?

[DelSkayn]

So most of the time you shouldn't really have to use ctx.spawn.
Just to illustrate how to work with futures and promises in rquickjs see the following example.

use std::time::Duration;

use rquickjs::{
    async_with, AsyncContext, AsyncRuntime, CatchResultExt, Coerced, Ctx, Error, Function, Result,
};

#[rquickjs::function]
fn print(txt: Coerced<String>) {
    println!("{}", txt.0);
}

#[rquickjs::function]
async fn read_file<'js>(path: String) -> Result<String> {
    tokio::fs::read_to_string(path).await.map_err(Error::Io)
}

#[rquickjs::function]
fn timeout<'js>(ctx: Ctx<'js>, count: u32, cb: Function<'js>) -> Result<()> {
    ctx.clone().spawn(async move {
        tokio::time::sleep(Duration::from_millis(count as u64)).await;
        if let Err(e) = cb.call::<_, ()>(()).catch(&ctx) {
            println!("Timeout call returned error: {}", e)
        }
    });
    Ok(())
}

#[tokio::main]
pub async fn main() {
    let rt = AsyncRuntime::new().unwrap();
    let ctx = AsyncContext::full(&rt).await.unwrap();

    async_with!(ctx => |ctx| {
        // define the functions for access from javascript
        ctx.globals().set("read_file", js_read_file).unwrap();
        ctx.globals().set("timeout", js_timeout).unwrap();
        ctx.globals().set("println", js_print).unwrap();

        // actually run some js.
        ctx.eval_promise(r#"
            // Read a file
            println(await read_file('Cargo.toml')); 

            // set a timeout
            timeout(500,() => {
                println("done!")
            })
        "#).catch(&ctx).unwrap().into_future::<()>().await.catch(&ctx).unwrap();

    })
    .await;

    // The above will return before running the timeout callback.
    // If all futures need to befinished you must drive the runtime until it is complete.
    // The call below ensures that no more futures are pending.
    rt.idle().await;
}

Most of the time you can just convert between rust futures and js promises and have rquickjs take care of running them.
Spawn is only necessary if you want to run things concurrently. So if you want to run a bunch of scripts you can spawn a bunch of eval calls. Note that spawning will generally be slower then just running the sequentially if the scripts you are running can all run without ever needing to wait on a future or promise.

[mrxsisyphus]

@DelSkayn
Thank you for your clear examples!

#[rquickjs::function]
fn timeout<'js>(ctx: Ctx<'js>, count: u32, cb: Function<'js>) -> Result<()> {
    ctx.clone().spawn(async move {
        tokio::time::sleep(Duration::from_millis(count as u64)).await;
        if let Err(e) = cb.call::<_, ()>(()).catch(&ctx) {
            println!("Timeout call returned error: {}", e);
        }
    });
    Ok(())
}

Does the cb here support accepting a JavaScript async function(promises) with a return value?
Is it reasonable to express it this way (I'm not sure if this is correct):

task_fn
    .call::<(), Promise>(()).catch(&ctx)?
    .into_future::<Value>()
    .await;

And if there is a return value, how can it be obtained after ctx.spawn?