1

I am having a play with async/await by using Tokio with the async/await feature enabled in my Cargo.toml (and the latest Rust nightly with 2018 edition):

tokio = { version = "0.1.11", features = ["async-await-preview"] }

I've run into an error I don't understand, which is reproduced in this minimal example:

#![feature(await_macro, async_await, futures_api)]
use tokio::prelude::*;

pub fn main() {
    tokio::run_async(async {
        let s: Option<Box<dyn Sink<SinkItem = u8, SinkError = ()> + Send + Sync + 'static>> = None;

        if let Some(sink) = s {
            await!(sink.send_async(100));
        }
    });
}

The error is:

error[E0277]: the trait bound `for<'r> (dyn futures::sink::Sink<SinkItem=u8, SinkError=()> + std::marker::Sync + std::marker::Send + 'r): futures::sink::Sink` is not satisfied
 --> src/main.rs:5:5
  |
5 |     tokio::run_async(async {
  |     ^^^^^^^^^^^^^^^^ the trait `for<'r> futures::sink::Sink` is not implemented for `dyn futures::sink::Sink<SinkItem=u8, SinkError=()> + std::marker::Sync + std::marker::Send`
  |
  = note: required because of the requirements on the impl of `futures::sink::Sink` for `std::boxed::Box<dyn futures::sink::Sink<SinkItem=u8, SinkError=()> + std::marker::Sync + std::marker::Send>`
  = note: required because it appears within the type `tokio_async_await::sink::send::Send<'_, std::boxed::Box<dyn futures::sink::Sink<SinkItem=u8, SinkError=()> + std::marker::Sync + std::marker::Send>>`
  = note: required because it appears within the type `for<'r, 's, 't0, 't1> {std::option::Option<std::boxed::Box<(dyn futures::sink::Sink<SinkItem=u8, SinkError=()> + std::marker::Sync + std::marker::Send + 'r)>>, std::boxed::Box<(dyn futures::sink::Sink<SinkItem=u8, SinkError=()> + std::marker::Sync + std::marker::Send + 's)>, tokio_async_await::sink::send::Send<'t0, std::boxed::Box<(dyn futures::sink::Sink<SinkItem=u8, SinkError=()> + std::marker::Sync + std::marker::Send + 't1)>>, ()}`
  = note: required because it appears within the type `[static generator@src/main.rs:5:28: 11:6 for<'r, 's, 't0, 't1> {std::option::Option<std::boxed::Box<(dyn futures::sink::Sink<SinkItem=u8, SinkError=()> + std::marker::Sync + std::marker::Send + 'r)>>, std::boxed::Box<(dyn futures::sink::Sink<SinkItem=u8, SinkError=()> + std::marker::Sync + std::marker::Send + 's)>, tokio_async_await::sink::send::Send<'t0, std::boxed::Box<(dyn futures::sink::Sink<SinkItem=u8, SinkError=()> + std::marker::Sync + std::marker::Send + 't1)>>, ()}]`
  = note: required because it appears within the type `std::future::GenFuture<[static generator@src/main.rs:5:28: 11:6 for<'r, 's, 't0, 't1> {std::option::Option<std::boxed::Box<(dyn futures::sink::Sink<SinkItem=u8, SinkError=()> + std::marker::Sync + std::marker::Send + 'r)>>, std::boxed::Box<(dyn futures::sink::Sink<SinkItem=u8, SinkError=()> + std::marker::Sync + std::marker::Send + 's)>, tokio_async_await::sink::send::Send<'t0, std::boxed::Box<(dyn futures::sink::Sink<SinkItem=u8, SinkError=()> + std::marker::Sync + std::marker::Send + 't1)>>, ()}]>`
  = note: required because it appears within the type `impl std::future::Future`
  = note: required by `tokio::async_await::run_async`

and it goes away if I remove the line beginning "await!".

When using Tokio without the experimental async/await support, my program is happy with the idea that a Box<dyn Sink> is a Sink, so I'm not really sure why the error crops up using async/await stuff.

What does the error mean? How can I get around this?

1

My fix was to wrap the Boxed Sinks into a newtype and implement the Sink trait on that. I think that perhaps Box<dyn Sink> does not implement Sink in this nightly, which is basically what the error message implies (I guess the async/await shims redefine Sink and don't implement it on Box).

My wrapper ended up looking like this:

struct BoxedSink<I, E>(Box<dyn Sink<SinkItem = I, SinkError = E> + Send + Sync + 'static>);

impl<I, E> Sink for BoxedSink<I, E> {
    type SinkItem = I;
    type SinkError = E;

    fn start_send(
        &mut self,
        input: Self::SinkItem,
    ) -> Result<AsyncSink<Self::SinkItem>, Self::SinkError> {
        self.0.start_send(input)
    }

    fn poll_complete(&mut self) -> Poll<(), Self::SinkError> {
        self.0.poll_complete()
    }
}

You have to wrap your Box<Sinks> in this to make them once again implement Sink.

  • Based on chatting with some of the implementors, it sounds like Sink is not an abstraction that is going to continue to exist in the new futures world. You can tell it hasn't been updated because it still has the split Item / Error associated types, unlike Future and Stream. – Shepmaster Nov 16 at 1:29
  • Thanks, that's interesting to know. I am maybe a bit surprised that Stream would continue to exist but not Sink (piping Streams to Sinks seems useful for instance, though easy with async/await I guess!), but Sink does feel like the least useful of the abstractions to keep around! – jsdw Nov 20 at 7:39

Your Answer

By clicking "Post Your Answer", you acknowledge that you have read our updated terms of service, privacy policy and cookie policy, and that your continued use of the website is subject to these policies.

Not the answer you're looking for? Browse other questions tagged or ask your own question.