Capability to make an existing Stream participate in cooperative scheduling

[pepijnve]

Is your feature request related to a problem? Please describe.

I'm trying to make a Stream that represents some resource participate in cooperative scheduling. From reading the Tokio source code I think this would boil down to a decorator that looks something like this.

fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
    let coop = ready!(tokio::task::coop::poll_proceed(cx));

    // do the actual Stream::poll_next
    let ret = self.poll_next_internal(cx);
    if ret.is_ready() {
        coop.made_progress();
    }
    ret
}

poll_proceed is not accessible so this is currently not possible. I don't think I can achieve the same result using consume_budget

Describe the solution you'd like

Some mechanism to make an existing pollable item participate in cooperative scheduling.

[pepijnve]

I will happily leave this to the experts, but maybe something like this?

fn with_consume_budget<T, F>(poll: F, cx: &mut Context<'_>) -> Poll<T> 
where F: FnMut(&mut Context<'_>) -> Poll<T> {
    let coop = ready!(tokio::task::coop::poll_proceed(cx));
    let ret = poll(cx);
    if ret.is_ready() {
        coop.made_progress();
    }
    ret
}

[Darksonn]

In my opinion, we should just expose poll_proceed.

[pepijnve]

That would be even better. I thought that might be too much to ask for.

[pepijnve]

In the meantime I found cooperative in the source code as well which is exactly what I was looking for.