Ensure AncestorHashesSerice selects an open port (#21919) (#21997)

(cherry picked from commit 7f6fb6937a)

Co-authored-by: carllin <wumu727@gmail.com>

core/src/repair_service.rs

@@ -201,6 +201,7 @@ impl RepairService {
         blockstore: Arc<Blockstore>,
         exit: Arc<AtomicBool>,
         repair_socket: Arc<UdpSocket>,
+        ancestor_hashes_socket: Arc<UdpSocket>,
         repair_info: RepairInfo,
         verified_vote_receiver: VerifiedVoteReceiver,
         outstanding_requests: Arc<RwLock<OutstandingShredRepairs>>,
@@ -225,11 +226,10 @@ impl RepairService {
                 .unwrap()
         };
 
-        let ancestor_hashes_request_socket = Arc::new(UdpSocket::bind("0.0.0.0:0").unwrap());
         let ancestor_hashes_service = AncestorHashesService::new(
             exit,
             blockstore,
-            ancestor_hashes_request_socket,
+            ancestor_hashes_socket,
             repair_info,
             ancestor_hashes_replay_update_receiver,
         );

core/src/retransmit_stage.rs

@@ -433,6 +433,7 @@ impl RetransmitStage {
         cluster_info: Arc<ClusterInfo>,
         retransmit_sockets: Arc<Vec<UdpSocket>>,
         repair_socket: Arc<UdpSocket>,
+        ancestor_hashes_socket: Arc<UdpSocket>,
         verified_receiver: Receiver<Vec<PacketBatch>>,
         exit: Arc<AtomicBool>,
         cluster_slots_update_receiver: ClusterSlotsUpdateReceiver,
@@ -486,6 +487,7 @@ impl RetransmitStage {
             verified_receiver,
             retransmit_sender,
             repair_socket,
+            ancestor_hashes_socket,
             exit,
             repair_info,
             leader_schedule_cache,

core/src/tvu.rs

@@ -82,6 +82,7 @@ pub struct Sockets {
     pub repair: UdpSocket,
     pub retransmit: Vec<UdpSocket>,
     pub forwards: Vec<UdpSocket>,
+    pub ancestor_hashes_requests: UdpSocket,
 }
 
 #[derive(Default)]
@@ -148,11 +149,13 @@ impl Tvu {
             fetch: fetch_sockets,
             retransmit: retransmit_sockets,
             forwards: tvu_forward_sockets,
+            ancestor_hashes_requests: ancestor_hashes_socket,
         } = sockets;
 
         let (fetch_sender, fetch_receiver) = channel();
 
         let repair_socket = Arc::new(repair_socket);
+        let ancestor_hashes_socket = Arc::new(ancestor_hashes_socket);
         let fetch_sockets: Vec<Arc<UdpSocket>> = fetch_sockets.into_iter().map(Arc::new).collect();
         let forward_sockets: Vec<Arc<UdpSocket>> =
             tvu_forward_sockets.into_iter().map(Arc::new).collect();
@@ -187,6 +190,7 @@ impl Tvu {
             cluster_info.clone(),
             Arc::new(retransmit_sockets),
             repair_socket,
+            ancestor_hashes_socket,
             verified_receiver,
             exit.clone(),
             cluster_slots_update_receiver,
@@ -461,6 +465,7 @@ pub mod tests {
                     retransmit: target1.sockets.retransmit_sockets,
                     fetch: target1.sockets.tvu,
                     forwards: target1.sockets.tvu_forwards,
+                    ancestor_hashes_requests: target1.sockets.ancestor_hashes_requests,
                 }
             },
             blockstore,

core/src/validator.rs

@@ -828,6 +828,11 @@ impl Validator {
                     .iter()
                     .map(|s| s.try_clone().expect("Failed to clone TVU forwards Sockets"))
                     .collect(),
+                ancestor_hashes_requests: node
+                    .sockets
+                    .ancestor_hashes_requests
+                    .try_clone()
+                    .expect("Failed to clone ancestor_hashes_requests socket"),
             },
             blockstore.clone(),
             ledger_signal_receiver,

core/src/window_service.rs

@@ -454,6 +454,7 @@ impl WindowService {
         verified_receiver: CrossbeamReceiver<Vec<PacketBatch>>,
         retransmit_sender: Sender<Vec<Shred>>,
         repair_socket: Arc<UdpSocket>,
+        ancestor_hashes_socket: Arc<UdpSocket>,
         exit: Arc<AtomicBool>,
         repair_info: RepairInfo,
         leader_schedule_cache: Arc<LeaderScheduleCache>,
@@ -479,6 +480,7 @@ impl WindowService {
             blockstore.clone(),
             exit.clone(),
             repair_socket,
+            ancestor_hashes_socket,
             repair_info,
             verified_vote_receiver,
             outstanding_requests.clone(),