Peek future for Peekable stream

[dingxiangfei2009]

Fix #1871

This pull request is an attempt to provide a sensible peeking behaviour for Peekable streams. Instead of exposing wires that resembles poll from Future, this construction wraps this exposure into a proper Future, with pinned borrow of the underlying Peekable. It works by swapping out the mutable borrow, attempting a poll, then either swapping the mutable borrow back in if Pending, or producing the Poll directly.

By the construction, peek() can be reinstated and the documentation becomes accurate again.

[Nemo157]

Changing poll_peek is a breaking change. It should be possible to implement the future while leaving poll_peek alone (just something like self.inner.as_mut().poll_peek(...)).

[dingxiangfei2009]

@Nemo157 Is it possible? self: Pin<&mut Peekable<_>> cannot be recovered in case "poll_peek" returns Poll::Pending.

[dingxiangfei2009]

Because of this, poll_peek at this current form is hardly of any use in and outside async block. In the current type signature, we cannot construct it in an FnMut and use it in crate::future::poll_fn, and we cannot construct concrete Future types that eventually resolves to a borrow of the "head" item... Unless it is intended that any Future polling on items from Peekable resolves immediately even if the underlying Stream is not ready, but this behaviour feels like a big breakage from the rest of the Futures constructible from this crate.

The point is to make polling on Peek can still eventually resolve in case the Peekable is not ready at the time of polling. Though I am interested to learn what is the original design, too.

[Nemo157]

Pin::as_mut reborrows the Pin<&mut _> for the duration of the poll_peek call.

[dingxiangfei2009]

@Nemo157 Sorry, backtracking a little here.

I just realised that it is still not possible to use the original poll_peek. The main difficulty is because Peek needs return a borrow of bits of Peekable.

Say we use the original poll_peek. I am playing with the following implementation of Future:

impl<'a, St> Future for Peek<'a, St>
where
    St: Stream,
{
    type Output = Option<&'a St::Item>;
    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        if let Some(mut peekable) = self.inner.take() {
            match peekable.as_mut().poll_peek(cx) {
                Poll::Pending => {
                    self.inner = Some(peekable);
                    Poll::Pending
                }
                Poll::Ready(poll) => Poll::Ready(poll),
            }
        } else {
            Poll::Pending
        }
    }
}

Now we start to see lifetime issues, because we are borrowing from the peekable before as_mut.

error[E0505]: cannot move out of `peekable` because it is borrowed
   --> futures-util/src/stream/stream/peek.rs:165:39
    |
156 | impl<'a, St> Future for Peek<'a, St>
    |      -- lifetime `'a` defined here
...
163 |             match peekable.as_mut().poll_peek(cx) {
    |                   -------- borrow of `peekable` occurs here
164 |                 Poll::Pending => {
165 |                     self.inner = Some(peekable);
    |                                       ^^^^^^^^ move out of `peekable` occurs here
...
168 |                 Poll::Ready(poll) => Poll::Ready(poll),
    |                                      ----------------- returning this value requires that `peekable` is borrowed for `'a`

error[E0515]: cannot return value referencing local variable `peekable`
   --> futures-util/src/stream/stream/peek.rs:168:38
    |
163 |             match peekable.as_mut().poll_peek(cx) {
    |                   -------- `peekable` is borrowed here
...
168 |                 Poll::Ready(poll) => Poll::Ready(poll),
    |                                      ^^^^^^^^^^^^^^^^^ returns a value referencing data owned by the current function

error: aborting due to 2 previous errors

[dingxiangfei2009]

There is a bit of tricky borrowing happening here when we stick to the old poll_peek. Pin::as_mut takes a mutable borrow, and in this case the first peekable is then mutably borrowed. The return value of poll_peek serves as a witness of this borrow, too, because it is a (potential) borrow of bits of Peekable. However, this mutable borrow only spans across Peek::poll, meaning that we cannot return this borrow of the head item, nor can we shelf the Peekable up for another poll.

Returning the mutable borrow back from poll_peek, however, will enable us to do both of them.

[dingxiangfei2009]

@Nemo157 I have been playing around with Pin::as_mut with no success. Maybe I am missing something, so it would be great if you have suggestions or hints. 🙏 😄

[Nemo157]

Yeah, this is actually a really nasty intersection of lifetimes (similar to poll_fill_buf which I just realised we also don't have a Future returning function for). I've tried a couple of things but the only way I found was by accessing the Peekable internals directly, which is basically equivalent to your implementation.

For backwards compatibility I think you should rename the internal method something like fn do_poll_peek, then provide the existing pub fn poll_peek which calls it and just discards the Left value.

[cramertj]

Nice! Can you add a test for this?

[dingxiangfei2009]

@cramertj Sure. A simple is added as an illustration of its function.

[cramertj]

LGTM modulo remaining CI failures, which look legit:

• doc issues
• clippy (this one isn't your fault-- feel free to either fix or ignore)

[cramertj]

Thanks!