windows issues

Author: ssrlive

.github/workflows/rust.yml

@@ -13,7 +13,7 @@ jobs:
         host_os:
           - ubuntu-latest
           - macos-latest
-          # - windows-latest
+          - windows-latest
 
     runs-on: ${{ matrix.host_os }}
 

Cargo.toml

@@ -11,6 +11,7 @@ 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

@@ -67,8 +67,12 @@ fn connect_to_server(address: &str, port: &u16) -> BoxResult<TcpStream> {
         let socket: socket2::Socket = socket2::Socket::from(stream);
         let keepalive = socket2::TcpKeepalive::new()
             .with_time(KEEPALIVE_DURATION)
-            .with_interval(KEEPALIVE_DURATION)
-            .with_retries(4);
+            .with_interval(KEEPALIVE_DURATION);
+        cfg_if::cfg_if! {
+            if #[cfg(unix)] {
+                let keepalive = keepalive.with_retries(4);
+            }
+        }
         socket.set_tcp_keepalive(&keepalive)?;
         socket.set_nodelay(true)?;
 

src/main.rs

@@ -41,8 +41,7 @@ fn main() -> BoxResult<()> {
                 std::process::exit(3);
             }
             true
-        })
-        .expect("unable to set SIGINT handler");
+        })?;
 
         log::debug!("beginning normal operation...");
         server::serve(&args)?;
@@ -57,8 +56,7 @@ fn main() -> BoxResult<()> {
                 std::process::exit(3);
             }
             true
-        })
-        .expect("unable to set SIGINT handler");
+        })?;
 
         log::debug!("connecting to server...");
         client::execute(&args)?;

src/protocol/communication.rs

@@ -27,7 +27,7 @@ use mio::{Events, Interest, Poll};
 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
+/// the communications channels typically exchange data every second, so 2s is reasonable to avoid excess noise
 pub const KEEPALIVE_DURATION: Duration = Duration::from_secs(2);
 
 /// how long to block on polling operations
@@ -101,8 +101,8 @@ fn receive_length(stream: &mut TcpStream, alive_check: fn() -> bool, results_han
     };
     poll.registry().deregister(stream)?;
     result
-    // 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(
     stream: &mut TcpStream,
@@ -166,8 +166,8 @@ fn receive_payload(
     };
     poll.registry().deregister(stream)?;
     result
-    // 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(
     stream: &mut TcpStream,

src/server.rs

@@ -45,10 +45,20 @@ static ALIVE: AtomicBool = AtomicBool::new(true);
 static CLIENTS: AtomicU16 = AtomicU16::new(0);
 
 fn tcp_stream_try_clone(stream: &TcpStream) -> BoxResult<TcpStream> {
-    use std::os::fd::{AsRawFd, BorrowedFd};
-    let fd = unsafe { BorrowedFd::borrow_raw(stream.as_raw_fd()) };
-    let fd = fd.try_clone_to_owned()?;
-    let socket: socket2::Socket = socket2::Socket::from(fd);
+    cfg_if::cfg_if! {
+        if #[cfg(unix)] {
+            use std::os::fd::{AsRawFd, BorrowedFd};
+            let fd = unsafe { BorrowedFd::borrow_raw(stream.as_raw_fd()) };
+            let fd = fd.try_clone_to_owned()?;
+            let socket: socket2::Socket = socket2::Socket::from(fd);
+        } else {
+            use std::os::windows::io::{AsRawSocket, BorrowedSocket};
+            let socket = unsafe { BorrowedSocket::borrow_raw(stream.as_raw_socket()) };
+            let socket = socket.try_clone_to_owned()?;
+            let socket: socket2::Socket = socket2::Socket::from(socket);
+        }
+    }
+
     let stream: std::net::TcpStream = socket.into();
     let socket = TcpStream::from_std(stream);
     Ok(socket)
@@ -359,14 +369,26 @@ pub fn serve(args: &Args) -> BoxResult<()> {
                 log::info!("connection from {}", address);
 
                 let mut stream = {
-                    use std::os::fd::{FromRawFd, IntoRawFd};
-                    let fd = stream.into_raw_fd();
-                    let socket: socket2::Socket = unsafe { socket2::Socket::from_raw_fd(fd) };
-
-                    let keepalive = socket2::TcpKeepalive::new()
-                        .with_time(KEEPALIVE_DURATION)
-                        .with_interval(KEEPALIVE_DURATION)
-                        .with_retries(4);
+                    cfg_if::cfg_if! {
+                        if #[cfg(unix)] {
+                            use std::os::fd::{FromRawFd, IntoRawFd};
+                            let fd = stream.into_raw_fd();
+                            let socket: socket2::Socket = unsafe { socket2::Socket::from_raw_fd(fd) };
+
+                            let keepalive = socket2::TcpKeepalive::new()
+                                .with_time(KEEPALIVE_DURATION)
+                                .with_interval(KEEPALIVE_DURATION)
+                                .with_retries(4);
+                        } else {
+                            // Only Windows supports raw sockets
+                            use std::os::windows::io::{FromRawSocket, IntoRawSocket};
+                            let socket = stream.into_raw_socket();
+                            let socket: socket2::Socket = unsafe { socket2::Socket::from_raw_socket(socket) };
+                            let keepalive = socket2::TcpKeepalive::new()
+                                .with_time(KEEPALIVE_DURATION)
+                                .with_interval(KEEPALIVE_DURATION);
+                        }
+                    }
                     socket.set_tcp_keepalive(&keepalive)?;
 
                     socket.set_nodelay(true)?;

src/stream/tcp.rs

@@ -18,8 +18,6 @@
  * along with rperf.  If not, see <https://www.gnu.org/licenses/>.
  */
 
-use nix::sys::socket::{setsockopt, sockopt::RcvBuf, sockopt::SndBuf};
-
 use crate::protocol::results::{get_unix_timestamp, IntervalResult, TcpReceiveResult, TcpSendResult};
 use crate::BoxResult;
 
@@ -192,6 +190,7 @@ pub mod receiver {
         }
     }
 
+    #[allow(dead_code)]
     pub struct TcpReceiver {
         active: AtomicBool,
         test_definition: super::TcpTestDefinition,
@@ -300,7 +299,8 @@ pub mod receiver {
 
                                     use std::os::fd::{FromRawFd, IntoRawFd};
                                     let raw_stream = unsafe { std::net::TcpStream::from_raw_fd(stream.into_raw_fd()) };
-                                    super::setsockopt(&raw_stream, super::RcvBuf, &self.receive_buffer)?;
+                                    use nix::sys::socket::{setsockopt, sockopt::RcvBuf};
+                                    setsockopt(&raw_stream, RcvBuf, &self.receive_buffer)?;
                                     stream = unsafe { TcpStream::from_raw_fd(raw_stream.into_raw_fd()) };
                                 }
 
@@ -457,6 +457,7 @@ pub mod sender {
     const WRITE_TIMEOUT: Duration = Duration::from_millis(50);
     const BUFFER_FULL_TIMEOUT: Duration = Duration::from_millis(1);
 
+    #[allow(dead_code)]
     pub struct TcpSender {
         active: bool,
         test_definition: super::TcpTestDefinition,
@@ -530,7 +531,8 @@ pub mod sender {
             #[cfg(unix)]
             if self.send_buffer != 0 {
                 log::debug!("setting send-buffer to {}...", self.send_buffer);
-                super::setsockopt(&raw_stream, super::SndBuf, &self.send_buffer)?;
+                use nix::sys::socket::{setsockopt, sockopt::SndBuf};
+                setsockopt(&raw_stream, SndBuf, &self.send_buffer)?;
             }
 
             raw_stream.set_write_timeout(Some(WRITE_TIMEOUT))?;

src/stream/udp.rs

@@ -18,8 +18,6 @@
  * along with rperf.  If not, see <https://www.gnu.org/licenses/>.
  */
 
-use nix::sys::socket::{setsockopt, sockopt::RcvBuf, sockopt::SndBuf};
-
 use crate::protocol::results::{get_unix_timestamp, IntervalResult, UdpReceiveResult, UdpSendResult};
 use crate::BoxResult;
 
@@ -217,6 +215,7 @@ pub mod receiver {
         socket: UdpSocket,
     }
     impl UdpReceiver {
+        #[allow(unused_variables)]
         pub fn new(
             test_definition: super::UdpTestDefinition,
             stream_idx: &u8,
@@ -231,7 +230,8 @@ pub mod receiver {
             #[cfg(unix)]
             if *receive_buffer != 0 {
                 log::debug!("setting receive-buffer to {}...", receive_buffer);
-                super::setsockopt(&socket, super::RcvBuf, receive_buffer)?;
+                use nix::sys::socket::{setsockopt, sockopt::RcvBuf};
+                setsockopt(&socket, RcvBuf, receive_buffer)?;
             }
             log::debug!("bound UDP receive socket for stream {}: {}", stream_idx, socket.local_addr()?);
 
@@ -514,7 +514,7 @@ pub mod sender {
         staged_packet: Vec<u8>,
     }
     impl UdpSender {
-        #[allow(clippy::too_many_arguments)]
+        #[allow(clippy::too_many_arguments, unused_variables)]
         pub fn new(
             test_definition: super::UdpTestDefinition,
             stream_idx: &u8,
@@ -538,7 +538,8 @@ pub mod sender {
             #[cfg(unix)]
             if *send_buffer != 0 {
                 log::debug!("setting send-buffer to {}...", send_buffer);
-                super::setsockopt(&socket, super::SndBuf, send_buffer)?;
+                use nix::sys::socket::{setsockopt, sockopt::SndBuf};
+                setsockopt(&socket, SndBuf, send_buffer)?;
             }
             socket.connect(socket_addr_receiver)?;
             log::debug!("connected UDP stream {} to {}", stream_idx, socket_addr_receiver);