Function respects_concurrency_limit

core/src/ordered_parallel.rs:169–200

()

    #[tokio::test]
    async fn respects_concurrency_limit() {
        use std::sync::atomic::{AtomicUsize, Ordering};
        use std::sync::Arc;

        let active = Arc::new(AtomicUsize::new(0));
        let max_active = Arc::new(AtomicUsize::new(0));
        let results = run_ordered_parallel_with_limit(vec![1_u8, 2, 3, 4], 2, {
            let active = Arc::clone(&active);
            let max_active = Arc::clone(&max_active);
            move |_index, value| {
                let active = Arc::clone(&active);
                let max_active = Arc::clone(&max_active);
                async move {
                    let now = active.fetch_add(1, Ordering::SeqCst) + 1;
                    max_active.fetch_max(now, Ordering::SeqCst);
                    tokio::time::sleep(std::time::Duration::from_millis(30)).await;
                    active.fetch_sub(1, Ordering::SeqCst);
                    value
                }
            }
        })
        .await;

        assert_eq!(max_active.load(Ordering::SeqCst), 2);
        assert_eq!(
            results
                .into_iter()
                .map(|result| result.output.unwrap())
                .collect::<Vec<_>>(),
            vec![1, 2, 3, 4]
        );
    }
}