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()
@@ -635,7 +636,7 @@ pub mod test {
             blockstore.clone(),
             bank_forks,
             StandardBroadcastRun::new(0),
-            None,
+            Arc::new(RwLock::new(None)),
         );
 
         MockBroadcastStage {

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

@@ -180,10 +180,22 @@ impl StandardBroadcastRun {
         let brecv = Arc::new(Mutex::new(brecv));
 
         //data
-        let _ = self.transmit(&srecv, cluster_info, sock, bank_forks, None);
+        let _ = self.transmit(
+            &srecv,
+            cluster_info,
+            sock,
+            bank_forks,
+            &Arc::new(RwLock::new(None)),
+        );
         let _ = self.record(&brecv, blockstore);
         //coding
-        let _ = self.transmit(&srecv, cluster_info, sock, bank_forks, None);
+        let _ = self.transmit(
+            &srecv,
+            cluster_info,
+            sock,
+            bank_forks,
+            &Arc::new(RwLock::new(None)),
+        );
         let _ = self.record(&brecv, blockstore);
         Ok(())
     }
@@ -378,7 +390,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 +476,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 +485,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<()> {

core/src/retransmit_stage.rs

@@ -178,7 +178,7 @@ fn retransmit(
     packet_hasher: &mut PacketHasher,
     max_slots: &MaxSlots,
     rpc_subscriptions: Option<&RpcSubscriptions>,
-    shred_receiver_addr: Option<SocketAddr>,
+    shred_receiver_address: &Arc<RwLock<Option<SocketAddr>>>,
 ) -> Result<(), RecvTimeoutError> {
     const RECV_TIMEOUT: Duration = Duration::from_secs(1);
     let mut shreds = shreds_receiver.recv_timeout(RECV_TIMEOUT)?;
@@ -257,7 +257,7 @@ fn retransmit(
                     socket_addr_space,
                     &sockets[index % sockets.len()],
                     stats,
-                    shred_receiver_addr,
+                    &shred_receiver_address.read().unwrap(),
                 );
                 (key.slot(), root_distance, num_nodes)
             })
@@ -277,7 +277,7 @@ fn retransmit(
                         socket_addr_space,
                         &sockets[index % sockets.len()],
                         stats,
-                        shred_receiver_addr,
+                        &shred_receiver_address.read().unwrap(),
                     );
                     (key.slot(), root_distance, num_nodes)
                 })
@@ -301,14 +301,14 @@ fn retransmit_shred(
     socket_addr_space: &SocketAddrSpace,
     socket: &UdpSocket,
     stats: &RetransmitStats,
-    shred_receiver_addr: Option<SocketAddr>,
+    shred_receiver_addr: &Option<SocketAddr>,
 ) -> (/*root_distance:*/ usize, /*num_nodes:*/ usize) {
     let mut compute_turbine_peers = Measure::start("turbine_start");
     let (root_distance, mut addrs) =
         cluster_nodes.get_retransmit_addrs(slot_leader, key, root_bank, DATA_PLANE_FANOUT);
 
     if let Some(addr) = shred_receiver_addr {
-        addrs.push(addr);
+        addrs.push(*addr);
     }
 
     let addrs: Vec<_> = addrs
@@ -360,7 +360,7 @@ pub fn retransmitter(
     shreds_receiver: Receiver<Vec</*shred:*/ Vec<u8>>>,
     max_slots: Arc<MaxSlots>,
     rpc_subscriptions: Option<Arc<RpcSubscriptions>>,
-    shred_receiver_addr: Option<SocketAddr>,
+    shred_receiver_addr: Arc<RwLock<Option<SocketAddr>>>,
 ) -> JoinHandle<()> {
     let cluster_nodes_cache = ClusterNodesCache::<RetransmitStage>::new(
         CLUSTER_NODES_CACHE_NUM_EPOCH_CAP,
@@ -393,7 +393,7 @@ pub fn retransmitter(
                 &mut packet_hasher,
                 &max_slots,
                 rpc_subscriptions.as_deref(),
-                shred_receiver_addr,
+                &shred_receiver_addr,
             ) {
                 Ok(()) => (),
                 Err(RecvTimeoutError::Timeout) => (),
@@ -416,7 +416,7 @@ impl RetransmitStage {
         retransmit_receiver: Receiver<Vec</*shred:*/ Vec<u8>>>,
         max_slots: Arc<MaxSlots>,
         rpc_subscriptions: Option<Arc<RpcSubscriptions>>,
-        shred_receiver_addr: Option<SocketAddr>,
+        shred_receiver_addr: Arc<RwLock<Option<SocketAddr>>>,
     ) -> Self {
         let retransmit_thread_handle = retransmitter(
             retransmit_sockets,

core/src/tpu.rs

@@ -116,7 +116,7 @@ impl Tpu {
         block_engine_config: Arc<Mutex<BlockEngineConfig>>,
         relayer_config: Arc<Mutex<RelayerConfig>>,
         tip_manager_config: TipManagerConfig,
-        shred_receiver_address: Option<SocketAddr>,
+        shred_receiver_address: Arc<RwLock<Option<SocketAddr>>>,
         shared_staked_nodes_overrides: Arc<RwLock<HashMap<Pubkey, u64>>>,
         tpu_enable_udp: bool,
         preallocated_bundle_cost: u64,

core/src/tvu.rs

@@ -131,7 +131,7 @@ impl Tvu {
         log_messages_bytes_limit: Option<usize>,
         connection_cache: &Arc<ConnectionCache>,
         prioritization_fee_cache: &Arc<PrioritizationFeeCache>,
-        shred_receiver_addr: Option<SocketAddr>,
+        shred_receiver_addr: Arc<RwLock<Option<SocketAddr>>>,
     ) -> Result<Self, String> {
         let TvuSockets {
             repair: repair_socket,
@@ -458,7 +458,7 @@ pub mod tests {
             None,
             &Arc::new(ConnectionCache::default()),
             &_ignored_prioritization_fee_cache,
-            None,
+            Arc::new(RwLock::new(None)),
         )
         .expect("assume success");
         exit.store(true, Ordering::Relaxed);

core/src/validator.rs

@@ -182,7 +182,8 @@ pub struct ValidatorConfig {
     pub replay_slots_concurrently: bool,
     pub relayer_config: Arc<Mutex<RelayerConfig>>,
     pub block_engine_config: Arc<Mutex<BlockEngineConfig>>,
-    pub shred_receiver_address: Option<SocketAddr>,
+    // Using Option inside RwLock is ugly, but only convenient way to allow toggle on/off
+    pub shred_receiver_address: Arc<RwLock<Option<SocketAddr>>>,
     pub tip_manager_config: TipManagerConfig,
     pub preallocated_bundle_cost: u64,
 }
@@ -252,7 +253,7 @@ impl Default for ValidatorConfig {
             replay_slots_concurrently: false,
             relayer_config: Arc::new(Mutex::new(RelayerConfig::default())),
             block_engine_config: Arc::new(Mutex::new(BlockEngineConfig::default())),
-            shred_receiver_address: None,
+            shred_receiver_address: Arc::new(RwLock::new(None)),
             tip_manager_config: TipManagerConfig::default(),
             preallocated_bundle_cost: u64::default(),
         }
@@ -1005,7 +1006,7 @@ impl Validator {
             config.runtime_config.log_messages_bytes_limit,
             &connection_cache,
             &prioritization_fee_cache,
-            config.shred_receiver_address,
+            config.shred_receiver_address.clone(),
         )?;
 
         let tpu = Tpu::new(
@@ -1044,7 +1045,7 @@ impl Validator {
             config.block_engine_config.clone(),
             config.relayer_config.clone(),
             config.tip_manager_config.clone(),
-            config.shred_receiver_address,
+            config.shred_receiver_address.clone(),
             config.staked_nodes_overrides.clone(),
             tpu_enable_udp,
             config.preallocated_bundle_cost,

local-cluster/src/validator_configs.rs

@@ -69,7 +69,7 @@ pub fn safe_clone_config(config: &ValidatorConfig) -> ValidatorConfig {
         replay_slots_concurrently: config.replay_slots_concurrently,
         relayer_config: config.relayer_config.clone(),
         block_engine_config: config.block_engine_config.clone(),
-        shred_receiver_address: config.shred_receiver_address,
+        shred_receiver_address: config.shred_receiver_address.clone(),
         tip_manager_config: config.tip_manager_config.clone(),
         preallocated_bundle_cost: config.preallocated_bundle_cost,
     }