From 919d2c3383b38e30b7454f49a8eaf2bd77c963b6 Mon Sep 17 00:00:00 2001
From: Vincent Picking <v.picking@gmx.de>
Date: Mon, 26 Feb 2024 15:40:09 +0100
Subject: [PATCH 1/5] Add single-connection parameter; Update README

---
 README.md             | 22 ++++++++++++++--------
 src/command.rs        |  7 ++++++-
 src/util/statistic.rs |  4 +++-
 3 files changed, 23 insertions(+), 10 deletions(-)

diff --git a/README.md b/README.md
index 2a57aca..bbfbe25 100644
--- a/README.md
+++ b/README.md
@@ -41,7 +41,7 @@ A network performance measurement tool
 * `-t`, `--time <TIME>` — Time to run the test
 
   Default value: `10`
-* `--with-gso` — Enable GSO on sending socket
+* `--with-gsro` — Enable GSO/GRO on socket
 
   Default value: `false`
 
@@ -53,12 +53,6 @@ A network performance measurement tool
 * `--with-mss <WITH_MSS>` — Set transmit buffer size. Gets overwritten by GSO/GRO buffer size if GSO/GRO is enabled
 
   Default value: `1472`
-* `--with-gro` — Enable GRO on receiving socket
-
-  Default value: `false`
-
-  Possible values: `true`, `false`
-
 * `--with-ip-frag` — Disable fragmentation on sending socket
 
   Default value: `false`
@@ -80,7 +74,13 @@ A network performance measurement tool
 * `--with-mmsg-amount <WITH_MMSG_AMOUNT>` — Amount of message packs of gso_buffers to send when using sendmmsg
 
   Default value: `1024`
-* `--without-non-blocking` — Enable non-blocking socket
+* `--with-socket-buffer` — Enable setting udp socket buffer size
+
+  Default value: `false`
+
+  Possible values: `true`, `false`
+
+* `--without-non-blocking` — Disable non-blocking socket
 
   Default value: `false`
 
@@ -95,6 +95,12 @@ A network performance measurement tool
 
   Possible values: `true`, `false`
 
+* `--single-connection` — Only one socket descriptor is used for all threads
+
+  Default value: `false`
+
+  Possible values: `true`, `false`
+
 * `--markdown-help`
 
   Possible values: `true`, `false`
diff --git a/src/command.rs b/src/command.rs
index bc12cdf..fe30dd9 100644
--- a/src/command.rs
+++ b/src/command.rs
@@ -83,6 +83,10 @@ pub struct nPerf {
     #[arg(long, default_value_t = false)]
     json: bool,
 
+    /// Only one socket descriptor is used for all threads
+    #[arg(long, default_value_t = false)]
+    single_connection: bool,
+
     #[arg(long, hide = true)]
     markdown_help: bool,
 }
@@ -226,7 +230,8 @@ impl nPerf {
             self.datagram_size, 
             packet_buffer_size, 
             socket_options, 
-            exchange_function
+            exchange_function,
+            self.single_connection
         ))
     }
 
diff --git a/src/util/statistic.rs b/src/util/statistic.rs
index cd96c4a..6eb3fe5 100644
--- a/src/util/statistic.rs
+++ b/src/util/statistic.rs
@@ -182,11 +182,12 @@ pub struct Parameter {
     pub packet_buffer_size: usize,
     pub socket_options: SocketOptions,
     pub exchange_function: super::ExchangeFunction,
+    pub single_connection: bool,
 }
 
 impl Parameter {
     #[allow(clippy::too_many_arguments)]
-    pub fn new(mode: super::NPerfMode, ip: std::net::Ipv4Addr, amount_threads: u16, amount_ports: u16, output_format: OutputFormat, io_model: super::IOModel, test_runtime_length: u64, mss: u32, datagram_size: u32, packet_buffer_size: usize, socket_options: SocketOptions, exchange_function: super::ExchangeFunction) -> Parameter {
+    pub fn new(mode: super::NPerfMode, ip: std::net::Ipv4Addr, amount_threads: u16, amount_ports: u16, output_format: OutputFormat, io_model: super::IOModel, test_runtime_length: u64, mss: u32, datagram_size: u32, packet_buffer_size: usize, socket_options: SocketOptions, exchange_function: super::ExchangeFunction, single_connection: bool) -> Parameter {
         Parameter {
             mode,
             ip,
@@ -200,6 +201,7 @@ impl Parameter {
             packet_buffer_size,
             socket_options,
             exchange_function,
+            single_connection
         }
     }
 }

From 111bcfd56d3555196c60f052e63d9308a283c9f1 Mon Sep 17 00:00:00 2001
From: Vincent Picking <v.picking@gmx.de>
Date: Mon, 26 Feb 2024 16:49:24 +0100
Subject: [PATCH 2/5] Change parameter name to single_socket; Add first version
 without any packet number checking or sync between the threads

---
 src/command.rs        | 35 +++++++++++++++++++++++++++--------
 src/net/socket.rs     |  2 +-
 src/node/client.rs    |  9 ++++++---
 src/node/server.rs    | 20 +++++++++++++++-----
 src/util/statistic.rs |  6 +++---
 5 files changed, 52 insertions(+), 20 deletions(-)

diff --git a/src/command.rs b/src/command.rs
index fe30dd9..b275ce3 100644
--- a/src/command.rs
+++ b/src/command.rs
@@ -3,7 +3,7 @@ use log::{debug, error, info};
 
 use std::{sync::mpsc::{self, Sender}, thread};
 
-use crate::{node::{client::Client, server::Server, Node}, util::NPerfMode};
+use crate::{net::socket::Socket, node::{client::Client, server::Server, Node}, util::NPerfMode};
 use crate::net::{self, socket_options::SocketOptions};
 use crate::util::{self, statistic::Statistic, ExchangeFunction};
 
@@ -85,7 +85,7 @@ pub struct nPerf {
 
     /// Only one socket descriptor is used for all threads
     #[arg(long, default_value_t = false)]
-    single_connection: bool,
+    single_socket: bool,
 
     #[arg(long, hide = true)]
     markdown_help: bool,
@@ -116,20 +116,39 @@ impl nPerf {
             let mut fetch_handle: Vec<thread::JoinHandle<()>> = Vec::new();
             let (tx, rx) = mpsc::channel();
     
+            // 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 {
+                    socket.connect().expect("Error connecting to remote host");
+                } else {
+                    socket.bind().expect("Error binding to local port");
+                }
+                Some(socket)
+            } else { 
+                None 
+            };
+
             for i in 0..self.parallel {
                 let tx: Sender<_> = tx.clone();
-                let port = if self.port != 45001 {
-                    info!("Port is set to different port than 45001. Incrementing port number is disabled.");
+                let port = if self.port != 45001 || self.single_socket {
+                    info!("Port is set to different port than 45001 or single_socket mode is enabled. Incrementing port number is disabled.");
                     self.port
                 } else {
                     self.port + i
                 };
-                
+
+                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, parameter))
+                        Box::new(Client::new(i as u64, parameter.ip, port, socket, parameter))
                     } else {
-                        Box::new(Server::new(parameter.ip, port, parameter))
+                        Box::new(Server::new(parameter.ip, port, socket, parameter))
                     };
     
                     match node.run(parameter.io_model) {
@@ -231,7 +250,7 @@ impl nPerf {
             packet_buffer_size, 
             socket_options, 
             exchange_function,
-            self.single_connection
+            self.single_socket
         ))
     }
 
diff --git a/src/net/socket.rs b/src/net/socket.rs
index 651967f..d7853d7 100644
--- a/src/net/socket.rs
+++ b/src/net/socket.rs
@@ -4,7 +4,7 @@ use std::{self, net::Ipv4Addr, io::Error};
 
 use super::socket_options::{self, SocketOptions};
 
-#[derive(Debug)]
+#[derive(Debug, Copy, Clone)]
 pub struct Socket {
     ip: Ipv4Addr,
     port: u16,
diff --git a/src/node/client.rs b/src/node/client.rs
index 6946a58..b65b6ea 100644
--- a/src/node/client.rs
+++ b/src/node/client.rs
@@ -16,9 +16,9 @@ pub struct Client {
 }
 
 impl Client {
-    pub fn new(test_id: u64, ip: Ipv4Addr, remote_port: u16, parameter: Parameter) -> Self {
+    pub fn new(test_id: u64, ip: Ipv4Addr, remote_port: u16, socket: Option<Socket>, parameter: Parameter) -> Self {
         info!("Current mode 'client' sending to remote host {}:{} with test ID {}", ip, remote_port, test_id);
-        let socket = Socket::new(ip, remote_port, parameter.socket_options).expect("Error creating socket");
+        let socket = socket.unwrap_or(Socket::new(ip, remote_port, parameter.socket_options).expect("Error creating socket"));
         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 {
@@ -141,7 +141,10 @@ impl Node for Client {
     fn run(&mut self, io_model: IOModel) -> Result<Statistic, &'static str> {
         self.fill_packet_buffers_with_repeating_pattern(); 
         self.add_message_headers();
-        self.socket.connect().expect("Error connecting to remote host"); 
+
+        if !self.statistic.parameter.single_socket {
+            self.socket.connect().expect("Error connecting to remote host"); 
+        }
 
         if let Ok(mss) = self.socket.get_mss() {
             info!("On the current socket the MSS is {}", mss);
diff --git a/src/node/server.rs b/src/node/server.rs
index 6c4750e..ebcd7de 100644
--- a/src/node/server.rs
+++ b/src/node/server.rs
@@ -16,9 +16,9 @@ pub struct Server {
 }
 
 impl Server {
-    pub fn new(ip: Ipv4Addr, local_port: u16, parameter: Parameter) -> Server {
+    pub fn new(ip: Ipv4Addr, local_port: u16, socket: Option<Socket>, parameter: Parameter) -> Server {
         info!("Current mode 'server' listening on {}:{}", ip, local_port);
-        let socket = Socket::new(ip, local_port, parameter.socket_options).expect("Error creating socket");
+        let socket = socket.unwrap_or(Socket::new(ip, local_port, parameter.socket_options).expect("Error creating socket"));
         let packet_buffer = Vec::from_iter((0..parameter.packet_buffer_size).map(|_| PacketBuffer::new(parameter.mss, parameter.datagram_size).expect("Error creating packet buffer")));
 
         Server {
@@ -133,7 +133,12 @@ impl Server {
                 Err("INIT_MESSAGE_RECEIVED")
             },
             MessageType::MEASUREMENT => { 
-                let measurement = self.measurements.get_mut(&test_id).expect("Error getting statistic in last message: test id not found");
+                let measurement = if let Some(x) = self.measurements.get_mut(&test_id) {
+                    x
+                } else {
+                    self.measurements.insert(test_id, Measurement::new(self.parameter));
+                    self.measurements.get_mut(&test_id).expect("Error getting statistic in measurement message: test id not found")
+                };
                 if !measurement.first_packet_received {
                     info!("First packet received from test {}!", test_id);
                     measurement.start_time = Instant::now();
@@ -157,7 +162,10 @@ impl Server {
 impl Node for Server { 
     fn run(&mut self, io_model: IOModel) -> Result<Statistic, &'static str>{
         info!("Current mode: server");
-        self.socket.bind().expect("Error binding socket");
+
+        if !self.parameter.single_socket {
+            self.socket.bind().expect("Error binding to local port");
+        }
 
         info!("Start server loop...");
         let mut read_fds: libc::fd_set = unsafe { self.socket.create_fdset() };
@@ -197,7 +205,9 @@ impl Node for Server {
             statistic.amount_syscalls += 1;
         }
 
-        self.socket.close()?;
+        if !self.parameter.single_socket {
+            self.socket.close()?;
+        }
 
         debug!("Finished receiving data from remote host");
         // Fold over all statistics, and calculate the final statistic
diff --git a/src/util/statistic.rs b/src/util/statistic.rs
index 6eb3fe5..0bb80ca 100644
--- a/src/util/statistic.rs
+++ b/src/util/statistic.rs
@@ -182,12 +182,12 @@ pub struct Parameter {
     pub packet_buffer_size: usize,
     pub socket_options: SocketOptions,
     pub exchange_function: super::ExchangeFunction,
-    pub single_connection: bool,
+    pub single_socket: bool,
 }
 
 impl Parameter {
     #[allow(clippy::too_many_arguments)]
-    pub fn new(mode: super::NPerfMode, ip: std::net::Ipv4Addr, amount_threads: u16, amount_ports: u16, output_format: OutputFormat, io_model: super::IOModel, test_runtime_length: u64, mss: u32, datagram_size: u32, packet_buffer_size: usize, socket_options: SocketOptions, exchange_function: super::ExchangeFunction, single_connection: bool) -> Parameter {
+    pub fn new(mode: super::NPerfMode, ip: std::net::Ipv4Addr, amount_threads: u16, amount_ports: u16, output_format: OutputFormat, io_model: super::IOModel, test_runtime_length: u64, mss: u32, datagram_size: u32, packet_buffer_size: usize, socket_options: SocketOptions, exchange_function: super::ExchangeFunction, single_socket: bool) -> Parameter {
         Parameter {
             mode,
             ip,
@@ -201,7 +201,7 @@ impl Parameter {
             packet_buffer_size,
             socket_options,
             exchange_function,
-            single_connection
+            single_socket
         }
     }
 }

From 6ee5053defdc28b0e09714fbec80009bbb2843d8 Mon Sep 17 00:00:00 2001
From: PickingUpPieces <v.picking@gmx.de>
Date: Tue, 27 Feb 2024 10:29:08 +0100
Subject: [PATCH 3/5] Catch EAGAIN when sending buffer is full; Add workaround
 for sendmmsg bug always returning 1;

---
 benchmarks/benchmark.py                       |  5 +--
 .../configs/send_methods_with_threads.json    |  3 +-
 src/command.rs                                |  5 +++
 src/net/socket.rs                             | 33 ++++++++++++++-----
 src/node/client.rs                            | 10 ++++++
 5 files changed, 44 insertions(+), 12 deletions(-)

diff --git a/benchmarks/benchmark.py b/benchmarks/benchmark.py
index f811b57..87a3b86 100644
--- a/benchmarks/benchmark.py
+++ b/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']
+    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']
     file_exists = os.path.isfile(csv_file_path)
 
     with open(csv_file_path, 'a', newline='') as csvfile:
@@ -162,7 +162,8 @@ def write_results_to_csv(test_results, test_name, csv_file_path):
                 'nonblocking': server_result['parameter']['socket_options']['nonblocking'],
                 'ip_fragmentation': client_result['parameter']['socket_options']['ip_fragmentation'],
                 'gso': client_result['parameter']['socket_options']['gso'],
-                'gro': server_result['parameter']['socket_options']['gro']
+                'gro': server_result['parameter']['socket_options']['gro'],
+                'single-socket': server_result['parameter']['single_socket']
             }
             writer.writerow(row)
 
diff --git a/benchmarks/configs/send_methods_with_threads.json b/benchmarks/configs/send_methods_with_threads.json
index 6dc4e2e..d535532 100644
--- a/benchmarks/configs/send_methods_with_threads.json
+++ b/benchmarks/configs/send_methods_with_threads.json
@@ -1,6 +1,6 @@
 {
     "parameters": {
-        "repetitions": 7,
+        "repetitions": 3,
         "ip": "0.0.0.0",
         "port": 45001,
         "datagram-size": 1472,
@@ -13,6 +13,7 @@
         "with-mmsg-amount": 20,
         "with-ip-frag": false,
         "without-non-blocking": false,
+	    "single-socket": true,
         "io-model": "select"
     },
     "sendmmsg": {
diff --git a/src/command.rs b/src/command.rs
index b275ce3..45a60d2 100644
--- a/src/command.rs
+++ b/src/command.rs
@@ -173,12 +173,17 @@ impl nPerf {
                 handle.join().unwrap(); 
                 stat 
             });
+
             info!("All threads finished!");
+            if let Some(socket) = socket {
+                socket.close().expect("Error closing socket");
+            }
     
             if statistic.amount_datagrams != 0 {
                 statistic.calculate_statistics();
                 statistic.print(parameter.output_format);
             }
+
             if !(self.run_infinite && parameter.mode == util::NPerfMode::Server) {
                 return Some(statistic);
             }
diff --git a/src/net/socket.rs b/src/net/socket.rs
index d7853d7..87e5d82 100644
--- a/src/net/socket.rs
+++ b/src/net/socket.rs
@@ -1,5 +1,5 @@
 
-use log::{info, trace, debug, error};
+use log::{debug, error, info, trace, warn};
 use std::{self, net::Ipv4Addr, io::Error};
 
 use super::socket_options::{self, SocketOptions};
@@ -8,7 +8,8 @@ use super::socket_options::{self, SocketOptions};
 pub struct Socket {
     ip: Ipv4Addr,
     port: u16,
-    socket: i32
+    socket: i32,
+    sendmmsg_econnrefused_counter: u16
 } 
 
 impl Socket {
@@ -20,7 +21,8 @@ impl Socket {
         Some(Socket {
             ip,
             port,
-            socket
+            socket,
+            sendmmsg_econnrefused_counter: 0
         })
     }
 
@@ -89,10 +91,9 @@ impl Socket {
         if buffer_len == 0 {
             error!("Buffer is empty");
             return Err("Buffer is empty");
-        } else {
-            debug!("Sending on socket {} with buffer size: {}", self.socket, buffer_len);
-            trace!("Buffer: {:?}", buffer)
         }
+        debug!("Sending on socket {} with buffer size: {}", self.socket, buffer_len);
+        trace!("Buffer: {:?}", buffer);
     
         let send_result = unsafe {
             libc::send(
@@ -145,6 +146,10 @@ impl Socket {
                     error!("Connection refused while trying to send data!");
                     return Err("ECONNREFUSED");
                 },
+                Some(libc::EAGAIN) => {
+                    warn!("Error EGAIN: Probably socket buffer is full!");
+                    return Err("EAGAIN");
+                },
                 _ => {
                     error!("Errno when trying to send data with sendmsg(): {}", errno);
                     return Err("Failed to send data");
@@ -156,7 +161,7 @@ impl Socket {
         Ok(send_result as usize)
     }
 
-    pub fn sendmmsg(&self, mmsgvec: &mut [libc::mmsghdr]) -> Result<usize, &'static str> {
+    pub fn sendmmsg(&mut self, mmsgvec: &mut [libc::mmsghdr]) -> Result<usize, &'static str> {
         let send_result: i32 = unsafe {
             libc::sendmmsg(
                 self.socket,
@@ -173,14 +178,24 @@ impl Socket {
                     error!("Connection refused while trying to send data!");
                     return Err("ECONNREFUSED");
                 },
+                Some(libc::EAGAIN) => {
+                    warn!("Error EGAIN: Probably socket buffer is full!");
+                    return Err("EAGAIN");
+                },
                 _ => {
                     error!("Errno when trying to send data with sendmmsg(): {}", errno);
                     return Err("Failed to send data");
                 }
             }
+        // sendmmsg() always returns 1, even when it should return ECONNREFUSED (when the server isn't up yet, similar to send()/sendmsg()). This is a workaround to detect ECONNREFUSED.
         } else if send_result == 1 && mmsgvec.len() > 1 {
-            error!("sendmmsg() returned 1, but mmsgvec.len() > 1. This probably means that the first message was sent successfully, but the second one failed. We assume that the server is not running.");
-            return Err("ECONNREFUSED");
+            if self.sendmmsg_econnrefused_counter > 5 {
+                error!("sendmmsg() returned 1, but mmsgvec.len() > 1. This probably means that the first message was sent successfully, but the second one failed. We assume that the server is not running.");
+                return Err("ECONNREFUSED");
+            } 
+            self.sendmmsg_econnrefused_counter += 1;
+        } else {
+            self.sendmmsg_econnrefused_counter = 0;
         }
     
         debug!("Sent {} mmsghdr(s)", send_result);
diff --git a/src/node/client.rs b/src/node/client.rs
index b65b6ea..28319d3 100644
--- a/src/node/client.rs
+++ b/src/node/client.rs
@@ -80,6 +80,11 @@ impl Client {
                 Ok(())
             },
             Err("ECONNREFUSED") => Err("Start the server first! Abort measurement..."),
+            Err("EAGAIN") => {
+                // Reset next_packet_id to the last packet_id that was sent
+                self.next_packet_id -= amount_datagrams;
+                Ok(())
+            },
             Err(x) => Err(x) 
         }
     }
@@ -104,6 +109,11 @@ impl Client {
                 Ok(())
             },
             Err("ECONNREFUSED") => Err("Start the server first! Abort measurement..."),
+            Err("EAGAIN") => {
+                // Reset next_packet_id to the last packet_id that was sent
+                self.next_packet_id -= amount_datagrams;
+                Ok(())
+            },
             Err(x) => Err(x)
         }
     }

From e3ad6ab5bdd7fdecaf20c06d8007dc5471b6805f Mon Sep 17 00:00:00 2001
From: PickingUpPieces <v.picking@gmx.de>
Date: Tue, 27 Feb 2024 11:13:37 +0100
Subject: [PATCH 4/5] Increase timeout to minimum of 120s

---
 src/command.rs | 14 ++++++++++----
 1 file changed, 10 insertions(+), 4 deletions(-)

diff --git a/src/command.rs b/src/command.rs
index bc12cdf..4cfe2a0 100644
--- a/src/command.rs
+++ b/src/command.rs
@@ -1,7 +1,7 @@
 use clap::Parser;
 use log::{debug, error, info};
 
-use std::{sync::mpsc::{self, Sender}, thread};
+use std::{cmp::max, sync::mpsc::{self, Sender}, thread};
 
 use crate::{node::{client::Client, server::Server, Node}, util::NPerfMode};
 use crate::net::{self, socket_options::SocketOptions};
@@ -143,10 +143,16 @@ impl nPerf {
     
             info!("Waiting for all threads to finish...");
             let mut statistic = fetch_handle.into_iter().fold(Statistic::new(parameter), |acc: Statistic, handle| { 
-                let stat = acc + match rx.recv_timeout(std::time::Duration::from_secs(parameter.test_runtime_length * 2)).expect("Timeout") {
-                    Some(x) => x,
-                    None => Statistic::new(parameter)
+                let stat = acc + match rx.recv_timeout(std::time::Duration::from_secs(max(parameter.test_runtime_length * 2, 120))) {
+                    Ok(x) => {
+                        match x {
+                        Some(x) => x,
+                        None => Statistic::new(parameter)
+                        }
+                    },
+                    Err(_) => Statistic::new(parameter)
                 };
+                    
                 handle.join().unwrap(); 
                 stat 
             });

From a877efcbbf7ce69399ed222f9057fcd8f8077239 Mon Sep 17 00:00:00 2001
From: Vincent Picking <v.picking@gmx.de>
Date: Fri, 1 Mar 2024 15:31:18 +0100
Subject: [PATCH 5/5] change unwrap_or to unwrap_or_else since lazily
 evaluated; Add test for single_socket;

---
 .vscode/launch.json           |  8 ++++----
 src/net/socket.rs             |  4 ++++
 src/node/client.rs            |  9 +++++++--
 src/node/server.rs            |  6 ++----
 tests/client_tests.rs         |  1 -
 tests/multithreading_tests.rs | 13 +++++++++++++
 6 files changed, 30 insertions(+), 11 deletions(-)

diff --git a/.vscode/launch.json b/.vscode/launch.json
index 7eaed18..7a4536f 100644
--- a/.vscode/launch.json
+++ b/.vscode/launch.json
@@ -22,7 +22,7 @@
             "env": {
                 "RUST_LOG": "TRACE",
             },
-            "args": ["client", "--with-mmsg", "--with-gso"],
+            "args": ["client", "--with-msg", "--single-socket", "--parallel", "2"],
             "cwd": "${workspaceFolder}",
             "preLaunchTask": "delay"
         },
@@ -44,7 +44,7 @@
             "env": {
                 "RUST_LOG": "TRACE",
             },
-            "args": ["server", "--with-mmsg", "--with-gro"],
+            "args": ["server", "--with-msg", "--single-socket", "--parallel", "2"],
             "cwd": "${workspaceFolder}"
         },
         {
@@ -55,7 +55,7 @@
                 "RUST_LOG": "INFO",
             },
             "program": "${workspaceFolder}/target/debug/nperf",
-            "args": ["client", "--with-msg", "--with-gso"],
+            "args": ["client", "--with-msg", "--single-socket", "--parallel", "2"],
             "cwd": "${workspaceFolder}",
             "stopOnEntry": false,
             "preLaunchTask": "delay"
@@ -68,7 +68,7 @@
                 "RUST_LOG": "INFO",
             },
             "program": "${workspaceFolder}/target/debug/nperf",
-            "args": ["server", "--with-msg", "--with-gro"],
+            "args": ["server", "--with-msg", "--single-socket", "--parallel", "2"],
             "cwd": "${workspaceFolder}",
             "stopOnEntry": false,
         },
diff --git a/src/net/socket.rs b/src/net/socket.rs
index 87e5d82..f2a2f4b 100644
--- a/src/net/socket.rs
+++ b/src/net/socket.rs
@@ -292,6 +292,10 @@ impl Socket {
         socket_options::get_mss(self.socket)
     }
 
+    pub fn get_socket_id(&self) -> i32 {
+        self.socket
+    }
+
     fn create_sockaddr(address: Ipv4Addr, port: u16) -> libc::sockaddr_in {
         let addr_u32: u32 = address.into(); 
     
diff --git a/src/node/client.rs b/src/node/client.rs
index 28319d3..67c328f 100644
--- a/src/node/client.rs
+++ b/src/node/client.rs
@@ -17,8 +17,8 @@ pub struct Client {
 
 impl Client {
     pub fn new(test_id: u64, ip: Ipv4Addr, remote_port: u16, socket: Option<Socket>, parameter: Parameter) -> Self {
-        info!("Current mode 'client' sending to remote host {}:{} with test ID {}", ip, remote_port, test_id);
-        let socket = socket.unwrap_or(Socket::new(ip, remote_port, parameter.socket_options).expect("Error creating socket"));
+        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());
         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 {
@@ -62,6 +62,11 @@ impl Client {
                 trace!("Sent datagram to remote host");
                 Ok(())
             },
+            Err("EAGAIN") => {
+                // Reset next_packet_id to the last packet_id that was sent
+                self.next_packet_id -= amount_datagrams;
+                Ok(())
+            },
             Err("ECONNREFUSED") => Err("Start the server first! Abort measurement..."),
             Err(x) => Err(x) 
         }
diff --git a/src/node/server.rs b/src/node/server.rs
index ebcd7de..fcff742 100644
--- a/src/node/server.rs
+++ b/src/node/server.rs
@@ -17,8 +17,8 @@ pub struct Server {
 
 impl Server {
     pub fn new(ip: Ipv4Addr, local_port: u16, socket: Option<Socket>, parameter: Parameter) -> Server {
-        info!("Current mode 'server' listening on {}:{}", ip, local_port);
-        let socket = socket.unwrap_or(Socket::new(ip, local_port, parameter.socket_options).expect("Error creating socket"));
+        let socket = socket.unwrap_or_else(|| Socket::new(ip, local_port, 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")));
 
         Server {
@@ -161,8 +161,6 @@ impl Server {
 
 impl Node for Server { 
     fn run(&mut self, io_model: IOModel) -> Result<Statistic, &'static str>{
-        info!("Current mode: server");
-
         if !self.parameter.single_socket {
             self.socket.bind().expect("Error binding to local port");
         }
diff --git a/tests/client_tests.rs b/tests/client_tests.rs
index 03a9cec..b85fc3c 100644
--- a/tests/client_tests.rs
+++ b/tests/client_tests.rs
@@ -19,7 +19,6 @@ fn test_client_sendmsg() -> Result<(), Box<dyn std::error::Error>>{
 
     let args = vec!["client", "--with-msg", "--port", "45002"];
     if let Some(x) = nperf::nPerf::new().set_args(args).exec() {
-        println!("{:?}", x);
         assert!(x.amount_datagrams > 10000);
     };
 
diff --git a/tests/multithreading_tests.rs b/tests/multithreading_tests.rs
index 9ae38ab..c9f8af1 100644
--- a/tests/multithreading_tests.rs
+++ b/tests/multithreading_tests.rs
@@ -25,3 +25,16 @@ 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(())
+}
\ No newline at end of file