sim link rates

1 files changed, 49 insertions, 1 deletions

diff --git a/src/transport/sim.rs b/src/transport/sim.rs
index 906d60a..09e7952 100644
--- a/src/transport/sim.rs
+++ b/src/transport/sim.rs
@@ -45,6 +45,10 @@ pub(crate) struct LinkParams {
     /// Messages of these kinds are always lost: a peer that answers some messages but
     /// not others (e.g. keepalives but not sync).
     pub(crate) drop: Vec<MessageKind>,
+    /// Bytes a second the link carries, if limited: one message at a time each way,
+    /// each taking its size at this rate, then the delay (the delay is the latency, as
+    /// half a round trip). None: as fast as any size.
+    pub(crate) rate: Option<u64>,
 }
 
 impl Default for LinkParams {
@@ -55,6 +59,7 @@ impl Default for LinkParams {
             duplicate: 0.0,
             delay: (Duration::from_millis(1), Duration::from_millis(5)),
             drop: Vec::new(),
+            rate: None,
         }
     }
 }
@@ -81,6 +86,8 @@ struct NetInner {
     rng: Rng,
     default_link: LinkParams,
     links: HashMap<(String, String), LinkParams>,
+    /// Per link with a rate, when it is free again (`LinkParams::rate`).
+    busy_until: HashMap<(String, String), tokio::time::Instant>,
     down: HashSet<String>,
     inboxes: HashMap<String, mpsc::UnboundedSender<Delivery>>,
     /// Per node, its current run (`restart`): what a transport of an earlier run sends
@@ -110,6 +117,7 @@ impl SimNet {
                 rng: Rng::new(seed),
                 default_link: LinkParams::default(),
                 links: HashMap::new(),
+                busy_until: HashMap::new(),
                 down: HashSet::new(),
                 inboxes: HashMap::new(),
                 runs: HashMap::new(),
@@ -202,8 +210,26 @@ impl SimNet {
             } else {
                 1
             };
+            // On a link with a rate: after those before it, as long as its size takes.
+            let queued = params
+                .rate
+                .filter(|rate| *rate > 0)
+                .map_or(Duration::ZERO, |rate| {
+                    let now = tokio::time::Instant::now();
+                    let link = (from.to_owned(), message.peer.clone());
+                    let start = inner
+                        .busy_until
+                        .get(&link)
+                        .copied()
+                        .filter(|free| *free > now)
+                        .unwrap_or(now);
+                    let free =
+                        start + Duration::from_secs_f64(message.body.len() as f64 / rate as f64);
+                    inner.busy_until.insert(link, free);
+                    free - now
+                });
             (0..copies)
-                .map(|_| inner.rng.duration_in(params.delay))
+                .map(|_| queued + inner.rng.duration_in(params.delay))
                 .collect::<Vec<_>>()
         };
         for delay in delays {
@@ -477,6 +503,28 @@ mod tests {
         assert_eq!(*at_a.lock().unwrap(), vec!["b"]);
     }
 
+    /// On a link with a rate, messages go one after another, each its size at the rate:
+    /// at a message every 100 ms, one of three is in after 150 ms, all after 400.
+    #[tokio::test]
+    async fn a_rate_queues_messages() {
+        let net = SimNet::new(1);
+        let a = net.transport("a");
+        let at_b = record(net.transport("b"));
+        let size = a.encode("b", &status()).unwrap().body.len() as u64;
+        net.set_default_link(LinkParams {
+            delay: (Duration::ZERO, Duration::ZERO),
+            rate: Some(size * 10),
+            ..LinkParams::default()
+        });
+        for _ in 0..3 {
+            a.send_message("b", &status()).await.unwrap();
+        }
+        tokio::time::sleep(Duration::from_millis(150)).await;
+        assert_eq!(at_b.lock().unwrap().len(), 1);
+        tokio::time::sleep(Duration::from_millis(250)).await;
+        assert_eq!(at_b.lock().unwrap().len(), 3);
+    }
+
     #[tokio::test]
     async fn inject_returns_once_handled() {
         let net = SimNet::new(1);