util: add a channel body by davidpdrsn · Pull Request #100 · hyperium/http-body

davidpdrsn · 2023-11-19T18:22:43Z

Adds a new body type that's backed by a tokio::sync::mpsc channel. Inspired by hyper's old Body::channel.

seanmonstar · 2023-11-20T21:43:06Z

http-body-util/src/channel.rs

+
+impl<D, E> Sender<D, E> {
+    /// Send a frame on the channel.
+    pub async fn send(&self, frame: Frame<D>) -> Result<(), SendError> {


Even though it's not required internally, should we make these all &mut self?

this is addressed in 4fed62f

seanmonstar · 2023-11-20T21:44:00Z

http-body-util/src/channel.rs

+/// A body backed by a channel.
+pub struct Channel<D, E = std::convert::Infallible> {
+    rx_frame: mpsc::Receiver<Frame<D>>,
+    rx_error: mpsc::Receiver<E>,


Would a oneshot channel work? It should be a smaller struct, and use less memory wen sending the error (the mpsc channel reserves blocks).

At first I thought it wouldn't be possible because one shot::Receiver doesn't have a poll_recv method but I suppose I can call Future::poll instead 🤔 I'll try that.

i ran into this issue myself, and noticed this thread here. i've opened tokio-rs/tokio#7059 to introduce a poll_recv(..) method for oneshot::Receiver<T>.

tokio-rs/tokio#7059 is now closed. to poll the receiver, it should be pinned and polled via Future::poll, rather than adding a poll_recv method matching mpsc::Receiver<T>.

this is addressed in 08aba95

Didn't mean to approve just yet

this commit adds a `rx.poll_recv(&mut cx)` to the public interface of `tokio::sync::oneshot::Receiver<T>`. this method has the following signature: ```rust // tokio/src/sync/oneshot.rs impl<T> Receiver<T> { pub fn poll_recv(&mut self, cx: &mut Context<'_>) -> Poll<Result<T, RecvError>> { // ... } } ``` this is similar to the `tokio::sync::mpsc::Receiver::poll_recv` and `tokio::sync::mpsc::UnboundedReceiver::poll_recv` methods, which have the following signature: ```rust // tokio/src/sync/mpsc/bounded.rs impl<T> Receiver<T> { pub fn poll_recv(&mut self, cx: &mut Context<'_>) -> Poll<Option<T>> { // ... } } ``` see: https://docs.rs/tokio/latest/tokio/sync/mpsc/struct.Receiver.html#method.poll_recv in particular, note the `&mut self` receiver of these methods, as opposed to the `Pin<&mut Self>` receiver in `Future::poll(..)`. today, a oneshot receiver must be pinned in order to be polled via `Future::poll(..)`. `tokio::sync::oneshot::Receiver::try_recv(..)` has an important but subtle difference from `poll_recv(..)`, alluded to in its documentation: > If a pending value exists in the channel, it is returned. If no value > has been sent, the current task will not be registered for future > notification. > > This function is useful to call from outside the context of an > asynchronous task. see hyperium/http-body#100 for an example use-case for this. if we *are* in the context of an asynchronous task, we may wish to poll on the receiver-end of the channel and register for future notification, indicating that we should be awoken later when a value is ready or when conditions yielding a spurious failure have passed. providing a means to poll a `&mut Receiver<T>` avoids the performance impact of boxing the receiver as an erased `dyn Future` trait object, or of using an `tokio::sync::mpsc::Receiver<T>`, or the ergonomic wrinkles of needing to rely on pin projection in asynchronous types that compose on top of oneshot channels. --- * https://docs.rs/tokio/latest/tokio/sync/mpsc/struct.Receiver.html#method.poll_recv * https://docs.rs/tokio/latest/tokio/sync/mpsc/struct.UnboundedReceiver.html#method.poll_recv * https://doc.rust-lang.org/stable/std/future/trait.Future.html#tymethod.poll * https://docs.rs/tokio/latest/tokio/sync/oneshot/struct.Receiver.html#method.try_recv * https://github.com/hyperium/http-body/pull/100/files#r1399818104 * hyperium/http-body#100

this applies a review suggestion here: https://github.com/hyperium/http-body/pull/100/files#r1399781061 this commit refactors the channel-backed body in hyperium#100, changing the `mpsc::Receiver<E>` used to transmit an error into a `oneshot::Receiver<E>`. this should improve memory usage, and make the channel a smaller structure. in order to achieve this, some minor adjustments are made: * use pin projection, projecting pinnedness to the oneshot receiver, polling it via `core::future::Future::poll(..)` to yield a body frame. * add `Debug` bounds were needed. as an alternative, see tokio-rs/tokio#7059, which proposed a `poll_recv(..)` inherent method for a oneshot channel receiver.