Skip to content
This repository has been archived by the owner on Jan 13, 2025. It is now read-only.

Commit

Permalink
Add --bind-address and --rpc-bind-address validator arguments
Browse files Browse the repository at this point in the history
  • Loading branch information
mvines committed Mar 4, 2020
1 parent d3b458d commit fb9d7d2
Show file tree
Hide file tree
Showing 12 changed files with 189 additions and 100 deletions.
11 changes: 6 additions & 5 deletions archiver-lib/src/archiver.rs
Original file line number Diff line number Diff line change
Expand Up @@ -47,7 +47,7 @@ use solana_storage_program::{
};
use std::{
io::{self, ErrorKind},
net::{SocketAddr, UdpSocket},
net::{IpAddr, Ipv4Addr, SocketAddr, UdpSocket},
path::{Path, PathBuf},
result,
sync::atomic::{AtomicBool, Ordering},
Expand Down Expand Up @@ -804,14 +804,15 @@ impl Archiver {
blockstore: &Arc<Blockstore>,
slots_per_segment: u64,
) -> Result<u64> {
let ip_addr = IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0));
// Create a client which downloads from the archiver and see that it
// can respond with shreds.
let start_slot = Self::get_archiver_segment_slot(archiver_info.storage_addr);
let start_slot = Self::get_archiver_segment_slot(ip_addr, archiver_info.storage_addr);
info!("Archiver download: start at {}", start_slot);

let exit = Arc::new(AtomicBool::new(false));
let (s_reader, r_reader) = channel();
let repair_socket = Arc::new(bind_in_range(VALIDATOR_PORT_RANGE).unwrap().1);
let repair_socket = Arc::new(bind_in_range(ip_addr, VALIDATOR_PORT_RANGE).unwrap().1);
let t_receiver = receiver(
repair_socket.clone(),
&exit,
Expand Down Expand Up @@ -908,8 +909,8 @@ impl Archiver {
true
}

fn get_archiver_segment_slot(to: SocketAddr) -> u64 {
let (_port, socket) = bind_in_range(VALIDATOR_PORT_RANGE).unwrap();
fn get_archiver_segment_slot(bind_ip_addr: IpAddr, to: SocketAddr) -> u64 {
let (_port, socket) = bind_in_range(bind_ip_addr, VALIDATOR_PORT_RANGE).unwrap();
socket
.set_read_timeout(Some(Duration::from_secs(5)))
.unwrap();
Expand Down
8 changes: 7 additions & 1 deletion archiver/src/main.rs
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,12 @@ use solana_core::{
contact_info::ContactInfo,
};
use solana_sdk::{commitment_config::CommitmentConfig, signature::Signer};
use std::{net::SocketAddr, path::PathBuf, process::exit, sync::Arc};
use std::{
net::{IpAddr, Ipv4Addr, SocketAddr},
path::PathBuf,
process::exit,
sync::Arc,
};

fn main() {
solana_logger::setup();
Expand Down Expand Up @@ -116,6 +121,7 @@ fn main() {
&identity_keypair.pubkey(),
&gossip_addr,
VALIDATOR_PORT_RANGE,
IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0)),
);

println!(
Expand Down
5 changes: 3 additions & 2 deletions bench-streamer/src/main.rs
Original file line number Diff line number Diff line change
Expand Up @@ -67,15 +67,16 @@ fn main() -> Result<()> {
}

let mut port = 0;
let mut addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0)), 0);
let ip_addr = IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0));
let mut addr = SocketAddr::new(ip_addr, 0);

let exit = Arc::new(AtomicBool::new(false));

let mut read_channels = Vec::new();
let mut read_threads = Vec::new();
let recycler = PacketsRecycler::default();
for _ in 0..num_sockets {
let read = solana_net_utils::bind_to(port, false).unwrap();
let read = solana_net_utils::bind_to(ip_addr, port, false).unwrap();
read.set_read_timeout(Some(Duration::new(1, 0))).unwrap();

addr = read.local_addr().unwrap();
Expand Down
8 changes: 5 additions & 3 deletions client/src/thin_client.rs
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,7 @@ use solana_sdk::{
};
use std::{
io,
net::{SocketAddr, UdpSocket},
net::{IpAddr, Ipv4Addr, SocketAddr, UdpSocket},
sync::{
atomic::{AtomicBool, AtomicUsize, Ordering},
RwLock,
Expand Down Expand Up @@ -603,7 +603,8 @@ impl AsyncClient for ThinClient {
}

pub fn create_client((rpc, tpu): (SocketAddr, SocketAddr), range: (u16, u16)) -> ThinClient {
let (_, transactions_socket) = solana_net_utils::bind_in_range(range).unwrap();
let (_, transactions_socket) =
solana_net_utils::bind_in_range(IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0)), range).unwrap();
ThinClient::new(rpc, tpu, transactions_socket)
}

Expand All @@ -612,7 +613,8 @@ pub fn create_client_with_timeout(
range: (u16, u16),
timeout: Duration,
) -> ThinClient {
let (_, transactions_socket) = solana_net_utils::bind_in_range(range).unwrap();
let (_, transactions_socket) =
solana_net_utils::bind_in_range(IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0)), range).unwrap();
ThinClient::new_socket_with_timeout(rpc, tpu, transactions_socket, timeout)
}

Expand Down
57 changes: 36 additions & 21 deletions core/src/cluster_info.rs
Original file line number Diff line number Diff line change
Expand Up @@ -1635,16 +1635,18 @@ impl ClusterInfo {
id: &Pubkey,
gossip_addr: &SocketAddr,
) -> (ContactInfo, UdpSocket, Option<TcpListener>) {
let bind_ip_addr = IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0));
let (port, (gossip_socket, ip_echo)) =
Node::get_gossip_port(gossip_addr, VALIDATOR_PORT_RANGE);
Node::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));

(contact_info, gossip_socket, Some(ip_echo))
}

/// A Node with dummy ports to spy on gossip via pull requests
pub fn spy_node(id: &Pubkey) -> (ContactInfo, UdpSocket, Option<TcpListener>) {
let (_, gossip_socket) = bind_in_range(VALIDATOR_PORT_RANGE).unwrap();
let bind_ip_addr = IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0));
let (_, gossip_socket) = bind_in_range(bind_ip_addr, VALIDATOR_PORT_RANGE).unwrap();
let contact_info = Self::spy_contact_info(id);

(contact_info, gossip_socket, None)
Expand Down Expand Up @@ -1759,16 +1761,18 @@ impl Node {
}
}
pub fn new_localhost_with_pubkey(pubkey: &Pubkey) -> Self {
let bind_ip_addr = IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0));
let tpu = UdpSocket::bind("127.0.0.1:0").unwrap();
let (gossip_port, (gossip, ip_echo)) = bind_common_in_range((1024, 65535)).unwrap();
let (gossip_port, (gossip, ip_echo)) =
bind_common_in_range(bind_ip_addr, (1024, 65535)).unwrap();
let gossip_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), gossip_port);
let tvu = UdpSocket::bind("127.0.0.1:0").unwrap();
let tvu_forwards = UdpSocket::bind("127.0.0.1:0").unwrap();
let tpu_forwards = UdpSocket::bind("127.0.0.1:0").unwrap();
let repair = UdpSocket::bind("127.0.0.1:0").unwrap();
let rpc_port = find_available_port_in_range((1024, 65535)).unwrap();
let rpc_port = find_available_port_in_range(bind_ip_addr, (1024, 65535)).unwrap();
let rpc_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), rpc_port);
let rpc_pubsub_port = find_available_port_in_range((1024, 65535)).unwrap();
let rpc_pubsub_port = find_available_port_in_range(bind_ip_addr, (1024, 65535)).unwrap();
let rpc_pubsub_addr =
SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), rpc_pubsub_port);

Expand Down Expand Up @@ -1811,45 +1815,52 @@ impl Node {
fn get_gossip_port(
gossip_addr: &SocketAddr,
port_range: PortRange,
bind_ip_addr: IpAddr,
) -> (u16, (UdpSocket, TcpListener)) {
if gossip_addr.port() != 0 {
(
gossip_addr.port(),
bind_common(gossip_addr.port(), false).unwrap_or_else(|e| {
bind_common(bind_ip_addr, gossip_addr.port(), false).unwrap_or_else(|e| {
panic!("gossip_addr bind_to port {}: {}", gossip_addr.port(), e)
}),
)
} else {
bind_common_in_range(port_range).expect("Failed to bind")
bind_common_in_range(bind_ip_addr, port_range).expect("Failed to bind")
}
}
fn bind(port_range: PortRange) -> (u16, UdpSocket) {
bind_in_range(port_range).expect("Failed to bind")
fn bind(bind_ip_addr: IpAddr, port_range: PortRange) -> (u16, UdpSocket) {
bind_in_range(bind_ip_addr, port_range).expect("Failed to bind")
}

pub fn new_with_external_ip(
pubkey: &Pubkey,
gossip_addr: &SocketAddr,
port_range: PortRange,
bind_ip_addr: IpAddr,
) -> Node {
let (gossip_port, (gossip, ip_echo)) = Self::get_gossip_port(gossip_addr, port_range);
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(port_range, 8).expect("tvu multi_bind");
let (tvu_port, tvu_sockets) =
multi_bind_in_range(bind_ip_addr, port_range, 8).expect("tvu multi_bind");

let (tvu_forwards_port, tvu_forwards_sockets) =
multi_bind_in_range(port_range, 8).expect("tvu_forwards multi_bind");
multi_bind_in_range(bind_ip_addr, port_range, 8).expect("tvu_forwards multi_bind");

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

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

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

let (repair_port, repair) = Self::bind(port_range);
let (serve_repair_port, serve_repair) = Self::bind(port_range);
let (repair_port, repair) = Self::bind(bind_ip_addr, port_range);
let (serve_repair_port, serve_repair) = Self::bind(bind_ip_addr, port_range);

let (_, broadcast) = multi_bind_in_range(port_range, 4).expect("broadcast multi_bind");
let (_, broadcast) =
multi_bind_in_range(bind_ip_addr, port_range, 4).expect("broadcast multi_bind");

let info = ContactInfo {
id: *pubkey,
Expand Down Expand Up @@ -1889,9 +1900,10 @@ impl Node {
pubkey: &Pubkey,
gossip_addr: &SocketAddr,
port_range: PortRange,
bind_ip_addr: IpAddr,
) -> Node {
let mut new = Self::new_with_external_ip(pubkey, gossip_addr, port_range);
let (storage_port, storage_socket) = Self::bind(port_range);
let mut new = Self::new_with_external_ip(pubkey, gossip_addr, port_range, bind_ip_addr);
let (storage_port, storage_socket) = Self::bind(bind_ip_addr, port_range);

new.info.storage_addr = SocketAddr::new(gossip_addr.ip(), storage_port);
new.sockets.storage = Some(storage_socket);
Expand Down Expand Up @@ -2025,6 +2037,7 @@ mod tests {
&Pubkey::new_rand(),
&socketaddr!(ip, 0),
VALIDATOR_PORT_RANGE,
IpAddr::V4(ip),
);

check_node_sockets(&node, IpAddr::V4(ip), VALIDATOR_PORT_RANGE);
Expand All @@ -2034,14 +2047,15 @@ mod tests {
fn new_with_external_ip_test_gossip() {
let ip = IpAddr::V4(Ipv4Addr::from(0));
let port = {
bind_in_range(VALIDATOR_PORT_RANGE)
bind_in_range(ip, VALIDATOR_PORT_RANGE)
.expect("Failed to bind")
.0
};
let node = Node::new_with_external_ip(
&Pubkey::new_rand(),
&socketaddr!(0, port),
VALIDATOR_PORT_RANGE,
ip,
);

check_node_sockets(&node, ip, VALIDATOR_PORT_RANGE);
Expand All @@ -2056,6 +2070,7 @@ mod tests {
&Pubkey::new_rand(),
&socketaddr!(ip, 0),
VALIDATOR_PORT_RANGE,
IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0)),
);

let ip = IpAddr::V4(ip);
Expand Down
8 changes: 6 additions & 2 deletions core/src/gossip_service.rs
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,7 @@ use solana_ledger::bank_forks::BankForks;
use solana_perf::recycler::Recycler;
use solana_sdk::pubkey::Pubkey;
use solana_sdk::signature::{Keypair, Signer};
use std::net::{SocketAddr, TcpListener, UdpSocket};
use std::net::{IpAddr, Ipv4Addr, SocketAddr, TcpListener, UdpSocket};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::mpsc::channel;
use std::sync::{Arc, RwLock};
Expand Down Expand Up @@ -163,7 +163,11 @@ pub fn get_multi_client(nodes: &[ContactInfo]) -> (ThinClient, usize) {
.collect();
let rpc_addrs: Vec<_> = addrs.iter().map(|addr| addr.0).collect();
let tpu_addrs: Vec<_> = addrs.iter().map(|addr| addr.1).collect();
let (_, transactions_socket) = solana_net_utils::bind_in_range(VALIDATOR_PORT_RANGE).unwrap();
let (_, transactions_socket) = solana_net_utils::bind_in_range(
IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0)),
VALIDATOR_PORT_RANGE,
)
.unwrap();
let num_nodes = tpu_addrs.len();
(
ThinClient::new_from_addrs(rpc_addrs, tpu_addrs, transactions_socket),
Expand Down
9 changes: 6 additions & 3 deletions core/src/local_vote_signer_service.rs
Original file line number Diff line number Diff line change
Expand Up @@ -15,9 +15,12 @@ pub struct LocalVoteSignerService {
impl LocalVoteSignerService {
#[allow(clippy::new_ret_no_self)]
pub fn new(port_range: PortRange) -> (Self, SocketAddr) {
let addr = solana_net_utils::find_available_port_in_range(port_range)
.map(|port| SocketAddr::new(IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0)), port))
.expect("Failed to find an available port for local vote signer service");
let ip_addr = IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0));
let addr = SocketAddr::new(
ip_addr,
solana_net_utils::find_available_port_in_range(ip_addr, port_range)
.expect("Failed to find an available port for local vote signer service"),
);
let exit = Arc::new(AtomicBool::new(false));
let thread_exit = exit.clone();
let thread = Builder::new()
Expand Down
6 changes: 4 additions & 2 deletions core/src/retransmit_stage.rs
Original file line number Diff line number Diff line change
Expand Up @@ -284,6 +284,7 @@ mod tests {
use solana_ledger::create_new_tmp_ledger;
use solana_net_utils::find_available_port_in_range;
use solana_sdk::pubkey::Pubkey;
use std::net::{IpAddr, Ipv4Addr};

#[test]
fn test_skip_repair() {
Expand All @@ -300,11 +301,12 @@ mod tests {
let bank_forks = Arc::new(RwLock::new(bank_forks));

let mut me = ContactInfo::new_localhost(&Pubkey::new_rand(), 0);
let port = find_available_port_in_range((8000, 10000)).unwrap();
let ip_addr = IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0));
let port = find_available_port_in_range(ip_addr, (8000, 10000)).unwrap();
let me_retransmit = UdpSocket::bind(format!("127.0.0.1:{}", port)).unwrap();
// need to make sure tvu and tpu are valid addresses
me.tvu_forwards = me_retransmit.local_addr().unwrap();
let port = find_available_port_in_range((8000, 10000)).unwrap();
let port = find_available_port_in_range(ip_addr, (8000, 10000)).unwrap();
me.tvu = UdpSocket::bind(format!("127.0.0.1:{}", port))
.unwrap()
.local_addr()
Expand Down
5 changes: 3 additions & 2 deletions core/src/rpc_service.rs
Original file line number Diff line number Diff line change
Expand Up @@ -262,9 +262,10 @@ mod tests {
let cluster_info = Arc::new(RwLock::new(ClusterInfo::new_with_invalid_keypair(
ContactInfo::default(),
)));
let ip_addr = IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0));
let rpc_addr = SocketAddr::new(
IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0)),
solana_net_utils::find_available_port_in_range((10000, 65535)).unwrap(),
ip_addr,
solana_net_utils::find_available_port_in_range(ip_addr, (10000, 65535)).unwrap(),
);
let bank_forks = Arc::new(RwLock::new(BankForks::new(bank.slot(), bank)));
let block_commitment_cache = Arc::new(RwLock::new(BlockCommitmentCache::default()));
Expand Down
7 changes: 5 additions & 2 deletions gossip/src/main.rs
Original file line number Diff line number Diff line change
Expand Up @@ -197,8 +197,11 @@ fn main() -> Result<(), Box<dyn error::Error>> {
let gossip_addr = SocketAddr::new(
gossip_host,
value_t!(matches, "gossip_port", u16).unwrap_or_else(|_| {
solana_net_utils::find_available_port_in_range((0, 1))
.expect("unable to find an available gossip port")
solana_net_utils::find_available_port_in_range(
IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0)),
(0, 1),
)
.expect("unable to find an available gossip port")
}),
);

Expand Down
Loading

0 comments on commit fb9d7d2

Please sign in to comment.