dedups gossip addresses, taking the one with highest weight

gossip/src/contact_info.rs

@@ -112,7 +112,9 @@ impl ContactInfo {
         let delay = 10 * 60 * 1000; // 10 minutes
         let now = timestamp() - delay + rng.gen_range(0, 2 * delay);
         let pubkey = pubkey.unwrap_or_else(solana_sdk::pubkey::new_rand);
-        ContactInfo::new_localhost(&pubkey, now)
+        let mut node = ContactInfo::new_localhost(&pubkey, now);
+        node.gossip.set_port(rng.gen_range(1024, u16::MAX));
+        node
     }
 
     #[cfg(test)]

gossip/src/crds_gossip.rs

@@ -17,6 +17,7 @@ use {
         duplicate_shred::{self, DuplicateShredIndex, LeaderScheduleFn, MAX_DUPLICATE_SHREDS},
         ping_pong::PingCache,
     },
+    itertools::Itertools,
     rayon::ThreadPool,
     solana_ledger::shred::Shred,
     solana_sdk::{
@@ -353,6 +354,22 @@ pub fn get_weight(max_weight: f32, time_since_last_selected: u32, stake: f32) ->
     1.0_f32.max(weight.min(max_weight))
 }
 
+// Dedups gossip addresses, keeping only the one with the highest weight.
+pub(crate) fn dedup_gossip_addresses<I, T: PartialOrd>(
+    nodes: I,
+) -> HashMap</*gossip:*/ SocketAddr, (/*weight:*/ T, ContactInfo)>
+where
+    I: IntoIterator<Item = (/*weight:*/ T, ContactInfo)>,
+{
+    nodes
+        .into_iter()
+        .into_grouping_map_by(|(_weight, node)| node.gossip)
+        .aggregate(|acc, _node_gossip, (weight, node)| match acc {
+            Some((ref w, _)) if w >= &weight => acc,
+            Some(_) | None => Some((weight, node)),
+        })
+}
+
 #[cfg(test)]
 mod test {
     use {

gossip/src/crds_gossip_pull.rs

@@ -17,7 +17,7 @@ use {
         cluster_info_metrics::GossipStats,
         contact_info::ContactInfo,
         crds::{Crds, GossipRoute, VersionedCrdsValue},
-        crds_gossip::{get_stake, get_weight},
+        crds_gossip::{self, get_stake, get_weight},
         crds_gossip_error::CrdsGossipError,
         crds_value::CrdsValue,
         ping_pong::PingCache,
@@ -244,22 +244,25 @@ impl CrdsGossipPull {
         );
         // Check for nodes which have responded to ping messages.
         let mut rng = rand::thread_rng();
-        let (weights, peers): (Vec<_>, Vec<_>) = {
+        let peers: Vec<_> = {
             let mut ping_cache = ping_cache.lock().unwrap();
             let mut pingf = move || Ping::new_rand(&mut rng, self_keypair).ok();
             let now = Instant::now();
             peers
                 .into_iter()
-                .filter_map(|(weight, peer)| {
+                .filter(|(_weight, peer)| {
                     let node = (peer.id, peer.gossip);
                     let (check, ping) = ping_cache.check(now, node, &mut pingf);
                     if let Some(ping) = ping {
                         pings.push((peer.gossip, ping));
                     }
-                    check.then_some((weight, peer))
+                    check
                 })
-                .unzip()
+                .collect()
         };
+        let (weights, peers): (Vec<_>, Vec<_>) = crds_gossip::dedup_gossip_addresses(peers)
+            .into_values()
+            .unzip();
         if peers.is_empty() {
             return Err(CrdsGossipError::NoPeers);
         }

gossip/src/crds_gossip_push.rs

@@ -16,7 +16,7 @@ use {
         cluster_info::{Ping, CRDS_UNIQUE_PUBKEY_CAPACITY},
         contact_info::ContactInfo,
         crds::{Crds, CrdsError, Cursor, GossipRoute},
-        crds_gossip::{get_stake, get_weight},
+        crds_gossip::{self, get_stake, get_weight},
         crds_value::CrdsValue,
         ping_pong::PingCache,
         received_cache::ReceivedCache,
@@ -299,22 +299,26 @@ impl CrdsGossipPush {
             socket_addr_space,
         );
         // Check for nodes which have responded to ping messages.
-        let (weights, peers): (Vec<_>, Vec<_>) = {
+        let peers: Vec<_> = {
             let mut ping_cache = ping_cache.lock().unwrap();
             let mut pingf = move || Ping::new_rand(&mut rng, self_keypair).ok();
             let now = Instant::now();
             peers
                 .into_iter()
-                .filter_map(|(weight, peer)| {
+                .filter(|(_weight, peer)| {
                     let node = (peer.id, peer.gossip);
                     let (check, ping) = ping_cache.check(now, node, &mut pingf);
                     if let Some(ping) = ping {
                         pings.push((peer.gossip, ping));
                     }
-                    check.then_some((weight, peer.id))
+                    check
                 })
-                .unzip()
+                .collect()
         };
+        let (weights, peers): (Vec<_>, Vec<_>) = crds_gossip::dedup_gossip_addresses(peers)
+            .into_values()
+            .map(|(weight, node)| (weight, node.id))
+            .unzip();
         if peers.is_empty() {
             return;
         }
@@ -572,7 +576,8 @@ mod tests {
 
         let active_set = push.active_set.read().unwrap();
         assert!(active_set.get(&value1.label().pubkey()).is_some());
-        let value2 = ContactInfo::new_localhost(&solana_sdk::pubkey::new_rand(), 0);
+        let mut value2 = ContactInfo::new_localhost(&solana_sdk::pubkey::new_rand(), 0);
+        value2.gossip.set_port(1245);
         ping_cache
             .lock()
             .unwrap()
@@ -608,8 +613,9 @@ mod tests {
             let active_set = push.active_set.read().unwrap();
             assert!(active_set.get(&value2.label().pubkey()).is_some());
         }
-        for _ in 0..push.num_active {
-            let value2 = ContactInfo::new_localhost(&solana_sdk::pubkey::new_rand(), 0);
+        for k in 0..push.num_active {
+            let mut value2 = ContactInfo::new_localhost(&solana_sdk::pubkey::new_rand(), 0);
+            value2.gossip.set_port(1246 + k as u16);
             ping_cache
                 .lock()
                 .unwrap()

gossip/src/crds_value.rs

@@ -148,7 +148,8 @@ impl CrdsData {
         // the mainnet crds table.
         match kind {
             0 => CrdsData::ContactInfo(ContactInfo::new_rand(rng, pubkey)),
-            1 => CrdsData::LowestSlot(rng.gen(), LowestSlot::new_rand(rng, pubkey)),
+            // Index for LowestSlot is deprecated and should be zero.
+            1 => CrdsData::LowestSlot(0, LowestSlot::new_rand(rng, pubkey)),
             2 => CrdsData::SnapshotHashes(SnapshotHashes::new_rand(rng, pubkey)),
             3 => CrdsData::AccountsHashes(SnapshotHashes::new_rand(rng, pubkey)),
             4 => CrdsData::Version(Version::new_rand(rng, pubkey)),
@@ -864,7 +865,7 @@ mod test {
             let index = rng.gen_range(0, keys.len());
             CrdsValue::new_rand(&mut rng, Some(&keys[index]))
         })
-        .take(2048)
+        .take(1 << 12)
         .collect();
         let mut currents = HashMap::new();
         for value in filter_current(&values) {
@@ -888,10 +889,12 @@ mod test {
             }
         }
         assert_eq!(count, currents.len());
-        // Currently CrdsData::new_rand is only implemented for 5 different
-        // kinds and excludes EpochSlots, and so the unique labels cannot be
-        // more than (5 + MAX_VOTES) times number of keys.
-        assert!(currents.len() <= keys.len() * (5 + MAX_VOTES as usize));
+        // Currently CrdsData::new_rand is implemented for:
+        //   AccountsHashes, ContactInfo, LowestSlot, SnapshotHashes, Version
+        //   EpochSlots x MAX_EPOCH_SLOTS
+        //   Vote x MAX_VOTES
+        let num_kinds = 5 + MAX_VOTES as usize + MAX_EPOCH_SLOTS as usize;
+        assert!(currents.len() <= keys.len() * num_kinds);
     }
 
     #[test]