rustfmt max_width = 140

Author: ssrlive

rustfmt.toml

@@ -0,0 +1 @@
+max_width = 140

src/args.rs

@@ -50,13 +50,7 @@ pub struct Args {
     pub reverse: bool,
 
     /// the format in which to deplay information (json, megabit/sec, megabyte/sec)
-    #[arg(
-        short,
-        long,
-        value_enum,
-        value_name = "format",
-        default_value = "megabit"
-    )]
+    #[arg(short, long, value_enum, value_name = "format", default_value = "megabit")]
     pub format: Format,
 
     /// use UDP rather than TCP

src/client.rs

@@ -22,14 +22,9 @@ use crate::{
     args,
     protocol::{
         communication::{receive, send, KEEPALIVE_DURATION},
-        messaging::{
-            prepare_begin, prepare_download_configuration, prepare_end,
-            prepare_upload_configuration,
-        },
-        results::ClientDoneResult,
-        results::{
-            IntervalResultBox, IntervalResultKind, TcpTestResults, TestResults, UdpTestResults,
-        },
+        messaging::{prepare_begin, prepare_download_configuration, prepare_end, prepare_upload_configuration},
+        results::{ClientDoneResult, ClientFailedResult},
+        results::{IntervalResultBox, IntervalResultKind, TcpTestResults, TestResults, UdpTestResults},
     },
     stream::{tcp, udp, TestStream},
 };
@@ -63,21 +58,12 @@ fn connect_to_server(address: &str, port: &u16) -> BoxResult<TcpStream> {
 
     let server_addr = destination.to_socket_addrs()?.next();
     if server_addr.is_none() {
-        return Err(Box::new(simple_error::simple_error!(
-            "unable to resolve {}",
-            address
-        )));
+        return Err(Box::new(simple_error::simple_error!("unable to resolve {}", address)));
     }
-    let raw_stream =
-        match std::net::TcpStream::connect_timeout(&server_addr.unwrap(), CONNECT_TIMEOUT) {
-            Ok(s) => s,
-            Err(e) => {
-                return Err(Box::new(simple_error::simple_error!(
-                    "unable to connect: {}",
-                    e
-                )))
-            }
-        };
+    let raw_stream = match std::net::TcpStream::connect_timeout(&server_addr.unwrap(), CONNECT_TIMEOUT) {
+        Ok(s) => s,
+        Err(e) => return Err(Box::new(simple_error::simple_error!("unable to connect: {}", e))),
+    };
     let stream = match TcpStream::from_stream(raw_stream) {
         Ok(s) => s,
         Err(e) => {
@@ -89,12 +75,8 @@ fn connect_to_server(address: &str, port: &u16) -> BoxResult<TcpStream> {
     };
     log::info!("connected to server");
 
-    stream
-        .set_nodelay(true)
-        .expect("cannot disable Nagle's algorithm");
-    stream
-        .set_keepalive(Some(KEEPALIVE_DURATION))
-        .expect("unable to set TCP keepalive");
+    stream.set_nodelay(true).expect("cannot disable Nagle's algorithm");
+    stream.set_keepalive(Some(KEEPALIVE_DURATION)).expect("unable to set TCP keepalive");
 
     Ok(stream)
 }
@@ -121,18 +103,10 @@ 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 cpu_affinity_manager = Arc::new(Mutex::new(
-        crate::utils::cpu_affinity::CpuAffinityManager::new(&args.affinity)?,
-    ));
+    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 cpu_affinity_manager = Arc::new(Mutex::new(crate::utils::cpu_affinity::CpuAffinityManager::new(&args.affinity)?));
 
     let display_json: bool;
     let display_bit: bool;
@@ -163,16 +137,14 @@ pub fn execute(args: &args::Args) -> BoxResult<()> {
 
     //scaffolding to track and relay the streams and stream-results associated with this test
     let stream_count = download_config.get("streams").unwrap().as_i64().unwrap() as usize;
-    let mut parallel_streams: Vec<Arc<Mutex<(dyn TestStream + Sync + Send)>>> =
-        Vec::with_capacity(stream_count);
+    let mut parallel_streams: Vec<Arc<Mutex<(dyn TestStream + Sync + Send)>>> = Vec::with_capacity(stream_count);
     let mut parallel_streams_joinhandles = Vec::with_capacity(stream_count);
     let (results_tx, results_rx): (
         std::sync::mpsc::Sender<IntervalResultBox>,
         std::sync::mpsc::Receiver<IntervalResultBox>,
     ) = channel();
 
-    let test_results: Mutex<Box<dyn TestResults>> =
-        prepare_test_results(is_udp, stream_count as u8);
+    let test_results: Mutex<Box<dyn TestResults>> = prepare_test_results(is_udp, stream_count as u8);
 
     //a closure used to pass results from stream-handlers to the test-result structure
     let mut results_handler = || -> BoxResult<()> {
@@ -193,10 +165,7 @@ pub fn execute(args: &args::Args) -> BoxResult<()> {
                     } else {
                         log::warn!("stream {} failed", result.get_stream_idx());
                     }
-                    tr.mark_stream_done(
-                        &result.get_stream_idx(),
-                        result.kind() == IntervalResultKind::ClientDone,
-                    );
+                    tr.mark_stream_done(&result.get_stream_idx(), result.kind() == IntervalResultKind::ClientDone);
                     if tr.count_in_progress_streams() == 0 {
                         complete = true;
 
@@ -226,10 +195,7 @@ pub fn execute(args: &args::Args) -> BoxResult<()> {
 
         if is_udp {
             //UDP
-            log::info!(
-                "preparing for reverse-UDP test with {} streams...",
-                stream_count
-            );
+            log::info!("preparing for reverse-UDP test with {} streams...", stream_count);
 
             let test_definition = udp::UdpTestDefinition::new(&download_config)?;
             for stream_idx in 0..stream_count {
@@ -246,10 +212,7 @@ pub fn execute(args: &args::Args) -> BoxResult<()> {
             }
         } else {
             //TCP
-            log::info!(
-                "preparing for reverse-TCP test with {} streams...",
-                stream_count
-            );
+            log::info!("preparing for reverse-TCP test with {} streams...", stream_count);
 
             let test_definition = tcp::TcpTestDefinition::new(&download_config)?;
             for stream_idx in 0..stream_count {
@@ -410,11 +373,9 @@ pub fn execute(args: &args::Args) -> BoxResult<()> {
                         },
                         Err(e) => {
                             log::error!("unable to process stream: {}", e);
-                            match c_results_tx.send(Box::new(
-                                crate::protocol::results::ClientFailedResult {
-                                    stream_idx: test.get_idx(),
-                                },
-                            )) {
+                            match c_results_tx.send(Box::new(ClientFailedResult {
+                                stream_idx: test.get_idx(),
+                            })) {
                                 Ok(_) => (),
                                 Err(e) => {
                                     log::error!("unable to report interval-failed-result: {}", e)
@@ -457,8 +418,7 @@ pub fn execute(args: &args::Args) -> BoxResult<()> {
                 "receive" | "send" => {
                     //receive/send-results from the server
                     if !display_json {
-                        let result =
-                            crate::protocol::results::interval_result_from_json(payload.clone())?;
+                        let result = crate::protocol::results::interval_result_from_json(payload.clone())?;
                         println!("{}", result.to_string(display_bit));
                     }
                     let mut tr = test_results.lock().unwrap();
@@ -480,9 +440,7 @@ pub fn execute(args: &args::Args) -> BoxResult<()> {
                                 _ => (), //not possible
                             }
                             tr.mark_stream_done_server(&(idx64 as u8));
-                            if tr.count_in_progress_streams() == 0
-                                && tr.count_in_progress_streams_server() == 0
-                            {
+                            if tr.count_in_progress_streams() == 0 && tr.count_in_progress_streams_server() == 0 {
                                 //all data gathered from both sides
                                 kill();
                             }
@@ -517,9 +475,7 @@ pub fn execute(args: &args::Args) -> BoxResult<()> {
         let mut stream = match (*ps).lock() {
             Ok(guard) => guard,
             Err(poisoned) => {
-                log::error!(
-                    "a stream-handler was poisoned; this indicates some sort of logic error"
-                );
+                log::error!("a stream-handler was poisoned; this indicates some sort of logic error");
                 poisoned.into_inner()
             }
         };
@@ -601,23 +557,14 @@ pub fn kill() -> bool {
 }
 fn start_kill_timer() {
     unsafe {
-        KILL_TIMER_RELATIVE_START_TIME = SystemTime::now()
-            .duration_since(UNIX_EPOCH)
-            .unwrap()
-            .as_secs_f64();
+        KILL_TIMER_RELATIVE_START_TIME = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs_f64();
     }
 }
 fn is_alive() -> bool {
     unsafe {
         if KILL_TIMER_RELATIVE_START_TIME != 0.0 {
             //initialised
-            if SystemTime::now()
-                .duration_since(UNIX_EPOCH)
-                .unwrap()
-                .as_secs_f64()
-                - KILL_TIMER_RELATIVE_START_TIME
-                >= KILL_TIMEOUT
-            {
+            if SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs_f64() - KILL_TIMER_RELATIVE_START_TIME >= KILL_TIMEOUT {
                 return false;
             }
         }

src/main.rs

@@ -39,9 +39,7 @@ fn main() {
         log::debug!("registering SIGINT handler...");
         ctrlc2::set_handler(move || {
             if server::kill() {
-                log::warn!(
-                    "shutdown requested; please allow a moment for any in-progress tests to stop"
-                );
+                log::warn!("shutdown requested; please allow a moment for any in-progress tests to stop");
             } else {
                 log::warn!("forcing shutdown immediately");
                 std::process::exit(3);
@@ -60,9 +58,7 @@ fn main() {
         log::debug!("registering SIGINT handler...");
         ctrlc2::set_handler(move || {
             if client::kill() {
-                log::warn!(
-                    "shutdown requested; please allow a moment for any in-progress tests to stop"
-                );
+                log::warn!("shutdown requested; please allow a moment for any in-progress tests to stop");
             } else {
                 log::warn!("forcing shutdown immediately");
                 std::process::exit(3);

src/protocol/communication.rs

@@ -51,21 +51,12 @@ pub fn send(stream: &mut TcpStream, message: &serde_json::Value) -> BoxResult<()
 }
 
 /// 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, results_handler: &mut dyn FnMut() -> BoxResult<()>) -> BoxResult<u16> {
     let mut cloned_stream = stream.try_clone()?;
 
     let mio_token = Token(0);
     let poll = Poll::new()?;
-    poll.register(
-        &cloned_stream,
-        mio_token,
-        Ready::readable(),
-        PollOpt::edge(),
-    )?;
+    poll.register(&cloned_stream, mio_token, Ready::readable(), PollOpt::edge())?;
     let mut events = Events::with_capacity(1); //only interacting with one stream
 
     let mut length_bytes_read = 0;
@@ -93,30 +84,22 @@ fn receive_length(
                         return Err(Box::new(simple_error::simple_error!("connection lost")));
                     } else {
                         //shutting down; a disconnect is expected
-                        return Err(Box::new(simple_error::simple_error!(
-                            "local shutdown requested"
-                        )));
+                        return Err(Box::new(simple_error::simple_error!("local shutdown requested")));
                     }
                 }
 
                 length_bytes_read += size;
                 if length_bytes_read == 2 {
                     let length = u16::from_be_bytes(length_spec);
-                    log::debug!(
-                        "received length-spec of {} from {}",
-                        length,
-                        stream.peer_addr()?
-                    );
+                    log::debug!("received length-spec of {} from {}", length, stream.peer_addr()?);
                     return Ok(length);
                 } else {
                     log::debug!("received partial length-spec from {}", stream.peer_addr()?);
                 }
             }
         }
     }
-    Err(Box::new(simple_error::simple_error!(
-        "system shutting down"
-    )))
+    Err(Box::new(simple_error::simple_error!("system shutting down")))
 }
 /// receives the data-value of a pending message over a client-server communications stream
 fn receive_payload(
@@ -129,12 +112,7 @@ fn receive_payload(
 
     let mio_token = Token(0);
     let poll = Poll::new()?;
-    poll.register(
-        &cloned_stream,
-        mio_token,
-        Ready::readable(),
-        PollOpt::edge(),
-    )?;
+    poll.register(&cloned_stream, mio_token, Ready::readable(), PollOpt::edge())?;
     let mut events = Events::with_capacity(1); //only interacting with one stream
 
     let mut bytes_read = 0;
@@ -162,9 +140,7 @@ fn receive_payload(
                         return Err(Box::new(simple_error::simple_error!("connection lost")));
                     } else {
                         // shutting down; a disconnect is expected
-                        return Err(Box::new(simple_error::simple_error!(
-                            "local shutdown requested"
-                        )));
+                        return Err(Box::new(simple_error::simple_error!("local shutdown requested")));
                     }
                 }
 
@@ -185,9 +161,7 @@ fn receive_payload(
             }
         }
     }
-    Err(Box::new(simple_error::simple_error!(
-        "system shutting down"
-    )))
+    Err(Box::new(simple_error::simple_error!("system shutting down")))
 }
 /// handles the full process of retrieving a message from a client-server communications stream
 pub fn receive(