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/07/05 08:25:52 UTC
(iggy) branch use_tcp_options_instead_of_socket created (now e032db28)
This is an automated email from the ASF dual-hosted git repository.
gkoszyk pushed a change to branch use_tcp_options_instead_of_socket
in repository https://gitbox.apache.org/repos/asf/iggy.git
at e032db28 refactor(io_uring): use tcp opts from compio instead of socket config
This branch includes the following new commits:
new e032db28 refactor(io_uring): use tcp opts from compio instead of socket config
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(iggy) 01/01: refactor(io_uring): use tcp opts from compio instead of socket config
Posted by gk...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
gkoszyk pushed a commit to branch use_tcp_options_instead_of_socket
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit e032db287875b965f95e452f4dacbf9f7802f066
Author: numinex <g....@gmail.com>
AuthorDate: Sat Jul 5 10:25:41 2025 +0200
refactor(io_uring): use tcp opts from compio instead of socket config
---
core/common/src/error/iggy_error.rs | 3 ++
core/server/src/shard/mod.rs | 2 +-
core/server/src/shard/transmission/event.rs | 5 ++-
core/server/src/tcp/tcp_listener.rs | 51 ++++++++++++++++++++++++-----
core/server/src/tcp/tcp_server.rs | 2 +-
5 files changed, 51 insertions(+), 12 deletions(-)
diff --git a/core/common/src/error/iggy_error.rs b/core/common/src/error/iggy_error.rs
index 334e14c7..53b4eb5b 100644
--- a/core/common/src/error/iggy_error.rs
+++ b/core/common/src/error/iggy_error.rs
@@ -464,6 +464,9 @@ pub enum IggyError {
ShardNotFound(u32, u32, u32) = 11000,
#[error("Shard communication error, shard ID: {0}")]
ShardCommunicationError(u16) = 11001,
+
+ #[error("Cannot bind to socket with addr: {0}")]
+ CannotBindToSocket(String) = 12000,
}
impl IggyError {
diff --git a/core/server/src/shard/mod.rs b/core/server/src/shard/mod.rs
index 2178d2b3..16b4459c 100644
--- a/core/server/src/shard/mod.rs
+++ b/core/server/src/shard/mod.rs
@@ -535,7 +535,7 @@ impl IggyShard {
*compression_algorithm,
*max_topic_size,
*replication_factor,
- shards_assignment.clone()
+ shards_assignment.clone(),
)
.await
}
diff --git a/core/server/src/shard/transmission/event.rs b/core/server/src/shard/transmission/event.rs
index 60a28dfc..c1a34c27 100644
--- a/core/server/src/shard/transmission/event.rs
+++ b/core/server/src/shard/transmission/event.rs
@@ -2,7 +2,10 @@ use std::net::SocketAddr;
use iggy_common::{CompressionAlgorithm, Identifier, IggyExpiry, MaxTopicSize};
-use crate::{shard::{namespace::IggyNamespace, ShardInfo}, streaming::clients::client_manager::Transport};
+use crate::{
+ shard::{ShardInfo, namespace::IggyNamespace},
+ streaming::clients::client_manager::Transport,
+};
#[derive(Debug)]
pub enum ShardEvent {
diff --git a/core/server/src/tcp/tcp_listener.rs b/core/server/src/tcp/tcp_listener.rs
index b541305b..f927cf33 100644
--- a/core/server/src/tcp/tcp_listener.rs
+++ b/core/server/src/tcp/tcp_listener.rs
@@ -17,33 +17,66 @@
*/
use crate::binary::sender::SenderKind;
+use crate::configs::tcp::TcpSocketConfig;
use crate::shard::IggyShard;
use crate::shard::transmission::event::ShardEvent;
use crate::streaming::clients::client_manager::Transport;
use crate::tcp::connection_handler::{handle_connection, handle_error};
+use compio::net::{TcpListener, TcpOpts};
+use error_set::ErrContext;
use futures::FutureExt;
use iggy_common::IggyError;
use socket2::Socket;
use std::net::SocketAddr;
+use std::num::TryFromIntError;
use std::rc::Rc;
use std::time::Duration;
use tracing::{error, info};
+async fn create_listener(
+ addr: SocketAddr,
+ config: &TcpSocketConfig,
+) -> Result<TcpListener, std::io::Error> {
+ // Required by the thread-per-core model...
+ // We create bunch of sockets on different threads, that bind to exactly the same address and port.
+ let opts = TcpOpts::new().set_reuse_address(true).set_reuse_port(true);
+ let opts = if config.override_defaults {
+ let recv_buffer_size = config
+ .recv_buffer_size
+ .as_bytes_u64()
+ .try_into()
+ .expect("Failed to parse recv_buffer_size for TCP socket");
+
+ let send_buffer_size = config
+ .send_buffer_size
+ .as_bytes_u64()
+ .try_into()
+ .expect("Failed to parse send_buffer_size for TCP socket");
+
+ opts.set_recv_buffer_size(recv_buffer_size)
+ .set_send_buffer_size(send_buffer_size)
+ .set_keepalive(config.keepalive)
+ .set_linger(config.linger.get_duration())
+ .set_no_delay(config.nodelay)
+ } else {
+ opts
+ };
+ TcpListener::bind_with_options(addr, opts).await
+}
+
pub async fn start(
server_name: &'static str,
addr: SocketAddr,
- socket: Socket,
+ config: &TcpSocketConfig,
shard: Rc<IggyShard>,
) -> Result<(), IggyError> {
- socket
- .bind(&addr.into())
- .expect("Failed to bind TCP listener");
- socket.listen(1024).unwrap();
- //TODO: Fix it later to use `TcpOpts` from compio.
- let listener: std::net::TcpListener = socket.into();
- let listener = compio::net::TcpListener::from_std(listener).unwrap();
+ let listener = create_listener(addr, config)
+ .await
+ .map_err(|_| IggyError::CannotBindToSocket(addr.to_string()))
+ .with_error_context(|err| {
+ format!("Failed to bind {server_name} server to address: {addr}, {err}")
+ })?;
info!("{server_name} server has started on: {:?}", addr);
-
loop {
let shutdown_check = async {
loop {
diff --git a/core/server/src/tcp/tcp_server.rs b/core/server/src/tcp/tcp_server.rs
index ac019bb0..6b299bff 100644
--- a/core/server/src/tcp/tcp_server.rs
+++ b/core/server/src/tcp/tcp_server.rs
@@ -44,7 +44,7 @@ pub async fn spawn_tcp_server(shard: Rc<IggyShard>) -> Result<(), IggyError> {
// TODO: Fixme -- storing addr of the server inside of the config for integration tests...
match shard.config.tcp.tls.enabled {
true => tcp_tls_listener::start(server_name, addr, socket, shard.clone()).await?,
- false => tcp_listener::start(server_name, addr, socket, shard.clone()).await?,
+ false => tcp_listener::start(server_name, addr, socket_config, shard.clone()).await?,
};
Ok(())