Admin Shred Receiver Change

core/benches/cluster_info.rs

@@ -79,7 +79,7 @@ fn broadcast_shreds_bench(bencher: &mut Bencher) {
             &cluster_info,
             &bank_forks,
             &SocketAddrSpace::Unspecified,
-            None,
+            &None,
         )
         .unwrap();
     });

core/benches/retransmit_stage.rs

@@ -121,7 +121,7 @@ fn bench_retransmitter(bencher: &mut Bencher) {
         shreds_receiver,
         Arc::default(), // solana_rpc::max_slots::MaxSlots
         None,
-        None,
+        Arc::new(RwLock::new(None)),
     );
 
     let mut index = 0;

core/src/broadcast_stage.rs

@@ -86,7 +86,7 @@ impl BroadcastStageType {
         blockstore: Arc<Blockstore>,
         bank_forks: Arc<RwLock<BankForks>>,
         shred_version: u16,
-        shred_receiver_addr: Option<SocketAddr>,
+        shred_receiver_address: Arc<RwLock<Option<SocketAddr>>>,
     ) -> BroadcastStage {
         match self {
             BroadcastStageType::Standard => BroadcastStage::new(
@@ -98,7 +98,7 @@ impl BroadcastStageType {
                 blockstore,
                 bank_forks,
                 StandardBroadcastRun::new(shred_version),
-                shred_receiver_addr,
+                shred_receiver_address,
             ),
 
             BroadcastStageType::FailEntryVerification => BroadcastStage::new(
@@ -110,7 +110,7 @@ impl BroadcastStageType {
                 blockstore,
                 bank_forks,
                 FailEntryVerificationBroadcastRun::new(shred_version),
-                None,
+                Arc::new(RwLock::new(None)),
             ),
 
             BroadcastStageType::BroadcastFakeShreds => BroadcastStage::new(
@@ -122,7 +122,7 @@ impl BroadcastStageType {
                 blockstore,
                 bank_forks,
                 BroadcastFakeShredsRun::new(0, shred_version),
-                None,
+                Arc::new(RwLock::new(None)),
             ),
 
             BroadcastStageType::BroadcastDuplicates(config) => BroadcastStage::new(
@@ -134,7 +134,7 @@ impl BroadcastStageType {
                 blockstore,
                 bank_forks,
                 BroadcastDuplicatesRun::new(shred_version, config.clone()),
-                None,
+                Arc::new(RwLock::new(None)),
             ),
         }
     }
@@ -155,7 +155,7 @@ trait BroadcastRun {
         cluster_info: &ClusterInfo,
         sock: &UdpSocket,
         bank_forks: &RwLock<BankForks>,
-        shred_receiver_addr: Option<SocketAddr>,
+        shred_receiver_address: &Arc<RwLock<Option<SocketAddr>>>,
     ) -> Result<()>;
     fn record(&mut self, receiver: &Mutex<RecordReceiver>, blockstore: &Blockstore) -> Result<()>;
 }
@@ -251,7 +251,7 @@ impl BroadcastStage {
         blockstore: Arc<Blockstore>,
         bank_forks: Arc<RwLock<BankForks>>,
         broadcast_stage_run: impl BroadcastRun + Send + 'static + Clone,
-        shred_receiver_addr: Option<SocketAddr>,
+        shred_receiver_address: Arc<RwLock<Option<SocketAddr>>>,
     ) -> Self {
         let (socket_sender, socket_receiver) = unbounded();
         let (blockstore_sender, blockstore_receiver) = unbounded();
@@ -283,6 +283,7 @@ impl BroadcastStage {
             let mut bs_transmit = broadcast_stage_run.clone();
             let cluster_info = cluster_info.clone();
             let bank_forks = bank_forks.clone();
+            let shred_receiver_address = shred_receiver_address.clone();
             let t = Builder::new()
                 .name("solBroadcastTx".to_string())
                 .spawn(move || loop {
@@ -291,7 +292,7 @@ impl BroadcastStage {
                         &cluster_info,
                         &sock,
                         &bank_forks,
-                        shred_receiver_addr,
+                        &shred_receiver_address,
                     );
                     let res = Self::handle_error(res, "solana-broadcaster-transmit");
                     if let Some(res) = res {
@@ -408,7 +409,7 @@ pub fn broadcast_shreds(
     cluster_info: &ClusterInfo,
     bank_forks: &RwLock<BankForks>,
     socket_addr_space: &SocketAddrSpace,
-    shred_receiver_addr: Option<SocketAddr>,
+    shred_receiver_address: &Option<SocketAddr>,
 ) -> Result<()> {
     let mut result = Ok(());
     let mut shred_select = Measure::start("shred_select");
@@ -418,7 +419,7 @@ pub fn broadcast_shreds(
     };
     let packets: Vec<_> = shreds
         .iter()
-        .filter_map(|s| Some((s.payload(), shred_receiver_addr?)))
+        .filter_map(|s| Some((s.payload(), (*shred_receiver_address)?)))
         .chain(
             shreds
                 .iter()

core/src/broadcast_stage/broadcast_duplicates_run.rs

@@ -10,7 +10,7 @@ use {
         signature::{Keypair, Signature, Signer},
         system_transaction,
     },
-    std::{collections::HashSet, net::SocketAddr},
+    std::collections::HashSet,
 };
 
 pub const MINIMUM_DUPLICATE_SLOT: Slot = 20;
@@ -266,7 +266,7 @@ impl BroadcastRun for BroadcastDuplicatesRun {
         cluster_info: &ClusterInfo,
         sock: &UdpSocket,
         bank_forks: &RwLock<BankForks>,
-        _shred_receiver_addr: Option<SocketAddr>,
+        _shred_receiver_addr: &Arc<RwLock<Option<SocketAddr>>>,
     ) -> Result<()> {
         let (shreds, _) = receiver.lock().unwrap().recv()?;
         if shreds.is_empty() {

core/src/broadcast_stage/broadcast_fake_shreds_run.rs

@@ -3,7 +3,6 @@ use {
     solana_entry::entry::Entry,
     solana_ledger::shred::{ProcessShredsStats, ReedSolomonCache, Shredder},
     solana_sdk::{hash::Hash, signature::Keypair},
-    std::net::SocketAddr,
 };
 
 #[derive(Clone)]
@@ -133,7 +132,7 @@ impl BroadcastRun for BroadcastFakeShredsRun {
         cluster_info: &ClusterInfo,
         sock: &UdpSocket,
         _bank_forks: &RwLock<BankForks>,
-        _shred_receiver_addr: Option<SocketAddr>,
+        _shred_receiver_addr: &Arc<RwLock<Option<SocketAddr>>>,
     ) -> Result<()> {
         for (data_shreds, batch_info) in receiver.lock().unwrap().iter() {
             let fake = batch_info.is_some();

core/src/broadcast_stage/fail_entry_verification_broadcast_run.rs

@@ -162,7 +162,7 @@ impl BroadcastRun for FailEntryVerificationBroadcastRun {
         cluster_info: &ClusterInfo,
         sock: &UdpSocket,
         bank_forks: &RwLock<BankForks>,
-        shred_receiver_addr: Option<SocketAddr>,
+        shred_receiver_address: &Arc<RwLock<Option<SocketAddr>>>,
     ) -> Result<()> {
         let (shreds, _) = receiver.lock().unwrap().recv()?;
         broadcast_shreds(
@@ -174,7 +174,7 @@ impl BroadcastRun for FailEntryVerificationBroadcastRun {
             cluster_info,
             bank_forks,
             cluster_info.socket_addr_space(),
-            shred_receiver_addr,
+            &shred_receiver_address.read().unwrap(),
         )
     }
     fn record(&mut self, receiver: &Mutex<RecordReceiver>, blockstore: &Blockstore) -> Result<()> {

core/src/broadcast_stage/standard_broadcast_run.rs

@@ -378,7 +378,7 @@ impl StandardBroadcastRun {
         shreds: Arc<Vec<Shred>>,
         broadcast_shred_batch_info: Option<BroadcastShredBatchInfo>,
         bank_forks: &RwLock<BankForks>,
-        shred_receiver_addr: Option<SocketAddr>,
+        shred_receiver_addr: &Option<SocketAddr>,
     ) -> Result<()> {
         trace!("Broadcasting {:?} shreds", shreds.len());
         let mut transmit_stats = TransmitShredsStats::default();
@@ -464,7 +464,7 @@ impl BroadcastRun for StandardBroadcastRun {
         cluster_info: &ClusterInfo,
         sock: &UdpSocket,
         bank_forks: &RwLock<BankForks>,
-        shred_receiver_addr: Option<SocketAddr>,
+        shred_receiver_address: &Arc<RwLock<Option<SocketAddr>>>,
     ) -> Result<()> {
         let (shreds, batch_info) = receiver.lock().unwrap().recv()?;
         self.broadcast(
@@ -473,7 +473,7 @@ impl BroadcastRun for StandardBroadcastRun {
             shreds,
             batch_info,
             bank_forks,
-            shred_receiver_addr,
+            &shred_receiver_address.read().unwrap(),
         )
     }
     fn record(&mut self, receiver: &Mutex<RecordReceiver>, blockstore: &Blockstore) -> Result<()> {