net-utils (& others): swap tests to non-overlapping port allocation

core/src/repair/ancestor_hashes_service.rs

@@ -923,7 +923,7 @@ mod test {
             blockstore::make_many_slot_entries, get_tmp_ledger_path,
             get_tmp_ledger_path_auto_delete, shred::Nonce,
         },
-        solana_net_utils::bind_to_unspecified,
+        solana_net_utils::sockets::bind_to_localhost_unique,
         solana_perf::packet::Packet,
         solana_runtime::bank_forks::BankForks,
         solana_signer::Signer,
@@ -1357,7 +1357,8 @@ mod test {
     impl ManageAncestorHashesState {
         fn new(bank_forks: Arc<RwLock<BankForks>>) -> Self {
             let ancestor_hashes_request_statuses = Arc::new(DashMap::new());
-            let ancestor_hashes_request_socket = Arc::new(bind_to_unspecified().unwrap());
+            let ancestor_hashes_request_socket =
+                Arc::new(bind_to_localhost_unique().expect("should bind"));
             let epoch_schedule = bank_forks
                 .read()
                 .unwrap()

core/src/repair/repair_service.rs

@@ -1276,9 +1276,8 @@ mod test {
             get_tmp_ledger_path_auto_delete,
             shred::max_ticks_per_n_shreds,
         },
-        solana_net_utils::{
-            bind_to_unspecified,
-            sockets::{bind_to, localhost_port_range_for_tests},
+        solana_net_utils::sockets::{
+            bind_to, bind_to_localhost_unique, localhost_port_range_for_tests,
         },
         solana_runtime::bank::Bank,
         solana_signer::Signer,
@@ -1662,7 +1661,7 @@ mod test {
         };
         let mut duplicate_slot_repair_statuses = HashMap::new();
         let dead_slot = 9;
-        let receive_socket = &bind_to_unspecified().unwrap();
+        let receive_socket = &bind_to_localhost_unique().expect("should bind");
         let duplicate_status = DuplicateSlotRepairStatus {
             correct_ancestor_to_repair: (dead_slot, Hash::default()),
             start_ts: u64::MAX,
@@ -1691,7 +1690,7 @@ mod test {
             &blockstore,
             &serve_repair,
             &mut RepairStats::default(),
-            &bind_to_unspecified().unwrap(),
+            &bind_to_localhost_unique().expect("should bind"),
             &None,
             &RwLock::new(OutstandingRequests::default()),
             &identity_keypair,
@@ -1717,7 +1716,7 @@ mod test {
             &blockstore,
             &serve_repair,
             &mut RepairStats::default(),
-            &bind_to_unspecified().unwrap(),
+            &bind_to_localhost_unique().expect("should bind"),
             &None,
             &RwLock::new(OutstandingRequests::default()),
             &identity_keypair,
@@ -1736,7 +1735,7 @@ mod test {
             &blockstore,
             &serve_repair,
             &mut RepairStats::default(),
-            &bind_to_unspecified().unwrap(),
+            &bind_to_localhost_unique().expect("should bind"),
             &None,
             &RwLock::new(OutstandingRequests::default()),
             &identity_keypair,
@@ -1751,7 +1750,10 @@ mod test {
         let bank_forks = BankForks::new_rw_arc(bank);
         let dummy_addr = Some((
             Pubkey::default(),
-            bind_to_unspecified().unwrap().local_addr().unwrap(),
+            bind_to_localhost_unique()
+                .expect("should bind")
+                .local_addr()
+                .unwrap(),
         ));
         let cluster_info = Arc::new(new_test_cluster_info());
         let ledger_path = get_tmp_ledger_path_auto_delete!();

net-utils/src/sockets.rs

@@ -1,22 +1,20 @@
+#[cfg(feature = "dev-context-only-utils")]
+use tokio::net::UdpSocket as TokioUdpSocket;
 use {
     crate::PortRange,
     log::warn,
     socket2::{Domain, SockAddr, Socket, Type},
     std::{
         io,
-        net::{IpAddr, SocketAddr, TcpListener, UdpSocket},
+        net::{IpAddr, Ipv4Addr, SocketAddr, TcpListener, UdpSocket},
         sync::atomic::{AtomicU16, Ordering},
     },
 };
-#[cfg(feature = "dev-context-only-utils")]
-use {std::net::Ipv4Addr, tokio::net::UdpSocket as TokioUdpSocket};
 // base port for deconflicted allocations
 const BASE_PORT: u16 = 5000;
 // how much to allocate per individual process.
 // we expect to have at most 64 concurrent tests in CI at any moment on a given host.
 const SLICE_PER_PROCESS: u16 = (u16::MAX - BASE_PORT) / 64;
-/// Retrieve a free 20-port slice for unit tests
-///
 /// When running under nextest, this will try to provide
 /// a unique slice of port numbers (assuming no other nextest processes
 /// are running on the same host) based on NEXTEST_TEST_GLOBAL_SLOT variable
@@ -25,9 +23,9 @@ const SLICE_PER_PROCESS: u16 = (u16::MAX - BASE_PORT) / 64;
 /// When running without nextest, this will only bump an atomic and eventually
 /// panic when it runs out of port numbers to assign.
 #[allow(clippy::arithmetic_side_effects)]
-pub fn localhost_port_range_for_tests() -> (u16, u16) {
+pub fn unique_port_range_for_tests(size: u16) -> (u16, u16) {
     static SLICE: AtomicU16 = AtomicU16::new(0);
-    let offset = SLICE.fetch_add(20, Ordering::Relaxed);
+    let offset = SLICE.fetch_add(size, Ordering::Relaxed);
     let start = offset
         + match std::env::var("NEXTEST_TEST_GLOBAL_SLOT") {
             Ok(slot) => {
@@ -40,8 +38,29 @@ pub fn localhost_port_range_for_tests() -> (u16, u16) {
             }
             Err(_) => BASE_PORT,
         };
-    assert!(start < u16::MAX - 20, "ran out of port numbers!");
-    (start, start + 20)
+    assert!(start < u16::MAX - size, "ran out of port numbers!");
+    (start, start + size)
+}
+
+/// Retrieve a free 20-port slice for unit tests
+///
+/// When running under nextest, this will try to provide
+/// a unique slice of port numbers (assuming no other nextest processes
+/// are running on the same host) based on NEXTEST_TEST_GLOBAL_SLOT variable
+/// The port ranges will be reused following nextest logic.
+///
+/// When running without nextest, this will only bump an atomic and eventually
+/// panic when it runs out of port numbers to assign.
+pub fn localhost_port_range_for_tests() -> (u16, u16) {
+    unique_port_range_for_tests(20)
+}
+
+/// Bind a `UdpSocket` to a unique port.
+pub fn bind_to_localhost_unique() -> io::Result<UdpSocket> {
+    bind_to(
+        IpAddr::V4(Ipv4Addr::LOCALHOST),
+        localhost_port_range_for_tests().0,
+    )
 }
 
 pub fn bind_gossip_port_in_range(
@@ -246,7 +265,11 @@ pub async fn bind_to_async(ip_addr: IpAddr, port: u16) -> io::Result<TokioUdpSoc
 
 #[cfg(feature = "dev-context-only-utils")]
 pub async fn bind_to_localhost_async() -> io::Result<TokioUdpSocket> {
-    bind_to_async(IpAddr::V4(Ipv4Addr::LOCALHOST), 0).await
+    bind_to_async(
+        IpAddr::V4(Ipv4Addr::LOCALHOST),
+        localhost_port_range_for_tests().0,
+    )
+    .await
 }
 
 #[cfg(feature = "dev-context-only-utils")]

rpc-test/tests/rpc.rs

@@ -10,7 +10,7 @@ use {
     solana_commitment_config::CommitmentConfig,
     solana_hash::Hash,
     solana_keypair::Keypair,
-    solana_net_utils::bind_to_unspecified,
+    solana_net_utils::sockets::bind_to_localhost_unique,
     solana_pubkey::Pubkey,
     solana_pubsub_client::nonblocking::pubsub_client::PubsubClient,
     solana_rent::Rent,
@@ -290,7 +290,7 @@ fn test_rpc_subscriptions() {
     let test_validator =
         TestValidator::with_no_fees_udp(alice.pubkey(), None, SocketAddrSpace::Unspecified);
 
-    let transactions_socket = bind_to_unspecified().unwrap();
+    let transactions_socket = bind_to_localhost_unique().expect("should bind");
     transactions_socket.connect(test_validator.tpu()).unwrap();
 
     let rpc_client = RpcClient::new(test_validator.rpc_url());

streamer/src/nonblocking/sendmmsg.rs

@@ -61,7 +61,7 @@ mod tests {
             sendmmsg::SendPktsError,
         },
         assert_matches::assert_matches,
-        solana_net_utils::sockets::{bind_to_localhost_async, bind_to_unspecified_async},
+        solana_net_utils::sockets::bind_to_localhost_async,
         solana_packet::PACKET_DATA_SIZE,
         std::{
             io::ErrorKind,
@@ -177,7 +177,7 @@ mod tests {
         ];
         let dest_refs: Vec<_> = vec![&ip4, &ip6, &ip4];
 
-        let sender = bind_to_unspecified_async().await.expect("bind");
+        let sender = bind_to_localhost_async().await.expect("should bind");
         let res = batch_send(&sender, &packet_refs[..]).await;
         assert_matches!(res, Err(SendPktsError::IoError(_, /*num_failed*/ 1)));
         let res = multi_target_send(&sender, &packets[0], &dest_refs).await;
@@ -189,7 +189,7 @@ mod tests {
         let packets: Vec<_> = (0..5).map(|_| vec![0u8; PACKET_DATA_SIZE]).collect();
         let ipv4local = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 8080);
         let ipv4broadcast = SocketAddr::new(IpAddr::V4(Ipv4Addr::BROADCAST), 8080);
-        let sender = bind_to_unspecified_async().await.expect("bind");
+        let sender = bind_to_localhost_async().await.expect("should bind");
 
         // test intermediate failures for batch_send
         let packet_refs: Vec<_> = vec![

streamer/src/sendmmsg.rs

@@ -247,7 +247,7 @@ mod tests {
             sendmmsg::{batch_send, multi_target_send, SendPktsError},
         },
         assert_matches::assert_matches,
-        solana_net_utils::{bind_to_localhost, bind_to_unspecified},
+        solana_net_utils::{bind_to_localhost, sockets::bind_to_localhost_unique},
         solana_packet::PACKET_DATA_SIZE,
         std::{
             io::ErrorKind,
@@ -362,7 +362,7 @@ mod tests {
         ];
         let dest_refs: Vec<_> = vec![&ip4, &ip6, &ip4];
 
-        let sender = bind_to_unspecified().expect("bind");
+        let sender = bind_to_localhost_unique().expect("should bind");
         let res = batch_send(&sender, packet_refs);
         assert_matches!(res, Err(SendPktsError::IoError(_, /*num_failed*/ 1)));
         let res = multi_target_send(&sender, &packets[0], &dest_refs);
@@ -374,7 +374,7 @@ mod tests {
         let packets: Vec<_> = (0..5).map(|_| vec![0u8; PACKET_DATA_SIZE]).collect();
         let ipv4local = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 8080);
         let ipv4broadcast = SocketAddr::new(IpAddr::V4(Ipv4Addr::BROADCAST), 8080);
-        let sender = bind_to_unspecified().expect("bind");
+        let sender = bind_to_localhost_unique().expect("should bind");
 
         // test intermediate failures for batch_send
         let packet_refs: Vec<_> = vec![

turbine/src/broadcast_stage/standard_broadcast_run.rs

@@ -499,7 +499,7 @@ mod test {
             get_tmp_ledger_path,
             shred::{max_ticks_per_n_shreds, DATA_SHREDS_PER_FEC_BLOCK},
         },
-        solana_net_utils::bind_to_unspecified,
+        solana_net_utils::sockets::bind_to_localhost_unique,
         solana_runtime::bank::Bank,
         solana_signer::Signer,
         solana_streamer::socket::SocketAddrSpace,
@@ -531,7 +531,7 @@ mod test {
             leader_keypair.clone(),
             SocketAddrSpace::Unspecified,
         ));
-        let socket = bind_to_unspecified().unwrap();
+        let socket = bind_to_localhost_unique().expect("should bind");
         let mut genesis_config = create_genesis_config(10_000).genesis_config;
         genesis_config.ticks_per_slot = max_ticks_per_n_shreds(num_shreds_per_slot, None) + 1;
 

validator/src/admin_rpc_service.rs

@@ -961,7 +961,7 @@ mod tests {
                 create_genesis_config, create_genesis_config_with_leader, GenesisConfigInfo,
             },
         },
-        solana_net_utils::bind_to_unspecified,
+        solana_net_utils::sockets::bind_to_localhost_unique,
         solana_program_option::COption,
         solana_program_pack::Pack,
         solana_pubkey::Pubkey,
@@ -1030,7 +1030,7 @@ mod tests {
                     vote_account,
                     repair_whitelist,
                     notifies: Arc::new(RwLock::new(KeyUpdaters::default())),
-                    repair_socket: Arc::new(bind_to_unspecified().unwrap()),
+                    repair_socket: Arc::new(bind_to_localhost_unique().expect("should bind")),
                     outstanding_repair_requests: Arc::<
                         RwLock<repair_service::OutstandingShredRepairs>,
                     >::default(),