Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

update socket config and create builder pattern #3929

Open
wants to merge 6 commits into
base: master
Choose a base branch
from
+273 −140
Open

update socket config and create builder pattern #3929

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
d0255d6
update socket config and create builder pattern
gregcusack Nov 13, 2024
c373d6f
reuse new code for deprecated methods
gregcusack Dec 6, 2024
cd98ecf
add default config for readwrite socket
gregcusack Dec 11, 2024
ba21675
leave default buffer size to OS
gregcusack Dec 12, 2024
296b463
switch to unified socket config by reuseport
gregcusack Dec 13, 2024
f567051
remove cfg gating and add derive copy
gregcusack Dec 13, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 3 additions & 2 deletions bench-streamer/src/main.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@
use {
clap::{crate_description, crate_name, Arg, Command},
crossbeam_channel::unbounded,
solana_net_utils::bind_to_unspecified,
solana_net_utils::{bind_to_unspecified, SocketConfig},
solana_streamer::{
packet::{Packet, PacketBatch, PacketBatchRecycler, PACKET_DATA_SIZE},
streamer::{receiver, PacketBatchReceiver, StreamerReceiveStats},
Expand Down Expand Up @@ -95,9 +95,10 @@ fn main() -> Result<()> {
let mut read_channels = Vec::new();
let mut read_threads = Vec::new();
let recycler = PacketBatchRecycler::default();
let (_port, read_sockets) = solana_net_utils::multi_bind_in_range(
let (_port, read_sockets) = solana_net_utils::multi_bind_in_range_with_config(
ip_addr,
(port, port + num_sockets as u16),
SocketConfig::default().reuseport(true),
num_sockets,
)
.unwrap();
Expand Down
1 change: 1 addition & 0 deletions connection-cache/Cargo.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@ rayon = { workspace = true }
solana-keypair = { workspace = true }
solana-measure = { workspace = true }
solana-metrics = { workspace = true }
solana-net-utils = { workspace = true }
solana-time-utils = { workspace = true }
solana-transaction-error = { workspace = true }
thiserror = { workspace = true }
Expand Down
15 changes: 11 additions & 4 deletions connection-cache/src/connection_cache.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -514,6 +514,7 @@ mod tests {
async_trait::async_trait,
rand::{Rng, SeedableRng},
rand_chacha::ChaChaRng,
solana_net_utils::SocketConfig,
solana_transaction_error::TransportResult,
std::{
net::{IpAddr, Ipv4Addr, SocketAddr, UdpSocket},
Expand Down Expand Up @@ -571,8 +572,11 @@ mod tests {
fn default() -> Self {
Self {
udp_socket: Arc::new(
solana_net_utils::bind_with_any_port(IpAddr::V4(Ipv4Addr::UNSPECIFIED))
.expect("Unable to bind to UDP socket"),
solana_net_utils::bind_with_any_port_with_config(
IpAddr::V4(Ipv4Addr::UNSPECIFIED),
SocketConfig::default(),
)
.expect("Unable to bind to UDP socket"),
),
}
}
Expand All @@ -582,8 +586,11 @@ mod tests {
fn new() -> Result<Self, ClientError> {
Ok(Self {
udp_socket: Arc::new(
solana_net_utils::bind_with_any_port(IpAddr::V4(Ipv4Addr::UNSPECIFIED))
.map_err(Into::<ClientError>::into)?,
solana_net_utils::bind_with_any_port_with_config(
IpAddr::V4(Ipv4Addr::UNSPECIFIED),
SocketConfig::default(),
)
.map_err(Into::<ClientError>::into)?,
),
})
}
Expand Down
174 changes: 105 additions & 69 deletions gossip/src/cluster_info.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -54,10 +54,10 @@ use {
solana_ledger::shred::Shred,
solana_measure::measure::Measure,
solana_net_utils::{
bind_common, bind_common_in_range, bind_in_range, bind_in_range_with_config,
bind_more_with_config, bind_to_localhost, bind_to_unspecified,
bind_common_in_range_with_config, bind_common_with_config, bind_in_range,
bind_in_range_with_config, bind_more_with_config, bind_to_localhost, bind_to_unspecified,
bind_two_in_range_with_offset_and_config, find_available_port_in_range,
multi_bind_in_range, PortRange, SocketConfig, VALIDATOR_PORT_RANGE,
multi_bind_in_range_with_config, PortRange, SocketConfig, VALIDATOR_PORT_RANGE,
},
solana_perf::{
data_budget::DataBudget,
Expand Down Expand Up @@ -2612,21 +2612,20 @@ impl Node {
let localhost_ip_addr = IpAddr::V4(Ipv4Addr::LOCALHOST);
let port_range = (1024, 65535);

let udp_config = SocketConfig { reuseport: false };
let quic_config = SocketConfig { reuseport: true };
let udp_config = SocketConfig::default();
let quic_config = SocketConfig::default().reuseport(true);
let ((_tpu_port, tpu), (_tpu_quic_port, tpu_quic)) =
bind_two_in_range_with_offset_and_config(
localhost_ip_addr,
port_range,
QUIC_PORT_OFFSET,
udp_config.clone(),
quic_config.clone(),
udp_config,
quic_config,
)
.unwrap();
let tpu_quic =
bind_more_with_config(tpu_quic, num_quic_endpoints, quic_config.clone()).unwrap();
let tpu_quic = bind_more_with_config(tpu_quic, num_quic_endpoints, quic_config).unwrap();
let (gossip_port, (gossip, ip_echo)) =
bind_common_in_range(localhost_ip_addr, port_range).unwrap();
bind_common_in_range_with_config(localhost_ip_addr, port_range, udp_config).unwrap();
let gossip_addr = SocketAddr::new(localhost_ip_addr, gossip_port);
let tvu = bind_to_localhost().unwrap();
let tvu_quic = bind_to_localhost().unwrap();
Expand All @@ -2635,13 +2634,12 @@ impl Node {
localhost_ip_addr,
port_range,
QUIC_PORT_OFFSET,
udp_config.clone(),
quic_config.clone(),
udp_config,
quic_config,
)
.unwrap();
let tpu_forwards_quic =
bind_more_with_config(tpu_forwards_quic, num_quic_endpoints, quic_config.clone())
.unwrap();
bind_more_with_config(tpu_forwards_quic, num_quic_endpoints, quic_config).unwrap();
let tpu_vote = bind_to_localhost().unwrap();
let tpu_vote_quic = bind_to_localhost().unwrap();
let tpu_vote_quic =
Expand Down Expand Up @@ -2730,21 +2728,19 @@ impl Node {
port_range: PortRange,
bind_ip_addr: IpAddr,
) -> (u16, (UdpSocket, TcpListener)) {
let config = SocketConfig::default();
if gossip_addr.port() != 0 {
(
gossip_addr.port(),
bind_common(bind_ip_addr, gossip_addr.port()).unwrap_or_else(|e| {
panic!("gossip_addr bind_to port {}: {}", gossip_addr.port(), e)
}),
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(bind_ip_addr, port_range).expect("Failed to bind")
bind_common_in_range_with_config(bind_ip_addr, port_range, config)
.expect("Failed to bind")
}
}
fn bind(bind_ip_addr: IpAddr, port_range: PortRange) -> (u16, UdpSocket) {
let config = SocketConfig { reuseport: false };
Self::bind_with_config(bind_ip_addr, port_range, config)
}

fn bind_with_config(
bind_ip_addr: IpAddr,
Expand All @@ -2762,49 +2758,65 @@ impl Node {
) -> Self {
let (gossip_port, (gossip, ip_echo)) =
Self::get_gossip_port(gossip_addr, port_range, bind_ip_addr);
let (tvu_port, tvu) = Self::bind(bind_ip_addr, port_range);
let (tvu_quic_port, tvu_quic) = Self::bind(bind_ip_addr, port_range);
let udp_config = SocketConfig { reuseport: false };
let quic_config = SocketConfig { reuseport: true };

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,
udp_config.clone(),
quic_config.clone(),
socket_config,
socket_config_reuseport,
)
.unwrap();
let tpu_quic: Vec<UdpSocket> =
bind_more_with_config(tpu_quic, DEFAULT_QUIC_ENDPOINTS, quic_config.clone()).unwrap();
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,
udp_config,
quic_config.clone(),
socket_config,
socket_config_reuseport,
)
.unwrap();
let tpu_forwards_quic = bind_more_with_config(
tpu_forwards_quic,
DEFAULT_QUIC_ENDPOINTS,
quic_config.clone(),
socket_config_reuseport,
)
.unwrap();
let (tpu_vote_port, tpu_vote) = Self::bind(bind_ip_addr, port_range);
let (tpu_vote_quic_port, tpu_vote_quic) = Self::bind(bind_ip_addr, port_range);
let tpu_vote_quic: Vec<UdpSocket> =
bind_more_with_config(tpu_vote_quic, DEFAULT_QUIC_ENDPOINTS, quic_config).unwrap();

let (_, retransmit_socket) = Self::bind(bind_ip_addr, port_range);
let (_, repair) = Self::bind(bind_ip_addr, port_range);
let (_, repair_quic) = Self::bind(bind_ip_addr, port_range);
let (serve_repair_port, serve_repair) = Self::bind(bind_ip_addr, port_range);
let (serve_repair_quic_port, serve_repair_quic) = Self::bind(bind_ip_addr, port_range);
let (_, broadcast) = Self::bind(bind_ip_addr, port_range);
let (_, ancestor_hashes_requests) = Self::bind(bind_ip_addr, port_range);
let (_, ancestor_hashes_requests_quic) = Self::bind(bind_ip_addr, port_range);

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 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();
Expand Down Expand Up @@ -2880,62 +2892,86 @@ impl Node {
let (gossip_port, (gossip, ip_echo)) =
Self::get_gossip_port(&gossip_addr, port_range, bind_ip_addr);

let (tvu_port, tvu_sockets) =
multi_bind_in_range(bind_ip_addr, port_range, num_tvu_sockets.get())
.expect("tvu multi_bind");
let (tvu_quic_port, tvu_quic) = Self::bind(bind_ip_addr, port_range);
let socket_config = SocketConfig::default();
let socket_config_reuseport = SocketConfig::default().reuseport(true);

let (tvu_port, tvu_sockets) = multi_bind_in_range_with_config(
bind_ip_addr,
port_range,
socket_config_reuseport,
num_tvu_sockets.get(),
)
.expect("tvu multi_bind");

let (tvu_quic_port, tvu_quic) =
Self::bind_with_config(bind_ip_addr, port_range, socket_config);

let (tpu_port, tpu_sockets) =
multi_bind_in_range(bind_ip_addr, port_range, 32).expect("tpu multi_bind");
multi_bind_in_range_with_config(bind_ip_addr, port_range, socket_config_reuseport, 32)
.expect("tpu multi_bind");

let quic_config = SocketConfig { reuseport: true };
let (_tpu_port_quic, tpu_quic) = Self::bind_with_config(
bind_ip_addr,
(tpu_port + QUIC_PORT_OFFSET, tpu_port + QUIC_PORT_OFFSET + 1),
quic_config.clone(),
socket_config_reuseport,
);
let tpu_quic =
bind_more_with_config(tpu_quic, num_quic_endpoints.get(), quic_config.clone()).unwrap();
bind_more_with_config(tpu_quic, num_quic_endpoints.get(), socket_config_reuseport)
.unwrap();

let (tpu_forwards_port, tpu_forwards_sockets) =
multi_bind_in_range(bind_ip_addr, port_range, 8).expect("tpu_forwards multi_bind");
multi_bind_in_range_with_config(bind_ip_addr, port_range, socket_config_reuseport, 8)
.expect("tpu_forwards multi_bind");

let (_tpu_forwards_port_quic, tpu_forwards_quic) = Self::bind_with_config(
bind_ip_addr,
(
tpu_forwards_port + QUIC_PORT_OFFSET,
tpu_forwards_port + QUIC_PORT_OFFSET + 1,
),
quic_config.clone(),
socket_config_reuseport,
);
let tpu_forwards_quic = bind_more_with_config(
tpu_forwards_quic,
num_quic_endpoints.get(),
quic_config.clone(),
socket_config_reuseport,
)
.unwrap();

let (tpu_vote_port, tpu_vote_sockets) =
multi_bind_in_range(bind_ip_addr, port_range, 1).expect("tpu_vote multi_bind");
multi_bind_in_range_with_config(bind_ip_addr, port_range, socket_config_reuseport, 1)
.expect("tpu_vote multi_bind");

let (tpu_vote_quic_port, tpu_vote_quic) = Self::bind(bind_ip_addr, port_range);
let (tpu_vote_quic_port, tpu_vote_quic) =
Self::bind_with_config(bind_ip_addr, port_range, socket_config);

let tpu_vote_quic =
bind_more_with_config(tpu_vote_quic, num_quic_endpoints.get(), quic_config.clone())
.unwrap();
let tpu_vote_quic = bind_more_with_config(
tpu_vote_quic,
num_quic_endpoints.get(),
socket_config_reuseport,
)
.unwrap();

let (_, retransmit_sockets) =
multi_bind_in_range(bind_ip_addr, port_range, 8).expect("retransmit multi_bind");
multi_bind_in_range_with_config(bind_ip_addr, port_range, socket_config_reuseport, 8)
.expect("retransmit multi_bind");

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 (_, repair) = Self::bind(bind_ip_addr, port_range);
let (_, repair_quic) = Self::bind(bind_ip_addr, port_range);
let (serve_repair_port, serve_repair) = Self::bind(bind_ip_addr, port_range);
let (serve_repair_quic_port, serve_repair_quic) = Self::bind(bind_ip_addr, port_range);
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) =
multi_bind_in_range(bind_ip_addr, port_range, 4).expect("broadcast multi_bind");
multi_bind_in_range_with_config(bind_ip_addr, port_range, socket_config_reuseport, 4)
.expect("broadcast multi_bind");

let (_, ancestor_hashes_requests) = Self::bind(bind_ip_addr, port_range);
let (_, ancestor_hashes_requests_quic) = Self::bind(bind_ip_addr, port_range);
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 mut info = ContactInfo::new(
*pubkey,
Expand Down
Loading
Loading