Discussion: should we await send call message in serve loop?

[4t145]

In the serve loop of serve_inner, we await each SinkExt::send call. Should we make them asynchronous? It should, this may lead a lot of changes, for we have to change the defination of InfoTransport, which allow transport to create a send future and executed asynchronously by tokio.

As far as I can see by now, the Stream trait is proper for receiving messages from peer, but Sink seems not suitable for sending messages to peer. Because it doesn't focus on the sending result of single messages. It could be a mistake I made.

@jokemanfire How do you think about this?

[jokemanfire]

Let me understand, what is the trait used for intotransport now
Sending side sink, receiving side stream. Do you think there is a problem with using the sink trait on the sending side now? Is there any relevant document that I can take a look at?
Do you mean you don't need to pay attention to the return results similar to 'let send_result = sink.send(JsonRpcMessage::notification(notification)).await;'

[4t145]

@jokemanfire

To get the corresponded result of SinkExt::send, the serve loop need to block on awaiting the result of send call. It's okay with io transport since we cannot write multi messages concurrently. But when it comes to the case like http request, we had to await the corresponded response of one request. After all, http request interface is not a sink.

[jokemanfire]

Understood. We need to consider concurrent message processing, right？

[4t145]

@jokemanfire Yes. And the error from Sink is the error of sink itself, not the error for a single call. Like a channel, when you get a error from calling send on a channel, you get a error of this channel like channnel blocking, channel closed or something else, it not the error of the send(). When we call send in serve_inner, what we want is exactly the error of this send call.

As what it says it futures lib, an error from a sink usually means the sink itself is broken.

[jokemanfire]

Reconstruct the into_transport ,just use channel not the trait, and io's impl wrap in the channel?

[4t145]

It could be something like this:

pub trait TransportSend<Item>: Send + Clone {
    type Error;
    fn send(&self, item: Item) -> impl IntoFuture<Output = Result<(), Self::Error>>;
}

impl<T, S> TransportSend<T> for SharedSink<S>
where
    S: Sink<T> + Send,
{
    type Error = <S as Sink<T>>::Error;
    fn send(&self, item: T) -> impl IntoFuture<Output = Result<(), Self::Error>> {
        let sink = self.sink.clone();
        async {
            let mut sink = sink.lock_owned().await;
            sink.send(item).await
        }
    }
}

pub struct SharedSink<S> {
    sink: Arc<tokio::sync::Mutex<S>>,
}

impl<S> SharedSink<S> {
    pub fn new(sink: S) -> Self {
        Self {
            sink: Arc::new(tokio::sync::Mutex::new(sink)),
        }
    }
}

impl<S> Clone for SharedSink<S> {
    fn clone(&self) -> Self {
        Self {
            sink: self.sink.clone(),
        }
    }
}

and

pub trait IntoTransport<R, E, A>: Send + 'static
where
    R: ServiceRole,
    E: std::error::Error + Send + 'static,
{
    fn into_transport(
        self,
    ) -> (
        impl TransportSend<TxJsonRpcMessage<R>, Error = E> + Send + 'static,
        impl Stream<Item = RxJsonRpcMessage<R>> + Send + 'static,
    );
}

This shouldn't import breaking change unless user implemented IntoTransport themselves

And I think the Clone may be unnecessary

---

update:

pub trait TransportSend<Item>: Send {
    type Error;
    fn send(
        &mut self,
        item: Item,
    ) -> impl Future<Output = Result<(), Self::Error>> + Send + 'static;
    fn close(&mut self) -> impl Future<Output = Result<(), Self::Error>> + Send;
}

impl<T, S> TransportSend<T> for SharedSink<S>
where
    S: Sink<T> + Send,
{
    type Error = <S as Sink<T>>::Error;
    fn send(&mut self, item: T) -> impl Future<Output = Result<(), Self::Error>> + Send + 'static {
        let sink = self.sink.clone();
        async {
            let mut sink = sink.lock_owned().await;
            sink.send(item).await
        }
    }
    fn close(&mut self) -> impl Future<Output = Result<(), Self::Error>> + Send {
        let sink = self.sink.clone();
        async {
            let mut sink = sink.lock_owned().await;
            sink.close().await
        }
    }
}

pub struct SharedSink<S> {
    sink: Arc<tokio::sync::Mutex<S>>,
}

impl<S> SharedSink<S> {
    pub fn new(sink: S) -> Self {
        Self {
            sink: Arc::new(tokio::sync::Mutex::new(sink)),
        }
    }
}

This should be enough

[jokemanfire]

This is the encapsulation of the sink, but this wrapper could not get the error like channel?

[4t145]

@jokemanfire Not only for sink, but also for others possible senders. And it's error is error per request. If there's connection issue, it should be included in TransportSend::Error.

[jokemanfire]

@jokemanfire Not only for sink, but also for others possible senders. And it's error is error per request. If there's connection issue, it should be included in TransportSend::Error.不仅对于 sink，其他可能的发送者也应如此。并且它的错误是每次请求的错误。如果有连接问题，应该包含在 TransportSend::Error 中。

I think that's well. :)

[jokemanfire]

and cc @4t145 https://github.com/orgs/modelcontextprotocol/discussions/354 this is a new proposal.

[4t145]

@jokemanfire I'am not going to reply this comment there: To be honeiest I don't think rust today has such flexibility and compatibility to play this role.

[jokemanfire]

I think we can we can take a look at the packaging results @4t145

[jokemanfire]

┌──────────────────────────────────────────────────┐
│                    server                                                                                    │
│                                                                                                                 │
│  ┌─────────┐     ┌─────────────┐     ┌────────┐  │     ┌──────────┐
│  │ task 1           │─→│                              │     │                   │     │     │          │
│  └─────────┘     │                             │     │                    │     │     │          │
│                                │ sink_proxy_tx         │─→│sink_proxy_rx│─→│sink│─→│ remote │
│  ┌─────────┐     │                             │     │                    │     │     │          │
│  │ task 2           │─→│                              │     │                   │      │     │         │
│  └─────────┘     └─────────────┘     └────────┘     │     └──────────┘
│                                                                                                                 │
└──────────────────────────────────────────────────┘