FuturesUnordered: Limit max value of yield_every

Author: taiki-e

futures-util/src/stream/futures_unordered/mod.rs

@@ -6,6 +6,7 @@
 use crate::task::AtomicWaker;
 use alloc::sync::{Arc, Weak};
 use core::cell::UnsafeCell;
+use core::cmp;
 use core::fmt::{self, Debug};
 use core::iter::FromIterator;
 use core::marker::PhantomData;
@@ -393,11 +394,14 @@ impl<Fut: Future> Stream for FuturesUnordered<Fut> {
         // caps the number of calls to `poll` on underlying futures a single call to
         // `poll_next` is allowed to make.
         //
-        // The value is the length of FuturesUnordered. This ensures that each
-        // future is polled only once at most per iteration.
+        // The value itself is chosen somewhat arbitrarily. It needs to be high enough
+        // that amortize wakeup and scheduling costs, but low enough that we do not
+        // starve other tasks for long.
+        //
+        // Each future is polled only once *at most* per iteration.
         //
         // See also https://github.com/rust-lang/futures-rs/issues/2047.
-        let yield_every = self.len();
+        let yield_every = cmp::min(self.len(), 32);
 
         // Keep track of how many child futures we have polled,
         // in case we want to forcibly yield.

futures/tests/stream_futures_unordered.rs

@@ -340,7 +340,7 @@ fn polled_only_once_at_most_per_iteration() {
 
     let mut tasks = FuturesUnordered::from_iter(vec![F::default(); 33]);
     assert!(tasks.poll_next_unpin(cx).is_pending());
-    assert_eq!(33, tasks.iter().filter(|f| f.polled).count());
+    assert_eq!(32, tasks.iter().filter(|f| f.polled).count());
 
     let mut tasks = FuturesUnordered::<F>::new();
     assert_eq!(Poll::Ready(None), tasks.poll_next_unpin(cx));