change: only add peers to active view after receiving neighbor messages

Author: Frando

src/proto.rs

@@ -178,7 +178,7 @@ mod test {
         // Now let node 3 join node 0.
         // Node 0 is full, so it will disconnect from either node 1 or node 2.
         network.command(3, t, Command::Join(vec![0]));
-        network.ticks(8);
+        network.ticks(10);
 
         // Confirm emitted events. There's two options because whether node 0 disconnects from
         // node 1 or node 2 is random.
@@ -213,7 +213,7 @@ mod test {
         let config = Config::default();
         let mut network = Network::new(Instant::now());
         let broadcast_ticks = 12;
-        let join_ticks = 12;
+        let join_ticks = 13;
         // build a network with 6 nodes
         let rng = rand_chacha::ChaCha12Rng::seed_from_u64(99);
         for i in 0..6 {

src/proto/hyparview.rs

@@ -392,46 +392,50 @@ where
         &mut self,
         sender: PI,
         message: ForwardJoin<PI>,
-        now: Instant,
+        _now: Instant,
         io: &mut impl IO<PI>,
     ) {
+        let peer_id = message.peer.id;
         // If the peer is already in our active view, we renew our neighbor relationship.
-        if self.active_view.contains(&message.peer.id) {
-            self.send_neighbor(message.peer.id, Priority::High, io);
+        if self.active_view.contains(&peer_id) {
+            self.insert_peer_info(message.peer, io);
+            self.send_neighbor(peer_id, Priority::High, io);
         }
         // "i) If the time to live is equal to zero or if the number of nodes in p’s active view is equal to one,
         // it will add the new node to its active view (7)"
         else if message.ttl.expired() || self.active_view.len() <= 1 {
-            self.add_active(
-                message.peer.id,
-                message.peer.data.clone(),
-                Priority::High,
-                now,
-                io,
-            );
-        }
-        // "ii) If the time to live is equal to PRWL, p will insert the new node into its passive view"
-        else if message.ttl == self.config.passive_random_walk_length {
-            self.add_passive(message.peer.id, message.peer.data.clone(), io);
-        }
-        // "iii) The time to live field is decremented."
-        // "iv) If, at this point, n has not been inserted
-        // in p’s active view, p will forward the request to a random node in its active view
-        // (different from the one from which the request was received)."
-        if !self.active_view.contains(&message.peer.id) {
-            match self
-                .active_view
-                .pick_random_without(&[&sender], &mut self.rng)
+            self.insert_peer_info(message.peer, io);
+            // Modification from paper: Instead of adding the peer directly to our active view,
+            // we only send the Neighbor message. We will add the peer to our active view once we receive a
+            // reply from our neighbor.
+            // This prevents us adding unreachable peers to our active view.
+            self.send_neighbor(peer_id, Priority::High, io);
+        } else {
+            // "ii) If the time to live is equal to PRWL, p will insert the new node into its passive view"
+            if message.ttl == self.config.passive_random_walk_length {
+                self.add_passive(peer_id, message.peer.data.clone(), io);
+            }
+            // "iii) The time to live field is decremented."
+            // "iv) If, at this point, n has not been inserted
+            // in p’s active view, p will forward the request to a random node in its active view
+            // (different from the one from which the request was received)."
+            if !self.active_view.contains(&peer_id)
+                && !self.pending_neighbor_requests.contains(&peer_id)
             {
-                None => {
-                    unreachable!("if the peer was not added, there are at least two peers in our active view.");
-                }
-                Some(next) => {
-                    let message = Message::ForwardJoin(ForwardJoin {
-                        peer: message.peer,
-                        ttl: message.ttl.next(),
-                    });
-                    io.push(OutEvent::SendMessage(*next, message));
+                match self
+                    .active_view
+                    .pick_random_without(&[&sender], &mut self.rng)
+                {
+                    None => {
+                        unreachable!("if the peer was not added, there are at least two peers in our active view.");
+                    }
+                    Some(next) => {
+                        let message = Message::ForwardJoin(ForwardJoin {
+                            peer: message.peer,
+                            ttl: message.ttl.next(),
+                        });
+                        io.push(OutEvent::SendMessage(*next, message));
+                    }
                 }
             }
         }
@@ -612,23 +616,19 @@ where
         if let Some(node) = self
             .passive_view
             .pick_random_without(&skip_peers, &mut self.rng)
+            .copied()
         {
             let priority = match self.active_view.is_empty() {
                 true => Priority::High,
                 false => Priority::Low,
             };
-            let message = Message::Neighbor(Neighbor {
-                priority,
-                data: self.me_data.clone(),
-            });
-            io.push(OutEvent::SendMessage(*node, message));
+            self.send_neighbor(node, priority, io);
             // schedule a timer that checks if the node replied with a neighbor message,
             // otherwise try again with another passive node.
             io.push(OutEvent::ScheduleTimer(
                 self.config.neighbor_request_timeout,
-                Timer::PendingNeighborRequest(*node),
+                Timer::PendingNeighborRequest(node),
             ));
-            self.pending_neighbor_requests.insert(*node);
         };
     }
 
@@ -716,11 +716,13 @@ where
     }
 
     fn send_neighbor(&mut self, peer: PI, priority: Priority, io: &mut impl IO<PI>) {
-        let message = Message::Neighbor(Neighbor {
-            priority,
-            data: self.me_data.clone(),
-        });
-        io.push(OutEvent::SendMessage(peer, message));
+        if self.pending_neighbor_requests.insert(peer) {
+            let message = Message::Neighbor(Neighbor {
+                priority,
+                data: self.me_data.clone(),
+            });
+            io.push(OutEvent::SendMessage(peer, message));
+        }
     }
 }
 

src/proto/state.rs

@@ -207,7 +207,7 @@ impl<PI: PeerIdentity, R: Rng + Clone> State<PI, R> {
         event: InEvent<PI>,
         now: Instant,
     ) -> impl Iterator<Item = OutEvent<PI>> + '_ {
-        trace!("in_event: {event:?}");
+        trace!("in : {event:?}");
         track_in_event(&event);
 
         let event: InEventMapped<PI> = event.into();
@@ -274,7 +274,7 @@ fn handle_out_event<PI: PeerIdentity>(
     conns: &mut ConnsMap<PI>,
     outbox: &mut Outbox<PI>,
 ) {
-    trace!("out_event: {event:?}");
+    trace!("out: {event:?}");
     match event {
         topic::OutEvent::SendMessage(to, message) => {
             outbox.push(OutEvent::SendMessage(to, Message { topic, message }))

src/proto/tests.rs

@@ -6,7 +6,7 @@ use bytes::Bytes;
 use n0_future::time::{Duration, Instant};
 use rand::Rng;
 use rand_core::SeedableRng;
-use tracing::{debug, warn};
+use tracing::{debug, trace_span, warn};
 
 use super::{
     util::TimerMap, Command, Config, Event, InEvent, OutEvent, PeerIdentity, State, Timer, TopicId,
@@ -62,7 +62,7 @@ fn push_back<PI: Eq + std::hash::Hash>(
     inqueues.get_mut(peer_pos).unwrap().push_back(event);
 }
 
-impl<PI: PeerIdentity + Ord, R: Rng + Clone> Network<PI, R> {
+impl<PI: PeerIdentity + Ord + std::fmt::Display, R: Rng + Clone> Network<PI, R> {
     pub fn push(&mut self, peer: State<PI, R>) {
         let idx = self.inqueues.len();
         self.inqueues.push(VecDeque::new());
@@ -98,6 +98,8 @@ impl<PI: PeerIdentity + Ord, R: Rng + Clone> Network<PI, R> {
 
     pub fn tick(&mut self) {
         self.time += self.tick_duration;
+        let tick =
+            self.time.duration_since(self.start).as_millis() / self.tick_duration.as_millis();
 
         // process timers
         for (_time, (idx, timer)) in self.timers.drain_until(&self.time) {
@@ -114,14 +116,14 @@ impl<PI: PeerIdentity + Ord, R: Rng + Clone> Network<PI, R> {
         for (idx, queue) in self.inqueues.iter_mut().enumerate() {
             let state = self.peers.get_mut(idx).unwrap();
             let peer = *state.me();
+            let span = trace_span!("tick", node = %peer, %tick);
+            let _guard = span.enter();
             while let Some(event) = queue.pop_front() {
                 if let InEvent::RecvMessage(from, _message) = &event {
                     self.conns.insert((*from, peer).into());
                 }
-                debug!(peer = ?peer, "IN  {event:?}");
                 let out = state.handle(event, self.time);
                 for event in out {
-                    debug!(peer = ?peer, "OUT {event:?}");
                     match event {
                         OutEvent::SendMessage(to, message) => {
                             let to_idx = *self.peers_by_address.get(&to).unwrap();