You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by gk...@apache.org on 2025/10/20 12:58:28 UTC
(iggy) branch io_uring_tpc updated: feat(io_uring): fix ws bugs (#2279)
This is an automated email from the ASF dual-hosted git repository.
gkoszyk pushed a commit to branch io_uring_tpc
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/io_uring_tpc by this push:
new 2e7b6398f feat(io_uring): fix ws bugs (#2279)
2e7b6398f is described below
commit 2e7b6398f3a745f4ce707898c4071bb09adbd84b
Author: Krishna Vishal <kr...@gmail.com>
AuthorDate: Mon Oct 20 18:28:24 2025 +0530
feat(io_uring): fix ws bugs (#2279)
Bugs fixed:
- TLS handshake failed: This is due to system scenario test expecting
plan WS server and in the config TLS for WS was enabled.
- Connection closed log noise: This was happening because the websocket
client in benchmark sends `Close` frame to server and drops the
connection without waiting for ack from server. Due to this server
throws `Io error BrokenPipe`. This is expected in real world usage and
we handle this on the server side and downgrade the specific errors to
warning.
---
Cargo.lock | 2 +-
Cargo.toml | 2 +-
.../websocket_config/websocket_client_config.rs | 14 +------
core/configs/server.toml | 2 +-
core/sdk/src/websocket/websocket_client.rs | 40 ++++++++++--------
core/server/src/configs/websocket.rs | 14 +------
core/server/src/websocket/connection_handler.rs | 47 ++++++++++++++++------
core/server/src/websocket/websocket_listener.rs | 15 ++++---
core/server/src/websocket/websocket_sender.rs | 36 ++++++++++++++---
.../server/src/websocket/websocket_tls_listener.rs | 20 +++++----
10 files changed, 116 insertions(+), 76 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index 8e8a5e223..987cc4a2c 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1591,7 +1591,7 @@ dependencies = [
[[package]]
name = "compio-ws"
version = "0.1.0"
-source = "git+https://github.com/krishvishal/compio-ws?branch=main#ca9240b6b0eb2206e8ae050580a053c5e9d0e877"
+source = "git+https://github.com/krishvishal/compio-ws#ca9240b6b0eb2206e8ae050580a053c5e9d0e877"
dependencies = [
"compio",
"compio-buf",
diff --git a/Cargo.toml b/Cargo.toml
index fb61475ae..8f9337063 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -161,7 +161,7 @@ compio-tls = { git = "https://github.com/compio-rs/compio.git", rev = "fe4243f0b
"rustls",
] }
compio-io = { git = "https://github.com/compio-rs/compio.git", rev = "fe4243f0b6811ebc325afd081c9b087b4d9817be" }
-compio-ws = { git = "https://github.com/krishvishal/compio-ws", branch = "main", features = [
+compio-ws = { git = "https://github.com/krishvishal/compio-ws", features = [
"rustls",
] }
tungstenite = "0.27.0"
diff --git a/core/common/src/types/configuration/websocket_config/websocket_client_config.rs b/core/common/src/types/configuration/websocket_config/websocket_client_config.rs
index a3250272b..58bc5d9e3 100644
--- a/core/common/src/types/configuration/websocket_config/websocket_client_config.rs
+++ b/core/common/src/types/configuration/websocket_config/websocket_client_config.rs
@@ -97,17 +97,7 @@ impl Default for WebSocketConfig {
impl WebSocketConfig {
/// Convert to tungstenite WebSocketConfig so we can use tungstenite defaults
- pub fn to_tungstenite_config(&self) -> Option<TungsteniteConfig> {
- if self.read_buffer_size.is_none()
- && self.write_buffer_size.is_none()
- && self.max_write_buffer_size.is_none()
- && self.max_message_size.is_none()
- && self.max_frame_size.is_none()
- && !self.accept_unmasked_frames
- {
- return None;
- }
-
+ pub fn to_tungstenite_config(&self) -> TungsteniteConfig {
let mut config = TungsteniteConfig::default();
if let Some(read_buf_size) = self.read_buffer_size {
@@ -132,7 +122,7 @@ impl WebSocketConfig {
config = config.accept_unmasked_frames(self.accept_unmasked_frames);
- Some(config)
+ config
}
}
diff --git a/core/configs/server.toml b/core/configs/server.toml
index d721bc033..ea1be6c66 100644
--- a/core/configs/server.toml
+++ b/core/configs/server.toml
@@ -607,7 +607,7 @@ enabled = true
address = "0.0.0.0:8092"
[websocket.tls]
-enabled = true
+enabled = false
self_signed = true
cert_file = "core/certs/iggy_cert.pem"
key_file = "core/certs/iggy_key.pem"
diff --git a/core/sdk/src/websocket/websocket_client.rs b/core/sdk/src/websocket/websocket_client.rs
index 44fcadaec..3f34351cc 100644
--- a/core/sdk/src/websocket/websocket_client.rs
+++ b/core/sdk/src/websocket/websocket_client.rs
@@ -277,7 +277,7 @@ impl WebSocketClient {
let tungstenite_config = self.config.ws_config.to_tungstenite_config();
let (websocket_stream, response) =
- match client_async_with_config(request, tcp_stream, tungstenite_config).await {
+ match client_async_with_config(request, tcp_stream, Some(tungstenite_config)).await {
Ok(result) => result,
Err(error) => {
error!("WebSocket handshake failed: {}", error);
@@ -312,18 +312,20 @@ impl WebSocketClient {
let tungstenite_config = self.config.ws_config.to_tungstenite_config();
debug!("Initiating WebSocket TLS connection to: {}", ws_url);
- println!("Initiating WebSocket TLS connection to: {}", ws_url);
- println!("tungstenite_config: {:?}", tungstenite_config);
- let (websocket_stream, response) =
- match connect_async_tls_with_config(ws_url, tungstenite_config, false, Some(connector))
- .await
- {
- Ok(result) => result,
- Err(error) => {
- error!("WebSocket TLS handshake failed: {}", error);
- return self.handle_connection_error(retry_count).await;
- }
- };
+ let (websocket_stream, response) = match connect_async_tls_with_config(
+ ws_url,
+ Some(tungstenite_config),
+ false,
+ Some(connector),
+ )
+ .await
+ {
+ Ok(result) => result,
+ Err(error) => {
+ error!("WebSocket TLS handshake failed: {}", error);
+ return self.handle_connection_error(retry_count).await;
+ }
+ };
debug!(
"WebSocket TLS connection established. Response status: {}",
@@ -451,9 +453,7 @@ impl WebSocketClient {
info!("{NAME} client: {client_address} is disconnecting from server...");
self.set_state(ClientState::Disconnected).await;
- if let Some(mut stream) = self.stream.lock().await.take() {
- let _ = stream.shutdown().await;
- }
+ self.stream.lock().await.take();
self.publish_event(DiagnosticEvent::Disconnected).await;
let now = IggyTimestamp::now();
@@ -469,7 +469,13 @@ impl WebSocketClient {
let client_address = self.get_client_address_value().await;
info!("Shutting down the {NAME} client: {client_address}");
- self.disconnect().await?;
+ self.set_state(ClientState::Disconnected).await;
+
+ let stream = self.stream.lock().await.take();
+ if let Some(mut stream) = stream {
+ let _ = stream.shutdown().await;
+ }
+
self.set_state(ClientState::Shutdown).await;
self.publish_event(DiagnosticEvent::Shutdown).await;
info!("{NAME} client: {client_address} has been shutdown.");
diff --git a/core/server/src/configs/websocket.rs b/core/server/src/configs/websocket.rs
index 541e8253e..2c9c996ce 100644
--- a/core/server/src/configs/websocket.rs
+++ b/core/server/src/configs/websocket.rs
@@ -50,17 +50,7 @@ pub struct WebSocketTlsConfig {
}
impl WebSocketConfig {
- pub fn to_tungstenite_config(&self) -> Option<TungsteniteConfig> {
- if self.read_buffer_size.is_none()
- && self.write_buffer_size.is_none()
- && self.max_write_buffer_size.is_none()
- && self.max_message_size.is_none()
- && self.max_frame_size.is_none()
- && !self.accept_unmasked_frames
- {
- return None;
- }
-
+ pub fn to_tungstenite_config(&self) -> TungsteniteConfig {
let mut config = TungsteniteConfig::default();
if let Some(read_buf_size_str) = &self.read_buffer_size
@@ -95,7 +85,7 @@ impl WebSocketConfig {
config = config.accept_unmasked_frames(self.accept_unmasked_frames);
- Some(config)
+ config
}
}
diff --git a/core/server/src/websocket/connection_handler.rs b/core/server/src/websocket/connection_handler.rs
index 33f7f26b5..2c2ec1ced 100644
--- a/core/server/src/websocket/connection_handler.rs
+++ b/core/server/src/websocket/connection_handler.rs
@@ -28,7 +28,7 @@ use futures::FutureExt;
use iggy_common::IggyError;
use std::io::ErrorKind;
use std::rc::Rc;
-use tracing::{debug, error, info};
+use tracing::{debug, error, info, warn};
const INITIAL_BYTES_LENGTH: usize = 4;
@@ -72,7 +72,6 @@ pub(crate) async fn handle_connection(
res?;
let code: u32 =
u32::from_le_bytes(code_buffer_out[0..INITIAL_BYTES_LENGTH].try_into().unwrap());
-
initial_buffer.clear();
code_buffer_out.clear();
length_buffer = initial_buffer;
@@ -89,17 +88,39 @@ pub(crate) async fn handle_connection(
);
}
Err(error) => {
- error!("Command was not handled successfully, session: {session}, error: {error}.");
- if let IggyError::ClientNotFound(_) = error {
- sender.send_error_response(error).await?;
- debug!("WebSocket error response was sent to: {session}.");
- error!("Session: {session} will be deleted.");
- return Err(ConnectionError::from(IggyError::ClientNotFound(
- session.client_id,
- )));
- } else {
- sender.send_error_response(error).await?;
- debug!("WebSocket error response was sent to: {session}.");
+ match error {
+ IggyError::TcpError | IggyError::ConnectionClosed | IggyError::Disconnected => {
+ warn!(
+ "Client {} closed connection during request processing",
+ session.client_id
+ );
+ return Err(ConnectionError::from(IggyError::ConnectionClosed));
+ }
+ IggyError::ClientNotFound(_) => {
+ error!("Command failed for session: {session}, error: {error}.");
+ sender.send_error_response(error.clone()).await?;
+ return Err(ConnectionError::from(error));
+ }
+ _ => {
+ error!("Command failed for session: {session}, error: {error}.");
+ // try to send error response, but if it fails due to broken pipe, that's ok?
+ match sender.send_error_response(error).await {
+ Ok(_) => {
+ debug!("WebSocket error response was sent to: {session}.");
+ }
+ Err(IggyError::ConnectionClosed) => {
+ warn!(
+ "Could not send error response to {} - client already disconnected",
+ session.client_id
+ );
+ return Err(ConnectionError::from(IggyError::ConnectionClosed));
+ }
+ Err(send_err) => {
+ error!("Failed to send error response: {send_err}");
+ return Err(ConnectionError::from(send_err));
+ }
+ }
+ }
}
}
}
diff --git a/core/server/src/websocket/websocket_listener.rs b/core/server/src/websocket/websocket_listener.rs
index b3fc5969f..2fdc601cb 100644
--- a/core/server/src/websocket/websocket_listener.rs
+++ b/core/server/src/websocket/websocket_listener.rs
@@ -81,7 +81,7 @@ pub async fn start(
config.accept_unmasked_frames
);
- accept_loop(listener, ws_config, shard, shutdown).await
+ accept_loop(listener, Some(ws_config), shard, shutdown).await
}
async fn accept_loop(
@@ -137,10 +137,15 @@ async fn accept_loop(
}
registry_clone.remove_connection(&client_id);
- if let Err(error) = sender_kind.shutdown().await {
- shard_error!(shard_clone.id, "Failed to shutdown WebSocket stream for client: {}, address: {}. {}", client_id, remote_addr, error);
- } else {
- shard_info!(shard_clone.id, "Successfully closed WebSocket stream for client: {}, address: {}.", client_id, remote_addr);
+ match sender_kind.shutdown().await {
+ Ok(_) => {
+ shard_info!(shard_clone.id, "Successfully closed WebSocket stream for client: {}, address: {}.", client_id, remote_addr);
+ }
+ Err(_) => {
+ // shutdown failures during client disconnect are expected and normal
+ // real errors would have been caught earlier in handle_connection
+ shard_debug!(shard_clone.id, "WebSocket shutdown completed with error for client: {} (likely client already disconnected)", client_id);
+ }
}
}
Err(error) => {
diff --git a/core/server/src/websocket/websocket_sender.rs b/core/server/src/websocket/websocket_sender.rs
index d83042bc5..7759378e0 100644
--- a/core/server/src/websocket/websocket_sender.rs
+++ b/core/server/src/websocket/websocket_sender.rs
@@ -26,7 +26,7 @@ use compio_ws::TungsteniteError;
use compio_ws::{WebSocketMessage as Message, WebSocketStream};
use iggy_common::IggyError;
use std::ptr;
-use tracing::debug;
+use tracing::{debug, warn};
const READ_BUFFER_CAPACITY: usize = 8192;
const WRITE_BUFFER_CAPACITY: usize = 8192;
@@ -52,10 +52,23 @@ impl WebSocketSender {
return Ok(());
}
let data = self.write_buffer.split().freeze();
- self.stream
- .send(Message::Binary(data))
- .await
- .map_err(|_| IggyError::TcpError)
+ debug!("WebSocket sending data: {:?}", data.to_vec());
+
+ self.stream.send(Message::Binary(data)).await.map_err(|e| {
+ debug!("WebSocket send error: {:?}", e);
+ match e {
+ TungsteniteError::ConnectionClosed | TungsteniteError::AlreadyClosed => {
+ IggyError::ConnectionClosed
+ }
+ TungsteniteError::Io(ref io_err)
+ if io_err.kind() == std::io::ErrorKind::BrokenPipe =>
+ {
+ warn!("Broken pipe detected (client closed connection)");
+ IggyError::ConnectionClosed
+ }
+ _ => IggyError::TcpError,
+ }
+ })
}
}
@@ -171,6 +184,17 @@ impl Sender for WebSocketSender {
self.stream
.send(Message::Binary(response_bytes.freeze()))
.await
- .map_err(|_| IggyError::TcpError)
+ .map_err(|e| match e {
+ TungsteniteError::ConnectionClosed | TungsteniteError::AlreadyClosed => {
+ IggyError::ConnectionClosed
+ }
+ TungsteniteError::Io(ref io_err)
+ if io_err.kind() == std::io::ErrorKind::BrokenPipe =>
+ {
+ warn!("Broken pipe in vectored send - client closed connection");
+ IggyError::ConnectionClosed
+ }
+ _ => IggyError::TcpError,
+ })
}
}
diff --git a/core/server/src/websocket/websocket_tls_listener.rs b/core/server/src/websocket/websocket_tls_listener.rs
index 60a03ee91..da6ae9d79 100644
--- a/core/server/src/websocket/websocket_tls_listener.rs
+++ b/core/server/src/websocket/websocket_tls_listener.rs
@@ -117,7 +117,6 @@ pub async fn start(
"WebSocket TLS Server",
local_addr
);
-
let ws_config = config.to_tungstenite_config();
shard_info!(
shard.id,
@@ -141,7 +140,7 @@ pub async fn start(
async fn accept_loop(
listener: TcpListener,
acceptor: TlsAcceptor,
- ws_config: Option<compio_ws::WebSocketConfig>,
+ ws_config: compio_ws::WebSocketConfig,
shard: Rc<IggyShard>,
shutdown: ShutdownToken,
) -> Result<(), IggyError> {
@@ -170,7 +169,7 @@ async fn accept_loop(
shard_info!(shard.id, "Accepted new TCP connection for WebSocket TLS handshake from: {}", remote_addr);
let shard_clone = shard.clone();
- let ws_config_clone = ws_config;
+ let ws_config_clone = ws_config;
let registry = shard.task_registry.clone();
let registry_clone = registry.clone();
@@ -179,7 +178,7 @@ async fn accept_loop(
Ok(tls_stream) => {
shard_info!(shard_clone.id, "TLS handshake successful for {}, performing WebSocket upgrade...", remote_addr);
- match accept_async_with_config(tls_stream, ws_config_clone).await {
+ match accept_async_with_config(tls_stream, Some(ws_config_clone)).await {
Ok(websocket) => {
info!("WebSocket TLS handshake successful from: {}", remote_addr);
@@ -202,10 +201,15 @@ async fn accept_loop(
}
registry_clone.remove_connection(&client_id);
- if let Err(error) = sender_kind.shutdown().await {
- shard_error!(shard_clone.id, "Failed to shutdown WebSocket TLS stream for client: {}, address: {}. {}", client_id, remote_addr, error);
- } else {
- shard_info!(shard_clone.id, "Successfully closed WebSocket TLS stream for client: {}, address: {}.", client_id, remote_addr);
+ match sender_kind.shutdown().await {
+ Ok(_) => {
+ shard_info!(shard_clone.id, "Successfully closed WebSocket TLS stream for client: {}, address: {}.", client_id, remote_addr);
+ }
+ Err(_) => {
+ // shutdown failures during client disconnect are expected and normal
+ // real errors would have been caught earlier in handle_connection
+ shard_debug!(shard_clone.id, "WebSocket TLS shutdown completed with error for client: {} (likely client already disconnected)", client_id);
+ }
}
}
Err(error) => {