Как я могу объединить все фьючерсы в векторе, не отменяя при неудаче, как join_all?

Question

У меня есть Vec фьючерса, созданного в результате вызова функции async. После добавления всех фьючерсов к вектору, я хотел бы подождать весь набор, получая либо список результатов, либо обратный вызов для каждого, который заканчивается.

Я мог бы просто l oop или итерируйте вектор фьючерсов и вызывайте .await в каждом будущем, и это позволило бы мне правильно обрабатывать ошибки, а не futures::future::join_all отменять другие, но я уверен, что есть идиоматизм c способ выполнить sh эту задачу.

Я также хочу иметь возможность обрабатывать фьючерсы по мере их завершения, поэтому, если я получу достаточно информации из первых нескольких, я могу отменить оставшиеся неполные фьючерсы и не ждать для них и отбросить их результаты, ошибка или нет. Это было бы невозможно, если бы я перебрал вектор по порядку.

Мне нужен обратный вызов (closure, et c), который позволяет мне накапливать результаты по мере их поступления, чтобы я мог может обрабатывать ошибки надлежащим образом или отменять оставшуюся часть фьючерсов (из-за обратного вызова), если я решу, что остальные из них мне не нужны.

Я могу сказать, что запрашивает головную боль у средства проверки заимствований: попытка изменить будущее в Vec в обратном вызове из асинхронного двигателя c.

Существует множество вопросов о переполнении стека и публикациях Reddit, в которых объясняется, как join_all включается в список фьючерсов, но отменяется остальное - в случае отказа, и как асинхронные c движки могут порождать потоки, а могут - нет или плохой дизайн, если они это делают.

Answer 1

Использовать futures::select_all:

use futures::future; // 0.3.4

type Error = Box<dyn std::error::Error + Send + Sync + 'static>;
type Result<T, E = Error> = std::result::Result<T, E>;

async fn might_fail(fails: bool) -> Result<i32> {
    if fails {
        Ok(42)
    } else {
        Err("boom".into())
    }
}

async fn many() -> Result<i32> {
    let raw_futs = vec![might_fail(true), might_fail(false), might_fail(true)];
    let unpin_futs: Vec<_> = raw_futs.into_iter().map(Box::pin).collect();
    let mut futs = unpin_futs;

    let mut sum = 0;

    while !futs.is_empty() {
        match future::select_all(futs).await {
            (Ok(val), _index, remaining) => {
                sum += val;
                futs = remaining;
            }
            (Err(_e), _index, remaining) => {
                // Ignoring all errors
                futs = remaining;
            }
        }

        if sum > 42 {
            // Early exit
            return Ok(sum);
        }
    }

    Ok(sum)
}

Опрашивает все фьючерсы в коллекции, возвращая первое, которое не ожидает, его индекс и оставшееся ожидание или незаполненные фьючерсы. Затем вы можете сопоставить на Result и обработать случай успеха или неудачи.

Звоните select_all внутри ЭЛ oop. Это дает вам возможность рано выйти из функции. Когда вы выходите, вектор futs сбрасывается, а отбрасывание будущего отменяет его.