Merge pull request #20 from PickingUpPieces/feature/add-reuseport-sender

Feature/add reuseport sender

Author: PickingUpPieces

benchmarks/benchmark.py

@@ -122,7 +122,7 @@ def run_test(run_config):
 
 def write_results_to_csv(test_results, test_name, csv_file_path):
     # FIXME: If new measurement parameters are added, the header should be updated
-    header = ['test_name', 'run_number', 'run_name', 'amount_threads_client', 'amount_threads_server', 'amount_used_ports_server', 'test_runtime_length', 'datagram_size', 'packet_buffer_size', 'exchange_function', 'io_model', 'total_data_gbyte', 'amount_datagrams', 'amount_data_bytes', 'amount_reordered_datagrams', 'amount_duplicated_datagrams', 'amount_omitted_datagrams', 'amount_syscalls', 'amount_io_model_syscalls', 'data_rate_gbit', 'packet_loss', 'nonblocking', 'ip_fragmentation', 'gso', 'gro', 'single-socket', 'receive_buffer_size', 'send_buffer_size']
+    header = ['test_name', 'run_number', 'run_name', 'amount_threads_client', 'amount_threads_server', 'amount_used_ports_server', 'test_runtime_length', 'datagram_size', 'packet_buffer_size', 'exchange_function', 'io_model', 'total_data_gbyte', 'amount_datagrams', 'amount_data_bytes', 'amount_reordered_datagrams', 'amount_duplicated_datagrams', 'amount_omitted_datagrams', 'amount_syscalls', 'amount_io_model_syscalls', 'data_rate_gbit', 'packet_loss', 'nonblocking', 'ip_fragmentation', 'reuseport', 'gso', 'gro', 'single-socket', 'receive_buffer_size', 'send_buffer_size']
     file_exists = os.path.isfile(csv_file_path)
 
     with open(csv_file_path, 'a', newline='') as csvfile:
@@ -161,6 +161,7 @@ def write_results_to_csv(test_results, test_name, csv_file_path):
                 'packet_loss': server_result['packet_loss'],
                 'nonblocking': server_result['parameter']['socket_options']['nonblocking'],
                 'ip_fragmentation': client_result['parameter']['socket_options']['ip_fragmentation'],
+                'reuseport': server_result['parameter']['socket_options']['reuseport'],
                 'gso': client_result['parameter']['socket_options']['gso'],
                 'gro': server_result['parameter']['socket_options']['gro'],
                 'single-socket': server_result['parameter']['single_socket'],

benchmarks/configs/benchmark_test_config.json

@@ -17,6 +17,7 @@
                 "with-mmsg": false,
                 "with-mmsg-amount": 1024,
                 "without-non-blocking": false,
+                "with-reuseport": false,
                 "io-model": "select"
             },
             "server": {
@@ -50,6 +51,7 @@
                 "with-mmsg": false,
                 "with-mmsg-amount": 1024,
                 "without-non-blocking": false,
+                "with-reuseport": false,
                 "io-model": "select"
             },
             "server": {
@@ -85,6 +87,7 @@
                 "with-mmsg": false,
                 "with-mmsg-amount": 1024,
                 "without-non-blocking": false,
+                "with-reuseport": false,
                 "io-model": "select"
             },
             "server": {
@@ -118,6 +121,7 @@
                 "with-mmsg": false,
                 "with-mmsg-amount": 1024,
                 "without-non-blocking": false,
+                "with-reuseport": false,
                 "io-model": "select"
             },
             "server": {

benchmarks/configs/send_methods_with_threads-max-mss.json

@@ -13,6 +13,7 @@
         "with-mmsg-amount": 20,
         "with-ip-frag": false,
         "without-non-blocking": false,
+        "with-reuseport": false,
         "io-model": "select"
     },
     "sendmmsg": {

benchmarks/configs/send_methods_with_threads.json

@@ -13,7 +13,8 @@
         "with-mmsg-amount": 20,
         "with-ip-frag": false,
         "without-non-blocking": false,
-	    "single-socket": true,
+        "with-reuseport": true,
+	    "single-socket": false,
         "io-model": "select"
     },
     "sendmmsg": {

benchmarks/configs/sendmmsg_mmsg-vec_with_threads.json

@@ -12,6 +12,7 @@
         "with-gsro": true,
         "with-ip-frag": false,
         "without-non-blocking": false,
+        "with-reuseport": false,
         "io-model": "select"
     },
     "|mmsg-vec| 1": {

src/command.rs

@@ -3,7 +3,7 @@ use log::{debug, error, info};
 
 use std::{cmp::max, sync::mpsc::{self, Sender}, thread};
 
-use crate::{net::socket::Socket, node::{client::Client, server::Server, Node}, util::NPerfMode};
+use crate::{net::socket::Socket, node::{client::Client, server::Server, Node}, util::NPerfMode, DEFAULT_CLIENT_PORT};
 use crate::net::{self, socket_options::SocketOptions};
 use crate::util::{self, statistic::Statistic, ExchangeFunction};
 
@@ -20,7 +20,7 @@ pub struct nPerf {
     ip: String,
 
     /// Port number to measure against/listen on. If port is defined with parallel mode, all client threads will measure against the same port. 
-    #[arg(short, long, default_value_t = crate::DEFAULT_PORT)]
+    #[arg(short, long, default_value_t = crate::DEFAULT_SERVER_PORT)]
     port: u16,
 
     /// Start multiple client/server threads in parallel. The port number will be incremented automatically.
@@ -87,6 +87,11 @@ pub struct nPerf {
     #[arg(long, default_value_t = false)]
     single_socket: bool,
 
+    // Enable reuseport option on socket. For parallel mode same port number for sending is used for all threads.
+    #[arg(long, default_value_t = false)]
+    with_reuseport: bool,
+
+    /// Show help in markdown format
     #[arg(long, hide = true)]
     markdown_help: bool,
 }
@@ -105,7 +110,7 @@ impl nPerf {
             None => { error!("Error running app"); return None; },
         };
 
-        match Self::parameter_check(&parameter) {
+        match self.parameter_check(&parameter) {
             false => { error!("Invalid parameter!"); return None; },
             true => {}
         }
@@ -118,17 +123,21 @@ impl nPerf {
 
             // If single-connection, creating the socket and bind to port/connect must happen before the threads are spawned
             let socket = if self.single_socket {
-                let socket = Socket::new(parameter.ip, self.port, parameter.socket_options).expect("Error creating socket");
                 if parameter.mode == util::NPerfMode::Client {
+                    let socket = Socket::new(parameter.ip, None, Some(self.port), parameter.socket_options).expect("Error creating socket");
                     socket.connect().expect("Error connecting to remote host");
+                    Some(socket)
                 } else {
+                    let socket = Socket::new(parameter.ip, Some(self.port), None, parameter.socket_options).expect("Error creating socket");
                     socket.bind().expect("Error binding to local port");
+                    Some(socket)
                 }
-                Some(socket)
             } else { 
                 None 
             };
 
+            let local_port_client: Option<u16> = if self.with_reuseport { Some(DEFAULT_CLIENT_PORT) } else { None };
+
             for i in 0..self.parallel {
                 let tx: Sender<_> = tx.clone();
                 let port = if self.port != 45001 || self.single_socket {
@@ -138,15 +147,12 @@ impl nPerf {
                     self.port + i
                 };
 
-                let i = if self.single_socket {
-                    0
-                } else {
-                    i
-                };
+                // Use same test id for all threads
+                let i = if self.single_socket { 0 } else { i };
 
                 fetch_handle.push(thread::spawn(move || {
                     let mut node:Box<dyn Node> = if parameter.mode == util::NPerfMode::Client {
-                        Box::new(Client::new(i as u64, parameter.ip, port, socket, parameter))
+                        Box::new(Client::new(i as u64, parameter.ip, local_port_client, port, socket, parameter))
                     } else {
                         Box::new(Server::new(parameter.ip, port, socket, parameter))
                     };
@@ -265,12 +271,21 @@ impl nPerf {
         ))
     }
 
-    fn parameter_check(parameter: &util::statistic::Parameter)-> bool {
+    fn parameter_check(&self, parameter: &util::statistic::Parameter)-> bool {
         if parameter.datagram_size > crate::MAX_UDP_DATAGRAM_SIZE {
             error!("UDP datagram size is too big! Maximum is {}", crate::MAX_UDP_DATAGRAM_SIZE);
             return false;
         }
 
+        if self.with_reuseport && self.single_socket {
+            error!("Reuseport and single socket option can't be used simultaneously!");
+            return false;
+        }
+
+        if self.with_reuseport && parameter.mode == util::NPerfMode::Server {
+            error!("Reuseport option is enabled, but it is only available for client mode!");
+        }
+
         true
     }
 
@@ -294,10 +309,11 @@ impl nPerf {
         SocketOptions::new(
             !self.without_non_blocking, 
             self.with_ip_frag, 
+            self.with_reuseport,
             gso, 
             gro, 
             recv_buffer_size, 
             send_buffer_size
         )
     }
-}
+}

src/lib.rs

@@ -14,7 +14,8 @@ const MAX_SOCKET_RECEIVE_BUFFER_SIZE: u32 = 212992 ; // 25MB; // The buffer size
 const DEFAULT_SOCKET_SEND_BUFFER_SIZE: u32 = 212992; 
 const DEFAULT_SOCKET_RECEIVE_BUFFER_SIZE: u32 = 212992; 
 const DEFAULT_DURATION: u64 = 10; // /* seconds */
-const DEFAULT_PORT: u16 = 45001;
+const DEFAULT_SERVER_PORT: u16 = 45001;
+const DEFAULT_CLIENT_PORT: u16 = 46001;
 const WAIT_CONTROL_MESSAGE: u64 = 200; // /* milliseconds */
 
 // /* Maximum datagram size UDP is (64K - 1) - IP and UDP header sizes */

src/net/socket.rs

@@ -7,20 +7,22 @@ use super::socket_options::{self, SocketOptions};
 #[derive(Debug, Copy, Clone)]
 pub struct Socket {
     ip: Ipv4Addr,
-    port: u16,
+    local_port: Option<u16>,
+    remote_port: Option<u16>,
     socket: i32,
     sendmmsg_econnrefused_counter: u16
 } 
 
 impl Socket {
-    pub fn new(ip: Ipv4Addr, port: u16, mut socket_options: SocketOptions) -> Option<Socket> {
+    pub fn new(ip: Ipv4Addr, local_port: Option<u16>, remote_port: Option<u16>, mut socket_options: SocketOptions) -> Option<Socket> {
         let socket = Self::create_socket()?; 
 
         socket_options.set_socket_options(socket).expect("Error updating socket options");
 
         Some(Socket {
             ip,
-            port,
+            local_port,
+            remote_port,
             socket,
             sendmmsg_econnrefused_counter: 0
         })
@@ -39,7 +41,11 @@ impl Socket {
 
 
     pub fn connect(&self) -> Result<(), &'static str> {
-        let sockaddr = Self::create_sockaddr(self.ip, self.port);
+        if self.local_port.is_some() {
+            self.bind()?;
+        }
+
+        let sockaddr = Self::create_sockaddr(self.ip, self.remote_port.unwrap());
 
         let connect_result = unsafe {
             libc::connect(
@@ -58,7 +64,7 @@ impl Socket {
     }
 
     pub fn bind(&self) -> Result<(), &'static str> {
-        let sockaddr = Self::create_sockaddr(self.ip, self.port);
+        let sockaddr = Self::create_sockaddr(self.ip, self.local_port.unwrap());
 
         let bind_result = unsafe {
             libc::bind(

src/net/socket_options.rs

@@ -7,17 +7,19 @@ use std::io::Error;
 pub struct SocketOptions {
     nonblocking: bool,
     ip_fragmentation: bool,
-    pub gso: Option<u32>,
-    pub gro: bool,
+    reuseport: bool,
+    gso: Option<u32>,
+    gro: bool,
     recv_buffer_size: Option<u32>,
     send_buffer_size: Option<u32>,
 }
 
 impl SocketOptions {
-    pub fn new(nonblocking: bool, ip_fragmentation: bool, gso: Option<u32>, gro: bool, recv_buffer_size: Option<u32>, send_buffer_size: Option<u32>) -> Self {
+    pub fn new(nonblocking: bool, ip_fragmentation: bool, reuseport: bool, gso: Option<u32>, gro: bool, recv_buffer_size: Option<u32>, send_buffer_size: Option<u32>) -> Self {
         SocketOptions {
             nonblocking,
             ip_fragmentation,
+            reuseport,
             gso,
             gro,
             recv_buffer_size,
@@ -36,9 +38,9 @@ impl SocketOptions {
         if let Some(size) = self.gso {
             set_gso(socket, size)?;
         }
-        if self.gro {
-            set_gro(socket)?;
-        }
+
+        set_gro(socket, self.gro)?;
+        set_reuseport(socket, self.reuseport)?;
 
         if let Some(size) = self.recv_buffer_size { 
             set_buffer_size(socket, size, libc::SO_SNDBUF)?;
@@ -145,9 +147,9 @@ fn set_gso(socket: i32, gso_size: u32) -> Result<(), &'static str> {
     set_socket_option(socket, libc::SOL_UDP, libc::UDP_SEGMENT, gso_size)
 }
 
-fn set_gro(socket: i32) -> Result<(), &'static str> {
-    let value = 1;
-    info!("Set socket option GRO to {}", value);
+fn set_gro(socket: i32, status: bool) -> Result<(), &'static str> {
+    let value: u32 = if status { 1 } else { 0 };
+    info!("Set socket option GRO to {}", status);
     set_socket_option(socket, libc::SOL_UDP, libc::UDP_GRO, value)
 }
 
@@ -167,4 +169,10 @@ pub fn get_mss(socket: i32) -> Result<u32, &'static str> {
 
 pub fn _get_gso_size(socket: i32) -> Result<u32, &'static str> {
     get_socket_option(socket, libc::SOL_UDP, libc::UDP_SEGMENT)
+}
+
+pub fn set_reuseport(socket: i32, status: bool) -> Result<(), &'static str> {
+    let value: u32 = if status { 1 } else { 0 };
+    info!("Set socket option REUSEPORT to {}", status);
+    set_socket_option(socket, libc::SOL_SOCKET, libc::SO_REUSEPORT, value)
 }

src/node/client.rs

@@ -16,9 +16,9 @@ pub struct Client {
 }
 
 impl Client {
-    pub fn new(test_id: u64, ip: Ipv4Addr, remote_port: u16, socket: Option<Socket>, parameter: Parameter) -> Self {
-        let socket: Socket = socket.unwrap_or_else(|| Socket::new(ip, remote_port, parameter.socket_options).expect("Error creating socket"));
-        info!("Current mode 'client' sending to remote host {}:{} with test ID {} on socketID {}", ip, remote_port, test_id, socket.get_socket_id());
+    pub fn new(test_id: u64, ip: Ipv4Addr, local_port: Option<u16>, remote_port: u16, socket: Option<Socket>, parameter: Parameter) -> Self {
+        let socket: Socket = socket.unwrap_or_else(|| Socket::new(ip, local_port, Some(remote_port), parameter.socket_options).expect("Error creating socket"));
+        info!("Current mode 'client' sending to remote host {}:{} from {}:{} with test ID {} on socketID {}", ip, remote_port, ip, local_port.unwrap_or_default(), test_id, socket.get_socket_id());
         let packet_buffer = Vec::from_iter((0..parameter.packet_buffer_size).map(|_| PacketBuffer::new(parameter.mss, parameter.datagram_size).expect("Error creating packet buffer")));
 
         Client {

src/node/server.rs

@@ -17,7 +17,7 @@ pub struct Server {
 
 impl Server {
     pub fn new(ip: Ipv4Addr, local_port: u16, socket: Option<Socket>, parameter: Parameter) -> Server {
-        let socket = socket.unwrap_or_else(|| Socket::new(ip, local_port, parameter.socket_options).expect("Error creating socket"));
+        let socket = socket.unwrap_or_else(|| Socket::new(ip, Some(local_port), None, parameter.socket_options).expect("Error creating socket"));
         info!("Current mode 'server' listening on {}:{} with socketID {}", ip, local_port, socket.get_socket_id());
         let packet_buffer = Vec::from_iter((0..parameter.packet_buffer_size).map(|_| PacketBuffer::new(parameter.mss, parameter.datagram_size).expect("Error creating packet buffer")));
 

tests/multithreading_tests.rs

@@ -25,16 +25,3 @@ fn multiple_clients_multiple_server() -> Result<(), Box<dyn std::error::Error>>{
     handle.join().unwrap();
     Ok(())
 }
-
-#[test]
-fn multiple_clients_multiple_server_single_socket() -> Result<(), Box<dyn std::error::Error>>{
-    let handle = common::start_nperf_server(Some(vec!["--parallel".to_string(), "2".to_string(), "--single-socket".to_string(), "--port".to_string(), "45003".to_string()]));
-
-    let args = vec!["client", "--with-msg", "--parallel", "2", "--port", "45003", "--single-socket"];
-    if let Some(x) = nperf::nPerf::new().set_args(args).exec() {
-        assert!(x.amount_datagrams > 10000);
-    };
-
-    handle.join().unwrap();
-    Ok(())
-}

tests/single_connection.rs

@@ -0,0 +1,28 @@
+
+mod common;
+
+#[test]
+fn test_client_reuseport() -> Result<(), Box<dyn std::error::Error>>{
+    let handle = common::start_nperf_server(Some(vec!["--port".to_string(), "45001".to_string(), "--parallel".to_string(), "2".to_string()]));
+
+    let args = vec!["client", "--port", "45001", "--parallel", "2", "--with-reuseport"];
+    if let Some(x) = nperf::nPerf::new().set_args(args).exec() {
+        assert!(x.amount_datagrams > 10000);
+    };
+
+    handle.join().unwrap();
+    Ok(())
+}
+
+#[test]
+fn multiple_clients_multiple_server_single_socket() -> Result<(), Box<dyn std::error::Error>>{
+    let handle = common::start_nperf_server(Some(vec!["--parallel".to_string(), "2".to_string(), "--single-socket".to_string(), "--port".to_string(), "45003".to_string()]));
+
+    let args = vec!["client", "--with-msg", "--parallel", "2", "--port", "45003", "--single-socket"];
+    if let Some(x) = nperf::nPerf::new().set_args(args).exec() {
+        assert!(x.amount_datagrams > 10000);
+    };
+
+    handle.join().unwrap();
+    Ok(())
+}