reorg of clusterinfo port init logic

Author: alexpyattaev

gossip/src/cluster_info.rs

@@ -51,10 +51,10 @@ use {
     solana_keypair::{signable::Signable, Keypair},
     solana_ledger::shred::Shred,
     solana_net_utils::{
-        bind_common_in_range_with_config, bind_common_with_config, bind_in_range,
-        bind_in_range_with_config, bind_more_with_config, bind_to_unspecified,
-        bind_two_in_range_with_offset_and_config, find_available_port_in_range,
-        multi_bind_in_range_with_config, PortRange, SocketConfig, VALIDATOR_PORT_RANGE,
+        bind_in_range, bind_in_range_with_config, bind_more_with_config, bind_to_unspecified,
+        bind_to_with_config, find_available_port_in_range, multi_bind_in_range_with_config,
+        sockets::{get_gossip_port, localhost_port_range_for_tests},
+        PortRange, SocketConfig, VALIDATOR_PORT_RANGE,
     },
     solana_perf::{
         data_budget::DataBudget,
@@ -90,7 +90,7 @@ use {
         rc::Rc,
         result::Result,
         sync::{
-            atomic::{AtomicBool, AtomicU16, Ordering},
+            atomic::{AtomicBool, Ordering},
             Arc, Mutex, RwLock, RwLockReadGuard,
         },
         thread::{sleep, Builder, JoinHandle},
@@ -2331,7 +2331,7 @@ impl ClusterInfo {
     ) -> (ContactInfo, UdpSocket, Option<TcpListener>) {
         let bind_ip_addr = IpAddr::V4(Ipv4Addr::UNSPECIFIED);
         let (port, (gossip_socket, ip_echo)) =
-            Node::get_gossip_port(gossip_addr, VALIDATOR_PORT_RANGE, bind_ip_addr);
+            get_gossip_port(gossip_addr, VALIDATOR_PORT_RANGE, bind_ip_addr);
         let contact_info =
             Self::gossip_contact_info(id, SocketAddr::new(gossip_addr.ip(), port), shred_version);
 
@@ -2405,70 +2405,23 @@ pub struct Node {
     pub sockets: Sockets,
 }
 
-const NODE_BASE_PORT: u16 = 40000;
 impl Node {
     /// create localhost node for tests
     pub fn new_localhost() -> Self {
         let pubkey = solana_pubkey::new_rand();
         Self::new_localhost_with_pubkey(&pubkey)
     }
 
-    fn localhost_port_range_for_tests() -> (u16, u16) {
-        static SLICE: AtomicU16 = AtomicU16::new(NODE_BASE_PORT);
-        let start = match std::env::var("NEXTEST_TEST_GLOBAL_SLOT") {
-            Ok(slot) => {
-                let slot: u16 = slot.parse().unwrap();
-                NODE_BASE_PORT + slot * 25
-            }
-            Err(_) => SLICE.fetch_add(100, Ordering::Relaxed),
-        };
-        assert!(start < 65500, "ran out of port numbers!");
-        (start, start + 25)
-    }
-
     /// create localhost node for tests with provided pubkey
     /// unlike the public IP version, this will also bind RPC sockets.
     pub fn new_localhost_with_pubkey(pubkey: &Pubkey) -> Self {
         let addr = IpAddr::V4(Ipv4Addr::LOCALHOST);
-        let port_range = Self::localhost_port_range_for_tests();
+        let port_range = localhost_port_range_for_tests();
         let gossip_port = find_available_port_in_range(addr, port_range)
             .expect("At least one open port should be available");
-        let config = NodeConfig {
-            gossip_addr: SocketAddr::new(addr, gossip_port),
-            port_range,
-            bind_ip_addr: addr,
-            public_tpu_addr: None,
-            public_tpu_forwards_addr: None,
-            num_tvu_receive_sockets: NonZero::new(1).unwrap(),
-            num_tvu_retransmit_sockets: NonZero::new(1).unwrap(),
-            num_quic_endpoints: NonZero::new(DEFAULT_QUIC_ENDPOINTS)
-                .expect("Number of QUIC endpoints can not be zero"),
-        };
-        let mut node = Self::new_with_external_ip(pubkey, config);
-        let rpc_port = find_available_port_in_range(addr, port_range).unwrap();
-        let rpc_pubsub_port = find_available_port_in_range(addr, port_range).unwrap();
-        node.info.set_rpc((addr, rpc_port)).unwrap();
-        node.info.set_rpc_pubsub((addr, rpc_pubsub_port)).unwrap();
-        node
-    }
-
-    fn get_gossip_port(
-        gossip_addr: &SocketAddr,
-        port_range: PortRange,
-        bind_ip_addr: IpAddr,
-    ) -> (u16, (UdpSocket, TcpListener)) {
-        let config = SocketConfig::default();
-        if gossip_addr.port() != 0 {
-            (
-                gossip_addr.port(),
-                bind_common_with_config(bind_ip_addr, gossip_addr.port(), config).unwrap_or_else(
-                    |e| panic!("gossip_addr bind_to port {}: {}", gossip_addr.port(), e),
-                ),
-            )
-        } else {
-            bind_common_in_range_with_config(bind_ip_addr, port_range, config)
-                .expect("Failed to bind")
-        }
+        let gossip_addr = SocketAddr::new(addr, gossip_port);
+        #[allow(deprecated)] // new_single_bind will be merged in this function
+        Self::new_single_bind(pubkey, &gossip_addr, port_range, addr)
     }
 
     fn bind_with_config(
@@ -2486,139 +2439,25 @@ impl Node {
         port_range: PortRange,
         bind_ip_addr: IpAddr,
     ) -> Self {
-        let (gossip_port, (gossip, ip_echo)) =
-            Self::get_gossip_port(gossip_addr, port_range, bind_ip_addr);
-
-        let socket_config = SocketConfig::default();
-        let socket_config_reuseport = SocketConfig::default().reuseport(true);
-        let (tvu_port, tvu) = Self::bind_with_config(bind_ip_addr, port_range, socket_config);
-        let (tvu_quic_port, tvu_quic) =
-            Self::bind_with_config(bind_ip_addr, port_range, socket_config);
-        let ((tpu_port, tpu), (_tpu_quic_port, tpu_quic)) =
-            bind_two_in_range_with_offset_and_config(
-                bind_ip_addr,
-                port_range,
-                QUIC_PORT_OFFSET,
-                socket_config,
-                socket_config_reuseport,
-            )
-            .unwrap();
-        let tpu_quic: Vec<UdpSocket> =
-            bind_more_with_config(tpu_quic, DEFAULT_QUIC_ENDPOINTS, socket_config_reuseport)
-                .unwrap();
-
-        let ((tpu_forwards_port, tpu_forwards), (_tpu_forwards_quic_port, tpu_forwards_quic)) =
-            bind_two_in_range_with_offset_and_config(
-                bind_ip_addr,
-                port_range,
-                QUIC_PORT_OFFSET,
-                socket_config,
-                socket_config_reuseport,
-            )
-            .unwrap();
-        let tpu_forwards_quic = bind_more_with_config(
-            tpu_forwards_quic,
-            DEFAULT_QUIC_ENDPOINTS,
-            socket_config_reuseport,
-        )
-        .unwrap();
-
-        let (tpu_vote_port, tpu_vote) =
-            Self::bind_with_config(bind_ip_addr, port_range, socket_config);
-        let (tpu_vote_quic_port, tpu_vote_quic) =
-            Self::bind_with_config(bind_ip_addr, port_range, socket_config);
-        let tpu_vote_quic: Vec<UdpSocket> = bind_more_with_config(
-            tpu_vote_quic,
-            DEFAULT_QUIC_ENDPOINTS,
-            socket_config_reuseport,
-        )
-        .unwrap();
-
-        let (_, retransmit_socket) =
-            Self::bind_with_config(bind_ip_addr, port_range, socket_config);
-        let (_, repair) = Self::bind_with_config(bind_ip_addr, port_range, socket_config);
-        let (_, repair_quic) = Self::bind_with_config(bind_ip_addr, port_range, socket_config);
-        let (serve_repair_port, serve_repair) =
-            Self::bind_with_config(bind_ip_addr, port_range, socket_config);
-        let (serve_repair_quic_port, serve_repair_quic) =
-            Self::bind_with_config(bind_ip_addr, port_range, socket_config);
-        let (_, broadcast) = Self::bind_with_config(bind_ip_addr, port_range, socket_config);
-        let (_, ancestor_hashes_requests) =
-            Self::bind_with_config(bind_ip_addr, port_range, socket_config);
-        let (_, ancestor_hashes_requests_quic) =
-            Self::bind_with_config(bind_ip_addr, port_range, socket_config);
-
+        let config = NodeConfig {
+            gossip_addr: *gossip_addr,
+            port_range,
+            bind_ip_addr,
+            public_tpu_addr: None,
+            public_tpu_forwards_addr: None,
+            num_tvu_receive_sockets: NonZero::new(1).unwrap(),
+            num_tvu_retransmit_sockets: NonZero::new(1).unwrap(),
+            num_quic_endpoints: NonZero::new(DEFAULT_QUIC_ENDPOINTS)
+                .expect("Number of QUIC endpoints can not be zero"),
+        };
+        let mut node = Self::new_with_external_ip(pubkey, config);
         let rpc_port = find_available_port_in_range(bind_ip_addr, port_range).unwrap();
         let rpc_pubsub_port = find_available_port_in_range(bind_ip_addr, port_range).unwrap();
-
-        // These are client sockets, so the port is set to be 0 because it must be ephimeral.
-        let tpu_vote_forwards_client =
-            Self::bind_with_config(bind_ip_addr, 0, socket_config).unwrap();
-
-        let addr = gossip_addr.ip();
-        let mut info = ContactInfo::new(
-            *pubkey,
-            timestamp(), // wallclock
-            0u16,        // shred_version
-        );
-        macro_rules! set_socket {
-            ($method:ident, $port:ident, $name:literal) => {
-                info.$method((addr, $port)).expect(&format!(
-                    "Operator must spin up node with valid {} address",
-                    $name
-                ))
-            };
-            ($method:ident, $protocol:ident, $port:ident, $name:literal) => {{
-                info.$method(contact_info::Protocol::$protocol, (addr, $port))
-                    .expect(&format!(
-                        "Operator must spin up node with valid {} address",
-                        $name
-                    ))
-            }};
-        }
-        set_socket!(set_gossip, gossip_port, "gossip");
-        set_socket!(set_tvu, UDP, tvu_port, "TVU");
-        set_socket!(set_tvu, QUIC, tvu_quic_port, "TVU QUIC");
-        set_socket!(set_tpu, tpu_port, "TPU");
-        set_socket!(set_tpu_forwards, tpu_forwards_port, "TPU-forwards");
-        set_socket!(set_tpu_vote, UDP, tpu_vote_port, "TPU-vote");
-        set_socket!(set_tpu_vote, QUIC, tpu_vote_quic_port, "TPU-vote QUIC");
-        set_socket!(set_rpc, rpc_port, "RPC");
-        set_socket!(set_rpc_pubsub, rpc_pubsub_port, "RPC-pubsub");
-        set_socket!(set_serve_repair, UDP, serve_repair_port, "serve-repair");
-        set_socket!(
-            set_serve_repair,
-            QUIC,
-            serve_repair_quic_port,
-            "serve-repair QUIC"
-        );
-
-        trace!("new ContactInfo: {:?}", info);
-
-        Node {
-            info,
-            sockets: Sockets {
-                gossip,
-                ip_echo: Some(ip_echo),
-                tvu: vec![tvu],
-                tvu_quic,
-                tpu: vec![tpu],
-                tpu_forwards: vec![tpu_forwards],
-                tpu_vote: vec![tpu_vote],
-                broadcast: vec![broadcast],
-                repair,
-                repair_quic,
-                retransmit_sockets: vec![retransmit_socket],
-                serve_repair,
-                serve_repair_quic,
-                ancestor_hashes_requests,
-                ancestor_hashes_requests_quic,
-                tpu_quic,
-                tpu_forwards_quic,
-                tpu_vote_quic,
-                tpu_vote_forwards_client,
-            },
-        }
+        node.info.set_rpc((bind_ip_addr, rpc_port)).unwrap();
+        node.info
+            .set_rpc_pubsub((bind_ip_addr, rpc_pubsub_port))
+            .unwrap();
+        node
     }
 
     pub fn new_with_external_ip(pubkey: &Pubkey, config: NodeConfig) -> Node {
@@ -2634,7 +2473,7 @@ impl Node {
         } = config;
 
         let (gossip_port, (gossip, ip_echo)) =
-            Self::get_gossip_port(&gossip_addr, port_range, bind_ip_addr);
+            get_gossip_port(&gossip_addr, port_range, bind_ip_addr);
 
         let socket_config = SocketConfig::default();
         let socket_config_reuseport = SocketConfig::default().reuseport(true);

net-utils/src/sockets.rs

@@ -0,0 +1,49 @@
+use {
+    crate::{bind_common_in_range_with_config, bind_common_with_config, PortRange, SocketConfig},
+    std::{
+        net::{IpAddr, SocketAddr, TcpListener, UdpSocket},
+        sync::atomic::{AtomicU16, Ordering},
+    },
+};
+
+const BASE_PORT: u16 = 20000;
+
+/// Retrieve a free 100-port slice for unit tests
+///
+/// When running under nextest, this is guaranteed 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 bump an atomic and eventually
+/// panic when it runs out of port numbers to assign.
+pub fn localhost_port_range_for_tests() -> (u16, u16) {
+    static SLICE: AtomicU16 = AtomicU16::new(BASE_PORT);
+    let start = match std::env::var("NEXTEST_TEST_GLOBAL_SLOT") {
+        Ok(slot) => {
+            let slot: u16 = slot.parse().unwrap();
+            BASE_PORT + slot * 100
+        }
+        Err(_) => SLICE.fetch_add(100, Ordering::Relaxed),
+    };
+    assert!(start < 65500, "ran out of port numbers!");
+    (start, start + 100)
+}
+
+pub fn get_gossip_port(
+    gossip_addr: &SocketAddr,
+    port_range: PortRange,
+    bind_ip_addr: IpAddr,
+) -> (u16, (UdpSocket, TcpListener)) {
+    let config = SocketConfig::default();
+    if gossip_addr.port() != 0 {
+        (
+            gossip_addr.port(),
+            bind_common_with_config(bind_ip_addr, gossip_addr.port(), config).unwrap_or_else(|e| {
+                panic!("gossip_addr bind_to port {}: {}", gossip_addr.port(), e)
+            }),
+        )
+    } else {
+        bind_common_in_range_with_config(bind_ip_addr, port_range, config).expect("Failed to bind")
+    }
+}