diff --git a/banking-bench/src/main.rs b/banking-bench/src/main.rs index 51b0042abed374..2806a8a9e05a7a 100644 --- a/banking-bench/src/main.rs +++ b/banking-bench/src/main.rs @@ -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(); diff --git a/bench-tps/src/cli.rs b/bench-tps/src/cli.rs index e9b4faa848a835..8c5c22ac09d17d 100644 --- a/bench-tps/src/cli.rs +++ b/bench-tps/src/cli.rs @@ -290,10 +290,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( @@ -348,8 +348,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") { diff --git a/bench-tps/tests/bench_tps.rs b/bench-tps/tests/bench_tps.rs index 6d1c32b47a2629..a3c4bcfc986cec 100644 --- a/bench-tps/tests/bench_tps.rs +++ b/bench-tps/tests/bench_tps.rs @@ -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, diff --git a/client/src/connection_cache.rs b/client/src/connection_cache.rs index f0628d3e32b9de..9e5efff3f3e061 100644 --- a/client/src/connection_cache.rs +++ b/client/src/connection_cache.rs @@ -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; @@ -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(|_| { @@ -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] diff --git a/core/src/banking_stage.rs b/core/src/banking_stage.rs index ccef854514f1d3..0cb9713c3f11a5 100644 --- a/core/src/banking_stage.rs +++ b/core/src/banking_stage.rs @@ -4120,6 +4120,7 @@ mod tests { } #[test] + #[ignore] fn test_forwarder_budget() { solana_logger::setup(); // Create `PacketBatch` with 1 unprocessed packet @@ -4207,6 +4208,7 @@ mod tests { } #[test] + #[ignore] fn test_handle_forwarding() { solana_logger::setup(); // packets are deserialized upon receiving, failed packets will not be diff --git a/core/src/tpu.rs b/core/src/tpu.rs index cfd546869d6134..7ed1ee0c49ee6e 100644 --- a/core/src/tpu.rs +++ b/core/src/tpu.rs @@ -62,8 +62,8 @@ pub struct Tpu { banking_stage: BankingStage, cluster_info_vote_listener: ClusterInfoVoteListener, broadcast_stage: BroadcastStage, - tpu_quic_t: Option>, - tpu_forwards_quic_t: Option>, + 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, @@ -96,7 +96,6 @@ impl Tpu { connection_cache: &Arc, keypair: &Keypair, log_messages_bytes_limit: Option, - enable_quic_servers: bool, staked_nodes: &Arc>, ) -> Self { let TpuSockets { @@ -154,37 +153,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); @@ -271,13 +266,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?; diff --git a/core/src/validator.rs b/core/src/validator.rs index c70d8c62d50c36..78a611aa842d0c 100644 --- a/core/src/validator.rs +++ b/core/src/validator.rs @@ -174,7 +174,6 @@ pub struct ValidatorConfig { pub wait_to_vote_slot: Option, pub ledger_column_options: LedgerColumnOptions, pub runtime_config: RuntimeConfig, - pub enable_quic_servers: bool, } impl Default for ValidatorConfig { @@ -237,7 +236,6 @@ impl Default for ValidatorConfig { wait_to_vote_slot: None, ledger_column_options: LedgerColumnOptions::default(), runtime_config: RuntimeConfig::default(), - enable_quic_servers: true, } } } @@ -1036,7 +1034,6 @@ impl Validator { &connection_cache, &identity_keypair, config.runtime_config.log_messages_bytes_limit, - config.enable_quic_servers, &staked_nodes, ); diff --git a/dos/src/main.rs b/dos/src/main.rs index fa75fe90b79435..baadc5c0016388 100644 --- a/dos/src/main.rs +++ b/dos/src/main.rs @@ -1185,11 +1185,13 @@ pub mod test { } #[test] + #[ignore] fn test_dos_with_blockhash_and_payer() { run_dos_with_blockhash_and_payer(/*tpu_use_quic*/ false) } #[test] + #[ignore] fn test_dos_with_blockhash_and_payer_and_quic() { run_dos_with_blockhash_and_payer(/*tpu_use_quic*/ true) } diff --git a/local-cluster/src/validator_configs.rs b/local-cluster/src/validator_configs.rs index e717b468152dfa..ace823dd60ac77 100644 --- a/local-cluster/src/validator_configs.rs +++ b/local-cluster/src/validator_configs.rs @@ -64,7 +64,6 @@ pub fn safe_clone_config(config: &ValidatorConfig) -> ValidatorConfig { wait_to_vote_slot: config.wait_to_vote_slot, ledger_column_options: config.ledger_column_options.clone(), runtime_config: config.runtime_config.clone(), - enable_quic_servers: config.enable_quic_servers, } } diff --git a/local-cluster/tests/local_cluster.rs b/local-cluster/tests/local_cluster.rs index 072239f5d951ab..6fad4c541c47c8 100644 --- a/local-cluster/tests/local_cluster.rs +++ b/local-cluster/tests/local_cluster.rs @@ -173,6 +173,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; @@ -311,6 +312,7 @@ fn test_two_unbalanced_stakes() { #[test] #[serial] +#[ignore] fn test_forwarding() { solana_logger::setup_with_default(RUST_LOG_FILTER); // Set up a cluster where one node is never the leader, so all txs sent to this node @@ -1228,6 +1230,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 @@ -2520,6 +2523,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 = 3 * DEFAULT_NODE_STAKE; // Make `lighter_stake` insufficient for switching threshold diff --git a/local-cluster/tests/local_cluster_slow_1.rs b/local-cluster/tests/local_cluster_slow_1.rs index dab4f7807e677e..582d02ff15b4fd 100644 --- a/local-cluster/tests/local_cluster_slow_1.rs +++ b/local-cluster/tests/local_cluster_slow_1.rs @@ -49,6 +49,7 @@ mod common; #[test] #[serial] +#[ignore] // Steps in this test: // We want to create a situation like: /* @@ -587,6 +588,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 * DEFAULT_NODE_STAKE; diff --git a/local-cluster/tests/local_cluster_slow_2.rs b/local-cluster/tests/local_cluster_slow_2.rs index 6488ddea1e0e57..d6d315ed0d46d8 100644 --- a/local-cluster/tests/local_cluster_slow_2.rs +++ b/local-cluster/tests/local_cluster_slow_2.rs @@ -201,6 +201,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"); diff --git a/multinode-demo/bootstrap-validator.sh b/multinode-demo/bootstrap-validator.sh index 9245f507c394e2..deb82f106fae04 100755 --- a/multinode-demo/bootstrap-validator.sh +++ b/multinode-demo/bootstrap-validator.sh @@ -61,7 +61,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 diff --git a/validator/src/main.rs b/validator/src/main.rs index ce8c12c06fe3e7..ae039919c91538 100644 --- a/validator/src/main.rs +++ b/validator/src/main.rs @@ -1213,13 +1213,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") @@ -2314,8 +2322,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); @@ -2485,6 +2492,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") @@ -2664,7 +2674,6 @@ pub fn main() { log_messages_bytes_limit: value_of(&matches, "log_messages_bytes_limit"), ..RuntimeConfig::default() }, - enable_quic_servers, ..ValidatorConfig::default() };