ping go out in their own tasks

1 files changed, 44 insertions, 6 deletions

diff --git a/src/transport/paths.rs b/src/transport/paths.rs
index a9bac49..6696e1a 100644
--- a/src/transport/paths.rs
+++ b/src/transport/paths.rs
@@ -284,6 +284,17 @@ impl Table {
         self.counts.probes_sent += 1;
     }
 
+    /// The PING with `nonce` to `address` (from `due`) goes now: its round trip counts
+    /// from here, not from when it was found due.
+    fn sending(&mut self, address: &str, nonce: u64, now: Instant) {
+        if let Some(path) = self.paths.get_mut(address)
+            && let Some((expected, sent)) = &mut path.probe
+            && *expected == nonce
+        {
+            *sent = now;
+        }
+    }
+
     /// A PONG with `nonce` from `address`: the round trip, if it answers our PING out.
     fn pong(&mut self, address: &str, nonce: u64, now: Instant) -> Option<Duration> {
         let path = self.paths.get_mut(address)?;
@@ -665,7 +676,9 @@ impl Paths {
         sent
     }
 
-    /// Pings the paths due (`Table::due`), every `PROBE_TICK`, until dropped.
+    /// Pings the paths due (`Table::due`), every `PROBE_TICK`, until dropped. Each PING
+    /// goes on its own, timed from when it goes: a send that takes long (a dial) neither
+    /// holds up the others nor counts in their round trips.
     async fn probe(self: Arc<Self>) {
         let mut tick = tokio::time::interval(PROBE_TICK);
         tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
@@ -676,12 +689,20 @@ impl Paths {
                 .lock_or_recover()
                 .due(Instant::now(), random_nonce);
             for (address, nonce) in due {
-                if let Err(error) = self.send_probe(&address, false, nonce).await {
-                    debug!(to = %address, error = %format!("{error:#}"), "Couldn't send a PING");
-                    self.table
+                let paths = Arc::clone(&self);
+                tokio::spawn(async move {
+                    paths
+                        .table
                         .lock_or_recover()
-                        .failed(&address, Instant::now());
-                }
+                        .sending(&address, nonce, Instant::now());
+                    if let Err(error) = paths.send_probe(&address, false, nonce).await {
+                        debug!(to = %address, error = %format!("{error:#}"), "Couldn't send a PING");
+                        paths
+                            .table
+                            .lock_or_recover()
+                            .failed(&address, Instant::now());
+                    }
+                });
             }
         }
     }
@@ -963,6 +984,23 @@ mod tests {
         assert!(table.paths.is_empty() && table.chosen.is_empty());
     }
 
+    /// A PING's round trip counts from when it goes, not from when it was found due.
+    #[test]
+    fn round_trips_count_from_the_send() {
+        let mut table = Table::default();
+        let now = Instant::now();
+        table.rank(paths(&["libp2p:x", "mycelium:x"]), now);
+        let due = table.due(now, || 7);
+        assert_eq!(due.len(), 2);
+        // Sent a second late (after another path's slow send).
+        let sent = now + Duration::from_secs(1);
+        table.sending("libp2p:x", 7, sent);
+        table.sending("libp2p:x", 8, sent + Duration::from_secs(1)); // not its PING
+        table.pong("libp2p:x", 7, sent + Duration::from_millis(100));
+        let rtt = table.info("libp2p:x", now).rtt_ms.unwrap();
+        assert!((rtt - 100.0).abs() < 1e-6, "{rtt}");
+    }
+
     /// The node's keepalive PING on a path is its probe: paths doesn't ping it too, and
     /// the PONG measures it.
     #[test]