Use duration for the link conditioner latency (#205)

* Use duration for the link conditioner latency

* Test a really bad connection using the link conditioner

This makes the link conditioner public and exposes an interface in the
socket.

Author: kstrafe

Cargo.toml

@@ -29,7 +29,8 @@ crossbeam-channel = "0.3"
 lazy_static = "1.1"
 log = "0.4"
 mio = "0.6"
-rand = "0.5"
+rand = "0.6"
+rand_pcg = "0.1"
 clap = { version = "2.32", features = ["yaml"], optional = true }
 env_logger = { version = "0.6", optional = true }
 

src/lib.rs

@@ -39,5 +39,5 @@ pub use self::throughput::ThroughputMonitoring;
 
 pub use self::config::Config;
 pub use self::error::{ErrorKind, Result};
-pub use self::net::{Socket, SocketEvent};
+pub use self::net::{LinkConditioner, Socket, SocketEvent};
 pub use self::packet::{DeliveryGuarantee, OrderingGuarantee, Packet};

src/net.rs

@@ -11,6 +11,7 @@ mod virtual_connection;
 pub mod constants;
 
 pub use self::events::SocketEvent;
+pub use self::link_conditioner::LinkConditioner;
 pub use self::quality::{NetworkQuality, RttMeasurer};
 pub use self::socket::Socket;
 pub use self::virtual_connection::VirtualConnection;

src/net/link_conditioner.rs

@@ -3,14 +3,20 @@
 //! networks. This is not in heavy use yet, hence the allowing dead code. These will be removed as our testing
 //! becomes more sophisticated.
 
-use rand::prelude::random;
+use rand::Rng;
+use rand_pcg::Pcg64Mcg as Random;
+use std::time::Duration;
 
-#[derive(Debug)]
+/// Network simulator. Used to simulate network conditions as dropped packets and packet delays.
+/// For use in [Socket::set_link_conditioner](crate::net::Socket::set_link_conditioner).
+#[derive(Clone, Debug)]
 pub struct LinkConditioner {
     // Value between 0 and 1, representing the % change a packet will be dropped on sending
     packet_loss: f64,
-    // Value in milliseconds, representing the delay imposed between packets
-    latency: u32,
+    // Duration of the delay imposed between packets
+    latency: Duration,
+    // Random number generator
+    random: Random,
 }
 
 impl LinkConditioner {
@@ -19,7 +25,8 @@ impl LinkConditioner {
     pub fn new() -> LinkConditioner {
         LinkConditioner {
             packet_loss: 0.0,
-            latency: 0,
+            latency: Duration::default(),
+            random: Random::new(0),
         }
     }
 
@@ -31,12 +38,18 @@ impl LinkConditioner {
 
     /// Sets the latency the link conditioner should apply to each packet
     #[allow(dead_code)]
-    pub fn set_latency(&mut self, latency: u32) {
+    pub fn set_latency(&mut self, latency: Duration) {
         self.latency = latency
     }
 
     /// Function that checks to see if a packet should be dropped or not
-    pub fn should_send(&self) -> bool {
-        random::<f64>() >= self.packet_loss
+    pub fn should_send(&mut self) -> bool {
+        self.random.gen_range(0.0, 1.0) >= self.packet_loss
+    }
+}
+
+impl Default for LinkConditioner {
+    fn default() -> Self {
+        Self::new()
     }
 }

src/net/socket.rs

@@ -126,6 +126,11 @@ impl Socket {
         }
     }
 
+    /// Set the link conditioner for this socket. See [LinkConditioner] for further details.
+    pub fn set_link_conditioner(&mut self, link_conditioner: Option<LinkConditioner>) {
+        self.link_conditioner = link_conditioner;
+    }
+
     /// Iterate through all of the idle connections based on `idle_connection_timeout` config and
     /// remove them from the active connections. For each connection removed, we will send a
     /// `SocketEvent::TimeOut` event to the `event_sender` channel.
@@ -238,8 +243,8 @@ impl Socket {
 
     // In the presence of a link conditioner, we would like it to determine whether or not we should
     // send a packet.
-    fn should_send_packet(&self) -> bool {
-        if let Some(link_conditioner) = &self.link_conditioner {
+    fn should_send_packet(&mut self) -> bool {
+        if let Some(link_conditioner) = &mut self.link_conditioner {
             link_conditioner.should_send()
         } else {
             true
@@ -277,7 +282,7 @@ impl Socket {
 mod tests {
     use crate::{
         net::constants::{ACKED_PACKET_HEADER, FRAGMENT_HEADER_SIZE, STANDARD_HEADER_SIZE},
-        Config, Packet, Socket, SocketEvent,
+        Config, LinkConditioner, Packet, Socket, SocketEvent,
     };
     use std::collections::HashSet;
     use std::net::{SocketAddr, UdpSocket};
@@ -779,4 +784,73 @@ mod tests {
         let time = Instant::now();
         server.manual_poll(time);
     }
+
+    #[test]
+    fn really_bad_network_keeps_chugging_along() {
+        let server_addr = "127.0.0.1:12320".parse::<SocketAddr>().unwrap();
+        let client_addr = "127.0.0.1:12321".parse::<SocketAddr>().unwrap();
+
+        let (mut server, server_sender, server_receiver) = Socket::bind(server_addr).unwrap();
+        let (mut client, client_sender, client_receiver) = Socket::bind(client_addr).unwrap();
+
+        let time = Instant::now();
+
+        // We give both the server and the client a really bad bidirectional link
+        let link_conditioner = {
+            let mut lc = LinkConditioner::new();
+            lc.set_packet_loss(0.9);
+            Some(lc)
+        };
+
+        client.set_link_conditioner(link_conditioner.clone());
+        server.set_link_conditioner(link_conditioner);
+
+        let mut set = HashSet::new();
+
+        // We chat 100 packets between the client and server, which will re-send any non-acked
+        // packets
+        let mut send_many_packets = |dummy: Option<u8>| {
+            for id in 0..100 {
+                client_sender
+                    .send(Packet::reliable_unordered(
+                        server_addr,
+                        vec![dummy.unwrap_or(id)],
+                    ))
+                    .unwrap();
+
+                server_sender
+                    .send(Packet::reliable_unordered(client_addr, vec![255]))
+                    .unwrap();
+
+                client.manual_poll(time);
+                server.manual_poll(time);
+
+                while let Ok(_) = client_receiver.try_recv() {}
+                while let Ok(event) = server_receiver.try_recv() {
+                    match event {
+                        SocketEvent::Packet(pkt) => {
+                            set.insert(pkt.payload()[0]);
+                        }
+                        SocketEvent::Timeout(_) => {
+                            panic!["Unable to time out, time has not advanced"]
+                        }
+                        SocketEvent::Connect(_) => {}
+                    }
+                }
+            }
+
+            return set.len();
+        };
+
+        // The first chatting sequence sends packets 0..100 from the client to the server. After
+        // this we just chat with a value of 255 so we don't accidentally overlap those chatting
+        // packets with the packets we want to ack.
+        assert_eq![42, send_many_packets(None)];
+        assert_eq![85, send_many_packets(Some(255))];
+        assert_eq![98, send_many_packets(Some(255))];
+        assert_eq![100, send_many_packets(Some(255))];
+
+        // 101 because we have 0..100 and 255 from the dummies
+        assert_eq![101, send_many_packets(Some(255))];
+    }
 }