Adding UDP customization and configurable through the config file

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "torrust-actix"
-version = "4.0.11"
+version = "4.0.13"
 edition = "2024"
 license = "AGPL-3.0"
 authors = [
@@ -46,6 +46,7 @@ serde = { version = "^1.0", features = ["derive"] }
 serde_json = { version = "^1.0", features = ["preserve_order"] }
 serde_millis = "^0.1"
 sha1 = "^0.10"
+socket2 = "^0.6"
 sqlx = { version = "^0.8", features = ["mysql", "postgres", "sqlite", "runtime-tokio-rustls"] }
 thiserror = "^2.0"
 tokio = { version = "^1.47", features = ["full"] }

README.md

@@ -172,6 +172,15 @@ UDP_0_THREADS <UINT64>
 
 ### ChangeLog
 
+#### v4.0.13
+* Added further UDP improvement by adding customization, also added to the config:
+  * Receive Buffer Size
+  * Send Buffer Size
+  * Reuse of Address
+
+#### v4.0.12
+* Updating libraries and cleanup code
+
 #### v4.0.11
 * Updating libraries
 * Adding healthcheck for Docker through Python check script

docker/Dockerfile

@@ -2,7 +2,7 @@ FROM rust:alpine
 
 RUN apk add git musl-dev curl pkgconfig openssl-dev openssl-libs-static
 RUN git clone https://github.com/Power2All/torrust-actix.git /app/torrust-actix
-RUN cd /app/torrust-actix && git checkout tags/v4.0.12
+RUN cd /app/torrust-actix && git checkout tags/v4.0.13
 WORKDIR /app/torrust-actix
 RUN cd /app/torrust-actix
 RUN cargo build --release && rm -Rf target/release/.fingerprint target/release/build target/release/deps target/release/examples target/release/incremental

src/config/impls/configuration.rs

@@ -123,6 +123,9 @@ impl Configuration {
                     enabled: true,
                     bind_address: String::from("0.0.0.0:6969"),
                     threads: available_parallelism().unwrap().get() as u64,
+                    receive_buffer_size: 134217728,
+                    send_buffer_size: 67108864,
+                    reuse_address: true
                 }
             ),
             api_server: vec!(

src/config/structs/udp_trackers_config.rs

@@ -4,5 +4,8 @@ use serde::{Deserialize, Serialize};
 pub struct UdpTrackersConfig {
     pub enabled: bool,
     pub bind_address: String,
-    pub threads: u64
+    pub threads: u64,
+    pub receive_buffer_size: usize,
+    pub send_buffer_size: usize,
+    pub reuse_address: bool
 }

src/main.rs

@@ -196,9 +196,12 @@ fn main() -> std::io::Result<()>
                     udp_check_host_and_port_used(udp_server_object.bind_address.clone());
                     let address: SocketAddr = udp_server_object.bind_address.parse().unwrap();
                     let threads: u64 = udp_server_object.threads;
+                    let recv_buffer_size: usize = udp_server_object.receive_buffer_size;
+                    let send_buffer_size: usize = udp_server_object.send_buffer_size;
+                    let reuse_address: bool = udp_server_object.reuse_address;
                     let tracker_clone = tracker.clone();
                     let tokio_udp = Arc::new(Builder::new_multi_thread().thread_name("udp").worker_threads(threads as usize).enable_all().build()?);
-                    udp_futures.push(udp_service(address, threads, tracker_clone, udp_rx.clone(), tokio_udp.clone()).await);
+                    udp_futures.push(udp_service(address, threads, recv_buffer_size, send_buffer_size, reuse_address, tracker_clone, udp_rx.clone(), tokio_udp.clone()).await);
                     udp_tokio_threads.push(tokio_udp.clone());
                 }
             }

src/udp/impls/udp_server.rs

@@ -3,6 +3,7 @@ use std::net::{Ipv4Addr, Ipv6Addr, SocketAddr};
 use std::sync::Arc;
 use std::time::SystemTime;
 use log::{debug, info};
+use socket2::{Socket, Domain, Type, Protocol};
 use tokio::net::UdpSocket;
 use crate::stats::enums::stats_event::StatsEvent;
 use crate::tracker::enums::torrent_peers_type::TorrentPeersType;
@@ -34,12 +35,23 @@ use crate::udp::udp::{MAX_PACKET_SIZE, MAX_SCRAPE_TORRENTS};
 
 impl UdpServer {
     #[tracing::instrument(level = "debug")]
-    pub async fn new(tracker: Arc<TorrentTracker>, bind_address: SocketAddr, threads: u64) -> tokio::io::Result<UdpServer>
+    pub async fn new(tracker: Arc<TorrentTracker>, bind_address: SocketAddr, threads: u64, recv_buffer_size: usize, send_buffer_size: usize, reuse_address: bool) -> tokio::io::Result<UdpServer>
     {
-        let socket = UdpSocket::bind(bind_address).await?;
+        let domain = if bind_address.is_ipv4() { Domain::IPV4 } else { Domain::IPV6 };
+        let socket = Socket::new(domain, Type::DGRAM, Some(Protocol::UDP))?;
+
+        socket.set_recv_buffer_size(recv_buffer_size).map_err(|e| tokio::io::Error::new(tokio::io::ErrorKind::Other, e))?;
+        socket.set_send_buffer_size(send_buffer_size).map_err(|e| tokio::io::Error::new(tokio::io::ErrorKind::Other, e))?;
+        socket.set_reuse_address(reuse_address).map_err(|e| tokio::io::Error::new(tokio::io::ErrorKind::Other, e))?;
+        socket.bind(&bind_address.into()).map_err(|e| tokio::io::Error::new(tokio::io::ErrorKind::Other, e))?;
+        socket.set_nonblocking(true).map_err(|e| tokio::io::Error::new(tokio::io::ErrorKind::Other, e))?;
+
+        // Convert to std::net::UdpSocket, then to tokio::net::UdpSocket
+        let std_socket: std::net::UdpSocket = socket.into();
+        let tokio_socket = UdpSocket::from_std(std_socket)?;
 
         Ok(UdpServer {
-            socket: Arc::new(socket),
+            socket: Arc::new(tokio_socket),
             threads,
             tracker,
         })

src/udp/udp.rs

@@ -11,9 +11,9 @@ pub const PROTOCOL_IDENTIFIER: i64 = 4_497_486_125_440;
 pub const MAX_SCRAPE_TORRENTS: u8 = 74;
 pub const MAX_PACKET_SIZE: usize = 1496;
 
-pub async fn udp_service(addr: SocketAddr, threads: u64, data: Arc<TorrentTracker>, rx: tokio::sync::watch::Receiver<bool>, tokio_udp: Arc<Runtime>) -> JoinHandle<()>
+pub async fn udp_service(addr: SocketAddr, threads: u64, recv_buffer_size: usize, send_buffer_size: usize, reuse_address: bool, data: Arc<TorrentTracker>, rx: tokio::sync::watch::Receiver<bool>, tokio_udp: Arc<Runtime>) -> JoinHandle<()>
 {
-    let udp_server = UdpServer::new(data, addr, threads).await.unwrap_or_else(|e| {
+    let udp_server = UdpServer::new(data, addr, threads, recv_buffer_size, send_buffer_size, reuse_address).await.unwrap_or_else(|e| {
         error!("Could not listen to the UDP port: {e}");
         exit(1);
     });