refine code

Cargo.toml

@@ -11,7 +11,6 @@ categories = ["network-utilities"]
 readme = "README.md"
 
 [dependencies]
-cfg-if = "1.0"
 chrono = "0.4"
 clap = { version = "4.4", features = ["derive", "wrap_help"] }
 core_affinity = "0.8"

src/client.rs

@@ -100,8 +100,8 @@ pub fn execute(args: &args::Args) -> BoxResult<()> {
     let mut complete = false;
 
     //config-parsing and pre-connection setup
-    let mut tcp_port_pool = tcp::receiver::TcpPortPool::new(args.tcp_port_pool.to_string(), args.tcp6_port_pool.to_string());
-    let mut udp_port_pool = udp::receiver::UdpPortPool::new(args.udp_port_pool.to_string(), args.udp6_port_pool.to_string());
+    let mut tcp_port_pool = tcp::receiver::TcpPortPool::new(&args.tcp_port_pool, &args.tcp6_port_pool);
+    let mut udp_port_pool = udp::receiver::UdpPortPool::new(&args.udp_port_pool, &args.udp6_port_pool);
 
     let cpu_affinity_manager = Arc::new(Mutex::new(crate::utils::cpu_affinity::CpuAffinityManager::new(&args.affinity)?));
 

src/protocol/communication.rs

@@ -42,10 +42,10 @@ pub fn send(stream: &mut TcpStream, message: &serde_json::Value) -> BoxResult<()
     let serialised_message = serde_json::to_vec(message)?;
 
     log::debug!(
-        "sending message of length {}, {:?}, to {}...",
+        "sending message to {}, length {}, {:?}...",
+        stream.peer_addr()?,
         serialised_message.len(),
         message,
-        stream.peer_addr()?
     );
     let mut output_buffer = vec![0_u8; serialised_message.len() + 2];
     output_buffer[..2].copy_from_slice(&(serialised_message.len() as u16).to_be_bytes());
@@ -71,21 +71,19 @@ pub fn send(stream: &mut TcpStream, message: &serde_json::Value) -> BoxResult<()
             }
         }
     }
-    Err(Box::new(simple_error::simple_error!(
-        "timed out while attempting to send status-message to {}",
-        stream.peer_addr()?
-    )))
+    let err = simple_error::simple_error!("timed out while attempting to send status-message to {}", stream.peer_addr()?);
+    Err(Box::new(err))
 }
 
 /// receives the length-count of a pending message over a client-server communications stream
-fn receive_length(stream: &mut TcpStream, alive_check: fn() -> bool, results_handler: &mut dyn FnMut() -> BoxResult<()>) -> BoxResult<u16> {
+fn receive_length(stream: &mut TcpStream, alive_check: fn() -> bool, handler: &mut dyn FnMut() -> BoxResult<()>) -> BoxResult<u16> {
     stream.set_read_timeout(Some(POLL_TIMEOUT)).expect("unable to set TCP read-timeout");
 
     let mut length_bytes_read = 0;
     let mut length_spec: [u8; 2] = [0; 2];
     while alive_check() {
         //waiting to find out how long the next message is
-        results_handler()?; //send any outstanding results between cycles
+        handler()?; //send any outstanding results between cycles
 
         let size = match stream.read(&mut length_spec[length_bytes_read..]) {
             Ok(size) => size,
@@ -156,7 +154,7 @@ fn receive_payload(
         if bytes_read == length as usize {
             match serde_json::from_slice(&buffer) {
                 Ok(v) => {
-                    log::debug!("received {:?} from {}", v, stream.peer_addr()?);
+                    log::debug!("received message from {}: {:?}", stream.peer_addr()?, v);
                     return Ok(v);
                 }
                 Err(e) => {

src/protocol/results.rs

@@ -58,7 +58,7 @@ pub trait IntervalResult {
     fn to_string(&self, bit: bool) -> String;
 }
 
-pub type IntervalResultBox = Box<dyn IntervalResult + Sync + Send>;
+pub type IntervalResultBox = Box<dyn IntervalResult + Sync + Send + 'static>;
 
 pub struct ClientDoneResult {
     pub stream_idx: u8,
@@ -477,7 +477,7 @@ impl IntervalResult for UdpSendResult {
     }
 }
 
-pub fn interval_result_from_json(value: serde_json::Value) -> BoxResult<Box<dyn IntervalResult>> {
+pub fn interval_result_from_json(value: serde_json::Value) -> BoxResult<IntervalResultBox> {
     match value.get("family") {
         Some(f) => match f.as_str() {
             Some(family) => match family {

src/server.rs

@@ -21,8 +21,7 @@
 use std::io;
 use std::net::{Shutdown, SocketAddr};
 use std::sync::atomic::{AtomicBool, AtomicU16, Ordering};
-use std::sync::mpsc::channel;
-use std::sync::{Arc, Mutex};
+use std::sync::{mpsc, Arc, Mutex};
 use std::thread;
 use std::time::Duration;
 
@@ -31,7 +30,7 @@ use std::net::{TcpListener, TcpStream};
 use crate::args::Args;
 use crate::protocol::communication::{receive, send};
 use crate::protocol::messaging::{prepare_connect, prepare_connect_ready};
-use crate::protocol::results::ServerDoneResult;
+use crate::protocol::results::{IntervalResultBox, ServerDoneResult};
 use crate::stream::{tcp, udp, TestStream};
 use crate::BoxResult;
 
@@ -56,10 +55,7 @@ fn handle_client(
     let mut parallel_streams: Vec<Arc<Mutex<(dyn TestStream + Sync + Send)>>> = Vec::new();
     let mut parallel_streams_joinhandles = Vec::new();
 
-    let (results_tx, results_rx): (
-        std::sync::mpsc::Sender<crate::protocol::results::IntervalResultBox>,
-        std::sync::mpsc::Receiver<crate::protocol::results::IntervalResultBox>,
-    ) = channel();
+    let (results_tx, results_rx) = mpsc::channel::<IntervalResultBox>();
 
     //a closure used to pass results from stream-handlers to the client-communication stream
     let mut forwarding_send_stream = stream.try_clone()?;
@@ -295,12 +291,12 @@ impl Drop for ClientThreadMonitor {
 pub fn serve(args: &Args) -> BoxResult<()> {
     //config-parsing and pre-connection setup
     let tcp_port_pool = Arc::new(Mutex::new(tcp::receiver::TcpPortPool::new(
-        args.tcp_port_pool.to_string(),
-        args.tcp6_port_pool.to_string(),
+        &args.tcp_port_pool,
+        &args.tcp6_port_pool,
     )));
     let udp_port_pool = Arc::new(Mutex::new(udp::receiver::UdpPortPool::new(
-        args.udp_port_pool.to_string(),
-        args.udp6_port_pool.to_string(),
+        &args.udp_port_pool,
+        &args.udp6_port_pool,
     )));
 
     let cpu_affinity_manager = Arc::new(Mutex::new(crate::utils::cpu_affinity::CpuAffinityManager::new(&args.affinity)?));
@@ -317,53 +313,54 @@ pub fn serve(args: &Args) -> BoxResult<()> {
     log::info!("server listening on {}", listener.local_addr()?);
 
     while is_alive() {
-        match listener.accept() {
-            Ok((mut stream, address)) => {
-                log::info!("connection from {}", address);
-
-                stream.set_nodelay(true).expect("cannot disable Nagle's algorithm");
-
-                #[cfg(unix)]
-                {
-                    use crate::protocol::communication::KEEPALIVE_DURATION;
-                    let keepalive_parameters = socket2::TcpKeepalive::new().with_time(KEEPALIVE_DURATION);
-                    let raw_socket = socket2::SockRef::from(&stream);
-                    raw_socket.set_tcp_keepalive(&keepalive_parameters)?;
-                }
-
-                let client_count = CLIENTS.fetch_add(1, Ordering::Relaxed) + 1;
-                if client_limit > 0 && client_count > client_limit {
-                    log::warn!("client-limit ({}) reached; disconnecting {}...", client_limit, address.to_string());
-                    stream.shutdown(Shutdown::Both).unwrap_or_default();
-                    CLIENTS.fetch_sub(1, Ordering::Relaxed);
-                } else {
-                    let c_cam = cpu_affinity_manager.clone();
-                    let c_tcp_port_pool = tcp_port_pool.clone();
-                    let c_udp_port_pool = udp_port_pool.clone();
-                    let thread_builder = thread::Builder::new().name(address.to_string());
-                    thread_builder.spawn(move || {
-                        //ensure the client is accounted-for even if the handler panics
-                        let _client_thread_monitor = ClientThreadMonitor {
-                            client_address: address.to_string(),
-                        };
-
-                        match handle_client(&mut stream, c_cam, c_tcp_port_pool, c_udp_port_pool) {
-                            Ok(_) => (),
-                            Err(e) => log::error!("error in client-handler: {}", e),
-                        }
-
-                        //in the event of panic, this will happen when the stream is dropped
-                        stream.shutdown(Shutdown::Both).unwrap_or_default();
-                    })?;
-                }
-            }
+        let (mut stream, address) = match listener.accept() {
+            Ok((stream, address)) => (stream, address),
             Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
-                //no pending clients
+                // no pending clients
                 thread::sleep(POLL_TIMEOUT);
+                continue;
             }
             Err(e) => {
                 return Err(Box::new(e));
             }
+        };
+
+        log::info!("connection from {}", address);
+
+        stream.set_nodelay(true).expect("cannot disable Nagle's algorithm");
+
+        #[cfg(unix)]
+        {
+            use crate::protocol::communication::KEEPALIVE_DURATION;
+            let keepalive_parameters = socket2::TcpKeepalive::new().with_time(KEEPALIVE_DURATION);
+            let raw_socket = socket2::SockRef::from(&stream);
+            raw_socket.set_tcp_keepalive(&keepalive_parameters)?;
+        }
+
+        let client_count = CLIENTS.fetch_add(1, Ordering::Relaxed) + 1;
+        if client_limit > 0 && client_count > client_limit {
+            log::warn!("client-limit ({}) reached; disconnecting {}...", client_limit, address.to_string());
+            stream.shutdown(Shutdown::Both).unwrap_or_default();
+            CLIENTS.fetch_sub(1, Ordering::Relaxed);
+        } else {
+            let c_cam = cpu_affinity_manager.clone();
+            let c_tcp_port_pool = tcp_port_pool.clone();
+            let c_udp_port_pool = udp_port_pool.clone();
+            let thread_builder = thread::Builder::new().name(address.to_string());
+            thread_builder.spawn(move || {
+                // ensure the client is accounted-for even if the handler panics
+                let _client_thread_monitor = ClientThreadMonitor {
+                    client_address: address.to_string(),
+                };
+
+                match handle_client(&mut stream, c_cam, c_tcp_port_pool, c_udp_port_pool) {
+                    Ok(_) => (),
+                    Err(e) => log::error!("error in client-handler: {}", e),
+                }
+
+                //in the event of panic, this will happen when the stream is dropped
+                stream.shutdown(Shutdown::Both).unwrap_or_default();
+            })?;
         }
     }
 

src/stream/mod.rs

@@ -21,7 +21,7 @@
 pub mod tcp;
 pub mod udp;
 
-use crate::BoxResult;
+use crate::{protocol::results::IntervalResultBox, BoxResult};
 
 pub const INTERVAL: std::time::Duration = std::time::Duration::from_secs(1);
 
@@ -30,7 +30,7 @@ pub const INTERVAL: std::time::Duration = std::time::Duration::from_secs(1);
 /// INTERVAL while gathering data.
 pub trait TestStream {
     /// gather data; returns None when the test is over
-    fn run_interval(&mut self) -> Option<BoxResult<crate::protocol::results::IntervalResultBox>>;
+    fn run_interval(&mut self) -> Option<BoxResult<IntervalResultBox>>;
     /// return the port associated with the test-stream; this may vary over the test's lifetime
     fn get_port(&self) -> BoxResult<u16>;
     /// returns the index of the test, used to match client and server data
@@ -39,7 +39,7 @@ pub trait TestStream {
     fn stop(&mut self);
 }
 
-fn parse_port_spec(port_spec: String) -> Vec<u16> {
+fn parse_port_spec(port_spec: &str) -> Vec<u16> {
     let mut ports = Vec::<u16>::new();
     if !port_spec.is_empty() {
         for range in port_spec.split(',') {