How to use mpsc::channel in single-send, probably single-read scenario?

[sasakiyori]

I use tonic+tokio to serve a grpc service. One of the rpc method is: Server read something (e.g. file) and streaming its appending data to client.

service XXService {
    rpc Tail(TailRequest) returns (stream TailResponse);
}

My simple idea is that create a mpsc::channel, move the sender to the data-read thread, move the receiver to the rpc thread (Of course it must only one rpc connection at the same time, seems stupid huh):

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let (sender, receiver) = mpsc::channel(1);
    // read thread
    read_buffer(sender);
    // grpc server
    let addr = "xxx".parse()?;
    let xx_service = XXService::new(receiver);
    Server::builder()
        .add_service(XXServiceServer::new(xx_service))
        .serve(addr)
        .await?;

    Ok(())
}

I want to use the reference of the receiver, so that I can directly pass it to the rpc response.

async fn tail(
        &self,
        request: Request<TailRequest>,
    ) -> Result<Response<Self::TailResponseStream>, Status> {
    // error, of course
    let reply_stream = ReceiverStream::new(self.receiver);
    Ok(Response::new(
            Box::pin(reply_stream) as Self::TailResponseStream
     ))
}

I tried to use Arc, and implement a Stream for it. But poll_next need mutable reference, so finally I must wrap a Mutex for it:

// stream
struct TailStream {
    inner: Arc<Mutex<Receiver<Result<TailResponse, Status>>>>,
}

impl Stream for TailStream {
    type Item = Result<TailResponse, Status>;

    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        self.inner.lock().unwrap().poll_recv(cx)
    }
}

impl TailStream {
    fn new(recv: Arc<Mutex<Receiver<Result<TailResponse, Status>>>>) -> Self {
        Self { inner: recv }
    }
}

// main
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let xx_service = XXService::new(Arc::new(Mutex::new(receiver)));
}

// rpc
async fn tail(
        &self,
        request: Request<TailRequest>,
    ) -> Result<Response<Self::TailResponseStream>, Status> {
    let reply_stream = TailStream::new(self.receiver.clone());
    Ok(Response::new(
            Box::pin(reply_stream) as Self::TailResponseStream
     ))
}

It works, but IMO the Mutex is definitely no need in such use case because of the rude prerequisite that only one connection will come. But I don't know how to remove the Mutex out.

---

Some other methods I thought about but not satisfied:

• create another channel for rpc response, and use self.reciver.recv(), then pass it to the sender of new channel.
• create a global value Option<Sender<...>>, wrapped by some thread safe semantics. The read thread get the newest sender whenever it want to send data. The channel will be created inside the rpc function. Sender can be easier to be sent to other threads as it implements Clone trait.

---

I am new to rust and tokio, hope to get help from anyone 🙏

[Darksonn]

You definitely don't want an Arc or Mutex around your receiver. I would expect something very close to Box::pin(ReceiverStream::new(receiver)) to work.

You haven't provided a complete example or an error message, so it's difficult to provide more complete advice than this.

[sasakiyori]

I create an example repo now: https://github.com/sasakiyori/tokio-discussion-6119.

I would expect something very close to Box::pin(ReceiverStream::new(receiver)) to work.

I have no idea right now but I will try more methods. Maybe I have sunk into the wrong way..

[Darksonn]

It's because the <XXService as XxService>::tail method tries to give up ownership of the receiver, but the method could be called multiple times. Since ownership can only be given up once, this is an error.

What do you want to happen if the method is called multiple times?

In your arc_mutex branch, what happens is that each message will be delivered to one of the streams. You can't predict which stream will get each message if you have multiple streams; it's more or less random.

[sasakiyori]

It's because the ::tail method tries to give up ownership of the receiver, but the method could be called multiple times. Since ownership can only be given up once, this is an error.

Yes.

In my use case, I will ensure only one connection on it, I will control it by AutomicBool or somehow.
I can do this easily in c/c++ without mutex, and I am not sure can rust goes around this.
Maybe the usage of tonic and tokio on this use case is not suitable at all, but I'm afraid there have some work-around methods indeed but I miss it just because I am a beginner.

That will be very nice if you can give me some advices or solutions :)

[Darksonn]

You can do that like this:

pub struct XXService {
    receiver: std::sync::Mutex<Option<Receiver<Result<TailResponse, Status>>>>,
}

impl XXService {
    fn take_receiver(&self) -> Option<Receiver<Result<TailResponse, Status>>> {
        self.receiver.lock().unwrap().take()
    }
}

This method will return the receiver on the first call, and None on any future calls.

[Darksonn]

You could also use the atomic-take crate instead of a mutex/option combo.

[sasakiyori]

Thank you very much! This is exactly what I want!!

If I want to reuse the receiver when the only one client disconnect and then reconnect, I think I can use Option::replace to solve it as well.