stream: avoid yielding in AllFuture and AnyFuture

[SabrinaJewson]

Motivation

Yielding can be expensive as the task is pushed to the back of the run queue.

Solution

Use a loop to continue to process items from the stream as much as possible, instead of yielding every time an item is produced by the stream.

[taiki-e]

Thanks for the PR! I think this is more efficient than the current implementation (related discussion: rust-lang/futures-rs#2224 (comment)), but this probably causes another problem like #2542...

Using coop::poll_proceed in loop may be the correct solution to that problem, but IIRC the stream module (tokio-stream) is intentionally not using it. EDIT: Oh, tokio-stream is another crate so it can't access the coop module.

cc @jonhoo

[taiki-e]

Given that other futures in tokio-stream (e.g., fold, collect) are implemented in the same way as this PR's implementation, I tend to think that accepting this PR without a workaround for the above problem is fine...

(That said, this may break existing code that depends on the current behavior so a minor version bump is needed.)

cc @carllerche @jonhoo @Darksonn: What do you think?

[Darksonn]

I am ok with doing something like this, but I want an upper limit on the number of iterations in that loop.

[SabrinaJewson]

Ok, I've limited the loop to 32 iterations. I chose 32 because it's the same value used in tokio_stream::iter, but it can really be anything.

[jonhoo]

I think this is a strict improvement over what was there, so this seems worth merging. @taiki-e is right that we still have the problem in #2542, but I think that's a separate issue. Ideally this should call into poll_proceed instead, but as Taiki observed we can't do that since coop is still private (for now).