diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/simtest/bench.rs | 2 | ||||
| -rw-r--r-- | src/transport/sim.rs | 40 |
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 { |