Fix network preemption (#45)

* Ensures connections are only established if there's
space to store them.

Also downgrades send_interval to u64 (u128 is not supported
by serde)

* cast elapsed to u64

Author: Stringy

scripts/network/workloads/network-client.toml

@@ -12,3 +12,4 @@ departure_rate = 0.1
 connections_static = 100
 connections_dyn_max = 1000
 preempt = true
+conns_per_addr = 1

scripts/network/workloads/network-server.toml

@@ -8,5 +8,6 @@ server = true
 address = "223.42.0.1"
 target_port = 1337
 connections_static = 100
-connections_dyn_max = 1000
-preempt = true
+connections_dyn_max = 100
+preempt = false
+conns_per_addr = 1

src/lib.rs

@@ -131,7 +131,7 @@ pub enum Workload {
         /// This parameter allows to control the overhead of sending data,
         /// so that it will not impact connections monitoring.
         #[serde(default = "default_network_send_interval")]
-        send_interval: u128,
+        send_interval: u64,
 
         /// Whether or not to wait for a connection to be removed before adding
         /// a new one, when the dynamic connection limit is reached.
@@ -166,7 +166,7 @@ fn default_conns_per_addr() -> u16 {
     100
 }
 
-fn default_network_send_interval() -> u128 {
+fn default_network_send_interval() -> u64 {
     100
 }
 

src/worker/network.rs

@@ -191,31 +191,32 @@ impl NetworkWorker {
                 let (local_addr, local_port) =
                     get_local_addr_port(addr, conns_per_addr, index);
 
-                socket
-                    .connect(
-                        iface.context(),
-                        (addr, target_port),
-                        (local_addr, local_port),
-                    )
-                    .unwrap();
-
-                let handle = sockets.add(socket);
                 let lifetime: f64 =
                     thread_rng().sample(Exp::new(departure_rate).unwrap());
 
                 // If we've reached the connections limit
-                if dynamic_sockets.len() == connections_dyn_max as usize {
-                    if preempt {
-                        let idx = thread_rng()
-                            .gen_range(0..connections_dyn_max as usize);
-                        let (key, _) = sockets.iter().nth(idx).unwrap();
-                        dynamic_sockets.remove(&key);
-                        close_sockets.push(key);
-
-                        dynamic_sockets
-                            .insert(handle, (SystemTime::now(), lifetime));
-                    }
-                } else {
+                if dynamic_sockets.len() == connections_dyn_max as usize
+                    && preempt
+                {
+                    let idx =
+                        thread_rng().gen_range(0..connections_dyn_max as usize);
+                    let (key, _) = sockets.iter().nth(idx).unwrap();
+                    dynamic_sockets.remove(&key);
+                    close_sockets.push(key);
+                }
+
+                // either we've just removed a socket and want to preempt
+                // or, we've have space and we're processing normally
+                if dynamic_sockets.len() < connections_dyn_max as usize {
+                    socket
+                        .connect(
+                            iface.context(),
+                            (addr, target_port),
+                            (local_addr, local_port),
+                        )
+                        .unwrap();
+
+                    let handle = sockets.add(socket);
                     dynamic_sockets
                         .insert(handle, (SystemTime::now(), lifetime));
                 }
@@ -273,7 +274,8 @@ impl NetworkWorker {
                 }
 
                 if socket.may_send() {
-                    let elapsed = send_timer.elapsed().unwrap().as_millis();
+                    let elapsed =
+                        send_timer.elapsed().unwrap().as_millis() as u64;
 
                     // Throttle sending data via connection, since the main
                     // purpose is to excercise connection monitoring.