Add stream::Zip trait

[yoshuawuyts]

This is a core iteration "order of execution" operation, enabling people to author:

after

let s1 = async_iter::repeat(0).take(5);
let s2 = async_iter::repeat(1).take(5);
let s3 = async_iter::repeat(2).take(5);

for await (n1, n2, n3) in (s1, s2, s3).zip() {
    assert_eq!((n1, n2, n3), (0, 1, 2));
}

This is better than the nested tuple approach of non-trait based zip.

Design

pub trait Zip {
    /// What's the return type of our stream?
    type Item;

    /// What stream do we return?
    type Stream: Stream<Item = Self::Item>;

    /// Combine multiple streams into a single stream.
    fn zip(self) -> Self::Stream;
}

Tasks

• Zip trait Init zip and chain traits #73
• impl Zip for vec trait Init zip and chain traits #73
• impl Zip for array trait Init zip and chain traits #73
• impl Zip for tuple
• benchmarks

[yoshuawuyts]

For the Zip variant: we may want to skip directly ahead to #22 in terms of implementation, creating an output slot of Container[MaybUninit<T>] where we write into, and then mem::swap on each iteration.

Metadata would need to be stored in an accompanying PollState slot. The main things to keep an eye will be to:

• ensure to implement the PinnedDrop trait to drop memory in the case of a cancellation
• switch between PollState::Pending and PollState::Ready while data is actively being yielded, and only switch to PollState::Consume at the end.

The semantics of Zip are that if at any point one of the streams yields Poll::Ready(None), all other streams are cancelled. We should create a case where one stream yields more times than the other streams so that we can exercise miri's checkers during CI.

[yoshuawuyts]

Filed a concern in rust-lang/rust#80094 since its existence requires us to author:

let mut s = Zip::zip([a, b, c]);

instead of:

let mut s = [a, b, c].zip();