sync: add `Receiver::poll_recv(..)` method #7059

cratelyn · 2024-12-31T22:40:01Z

this commit adds a rx.poll_recv(cx) method to the public interface of tokio::sync::oneshot::Receiver<T>.

this method has the following signature:

// 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:

// tokio/src/sync/mpsc/bounded.rs
impl<T> Receiver<T> {
    pub fn poll_recv(&mut self, cx: &mut Context<'_>) -> Poll<Option<T>> {
        // ...
    }
}

see:

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.

Motivation

Solution

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

seanmonstar · 2024-12-31T23:21:48Z

If Receiver is Unpin, then it should be easy to wrap the borrow in a Pin and just call Future::poll. Is that not possible?

cratelyn · 2025-01-02T13:20:36Z

i'll close this. a receiver can be pinned, as described above. while there isn't any strict need for it to be pinned, i'm not especially attached to this proposal.

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.

* Add channel body * review: use `sync::oneshot` for error channel this applies a review suggestion here: https://github.com/hyperium/http-body/pull/100/files#r1399781061 this commit refactors the channel-backed body in #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. * review: use `&mut self` method receivers this applies a review suggestion here: https://github.com/hyperium/http-body/pull/100/files#r1399780355 this commit refactors the channel-backed body in #100, changing the signature of `send_*` methods on the sender to require a mutable reference. * review: fix `<Channel<D, E> as Body>::poll_frame()` see: #140 (comment) this commit adds test coverage exposing the bug, and tightens the pattern used to match frames yielded by the data channel. now, when the channel is closed, a `None` will flow onwards and poll the error channel. `None` will be returned when the error channel is closed, which also indicates that the associated `Sender` has been dropped. --------- Co-authored-by: David Pedersen <[email protected]>

note: this commit will not compile, code changes are intentionally elided from this commit. this commit upgrades hyper, http, tonic, prost, related dependencies, and their assorted cargo features. see <linkerd/linkerd2#8733>. see also: * #3379 * #3380 * #3382 * #3405 * hyperium/hyper#3796 * #3411 * #3421 * #3427 * #3428 * #3432 * #3433 * #3444 * #3445 * #3454 * #3455 * #3456 * #3457 * #3461 * #3459 * #3465 * #3466 * #3467 * #3468 * linkerd/linkerd2-proxy-api#421 * linkerd/linkerd2#13492 * linkerd/linkerd2#13493 * hyperium/hyper#3816 * #3472 * #3473 * #3479 * tokio-rs/tokio#7059 * #3509 * hyperium/http-body#140 * #3515 * hyperium/http-body#141 * #3530 * #3531 * #3540 * #3556 * #3558 * #3559 * #3564 * #3567 * #3573 * #3583 * hyperium/http-body#144 * #3585 * #3586 * #3597 * #3598 * #3611 * #3614 * #3615 * #3616 * #3647 * #3651 * #3653 * #3654 * #3655 * #3656 * #3657 * #3660 * #3671 * #3672 * #3673 * #3676 * hyperium/http-body#147 * #3692 * #3699 * #3700 * #3701 * #3708 * linkerd/drain-rs#36 * #3715 * #3717 * eminence/procfs#340 --- squash: chore(deps): add hyper-util workspace dependency chore(deps): add http-body-util workspace dependency chore(deps): upgrade linkerd2-proxy-api this commit represents main as of linkerd/linkerd2-proxy-api#421. Signed-off-by: katelyn martin <[email protected]>

## ✨ chore(deps): upgrade to hyperium 1.x crates this branch performs an exciting upgrade for our proxy. this branch upgrades a number of our dependencies so that we use the 1.0 release family of the `hyper` http framework, and its ecosystem. see the [v1.0 announcement][hyper-v1] for more information. this branch upgrades the following dependencies: * `h2`: 0.3 -> 0.4 * `http`: 0.2 -> 1 * `http-body`: 0.4 -> 1 * `hyper`: 0.14.32 -> 1 * `prost`: 0.12 -> 0.13 * `prost-build`: 0.12 -> 0.13 * `prost-types`: 0.12 -> 0.13 * `tonic`: 0.10 -> 0.12 * `tonic-build`: 0.10 -> 0.12 a `hyper-util` dependency is added, which provides among other things, legacy-compatible interfaces such as `hyper_util::client::legacy::Client`, or glue to use `hyper` with the tokio runtime. see <https://docs.rs/hyper-util/latest/hyper_util/> for more information. a `http-body-util` dependency is added, which provides a `BodyExt` trait and a channel-backed body for use in unit tests. the `deprecated` feature flag that was active on our `0.14` hyper dependency has been removed, along with the `stream` and `runtime` feature flags. the `linkerd2-proxy-api` dependency is updated. see: <linkerd/linkerd2-proxy-api#421> ### 📝 notes for review bear particular attention to changes involving `http_body::Body` middleware. the change from two separate `poll_data()` and `poll_trailers()` functions, to a single `poll_frame()` method, induces some subtle changes to various pieces of middleware. also bear in mind that failing to set a timer, in our case `hyper_util::rt::TokioTimer`, can cause http/2 clients, or http/1 and http/2 servers, to panic. make sure that any uses of `hyper::server::conn::http1::Builder`, `hyper::client::conn::http1::Builder`, or `hyper::client::conn::http2::Builder` install a timer. ### ❗ breaking change: `l5d-proxy-error` values the `l5d-proxy-error` header can be examined to observe the cause of proxy errors encountered when sending meshed traffic. by virtue of this using a newer `hyper` client in the proxy, some error messages may in turn look different. for example, an error like `"connect timed out after 1s"` may now appear as `"client error (Connect)"`. ### 📚 other notes this work, by virtue of touching so many parts of the system, is carried out in distinct commits. an initial commit upgrades the dependencies at th workspace level. subsequent commits will not compile if the `--workspace` flag is provided, but the intent of this branch is to update each crate individually. use commands like, e.g. `cargo check --tests -p linkerd-proxy-http` to build particular crates at intermediate commits within this branch. this commit is also only the final leaf in an _extended_ line of work. this has been done to mitigate the effort of reviewing this change, and the risk of churn in the event of any unanticipated errors. see the top-level comment in linkerd/linkerd2#8733 for an overview of all of the work that brought us to this juncture. [hyper-v1]: https://seanmonstar.com/blog/hyper-v1/ --- * chore(deps): upgrade to hyper 1.x note: this commit will not compile, code changes are intentionally elided from this commit. this commit upgrades hyper, http, tonic, prost, related dependencies, and their assorted cargo features. see <linkerd/linkerd2#8733>. see also: * #3379 * #3380 * #3382 * #3405 * hyperium/hyper#3796 * #3411 * #3421 * #3427 * #3428 * #3432 * #3433 * #3444 * #3445 * #3454 * #3455 * #3456 * #3457 * #3461 * #3459 * #3465 * #3466 * #3467 * #3468 * linkerd/linkerd2-proxy-api#421 * linkerd/linkerd2#13492 * linkerd/linkerd2#13493 * hyperium/hyper#3816 * #3472 * #3473 * #3479 * tokio-rs/tokio#7059 * #3509 * hyperium/http-body#140 * #3515 * hyperium/http-body#141 * #3530 * #3531 * #3540 * #3556 * #3558 * #3559 * #3564 * #3567 * #3573 * #3583 * hyperium/http-body#144 * #3585 * #3586 * #3597 * #3598 * #3611 * #3614 * #3615 * #3616 * #3647 * #3651 * #3653 * #3654 * #3655 * #3656 * #3657 * #3660 * #3671 * #3672 * #3673 * #3676 * hyperium/http-body#147 * #3692 * #3699 * #3700 * #3701 * #3708 * linkerd/drain-rs#36 * #3715 * #3717 * eminence/procfs#340 --- squash: chore(deps): add hyper-util workspace dependency chore(deps): add http-body-util workspace dependency chore(deps): upgrade linkerd2-proxy-api this commit represents main as of linkerd/linkerd2-proxy-api#421. Signed-off-by: katelyn martin <[email protected]> * chore(http/box): upgrade to hyper 1.x Signed-off-by: katelyn martin <[email protected]> * chore(hyper-balance): upgrade to hyper 1.x Signed-off-by: katelyn martin <[email protected]> * chore(http/retain): ugrade to hyper 1.x Signed-off-by: katelyn martin <[email protected]> * chore(http/stream-timeouts): upgrade to hyper 1.x Signed-off-by: katelyn martin <[email protected]> * chore(http/classify): upgrade to hyper 1.x Signed-off-by: katelyn martin <[email protected]> * chore(http/upgrade): upgrade to hyper 1.x NOTE: there is a comment noting that the upgrade middleware does not expect to be cloneable. it is unfortunately, however, at odds with the new bounds expected of extensions. so, `Http11Upgrade` is now Clone'able, but a comment is left in place noting this weakened invariant. it's worth investigating how upgrades have changed since, in more detail, but for the current moment we are interested in being especially conservative about changing behavior, and focusing on api changes like `Body::poll_frame(..)`. Signed-off-by: katelyn martin <[email protected]> * chore(metrics): upgrade to hyper 1.x a brief note; this commit happened to tickle an unfortunate sharp edge in `BoxBody` and `Full`'s respective constructors. type inference could not figure out how to construct the body, so we refrain from boxing the response body now. Signed-off-by: katelyn martin <[email protected]> * chore(http/metrics): upgrade to hyper 1.x Signed-off-by: katelyn martin <[email protected]> * chore(http/prom): upgrade to hyper 1.x Signed-off-by: katelyn martin <[email protected]> * chore(http/insert): upgrade to hyper 1.x Signed-off-by: katelyn martin <[email protected]> * chore(http/retry): deprecate linkerd-http-body-compat Signed-off-by: katelyn martin <[email protected]> * chore(mock/http-body): upgrade to hyper 1.x Signed-off-by: katelyn martin <[email protected]> * chore(http/retry): upgrade to hyper 1.x Signed-off-by: katelyn martin <[email protected]> * chore(proxy/tap): upgrade to hyper 1.x Signed-off-by: katelyn martin <[email protected]> * chore(proxy/http): update to hyper 1.x Signed-off-by: katelyn martin <[email protected]> * chore(app/core): upgrade to hyper 1.x Signed-off-by: katelyn martin <[email protected]> * chore(app/test): upgrade to hyper 1.x Signed-off-by: katelyn martin <[email protected]> * chore(app/admin): upgrade to hyper 1.x Signed-off-by: katelyn martin <[email protected]> * chore(app/outbound): upgrade to hyper 1.x Signed-off-by: katelyn martin <[email protected]> * chore(app/inbound): upgrade to hyper 1.x Signed-off-by: katelyn martin <[email protected]> * chore(app/integration): upgrade to hyper 1.x Signed-off-by: katelyn martin <[email protected]> * chore(app): upgrade to hyper 1.x Signed-off-by: katelyn martin <[email protected]> * chore(transport-header): update generated code Signed-off-by: katelyn martin <[email protected]> * chore(spiffe-proto): update generated code Signed-off-by: katelyn martin <[email protected]> * chore(opencensus-proto): update generated code Signed-off-by: katelyn martin <[email protected]> * chore(opentelemetry-proto): update generated code Signed-off-by: katelyn martin <[email protected]> * chore(deny.toml): update cargo-deny directives this commit updates the contents of `deny.toml`. Signed-off-by: katelyn martin <[email protected]> * chore: `compile` has been renamed to `compile_protos` this addresses deprecation warnings, updating calls to a function that has since been renamed. Signed-off-by: katelyn martin <[email protected]> * chore(deps): remove `linkerd-http-body-compat` dependencies this commit removes this crate, which we added to future proof code for this upgrade, from its dependents. Signed-off-by: katelyn martin <[email protected]> * chore(http/body-compat): remove `linkerd-http-body-compat` crate Signed-off-by: katelyn martin <[email protected]> * chore(deps): update to drain 0.2.1 see linkerd/drain-rs#41. Signed-off-by: katelyn martin <[email protected]> --------- Signed-off-by: katelyn martin <[email protected]>

github-actions bot added the R-loom-sync Run loom sync tests on this PR label Dec 31, 2024

This was referenced Dec 31, 2024

util: add a channel body hyperium/http-body#100

Closed

proxy: Figure out the plan for Hyper 1.0 migration linkerd/linkerd2#8733

Open

Darksonn added A-tokio Area: The main tokio crate M-sync Module: tokio/sync labels Jan 1, 2025

cratelyn closed this Jan 2, 2025

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

sync: add `Receiver::poll_recv(..)` method #7059

sync: add `Receiver::poll_recv(..)` method #7059

cratelyn commented Dec 31, 2024

seanmonstar commented Dec 31, 2024

cratelyn commented Jan 2, 2025

sync: add Receiver::poll_recv(..) method #7059

sync: add Receiver::poll_recv(..) method #7059

Conversation

cratelyn commented Dec 31, 2024

Motivation

Solution

seanmonstar commented Dec 31, 2024

cratelyn commented Jan 2, 2025

sync: add `Receiver::poll_recv(..)` method #7059

sync: add `Receiver::poll_recv(..)` method #7059