Skip to content

Commit

Permalink
Enable QUIC client by default. Add arg to disable QUIC client. Take 2 (
Browse files Browse the repository at this point in the history
…#26927)

* Enable QUIC client by default. Add arg to disable QUIC client.
* Deprecate --disable-quic-servers arg
* Add #[ignore] annotation to failing tests
  • Loading branch information
willhickey authored and lijunwangs committed Sep 14, 2022
1 parent 6cd38e5 commit 4fa55d2
Show file tree
Hide file tree
Showing 14 changed files with 86 additions and 64 deletions.
6 changes: 3 additions & 3 deletions banking-bench/src/main.rs
Original file line number Diff line number Diff line change
Expand Up @@ -214,10 +214,10 @@ fn main() {
.help("Number of threads to use in the banking stage"),
)
.arg(
Arg::new("tpu_use_quic")
.long("tpu-use-quic")
Arg::new("tpu_disable_quic")
.long("tpu-disable-quic")
.takes_value(false)
.help("Forward messages to TPU using QUIC"),
.help("Disable forwarding messages to TPU using QUIC"),
)
.get_matches();

Expand Down
10 changes: 5 additions & 5 deletions bench-tps/src/cli.rs
Original file line number Diff line number Diff line change
Expand Up @@ -270,10 +270,10 @@ pub fn build_args<'a, 'b>(version: &'b str) -> App<'a, 'b> {
.help("Submit transactions with a TpuClient")
)
.arg(
Arg::with_name("tpu_use_quic")
.long("tpu-use-quic")
Arg::with_name("tpu_disable_quic")
.long("tpu-disable-quic")
.takes_value(false)
.help("Submit transactions via QUIC; only affects ThinClient (default) \
.help("Do not submit transactions via QUIC; only affects ThinClient (default) \
or TpuClient sends"),
)
.arg(
Expand Down Expand Up @@ -328,8 +328,8 @@ pub fn extract_args(matches: &ArgMatches) -> Config {
args.external_client_type = ExternalClientType::RpcClient;
}

if matches.is_present("tpu_use_quic") {
args.use_quic = true;
if matches.is_present("tpu_disable_quic") {
args.use_quic = false;
}

if let Some(v) = matches.value_of("tpu_connection_pool_size") {
Expand Down
1 change: 1 addition & 0 deletions bench-tps/tests/bench_tps.rs
Original file line number Diff line number Diff line change
Expand Up @@ -129,6 +129,7 @@ fn test_bench_tps_test_validator(config: Config) {

#[test]
#[serial]
#[ignore]
fn test_bench_tps_local_cluster_solana() {
test_bench_tps_local_cluster(Config {
tx_count: 100,
Expand Down
30 changes: 23 additions & 7 deletions client/src/connection_cache.rs
Original file line number Diff line number Diff line change
Expand Up @@ -32,7 +32,7 @@ static MAX_CONNECTIONS: usize = 1024;

/// Used to decide whether the TPU and underlying connection cache should use
/// QUIC connections.
pub const DEFAULT_TPU_USE_QUIC: bool = false;
pub const DEFAULT_TPU_USE_QUIC: bool = true;

/// Default TPU connection pool size per remote address
pub const DEFAULT_TPU_CONNECTION_POOL_SIZE: usize = 4;
Expand Down Expand Up @@ -683,6 +683,11 @@ mod tests {
// be lazy and not connect until first use or handle connection errors somehow
// (without crashing, as would be required in a real practical validator)
let connection_cache = ConnectionCache::default();
let port_offset = if connection_cache.use_quic() {
QUIC_PORT_OFFSET
} else {
0
};
let addrs = (0..MAX_CONNECTIONS)
.into_iter()
.map(|_| {
Expand All @@ -695,18 +700,29 @@ mod tests {
let map = connection_cache.map.read().unwrap();
assert!(map.len() == MAX_CONNECTIONS);
addrs.iter().for_each(|a| {
let conn = &map.get(a).expect("Address not found").connections[0];
let conn = conn.new_blocking_connection(*a, connection_cache.stats.clone());
assert!(a.ip() == conn.tpu_addr().ip());
let port = a
.port()
.checked_add(port_offset)
.unwrap_or_else(|| a.port());
let addr = &SocketAddr::new(a.ip(), port);

let conn = &map.get(addr).expect("Address not found").connections[0];
let conn = conn.new_blocking_connection(*addr, connection_cache.stats.clone());
assert!(addr.ip() == conn.tpu_addr().ip());
});
}

let addr = get_addr(&mut rng);
connection_cache.get_connection(&addr);
let addr = &get_addr(&mut rng);
connection_cache.get_connection(addr);

let port = addr
.port()
.checked_add(port_offset)
.unwrap_or_else(|| addr.port());
let addr_with_quic_port = SocketAddr::new(addr.ip(), port);
let map = connection_cache.map.read().unwrap();
assert!(map.len() == MAX_CONNECTIONS);
let _conn = map.get(&addr).expect("Address not found");
let _conn = map.get(&addr_with_quic_port).expect("Address not found");
}

#[test]
Expand Down
2 changes: 1 addition & 1 deletion client/src/nonblocking/quic_client.rs
Original file line number Diff line number Diff line change
Expand Up @@ -332,7 +332,7 @@ impl QuicClient {
"Cannot make 0rtt connection to {}, error {:}",
self.addr, err
);
return Err(QuicError::WriteError(err));
return Err(err);
}
}
} else {
Expand Down
2 changes: 2 additions & 0 deletions core/src/banking_stage.rs
Original file line number Diff line number Diff line change
Expand Up @@ -4123,6 +4123,7 @@ mod tests {
}

#[test]
#[ignore]
fn test_forwarder_budget() {
solana_logger::setup();
// Create `PacketBatch` with 1 unprocessed packet
Expand Down Expand Up @@ -4210,6 +4211,7 @@ mod tests {
}

#[test]
#[ignore]
fn test_handle_forwarding() {
solana_logger::setup();
// packets are deserialized upon receiving, failed packets will not be
Expand Down
69 changes: 30 additions & 39 deletions core/src/tpu.rs
Original file line number Diff line number Diff line change
Expand Up @@ -66,8 +66,8 @@ pub struct Tpu {
banking_stage: BankingStage,
cluster_info_vote_listener: ClusterInfoVoteListener,
broadcast_stage: BroadcastStage,
tpu_quic_t: Option<thread::JoinHandle<()>>,
tpu_forwards_quic_t: Option<thread::JoinHandle<()>>,
tpu_quic_t: thread::JoinHandle<()>,
tpu_forwards_quic_t: thread::JoinHandle<()>,
find_packet_sender_stake_stage: FindPacketSenderStakeStage,
vote_find_packet_sender_stake_stage: FindPacketSenderStakeStage,
staked_nodes_updater_service: StakedNodesUpdaterService,
Expand Down Expand Up @@ -99,7 +99,6 @@ impl Tpu {
cost_model: &Arc<RwLock<CostModel>>,
connection_cache: &Arc<ConnectionCache>,
keypair: &Keypair,
enable_quic_servers: bool,
staked_nodes: &Arc<RwLock<StakedNodes>>,
) -> Self {
let TpuSockets {
Expand Down Expand Up @@ -157,37 +156,33 @@ impl Tpu {
let (verified_sender, verified_receiver) = unbounded();

let stats = Arc::new(StreamStats::default());
let tpu_quic_t = enable_quic_servers.then(|| {
spawn_server(
transactions_quic_sockets,
keypair,
cluster_info.my_contact_info().tpu.ip(),
packet_sender,
exit.clone(),
MAX_QUIC_CONNECTIONS_PER_PEER,
staked_nodes.clone(),
MAX_STAKED_CONNECTIONS,
MAX_UNSTAKED_CONNECTIONS,
stats.clone(),
)
.unwrap()
});
let tpu_quic_t = spawn_server(
transactions_quic_sockets,
keypair,
cluster_info.my_contact_info().tpu.ip(),
packet_sender,
exit.clone(),
MAX_QUIC_CONNECTIONS_PER_PEER,
staked_nodes.clone(),
MAX_STAKED_CONNECTIONS,
MAX_UNSTAKED_CONNECTIONS,
stats.clone(),
)
.unwrap();

let tpu_forwards_quic_t = enable_quic_servers.then(|| {
spawn_server(
transactions_forwards_quic_sockets,
keypair,
cluster_info.my_contact_info().tpu_forwards.ip(),
forwarded_packet_sender,
exit.clone(),
MAX_QUIC_CONNECTIONS_PER_PEER,
staked_nodes.clone(),
MAX_STAKED_CONNECTIONS.saturating_add(MAX_UNSTAKED_CONNECTIONS),
0, // Prevent unstaked nodes from forwarding transactions
stats,
)
.unwrap()
});
let tpu_forwards_quic_t = spawn_server(
transactions_forwards_quic_sockets,
keypair,
cluster_info.my_contact_info().tpu_forwards.ip(),
forwarded_packet_sender,
exit.clone(),
MAX_QUIC_CONNECTIONS_PER_PEER,
staked_nodes.clone(),
MAX_STAKED_CONNECTIONS.saturating_add(MAX_UNSTAKED_CONNECTIONS),
0, // Prevent unstaked nodes from forwarding transactions
stats,
)
.unwrap();

let sigverify_stage = {
let verifier = TransactionSigVerifier::new(verified_sender);
Expand Down Expand Up @@ -289,13 +284,9 @@ impl Tpu {
self.find_packet_sender_stake_stage.join(),
self.vote_find_packet_sender_stake_stage.join(),
self.staked_nodes_updater_service.join(),
self.tpu_quic_t.join(),
self.tpu_forwards_quic_t.join(),
];
if let Some(tpu_quic_t) = self.tpu_quic_t {
tpu_quic_t.join()?;
}
if let Some(tpu_forwards_quic_t) = self.tpu_forwards_quic_t {
tpu_forwards_quic_t.join()?;
}
let broadcast_result = self.broadcast_stage.join();
for result in results {
result?;
Expand Down
3 changes: 0 additions & 3 deletions core/src/validator.rs
Original file line number Diff line number Diff line change
Expand Up @@ -172,7 +172,6 @@ pub struct ValidatorConfig {
pub accounts_shrink_ratio: AccountShrinkThreshold,
pub wait_to_vote_slot: Option<Slot>,
pub ledger_column_options: LedgerColumnOptions,
pub enable_quic_servers: bool,
}

impl Default for ValidatorConfig {
Expand Down Expand Up @@ -236,7 +235,6 @@ impl Default for ValidatorConfig {
accounts_db_config: None,
wait_to_vote_slot: None,
ledger_column_options: LedgerColumnOptions::default(),
enable_quic_servers: true,
}
}
}
Expand Down Expand Up @@ -996,7 +994,6 @@ impl Validator {
&cost_model,
&connection_cache,
&identity_keypair,
config.enable_quic_servers,
&staked_nodes,
);

Expand Down
1 change: 0 additions & 1 deletion local-cluster/src/validator_configs.rs
Original file line number Diff line number Diff line change
Expand Up @@ -65,7 +65,6 @@ pub fn safe_clone_config(config: &ValidatorConfig) -> ValidatorConfig {
accounts_db_config: config.accounts_db_config.clone(),
wait_to_vote_slot: config.wait_to_vote_slot,
ledger_column_options: config.ledger_column_options.clone(),
enable_quic_servers: config.enable_quic_servers,
}
}

Expand Down
4 changes: 4 additions & 0 deletions local-cluster/tests/local_cluster.rs
Original file line number Diff line number Diff line change
Expand Up @@ -151,6 +151,7 @@ fn test_spend_and_verify_all_nodes_3() {

#[test]
#[serial]
#[ignore]
fn test_local_cluster_signature_subscribe() {
solana_logger::setup_with_default(RUST_LOG_FILTER);
let num_nodes = 2;
Expand Down Expand Up @@ -303,6 +304,7 @@ fn test_two_unbalanced_stakes() {

#[test]
#[serial]
#[ignore]
fn test_forwarding() {
// Set up a cluster where one node is never the leader, so all txs sent to this node
// will be have to be forwarded in order to be confirmed
Expand Down Expand Up @@ -1085,6 +1087,7 @@ fn test_incremental_snapshot_download_with_crossing_full_snapshot_interval_at_st
#[allow(unused_attributes)]
#[test]
#[serial]
#[ignore]
fn test_snapshot_restart_tower() {
solana_logger::setup_with_default(RUST_LOG_FILTER);
// First set up the cluster with 2 nodes
Expand Down Expand Up @@ -2145,6 +2148,7 @@ fn run_test_load_program_accounts_partition(scan_commitment: CommitmentConfig) {

#[test]
#[serial]
#[ignore]
fn test_votes_land_in_fork_during_long_partition() {
let total_stake = 100;
// Make `lighter_stake` insufficient for switching threshold
Expand Down
2 changes: 2 additions & 0 deletions local-cluster/tests/local_cluster_slow_1.rs
Original file line number Diff line number Diff line change
Expand Up @@ -49,6 +49,7 @@ mod common;

#[test]
#[serial]
#[ignore]
// Steps in this test:
// We want to create a situation like:
/*
Expand Down Expand Up @@ -594,6 +595,7 @@ fn test_duplicate_shreds_broadcast_leader() {

#[test]
#[serial]
#[ignore]
fn test_switch_threshold_uses_gossip_votes() {
solana_logger::setup_with_default(RUST_LOG_FILTER);
let total_stake = 100;
Expand Down
1 change: 1 addition & 0 deletions local-cluster/tests/local_cluster_slow_2.rs
Original file line number Diff line number Diff line change
Expand Up @@ -203,6 +203,7 @@ fn test_leader_failure_4() {

#[test]
#[serial]
#[ignore]
fn test_ledger_cleanup_service() {
solana_logger::setup_with_default(RUST_LOG_FILTER);
error!("test_ledger_cleanup_service");
Expand Down
2 changes: 1 addition & 1 deletion multinode-demo/bootstrap-validator.sh
Original file line number Diff line number Diff line change
Expand Up @@ -58,7 +58,7 @@ while [[ -n $1 ]]; do
elif [[ $1 = --enable-rpc-bigtable-ledger-storage ]]; then
args+=("$1")
shift
elif [[ $1 = --tpu-use-quic ]]; then
elif [[ $1 = --tpu-disable-quic ]]; then
args+=("$1")
shift
elif [[ $1 = --rpc-send-batch-ms ]]; then
Expand Down
17 changes: 13 additions & 4 deletions validator/src/main.rs
Original file line number Diff line number Diff line change
Expand Up @@ -1166,13 +1166,21 @@ pub fn main() {
Arg::with_name("tpu_use_quic")
.long("tpu-use-quic")
.takes_value(false)
.hidden(true)
.conflicts_with("tpu_disable_quic")
.help("Use QUIC to send transactions."),
)
.arg(
Arg::with_name("tpu_disable_quic")
.long("tpu-disable-quic")
.takes_value(false)
.help("Do not use QUIC to send transactions."),
)
.arg(
Arg::with_name("disable_quic_servers")
.long("disable-quic-servers")
.takes_value(false)
.help("Disable QUIC TPU servers"),
.hidden(true)
)
.arg(
Arg::with_name("enable_quic_servers")
Expand Down Expand Up @@ -2236,8 +2244,7 @@ pub fn main() {
let restricted_repair_only_mode = matches.is_present("restricted_repair_only_mode");
let accounts_shrink_optimize_total_space =
value_t_or_exit!(matches, "accounts_shrink_optimize_total_space", bool);
let tpu_use_quic = matches.is_present("tpu_use_quic");
let enable_quic_servers = !matches.is_present("disable_quic_servers");
let tpu_use_quic = !matches.is_present("tpu_disable_quic");
let tpu_connection_pool_size = value_t_or_exit!(matches, "tpu_connection_pool_size", usize);

let shrink_ratio = value_t_or_exit!(matches, "accounts_shrink_ratio", f64);
Expand Down Expand Up @@ -2410,6 +2417,9 @@ pub fn main() {
if matches.is_present("enable_quic_servers") {
warn!("--enable-quic-servers is now the default behavior. This flag is deprecated and can be removed from the launch args");
}
if matches.is_present("disable_quic_servers") {
warn!("--disable-quic-servers is deprecated. The quic server cannot be disabled.");
}

let rpc_bigtable_config = if matches.is_present("enable_rpc_bigtable_ledger_storage")
|| matches.is_present("enable_bigtable_ledger_upload")
Expand Down Expand Up @@ -2579,7 +2589,6 @@ pub fn main() {
tpu_coalesce_ms,
no_wait_for_vote_to_start_leader: matches.is_present("no_wait_for_vote_to_start_leader"),
accounts_shrink_ratio,
enable_quic_servers,
..ValidatorConfig::default()
};

Expand Down

0 comments on commit 4fa55d2

Please sign in to comment.