make rperf as a lib

Author: ssrlive

Cargo.toml

@@ -14,7 +14,7 @@ readme = "README.md"
 chrono = "0.4"
 clap = { version = "4.4", features = ["derive", "wrap_help"] }
 core_affinity = "0.8"
-ctrlc2 = "3.5"
+ctrlc2 = { version = "3.5", features = ["termination"] }
 env_logger = "0.10"
 log = { version = "0.4", features = ["std"] }
 mio = { version = "0.8", features = ["log", "os-poll", "net"] }

src/client.rs

@@ -27,6 +27,7 @@ use crate::{
         results::{IntervalResultBox, IntervalResultKind, TcpTestResults, TestResults, UdpTestResults},
     },
     stream::{tcp, udp, TestStream},
+    BoxResult,
 };
 use mio::net::TcpStream;
 use std::{
@@ -40,8 +41,6 @@ use std::{
     time::{Duration, SystemTime, UNIX_EPOCH},
 };
 
-type BoxResult<T> = Result<T, Box<dyn std::error::Error + Send + Sync + 'static>>;
-
 /// when false, the system is shutting down
 static ALIVE: AtomicBool = AtomicBool::new(true);
 

src/lib.rs

@@ -0,0 +1,8 @@
+pub mod args;
+pub mod client;
+pub(crate) mod protocol;
+pub mod server;
+pub(crate) mod stream;
+pub(crate) mod utils;
+
+pub(crate) type BoxResult<T> = Result<T, Box<dyn std::error::Error + Send + Sync + 'static>>;

src/main.rs

@@ -18,12 +18,7 @@
  * along with rperf.  If not, see <https://www.gnu.org/licenses/>.
  */
 
-mod args;
-mod client;
-mod protocol;
-mod server;
-mod stream;
-mod utils;
+use rperf::{args, client, server};
 
 fn main() {
     use clap::Parser;
@@ -37,7 +32,7 @@ fn main() {
 
     if args.server {
         log::debug!("registering SIGINT handler...");
-        ctrlc2::set_handler(move || {
+        let exiting = ctrlc2::set_handler(move || {
             if server::kill() {
                 log::warn!("shutdown requested; please allow a moment for any in-progress tests to stop");
             } else {
@@ -50,6 +45,7 @@ fn main() {
 
         log::debug!("beginning normal operation...");
         let service = server::serve(&args);
+        exiting.join().expect("unable to join SIGINT handler thread");
         if service.is_err() {
             log::error!("unable to run server: {}", service.unwrap_err());
             std::process::exit(4);

src/protocol/communication.rs

@@ -24,7 +24,7 @@ use std::time::Duration;
 use mio::net::TcpStream;
 use mio::{Events, Interest, Poll, Token};
 
-type BoxResult<T> = Result<T, Box<dyn std::error::Error + Send + Sync + 'static>>;
+use crate::BoxResult;
 
 /// how long to wait for keepalive events
 // the communications channels typically exchange data every second, so 2s is reasonable to avoid excess noise

src/protocol/messaging.rs

@@ -18,7 +18,7 @@
  * along with rperf.  If not, see <https://www.gnu.org/licenses/>.
  */
 
-type BoxResult<T> = Result<T, Box<dyn std::error::Error + Send + Sync + 'static>>;
+use crate::BoxResult;
 
 /// prepares a message used to tell the server to begin operations
 pub fn prepare_begin() -> serde_json::Value {

src/protocol/results.rs

@@ -18,13 +18,11 @@
  * along with rperf.  If not, see <https://www.gnu.org/licenses/>.
  */
 
+use crate::BoxResult;
+use serde::{Deserialize, Serialize};
 use std::collections::{HashMap, HashSet};
 use std::time::{SystemTime, UNIX_EPOCH};
 
-use serde::{Deserialize, Serialize};
-
-type BoxResult<T> = Result<T, Box<dyn std::error::Error + Send + Sync + 'static>>;
-
 /* This module contains structures used to represent and collect the results of tests.
  * Since everything is basically just a data-container with representation methods,
  * it isn't extensively documented.

src/server.rs

@@ -34,8 +34,7 @@ use crate::protocol::communication::{receive, send, KEEPALIVE_DURATION};
 use crate::protocol::messaging::{prepare_connect, prepare_connect_ready};
 use crate::protocol::results::ServerDoneResult;
 use crate::stream::{tcp, udp, TestStream};
-
-type BoxResult<T> = Result<T, Box<dyn std::error::Error + Send + Sync + 'static>>;
+use crate::BoxResult;
 
 const POLL_TIMEOUT: Duration = Duration::from_millis(500);
 
@@ -335,7 +334,14 @@ pub fn serve(args: &Args) -> BoxResult<()> {
     let mut events = Events::with_capacity(32);
 
     while is_alive() {
-        poll.poll(&mut events, Some(POLL_TIMEOUT))?;
+        if let Err(err) = poll.poll(&mut events, Some(POLL_TIMEOUT)) {
+            if err.kind() == std::io::ErrorKind::Interrupted {
+                log::debug!("Poll interrupted: \"{err}\", ignored, continue polling");
+                continue;
+            }
+            log::error!("Poll error: {}", err);
+            break;
+        }
         for event in events.iter() {
             event.token();
             loop {

src/stream/mod.rs

@@ -21,7 +21,7 @@
 pub mod tcp;
 pub mod udp;
 
-type BoxResult<T> = Result<T, Box<dyn std::error::Error + Send + Sync + 'static>>;
+use crate::BoxResult;
 
 pub const INTERVAL: std::time::Duration = std::time::Duration::from_secs(1);
 

src/stream/tcp.rs

@@ -21,11 +21,10 @@
 use nix::sys::socket::{setsockopt, sockopt::RcvBuf, sockopt::SndBuf};
 
 use crate::protocol::results::{get_unix_timestamp, IntervalResult, TcpReceiveResult, TcpSendResult};
+use crate::BoxResult;
 
 use super::{parse_port_spec, TestStream, INTERVAL};
 
-type BoxResult<T> = Result<T, Box<dyn std::error::Error + Send + Sync + 'static>>;
-
 pub const TEST_HEADER_SIZE: usize = 16;
 
 #[derive(Clone)]