From dfa52f1ed5e04206b187a2e3002c3f05765ac552 Mon Sep 17 00:00:00 2001 From: Niklas Adolfsson Date: Thu, 28 Sep 2023 11:22:53 +0200 Subject: [PATCH] fix(server): remove `bounded channel` check (#1209) This was intended to propogate the backpressure all the way down the underlying socket but it's weird and doesn't work very well. For subscriptions the backpressure will be handled by implementation itself and just rely on that now. --- CHANGELOG.md | 11 +++++++++++ server/src/transport/ws.rs | 29 ----------------------------- 2 files changed, 11 insertions(+), 29 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 8f4ac3862b..8ffded3b44 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,17 @@ The format is based on [Keep a Changelog]. [Keep a Changelog]: http://keepachangelog.com/en/1.0.0/ +## [v0.20.2] - 2023-10-13 + +This release removes the bounded buffer check which was intended to provide +backpressure all the way down to the TCP layer but it didn't work well. + +For subscriptions the backpressure will be handled by implementation itself +and just rely on that. + +### [Changed] +- server: remove bounded channel check ([#1209](https://github.com/paritytech/jsonrpsee/pull/1209)) + ## [v0.20.1] - 2023-09-15 This release adds support for `synchronous subscriptions` and fixes a leak in WebSocket server diff --git a/server/src/transport/ws.rs b/server/src/transport/ws.rs index 8fa76fbac1..886c034770 100644 --- a/server/src/transport/ws.rs +++ b/server/src/transport/ws.rs @@ -277,19 +277,6 @@ pub(crate) async fn background_task(sender: Sender, mut receiver: Rec let result = loop { data.clear(); - // This is a guard to ensure that the underlying socket is only read if there is space in - // the buffer for messages to be sent back to them. - // - // Thus, this check enforces that if the client can't keep up with receiving messages, - // then no new messages will be read from them. - // - // TCP retransmission mechanism will take of the rest and adjust the window size accordingly. - let Some(stop) = wait_until_connection_buffer_has_capacity(&sink, stopped).await else { - break Ok(Shutdown::ConnectionClosed); - }; - - stopped = stop; - match try_recv(&mut receiver, &mut data, stopped).await { Receive::Shutdown => break Ok(Shutdown::Stopped), Receive::Ok(stop) => { @@ -411,22 +398,6 @@ enum Receive { Ok(S), } -// Wait until there is capacity in connection buffer to send one message. -// -// Fails if the server was stopped. -async fn wait_until_connection_buffer_has_capacity(sink: &MethodSink, stopped: S) -> Option -where - S: Future + Unpin, -{ - let reserve = sink.has_capacity(); - tokio::pin!(reserve); - - match futures_util::future::select(reserve, stopped).await { - Either::Left((Ok(_), s)) => Some(s), - _ => None, - } -} - /// Attempts to read data from WebSocket fails if the server was stopped. async fn try_recv(receiver: &mut Receiver, data: &mut Vec, stopped: S) -> Receive where