Remove Sync requirement for Streaming

[parasyte]

This is a followup to the conversation in #81 (comment) - I decided to create a new ticket to raise awareness.

Bug Report

Version

tonic = "0.1.0-alpha.5"

Platform

Darwin JayMBP-2.local 18.7.0 Darwin Kernel Version 18.7.0: Sat Oct 12 00:02:19 PDT 2019; root:xnu-4903.278.12~1/RELEASE_X86_64 x86_64

Description

In older releases, we're able to use non-Sync futures, like hyper::client::ResponseFuture to yield items in a streaming response. #84 added a Sync trait bound to the pinned Stream trait object returned by the service impl. The following code works with the previous version of tonic:

Works with 0.1.0-alpha.4

#[derive(Debug)]
pub struct RouteGuide {
    client: Client<HttpConnector, Body>,
}

#[tonic::async_trait]
impl server::RouteGuide for RouteGuide {
    type RouteChatStream =
        Pin<Box<dyn Stream<Item = Result<RouteNote, Status>> + Send + 'static>>;

    async fn route_chat(
        &self,
        request: Request<tonic::Streaming<RouteNote>>,
    ) -> Result<Response<Self::RouteChatStream>, Status> {
        println!("RouteChat");

        let stream = request.into_inner();
        let client = self.client.clone();

        let output = async_stream::try_stream! {
            futures::pin_mut!(stream);

            while let Some(note) = stream.next().await {
                let _note = note?;

                // Make a simple HTTP request. What could possibly go wrong?
                let res = client.get(hyper::Uri::from_static("http://httpbin.org/get")).await;

                // Receive the response as a byte stream
                let mut body = res.unwrap().into_body();
                let mut bytes = Vec::new();
                while let Some(chunk) = body.next().await {
                    bytes.extend(chunk.map_err(|_| Status::new(tonic::Code::Internal, "Error"))?);
                }
                let message = String::from_utf8_lossy(&bytes).to_string();

                let note = RouteNote {
                    location: None,
                    message,
                };

                yield note;
            }
        };

        Ok(Response::new(Box::pin(output)
            as Pin<
                Box<dyn Stream<Item = Result<RouteNote, Status>> + Send + 'static>,
            >))
    }
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let addr = "[::1]:10000".parse().unwrap();

    println!("Listening on: {}", addr);

    let client = hyper::client::Client::new();
    let route_guide = RouteGuide {
        client,
    };

    let svc = server::RouteGuideServer::new(route_guide);

    Server::builder().serve(addr, svc).await?;

    Ok(())
}

And the updated code now fails:

Fails with 0.1.0-alpha.5

#[derive(Debug)]
pub struct RouteGuide {
    client: Client<HttpConnector, Body>,
}

#[tonic::async_trait]
impl server::RouteGuide for RouteGuide {
    type RouteChatStream =
        Pin<Box<dyn Stream<Item = Result<RouteNote, Status>> + Send + Sync + 'static>>;

    async fn route_chat(
        &self,
        request: Request<tonic::Streaming<RouteNote>>,
    ) -> Result<Response<Self::RouteChatStream>, Status> {
        println!("RouteChat");

        let stream = request.into_inner();
        let client = self.client.clone();

        let output = async_stream::try_stream! {
            futures::pin_mut!(stream);

            while let Some(note) = stream.next().await {
                let _note = note?;

                // Make a simple HTTP request. What could possibly go wrong?
                let res = client.get(hyper::Uri::from_static("http://httpbin.org/get")).await;

                // Receive the response as a byte stream
                let mut body = res.unwrap().into_body();
                let mut bytes = Vec::new();
                while let Some(chunk) = body.next().await {
                    bytes.extend(chunk.map_err(|_| Status::new(tonic::Code::Internal, "Error"))?);
                }
                let message = String::from_utf8_lossy(&bytes).to_string();

                let note = RouteNote {
                    location: None,
                    message,
                };

                yield note;
            }
        };

        Ok(Response::new(Box::pin(output)
            as Pin<
                Box<dyn Stream<Item = Result<RouteNote, Status>> + Send + Sync + 'static>,
            >))
    }
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let addr = "[::1]:10000".parse().unwrap();

    println!("Listening on: {}", addr);

    let client = hyper::client::Client::new();
    let route_guide = RouteGuide {
        client,
    };

    let svc = server::RouteGuideServer::new(route_guide);

    Server::builder().add_service(svc).serve(addr).await?;

    Ok(())
}

Compile error

error[E0277]: `(dyn core::future::future::Future<Output = std::result::Result<http::response::Response<hyper::body::body::Body>, hyper::error::Error>> + std::marker::Send + 'static)` cannot be shared between threads safely
  |
  = help: the trait `std::marker::Sync` is not implemented for `(dyn core::future::future::Future<Output = std::result::Result<http::response::Response<hyper::body::body::Body>, hyper::error::Error>> + std::marker::Send + 'static)`
  = note: required because of the requirements on the impl of `std::marker::Sync` for `std::ptr::Unique<(dyn core::future::future::Future<Output = std::result::Result<http::response::Response<hyper::body::body::Body>, hyper::error::Error>> + std::marker::Send + 'static)>`
  = note: required because it appears within the type `std::boxed::Box<(dyn core::future::future::Future<Output = std::result::Result<http::response::Response<hyper::body::body::Body>, hyper::error::Error>> + std::marker::Send + 'static)>`
  = note: required because it appears within the type `std::pin::Pin<std::boxed::Box<(dyn core::future::future::Future<Output = std::result::Result<http::response::Response<hyper::body::body::Body>, hyper::error::Error>> + std::marker::Send + 'static)>>`
  = note: required because it appears within the type `hyper::client::ResponseFuture`
  = note: required because it appears within the type `for<'r, 's, 't0, 't1, 't2, 't3, 't4, 't5, 't6, 't7, 't8, 't9, 't10, 't11, 't12, 't13, 't14> {tonic::codec::decode::Streaming<routeguide::RouteNote>, std::pin::Pin<&'r mut tonic::codec::decode::Streaming<routeguide::RouteNote>>, &'s mut std::pin::Pin<&'t0 mut tonic::codec::decode::Streaming<routeguide::RouteNote>>, std::pin::Pin<&'t1 mut tonic::codec::decode::Streaming<routeguide::RouteNote>>, futures_util::stream::next::Next<'t2, std::pin::Pin<&'t3 mut tonic::codec::decode::Streaming<routeguide::RouteNote>>>, futures_util::stream::next::Next<'t4, std::pin::Pin<&'t5 mut tonic::codec::decode::Streaming<routeguide::RouteNote>>>, (), std::option::Option<std::result::Result<routeguide::RouteNote, tonic::status::Status>>, std::result::Result<routeguide::RouteNote, tonic::status::Status>, std::result::Result<routeguide::RouteNote, tonic::status::Status>, tonic::status::Status, &'t6 mut async_stream::yielder::Sender<std::result::Result<routeguide::RouteNote, tonic::status::Status>>, async_stream::yielder::Sender<std::result::Result<routeguide::RouteNote, tonic::status::Status>>, std::result::Result<routeguide::RouteNote, tonic::status::Status>, impl core::future::future::Future, impl core::future::future::Future, (), routeguide::RouteNote, &'t7 hyper::client::Client<hyper::client::connect::http::HttpConnector>, hyper::client::Client<hyper::client::connect::http::HttpConnector>, &'t8 str, http::uri::Uri, hyper::client::ResponseFuture, hyper::client::ResponseFuture, (), std::result::Result<http::response::Response<hyper::body::body::Body>, hyper::error::Error>, hyper::body::body::Body, std::vec::Vec<u8>, &'t9 mut hyper::body::body::Body, hyper::body::body::Body, impl core::future::future::Future, impl core::future::future::Future, (), std::option::Option<std::result::Result<hyper::body::chunk::Chunk, hyper::error::Error>>, std::result::Result<hyper::body::chunk::Chunk, hyper::error::Error>, &'t12 mut std::vec::Vec<u8>, std::vec::Vec<u8>, std::result::Result<hyper::body::chunk::Chunk, hyper::error::Error>, [closure@<::async_stream::try_stream macros>:8:25: 8:54], std::result::Result<hyper::body::chunk::Chunk, tonic::status::Status>, &'t13 mut async_stream::yielder::Sender<std::result::Result<routeguide::RouteNote, tonic::status::Status>>, impl core::future::future::Future, (), std::string::String, routeguide::RouteNote, &'t14 mut async_stream::yielder::Sender<std::result::Result<routeguide::RouteNote, tonic::status::Status>>, impl core::future::future::Future, ()}`  = note: required because it appears within the type `[static generator@<::async_stream::try_stream macros>:7:10: 11:11 stream:tonic::codec::decode::Streaming<routeguide::RouteNote>, __yield_tx:async_stream::yielder::Sender<std::result::Result<routeguide::RouteNote, tonic::status::Status>>, client:hyper::client::Client<hyper::client::connect::http::HttpConnector> for<'r, 's, 't0, 't1, 't2, 't3, 't4, 't5, 't6, 't7, 't8, 't9, 't10, 't11, 't12, 't13, 't14> {tonic::codec::decode::Streaming<routeguide::RouteNote>, std::pin::Pin<&'r mut tonic::codec::decode::Streaming<routeguide::RouteNote>>, &'s mut std::pin::Pin<&'t0 mut tonic::codec::decode::Streaming<routeguide::RouteNote>>, std::pin::Pin<&'t1 mut tonic::codec::decode::Streaming<routeguide::RouteNote>>, futures_util::stream::next::Next<'t2, std::pin::Pin<&'t3 mut tonic::codec::decode::Streaming<routeguide::RouteNote>>>, futures_util::stream::next::Next<'t4, std::pin::Pin<&'t5 mut tonic::codec::decode::Streaming<routeguide::RouteNote>>>, (), std::option::Option<std::result::Result<routeguide::RouteNote, tonic::status::Status>>, std::result::Result<routeguide::RouteNote, tonic::status::Status>, std::result::Result<routeguide::RouteNote, tonic::status::Status>, tonic::status::Status, &'t6 mut async_stream::yielder::Sender<std::result::Result<routeguide::RouteNote, tonic::status::Status>>, async_stream::yielder::Sender<std::result::Result<routeguide::RouteNote, tonic::status::Status>>, std::result::Result<routeguide::RouteNote, tonic::status::Status>, impl core::future::future::Future, impl core::future::future::Future, (), routeguide::RouteNote, &'t7 hyper::client::Client<hyper::client::connect::http::HttpConnector>, hyper::client::Client<hyper::client::connect::http::HttpConnector>, &'t8 str, http::uri::Uri, hyper::client::ResponseFuture, hyper::client::ResponseFuture, (), std::result::Result<http::response::Response<hyper::body::body::Body>, hyper::error::Error>, hyper::body::body::Body, std::vec::Vec<u8>, &'t9 mut hyper::body::body::Body, hyper::body::body::Body, impl core::future::future::Future, impl core::future::future::Future, (), std::option::Option<std::result::Result<hyper::body::chunk::Chunk, hyper::error::Error>>, std::result::Result<hyper::body::chunk::Chunk, hyper::error::Error>, &'t12 mut std::vec::Vec<u8>, std::vec::Vec<u8>, std::result::Result<hyper::body::chunk::Chunk, hyper::error::Error>, [closure@<::async_stream::try_stream macros>:8:25: 8:54], std::result::Result<hyper::body::chunk::Chunk, tonic::status::Status>, &'t13 mut async_stream::yielder::Sender<std::result::Result<routeguide::RouteNote, tonic::status::Status>>, impl core::future::future::Future, (), std::string::String, routeguide::RouteNote, &'t14 mut async_stream::yielder::Sender<std::result::Result<routeguide::RouteNote, tonic::status::Status>>, impl core::future::future::Future, ()}]`  = note: required because it appears within the type `std::future::GenFuture<[static generator@<::async_stream::try_stream macros>:7:10: 11:11 stream:tonic::codec::decode::Streaming<routeguide::RouteNote>, __yield_tx:async_stream::yielder::Sender<std::result::Result<routeguide::RouteNote, tonic::status::Status>>, client:hyper::client::Client<hyper::client::connect::http::HttpConnector> for<'r, 's, 't0, 't1, 't2, 't3, 't4, 't5, 't6, 't7, 't8, 't9, 't10, 't11, 't12, 't13, 't14> {tonic::codec::decode::Streaming<routeguide::RouteNote>, std::pin::Pin<&'r mut tonic::codec::decode::Streaming<routeguide::RouteNote>>, &'s mut std::pin::Pin<&'t0 mut tonic::codec::decode::Streaming<routeguide::RouteNote>>, std::pin::Pin<&'t1 mut tonic::codec::decode::Streaming<routeguide::RouteNote>>, futures_util::stream::next::Next<'t2, std::pin::Pin<&'t3 mut tonic::codec::decode::Streaming<routeguide::RouteNote>>>, futures_util::stream::next::Next<'t4, std::pin::Pin<&'t5 mut tonic::codec::decode::Streaming<routeguide::RouteNote>>>, (), std::option::Option<std::result::Result<routeguide::RouteNote, tonic::status::Status>>, std::result::Result<routeguide::RouteNote, tonic::status::Status>, std::result::Result<routeguide::RouteNote, tonic::status::Status>, tonic::status::Status, &'t6 mut async_stream::yielder::Sender<std::result::Result<routeguide::RouteNote, tonic::status::Status>>, async_stream::yielder::Sender<std::result::Result<routeguide::RouteNote, tonic::status::Status>>, std::result::Result<routeguide::RouteNote, tonic::status::Status>, impl core::future::future::Future, impl core::future::future::Future, (), routeguide::RouteNote, &'t7 hyper::client::Client<hyper::client::connect::http::HttpConnector>, hyper::client::Client<hyper::client::connect::http::HttpConnector>, &'t8 str, http::uri::Uri, hyper::client::ResponseFuture, hyper::client::ResponseFuture, (), std::result::Result<http::response::Response<hyper::body::body::Body>, hyper::error::Error>, hyper::body::body::Body, std::vec::Vec<u8>, &'t9 mut hyper::body::body::Body, hyper::body::body::Body, impl core::future::future::Future, impl core::future::future::Future, (), std::option::Option<std::result::Result<hyper::body::chunk::Chunk, hyper::error::Error>>, std::result::Result<hyper::body::chunk::Chunk, hyper::error::Error>, &'t12 mut std::vec::Vec<u8>, std::vec::Vec<u8>, std::result::Result<hyper::body::chunk::Chunk, hyper::error::Error>, [closure@<::async_stream::try_stream macros>:8:25: 8:54], std::result::Result<hyper::body::chunk::Chunk, tonic::status::Status>, &'t13 mut async_stream::yielder::Sender<std::result::Result<routeguide::RouteNote, tonic::status::Status>>, impl core::future::future::Future, (), std::string::String, routeguide::RouteNote, &'t14 mut async_stream::yielder::Sender<std::result::Result<routeguide::RouteNote, tonic::status::Status>>, impl core::future::future::Future, ()}]>`
  = note: required because it appears within the type `impl core::future::future::Future`
  = note: required because it appears within the type `async_stream::async_stream::AsyncStream<std::result::Result<routeguide::RouteNote, tonic::status::Status>, impl core::future::future::Future>`
  = note: required for the cast to the object type `dyn futures_core::stream::Stream<Item = std::result::Result<routeguide::RouteNote, tonic::status::Status>> + std::marker::Send + std::marker::Sync`

[LucioFranco]

So I'm not sure if this is a bug because technically it was intentional but I'm not sure how to get both working. Thanks for opening this.

[ghost]

I have a similar case, I am using hyper client in a response stream, I get:

the trait std::marker::Sync is not implemented for (dyn core::future::future::Future<Output = std::result::Result<http::response::Response<hyper::body::body::Body>, hyper::error::Error>> + std::marker::Send + 'static)

it works with 0.1.0-alpha.4 as mentioned in OP, and fails in 0.1.0-alpha.6. I am not sure how to work around that problem.

[LucioFranco]

So I believe using a channel should work as a temporary solution. @abdul-rehman0 would you be able to use a channel and a spawned task instead?

[ghost]

@LucioFranco using a channel and spawn works for me, Thanks.

Changed the following code:

let client = Client::new(URL);
let response: PreferencesResponse = client.get(&[("to", "to")]).await.unwrap();

To:

let (response_sender, response_receiver) = oneshot::channel::<PreferencesResponse>();
tokio::spawn(async move {
    let client = Client::new(URL);
    let response: PreferencesResponse = client.get(&[("to", "to")]).await.unwrap();
    response_sender.send(response);
});
let response = response_receiver.await.unwrap();

[LucioFranco]

Cool, I think this is probably the correct solution for now. I think it makes sense to cover both send and sync for now in the trait object since users can just use a channel.

[LucioFranco]

I think the above solution works well for this, so I will close this issue, feel free to reopen if you have any more questions.

[rdettai]

I do understand the workaround with channels, but I can't get my head around why the response stream needs to be Sync in the first place.

@LucioFranco you say the change to Sync was intentional, can you explain what was the motivation?

Thanks for your help! These explanations will really help me in my (long) journey to enlightenment about rust async 😄

[MikailBag]

IIUC another way to fix is

pub struct AlwaysSync<T>(T);
// SAFETY: This is essentially mutex, but without `lock()`.
// alternative justification: &AlwaysSync can't be upgrared to &T.
unsafe impl<T: Send> Sync for AlwaysSync<T> {}

impl AlwaysSync {
     fn get_mut(&mut self) -> &mut T {
         &mut self.0
     }

     fn get_mut_pinned(self: Pin<&mut Self>) -> Pin<&mut T> {
         // SAFETY: just a pin projection
         unsafe {Pin::map_unchecked_map(self, |s| &mut s.0)}
     }
}

impl<T: Stream> Stream for AlwaysSync<T> {
    type Item = T::Item;
    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        self.get_mut_pinned().poll_next(cx)
    } 
}

[LucioFranco]

@LucioFranco you say the change to Sync was intentional, can you explain what was the motivation?

Internally, we Box<dyn Trait> things, and unfortunately if one person needs it to be Sync (warp needs this for example) then we need it always to be sync. Its unfortunate but because we trait object it, we are very restricted. But the cost here is worth the benefit.

[MikailBag]

I believe there is no need to require Sync from dyn Stream objects, if they are wrapped in AlwaysSync.

If I have enough time, I'll try to make a PR relaxing Sync bounds.

[LucioFranco]

@MikailBag Yeah, I am not sure how that adds up fully with regards to rust's safety rules, but personally I don't think its worth it. A channel works well.

[rdettai]

Internally, we Box things, and unfortunately if one person needs it to be Sync (warp needs this for example) then we need it always to be sync.

Thanks for taking the time to answer. Not sure I understand how it all fits together, but I guess I won't without looking at the code in more details. 😃

[MikailBag]

So after some considerations, I now think that this pattern/problem is pretty popular. I published a tiny crate: https://github.com/MikailBag/weird-mutex (will publish to crates-io if someone asks for that) that allows one to avoid Sync bounds without writing unsafe code. For example, see [https://github.com/hyperium/tonic/compare/master...MikailBag:unsync] for patch deleting Sync bounds from streams.

Without patching Tonic code, one can wrap their streams in WeirdMutex and these streams will become Sync.

[akiradeveloper]

@MikailBag There is already something like it: https://github.com/Actyx/sync_wrapper

and I've send a PR: Actyx/sync_wrapper#1

[hjfreyer]

I understand there's kind of a workaround now, but given that async_trait produces futures that aren't Sync, this is a pretty fundamental incompatibility between two libraries that basically have to be used together. While it seems like "just drop the Sync requirement" isn't a viable solution, I'd still say there's a problem that needs solving here, so I'd advocate for re-opening this issue.

[hjfreyer]

Also, did we ever get an answer for why a Stream could possibly need Sync when there's nothing you can do with a &Stream?

[parasyte]

The original motivation for making Streaming<T> Sync in the first place was warp's SSE filter... Which removed the Sync trait bound in August: seanmonstar/warp#683

@LucioFranco I think this trait bound can be removed on Tonic now.

[hjfreyer]

Yay! Can we reopen this issue and get rid of the trait bound? + Sync is already spreading like a virus through my project and it'd be lovely to head that off.

[LucioFranco]

Okay, I've reopened this, I will revisit this in the next release of tonic.

[davidpdrsn]

Hm I'm not sure this is possible since hyper's with_graceful_shutdown server requires the body to be Sync https://github.com/hyperium/hyper/blob/master/src/server/shutdown.rs#L53 and since tonics body is built from a stream object that needs to Sync as well.

Tonic uses with_graceful_shutdown here

tonic/tonic/src/transport/server/mod.rs

Line 400 in 31936e0

.with_graceful_shutdown(signal)

[MikailBag]

One can always "upgrade" Box<dyn Stream + Send> to Box<dyn Stream + Send + Sync> using e.g. this library (or just write the same unsafe impl Sync themselves).

[stuhood]

The SyncWrapper type is the conclusion of this thread: https://internals.rust-lang.org/t/what-shall-sync-mean-across-an-await/12020/31 ... they argue that hyper should likely be internally using that type for its Body to remove the Sync requirement, which would ease things here as well.

[LucioFranco]

Relevant commit in hyper hyperium/hyper@1d553e5