Function run_ordered_parallel_with_limit

core/src/ordered_parallel.rs:46–112

(
    items: Vec<I>,
    max_concurrency: usize,
    run: F,
)

}

pub(crate) async fn run_ordered_parallel_with_limit<I, F, Fut, T>(
    items: Vec<I>,
    max_concurrency: usize,
    run: F,
) -> Vec<OrderedParallelResult<T>>
where
    I: Send + 'static,
    F: Fn(usize, I) -> Fut + Clone + Send + 'static,
    Fut: Future<Output = T> + Send + 'static,
    T: Send + 'static,
{
    let mut join_set = JoinSet::new();
    let mut pending = items.into_iter().enumerate();
    let max_concurrency = max_concurrency.max(1);
    let mut active = 0usize;

    while active < max_concurrency {
        let Some((index, item)) = pending.next() else {
            break;
        };
        let run = run.clone();
        join_set.spawn(async move {
            let output = AssertUnwindSafe(run(index, item))
                .catch_unwind()
                .await
                .map_err(|payload| {
                    OrderedParallelError::Panicked(panic_payload_to_string(payload))
                });
            OrderedParallelResult { index, output }
        });
        active += 1;
    }

    let mut results = Vec::new();
    while active > 0 {
        if let Some(result) = join_set.join_next().await {
            active -= 1;
            match result {
                Ok(result) => results.push(result),
                Err(error) => results.push(OrderedParallelResult {
                    index: usize::MAX,
                    output: Err(OrderedParallelError::JoinFailed(error.to_string())),
                }),
            }
        }

        while active < max_concurrency {
            let Some((index, item)) = pending.next() else {
                break;
            };
            let run = run.clone();
            join_set.spawn(async move {
                let output = AssertUnwindSafe(run(index, item))
                    .catch_unwind()
                    .await
                    .map_err(|payload| {
                        OrderedParallelError::Panicked(panic_payload_to_string(payload))
                    });