Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

fix(server): remove MethodSinkPermit to fix backpressure issue on concurrent subscriptions #1126

Merged
merged 5 commits into from
May 18, 2023
Merged

fix(server): remove MethodSinkPermit to fix backpressure issue on concurrent subscriptions #1126

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
5215f76
fix(server): don't use `Permit` for responses
niklasad1 May 17, 2023
3b8b236
cleanup code
niklasad1 May 18, 2023
180a867
Update server/src/transport/ws.rs
niklasad1 May 18, 2023
1b63c86
Update core/src/server/helpers.rs
niklasad1 May 18, 2023
2c2cdde
cargo fmt
niklasad1 May 18, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
44 changes: 15 additions & 29 deletions core/src/server/helpers.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,7 +34,7 @@ use jsonrpsee_types::error::{
use jsonrpsee_types::{Id, InvalidRequest, Response, ResponsePayload};
use serde::Serialize;
use serde_json::value::to_raw_value;
use tokio::sync::mpsc::{self, OwnedPermit};
use tokio::sync::mpsc;

use super::{DisconnectError, SendTimeoutError, SubscriptionMessage, TrySendError};

Expand Down Expand Up @@ -139,45 +139,31 @@ impl MethodSink {
self.tx.send(msg).await.map_err(Into::into)
}

/// Send a JSON-RPC error to the client
pub async fn send_error<'a>(&self, id: Id<'a>, err: ErrorObject<'a>) -> Result<(), DisconnectError> {
let json =
serde_json::to_string(&Response::new(ResponsePayload::<()>::Error(err), id)).expect("valid JSON; qed");

self.send(json).await
}

/// Similar to to `MethodSink::send` but only waits for a limited time.
pub async fn send_timeout(&self, msg: String, timeout: Duration) -> Result<(), SendTimeoutError> {
tx_log_from_str(&msg, self.max_log_length);
self.tx.send_timeout(msg, timeout).await.map_err(Into::into)
}

/// Waits for channel capacity. Once capacity to send one message is available, it is reserved for the caller.
pub async fn reserve(&self) -> Result<MethodSinkPermit, DisconnectError> {
match self.tx.clone().reserve_owned().await {
Ok(permit) => Ok(MethodSinkPermit { tx: permit, max_log_length: self.max_log_length }),
/// Waits for channel capacity. Once capacity to send one message is available.
pub async fn has_capacity(&self) -> Result<(), DisconnectError> {
match self.tx.reserve().await {
// The permit is thrown away here because it's just
// a way to ensure that underlying buffer is not empty.
niklasad1 marked this conversation as resolved.
Show resolved Hide resolved
Ok(_) => Ok(()),
Err(_) => Err(DisconnectError(SubscriptionMessage::empty())),
}
}
}

/// A method sink with reserved spot in the bounded queue.
#[derive(Debug)]
pub struct MethodSinkPermit {
tx: OwnedPermit<String>,
max_log_length: u32,
}

impl MethodSinkPermit {
/// Send a JSON-RPC error to the client
pub fn send_error(self, id: Id, err: ErrorObject) {
let json = serde_json::to_string(&Response::new(ResponsePayload::<()>::Error(err.into_owned()), id))
.expect("valid JSON; qed");

self.send_raw(json)
}

/// Send a raw JSON-RPC message to the client, `MethodSink` does not check the validity
/// of the JSON being sent.
pub fn send_raw(self, json: String) {
self.tx.send(json.clone());
tx_log_from_str(&json, self.max_log_length);
}
}

/// Figure out if this is a sufficiently complete request that we can extract an [`Id`] out of, or just plain
/// unparseable garbage.
pub fn prepare_error(data: &[u8]) -> (Id<'_>, ErrorCode) {
Expand Down
2 changes: 1 addition & 1 deletion core/src/server/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,7 +38,7 @@ mod rpc_module;
mod subscription;

pub use error::*;
pub use helpers::{BatchResponseBuilder, BoundedWriter, MethodResponse, MethodSink, MethodSinkPermit};
pub use helpers::{BatchResponseBuilder, BoundedWriter, MethodResponse, MethodSink};
pub use host_filtering::*;
pub use rpc_module::*;
pub use subscription::*;
Expand Down
47 changes: 23 additions & 24 deletions server/src/transport/ws.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,9 +12,7 @@ use hyper::upgrade::Upgraded;
use jsonrpsee_core::server::helpers::{
batch_response_error, prepare_error, BatchResponseBuilder, MethodResponse, MethodSink,
};
use jsonrpsee_core::server::{
BoundedSubscriptions, CallOrSubscription, MethodCallback, MethodSinkPermit, Methods, SubscriptionState,
};
use jsonrpsee_core::server::{BoundedSubscriptions, CallOrSubscription, MethodCallback, Methods, SubscriptionState};
use jsonrpsee_core::tracing::{rx_log_from_json, tx_log_from_str};
use jsonrpsee_core::traits::IdProvider;
use jsonrpsee_core::{Error, JsonRawValue};
Expand Down Expand Up @@ -263,14 +261,19 @@ pub(crate) async fn background_task<L: Logger>(sender: Sender, mut receiver: Rec
let result = loop {
data.clear();

let sink_permit = match wait_for_permit(&sink, stopped).await {
Some((permit, stop)) => {
stopped = stop;
permit
}
None => break Ok(Shutdown::ConnectionClosed),
// This is a guard to ensure that underlying socket is only "read" if connection buffer has capacity
// to send further messages.
//
// This, this check enforces that if the client can't keep up with connection
// no new messages are handled before the existing ones are read.
//
// TCP retransmission mechanism will take of the rest and adjust the window size accordingly.
niklasad1 marked this conversation as resolved.
Show resolved Hide resolved
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) => {
Expand All @@ -290,7 +293,9 @@ pub(crate) async fn background_task<L: Logger>(sender: Sender, mut receiver: Rec
current,
maximum
);
sink_permit.send_error(Id::Null, reject_too_big_request(max_request_body_size));
if sink.send_error(Id::Null, reject_too_big_request(max_request_body_size)).await.is_err() {
break Ok(Shutdown::ConnectionClosed);
}

continue;
}
Expand All @@ -310,7 +315,6 @@ pub(crate) async fn background_task<L: Logger>(sender: Sender, mut receiver: Rec
max_log_length,
max_response_body_size,
sink: sink.clone(),
sink_permit,
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I don't really understand the point of acquiring the permis above now? Might it be safer to not try to reserve a slot at all, and just to .await until there is an opening each time we want to send an actual message?

Copy link
Member Author

@niklasad1 niklasad1 May 17, 2023 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

without "the reserve" the server will read the underlying socket and then the client is not "forced" to read its end of the socket to send new messages.

That's is the entire reason of reserve here.

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

otherwise, it will not propagated all the way down to the TCP level when the window size is adjusted etc

Copy link
Member Author

@niklasad1 niklasad1 May 17, 2023 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I get your concern, this permit could interfere with the stuff trying to send, so we could do:

		// Wait until there is a slot in the bounded channel.
		//
		// This will force the client to read socket on the other side
		// otherwise the socket will not be read again.
		let shutdown = loop {
			if stop_handle.shutdown_requested() {
				break true;
			}

			if sink.capacity() != 0 {
				break false;
			}
		};

		if shutdown {
			break Ok(Shutdown::Stopped);
		}

but feels a bit annoying to waste cycles doing busy looping :)

id_provider: id_provider.clone(),
logger: logger.clone(),
data: std::mem::take(&mut data),
Expand Down Expand Up @@ -403,21 +407,18 @@ enum Receive<S> {
Ok(S),
}

// Wait until there is a slot in the bounded channel.
//
// This will force the client to read socket on the other side
// otherwise the socket will not be read again.
// Wait until there is capacity in connection buffer to send one message.
//
// Fails if the server was stopped.
async fn wait_for_permit<S>(sink: &MethodSink, stopped: S) -> Option<(MethodSinkPermit, S)>
async fn wait_until_connection_buffer_has_capacity<S>(sink: &MethodSink, stopped: S) -> Option<S>
where
S: Future<Output = ()> + Unpin,
{
let reserve = sink.reserve();
let reserve = sink.has_capacity();
tokio::pin!(reserve);

match futures_util::future::select(reserve, stopped).await {
Either::Left((Ok(sink), s)) => Some((sink, s)),
Either::Left((Ok(_), s)) => Some(s),
_ => None,
}
}
Expand Down Expand Up @@ -461,7 +462,6 @@ struct ExecuteCallParams<L: Logger> {
max_response_body_size: u32,
max_log_length: u32,
sink: MethodSink,
sink_permit: MethodSinkPermit,
logger: L,
}

Expand All @@ -471,7 +471,6 @@ async fn execute_unchecked_call<L: Logger>(params: ExecuteCallParams<L>) {
conn_id,
data,
sink,
sink_permit,
max_response_body_size,
max_log_length,
methods,
Expand Down Expand Up @@ -505,7 +504,7 @@ async fn execute_unchecked_call<L: Logger>(params: ExecuteCallParams<L>) {

CallOrSubscription::Call(r) => {
logger.on_response(&r.result, request_start, TransportProtocol::WebSocket);
sink_permit.send_raw(r.result);
_ = sink.send(r.result).await;
Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this will only fail if the connection is closed.

}
}
}
Expand All @@ -518,7 +517,7 @@ async fn execute_unchecked_call<L: Logger>(params: ExecuteCallParams<L>) {
ErrorObject::borrowed(BATCHES_NOT_SUPPORTED_CODE, &BATCHES_NOT_SUPPORTED_MSG, None),
);
logger.on_response(&response.result, request_start, TransportProtocol::WebSocket);
sink_permit.send_raw(response.result);
_ = sink.send(response.result).await;
return;
}
BatchRequestConfig::Limit(limit) => limit as usize,
Expand All @@ -542,11 +541,11 @@ async fn execute_unchecked_call<L: Logger>(params: ExecuteCallParams<L>) {
if let Some(response) = response {
tx_log_from_str(&response, max_log_length);
logger.on_response(&response, request_start, TransportProtocol::WebSocket);
sink_permit.send_raw(response);
_ = sink.send(response).await;
}
}
_ => {
sink_permit.send_error(Id::Null, ErrorCode::ParseError.into());
_ = sink.send_error(Id::Null, ErrorCode::ParseError.into()).await;
}
};
}
Expand Down