revises turbine peers shuffling order

core/src/broadcast_stage.rs

@@ -17,7 +17,7 @@ use {
         Sender as CrossbeamSender,
     },
     itertools::Itertools,
-    solana_gossip::cluster_info::{ClusterInfo, ClusterInfoError},
+    solana_gossip::cluster_info::{ClusterInfo, ClusterInfoError, DATA_PLANE_FANOUT},
     solana_ledger::{blockstore::Blockstore, shred::Shred},
     solana_measure::measure::Measure,
     solana_metrics::{inc_new_counter_error, inc_new_counter_info},
@@ -33,6 +33,7 @@ use {
     },
     std::{
         collections::HashMap,
+        iter::repeat,
         net::UdpSocket,
         sync::{
             atomic::{AtomicBool, Ordering},
@@ -412,8 +413,6 @@ pub fn broadcast_shreds(
 ) -> Result<()> {
     let mut result = Ok(());
     let mut shred_select = Measure::start("shred_select");
-    // Only the leader broadcasts shreds.
-    let leader = cluster_info.id();
     let (root_bank, working_bank) = {
         let bank_forks = bank_forks.read().unwrap();
         (bank_forks.root_bank(), bank_forks.working_bank())
@@ -427,12 +426,13 @@ pub fn broadcast_shreds(
                 cluster_nodes_cache.get(slot, &root_bank, &working_bank, cluster_info);
             update_peer_stats(&cluster_nodes, last_datapoint_submit);
             let root_bank = root_bank.clone();
-            shreds.filter_map(move |shred| {
-                let seed = shred.seed(leader, &root_bank);
-                let node = cluster_nodes.get_broadcast_peer(seed)?;
-                socket_addr_space
-                    .check(&node.tvu)
-                    .then(|| (&shred.payload, node.tvu))
+            shreds.flat_map(move |shred| {
+                repeat(&shred.payload).zip(cluster_nodes.get_broadcast_addrs(
+                    shred,
+                    &root_bank,
+                    DATA_PLANE_FANOUT,
+                    socket_addr_space,
+                ))
             })
         })
         .collect();

core/src/broadcast_stage/broadcast_duplicates_run.rs

@@ -3,6 +3,7 @@ use {
     crate::cluster_nodes::ClusterNodesCache,
     itertools::Itertools,
     solana_entry::entry::Entry,
+    solana_gossip::cluster_info::DATA_PLANE_FANOUT,
     solana_ledger::shred::Shredder,
     solana_sdk::{
         hash::Hash,
@@ -246,6 +247,11 @@ impl BroadcastRun for BroadcastDuplicatesRun {
             (bank_forks.root_bank(), bank_forks.working_bank())
         };
         let self_pubkey = cluster_info.id();
+        let nodes: Vec<_> = cluster_info
+            .all_peers()
+            .into_iter()
+            .map(|(node, _)| node)
+            .collect();
 
         // Creat cluster partition.
         let cluster_partition: HashSet<Pubkey> = {
@@ -273,8 +279,11 @@ impl BroadcastRun for BroadcastDuplicatesRun {
         let packets: Vec<_> = shreds
             .iter()
             .filter_map(|shred| {
-                let seed = shred.seed(self_pubkey, &root_bank);
-                let node = cluster_nodes.get_broadcast_peer(seed)?;
+                let addr = cluster_nodes
+                    .get_broadcast_addrs(shred, &root_bank, DATA_PLANE_FANOUT, socket_addr_space)
+                    .first()
+                    .copied()?;
+                let node = nodes.iter().find(|node| node.tvu == addr)?;
                 if !socket_addr_space.check(&node.tvu) {
                     return None;
                 }