FuturesUnordered: Limit max value of yield_every

[taiki-e]

UPDATE(taiki-e): Note that this is not an approach we currently use; see #2551 for an approach we currently use.

---

Fixes #2526

The initial version of the #2333 did the same thing (tokio-rs/tokio#3493 (comment)), but at the time I thought it was unnecessary and removed it.

before

n: 10000, time: 315ms
n: 20000, time: 1235ms
n: 40000, time: 4753ms
n: 80000, time: 18912ms
(timeout)

after

n: 10000, time: 29ms
n: 20000, time: 55ms
n: 40000, time: 104ms
n: 80000, time: 202ms
n: 160000, time: 392ms

A little slower than the version using unconstrained, but a little faster than the version using tokio::spawn, on my machine.

with unconstrained

n: 10000, time: 27ms
n: 20000, time: 51ms
n: 40000, time: 93ms
n: 80000, time: 179ms
n: 160000, time: 361ms

with tokio::spawn

n: 10000, time: 50ms
n: 20000, time: 100ms
n: 40000, time: 168ms
n: 80000, time: 340ms
n: 160000, time: 741ms

fyi @jonhoo

[psarna]

Sounds very reasonable to have a constant cap on the number of polls in order to avoid quadratic behavior in Tokio (and I guess also other runtimes that implement preemption). Thanks! Crossing my fingers that it get accepted and backported.

[pkolaczk]

Is there any branch with changes backported to 0.3? I tried to use it but I couldn't make it work with tokio IntervalStream:

error[E0599]: the method `map` exists for struct `IntervalStream`, but its trait bounds were not satisfied
  --> src/main.rs:90:22
   |
90 |                     .map(|_| deadline.next())
   |                      ^^^ method cannot be called on `IntervalStream` due to unsatisfied trait bounds
   |
  ::: /home/pkolaczk/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-stream-0.1.8/src/wrappers/interval.rs:12:1
   |
12 | pub struct IntervalStream {
   | -------------------------
   | |
   | doesn't satisfy `IntervalStream: FutureExt`
   | doesn't satisfy `IntervalStream: futures::Future`
   | doesn't satisfy `IntervalStream: futures::StreamExt`
   | doesn't satisfy `IntervalStream: futures::Stream`
   | doesn't satisfy `IntervalStream: std::iter::Iterator`
   |
   = note: the following trait bounds were not satisfied:
           `IntervalStream: futures::Future`
           which is required by `IntervalStream: FutureExt`
           `IntervalStream: futures::Stream`
           which is required by `IntervalStream: futures::StreamExt`
           `&IntervalStream: futures::Future`
           which is required by `&IntervalStream: FutureExt`
           `&IntervalStream: futures::Stream`
           which is required by `&IntervalStream: futures::StreamExt`
           `&mut IntervalStream: futures::Future`
           which is required by `&mut IntervalStream: FutureExt`
           `&mut IntervalStream: futures::Stream`
           which is required by `&mut IntervalStream: futures::StreamExt`
           `IntervalStream: std::iter::Iterator`
           which is required by `&mut IntervalStream: std::iter::Iterator`
   = help: items from traits can only be used if the trait is in scope
help: the following traits are implemented but not in scope; perhaps add a `use` for one of them:
   |
1  | use std::iter::Iterator;
   |
1  | use futures::FutureExt;
   |
1  | use futures::StreamExt;
   |
1  | use futures_util::future::future::FutureExt;
   |
     and 5 other candidates

I do have use futures::StreamExt in my code...

[jonhoo]

I'd personally make this a const the same way it was in the pre-#2333 implementation, and include a bit more docs (you can probably copy-paste that too from the old code), but otherwise 👍 for this change.

[taiki-e]

@jonhoo Thanks for the review! I've re-added the constant along with docs that was removed in #2333, and added docs about #2333 and this issue.

[taiki-e]

Published in 0.3.19.

[nagisa]

Shouldn't this test be adjusted to ensure that eventually all of the contained futures are polled?

[taiki-e]

The purpose of the test is to make sure that the same future is not polled multiple times in a single poll_next call (#2333), so I think it's fine as is.

futures-rs/futures/tests/stream_futures_unordered.rs

Line 316 in d3747e6

fn polled_only_once_at_most_per_iteration() {