Waiting for a list of futures in Rust

Question

I am attempting to make a future that continuously finds new work to do and then maintains a set of futures for those work items. I would like to make sure neither my main future that finds work to be blocked for long periods of time and to have my work being done concurrently.

Here is a rough overview of what I am trying to do. Specifically isDone does not exist and also from what I can understand from the docs isn't necessarily a valid way to use futures in Rust. What is the idomatic way of doing this kind of thing?

use std::collections::HashMap;
use tokio::runtime::Runtime;

async fn find_work() -> HashMap<i64, String> {
    // Go read from the DB or something...
    let mut work = HashMap::new();
    work.insert(1, "test".to_string());
    work.insert(2, "test".to_string());
    return work;
}

async fn do_work(id: i64, value: String) -> () {
    // Result<(), Error> {
    println!("{}: {}", id, value);
}

async fn async_main() -> () {
    let mut pending_work = HashMap::new();
    loop {
        for (id, value) in find_work().await {
            if !pending_work.contains_key(&id) {
                let fut = do_work(id, value);
                pending_work.insert(id, fut);
            }
        }

        pending_work.retain(|id, fut| {
            if isDone(fut) {
                // do something with the result
                false
            } else {
                true
            }
        });
    }
}

fn main() {
    let runtime = Runtime::new().unwrap();
    let exec = runtime.executor();

    exec.spawn(async_main());
    runtime.shutdown_on_idle();
}

Ransom

Maybe you are looking for something like join!() or select!() – Perfumer

Thanks for the suggestion, I looked at those but unfortunately join! blocks my main loop that needs to check for new work and both select and join take in a fixed number of futures rather than a map or vec so they feel a little clumsy at best to use. I will try select out and see if it works well. – Ransom

Futures perform work only when they are polled, so maybe you should just poll() them ? If the future is done you'll get Poll::Ready, otherwise - Poll::Pending which is the same as your isDone() method ecept that it returns an enum with two variants instead of true/false – Perfumer

@Nick, did you ever find an answer to this? – Eventual

Not exactly as described but I ended up using futures::future::join_all (docs.rs/futures/0.3.5/futures/future/fn.join_all.html) which can take any IntoInterator and that worked well enough for me. – Ransom

Actually while grabbing the latest documentation link for my comment above I noticed this: docs.rs/futures/0.3.5/futures/stream/… Which appears to be exactly what I was looking for – Ransom

@Ransom post that as an answer – Tertius

Answer 1

As @Nick already answered, you can use futures::future::join_all from the futures crate.

Calcutta