tor links in order

2 files changed, 41 insertions, 1 deletions

diff --git a/src/simtest/bench.rs b/src/simtest/bench.rs
index 4d34e4b..08aa365 100644
--- a/src/simtest/bench.rs
+++ b/src/simtest/bench.rs
@@ -135,6 +135,7 @@ fn mycelium_like(sim: &Sim, ids: &[String]) {
 /// (Tor carries streams reliably). The simulated net can't make one link slow for a
 /// while only, so the circuits that are slow at times are slow throughout here. With a
 /// first wait for answers of 5 s, as Tor's transport waits longer (30 s in production).
+/// In order, as a circuit carries a stream: a message never overtakes an earlier one.
 fn tor_like(sim: &Sim, ids: &[String]) {
     let mut rng = crate::testrand::Rng::new(9);
     for (i, from) in ids.iter().enumerate() {
@@ -147,6 +148,7 @@ fn tor_like(sim: &Sim, ids: &[String]) {
             let half = Duration::from_secs_f64(rtt_ms / 2.0 / 1000.0);
             let link = LinkParams {
                 delay: (half * 4 / 5, half * 6 / 5),
+                in_order: true,
                 ..LinkParams::default()
             };
             sim.net.set_link(from, to, link.clone());
diff --git a/src/transport/sim.rs b/src/transport/sim.rs
index 06f15f6..40a1f7d 100644
--- a/src/transport/sim.rs
+++ b/src/transport/sim.rs
@@ -40,7 +40,7 @@ pub(crate) struct LinkParams {
     pub(crate) loss: f64,
     /// Probability that a message arrives twice (each copy with its own delay).
     pub(crate) duplicate: f64,
-    /// Delay, uniform in this range. Overlapping delays reorder messages.
+    /// Delay, uniform in this range. Overlapping delays reorder messages, unless `in_order`.
     pub(crate) delay: (Duration, Duration),
     /// Messages of these kinds are always lost: a peer that answers some messages but
     /// not others (e.g. keepalives but not sync).
@@ -49,6 +49,9 @@ pub(crate) struct LinkParams {
     /// 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>,
+    /// Messages arrive in the order they were sent, as on a stream (Tor's circuits,
+    /// libp2p's streams): one that would overtake waits for those before it.
+    pub(crate) in_order: bool,
 }
 
 impl Default for LinkParams {
@@ -60,6 +63,7 @@ impl Default for LinkParams {
             delay: (Duration::from_millis(1), Duration::from_millis(5)),
             drop: Vec::new(),
             rate: None,
+            in_order: false,
         }
     }
 }
@@ -88,6 +92,9 @@ struct NetInner {
     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>,
+    /// Per link in order, what it carries, each with when it would arrive: one task
+    /// delivers them one after the other (`LinkParams::in_order`).
+    in_order: HashMap<(String, String), mpsc::UnboundedSender<(tokio::time::Instant, Delivery)>>,
     down: HashSet<String>,
     inboxes: HashMap<String, mpsc::UnboundedSender<Delivery>>,
     /// Per node, its current run (`restart`): what a transport of an earlier run sends
@@ -118,6 +125,7 @@ impl SimNet {
                 default_link: LinkParams::default(),
                 links: HashMap::new(),
                 busy_until: HashMap::new(),
+                in_order: HashMap::new(),
                 down: HashSet::new(),
                 inboxes: HashMap::new(),
                 runs: HashMap::new(),
@@ -244,6 +252,36 @@ impl SimNet {
             let delays = (0..copies)
                 .map(|_| queued + inner.rng.duration_in(params.delay))
                 .collect::<Vec<_>>();
+            if params.in_order {
+                let link = (from.to_owned(), message.peer.clone());
+                let now = tokio::time::Instant::now();
+                let net = self.clone();
+                let link_tx = inner.in_order.entry(link.clone()).or_insert_with(|| {
+                    let (tx, mut rx) =
+                        mpsc::unbounded_channel::<(tokio::time::Instant, Delivery)>();
+                    let to = link.1;
+                    tokio::spawn(async move {
+                        // Each after those before it: a message that would arrive
+                        // earlier waits (`sleep_until` a past instant returns at once).
+                        while let Some((at, delivery)) = rx.recv().await {
+                            tokio::time::sleep_until(at).await;
+                            net.arrive(&to, delivery);
+                        }
+                    });
+                    tx
+                });
+                for delay in delays {
+                    let delivery = Delivery {
+                        from: from.to_owned(),
+                        body: message.body.clone(),
+                        payload_len: message.payload_len,
+                        handled: None,
+                    };
+                    // The task runs as long as the runtime: the send doesn't fail.
+                    let _ = link_tx.send((now + delay, delivery));
+                }
+                return wait;
+            }
             (delays, wait)
         };
         for delay in delays {