From 48ab88a2af2b24db28688077cab4cde3c1153cf4 Mon Sep 17 00:00:00 2001 From: Michael Vines Date: Tue, 7 Jan 2020 14:18:34 -0700 Subject: [PATCH] Add --wait-for-super-majority to facilitate asynchronous cluster restarts --- core/src/validator.rs | 64 +++++++++++++++++++++++++++++++++++++------ validator/src/main.rs | 7 +++++ 2 files changed, 63 insertions(+), 8 deletions(-) diff --git a/core/src/validator.rs b/core/src/validator.rs index 333b42434..d58dd6a19 100644 --- a/core/src/validator.rs +++ b/core/src/validator.rs @@ -48,7 +48,8 @@ use std::{ sync::atomic::{AtomicBool, Ordering}, sync::mpsc::Receiver, sync::{Arc, Mutex, RwLock}, - thread::Result, + thread::{sleep, Result}, + time::Duration, }; #[derive(Clone, Debug)] @@ -67,6 +68,7 @@ pub struct ValidatorConfig { pub broadcast_stage_type: BroadcastStageType, pub partition_cfg: Option, pub fixed_leader_schedule: Option, + pub wait_for_super_majority: bool, } impl Default for ValidatorConfig { @@ -86,6 +88,7 @@ impl Default for ValidatorConfig { broadcast_stage_type: BroadcastStageType::Standard, partition_cfg: None, fixed_leader_schedule: None, + wait_for_super_majority: false, } } } @@ -293,14 +296,7 @@ impl Validator { if config.snapshot_config.is_some() { poh_recorder.set_bank(&bank); } - let poh_recorder = Arc::new(Mutex::new(poh_recorder)); - let poh_service = PohService::new(poh_recorder.clone(), &poh_config, &exit); - assert_eq!( - blocktree.new_shreds_signals.len(), - 1, - "New shred signal for the TVU should be the same as the clear bank signal." - ); let ip_echo_server = solana_net_utils::ip_echo_server(node.sockets.ip_echo.unwrap()); @@ -321,6 +317,22 @@ impl Validator { .set_entrypoint(entrypoint_info.clone()); } + if config.wait_for_super_majority { + info!( + "Waiting more than 66% of activated stake at slot {} to be in gossip...", + bank.slot() + ); + loop { + let gossip_stake_percent = get_stake_percent_in_gossip(&bank, &cluster_info); + + info!("{}% of activated stake in gossip", gossip_stake_percent,); + if gossip_stake_percent > 66 { + break; + } + sleep(Duration::new(1, 0)); + } + } + let sockets = Sockets { repair: node .sockets @@ -353,6 +365,13 @@ impl Validator { Some(voting_keypair.clone()) }; + let poh_service = PohService::new(poh_recorder.clone(), &poh_config, &exit); + assert_eq!( + blocktree.new_shreds_signals.len(), + 1, + "New shred signal for the TVU should be the same as the clear bank signal." + ); + let tvu = Tvu::new( vote_account, voting_keypair, @@ -584,6 +603,35 @@ pub fn new_validator_for_tests() -> (Validator, ContactInfo, Keypair, PathBuf) { (node, contact_info, mint_keypair, ledger_path) } +// Get the activated stake percentage (based on the provided bank) that is visible in gossip +fn get_stake_percent_in_gossip( + bank: &Arc, + cluster_info: &Arc>, +) -> u64 { + let mut gossip_stake = 0; + let mut total_activated_stake = 0; + let tvu_peers = cluster_info.read().unwrap().tvu_peers(); + + for (activated_stake, vote_account) in bank.vote_accounts().values() { + let vote_state = + solana_vote_program::vote_state::VoteState::from(&vote_account).unwrap_or_default(); + total_activated_stake += activated_stake; + if tvu_peers + .iter() + .any(|peer| peer.id == vote_state.node_pubkey) + { + trace!( + "observed {} in gossip, (activated_stake={})", + vote_state.node_pubkey, + activated_stake + ); + gossip_stake += activated_stake; + } + } + + gossip_stake * 100 / total_activated_stake +} + #[cfg(test)] mod tests { use super::*; diff --git a/validator/src/main.rs b/validator/src/main.rs index 02fe1be16..01ad149ba 100644 --- a/validator/src/main.rs +++ b/validator/src/main.rs @@ -537,6 +537,12 @@ pub fn main() { .takes_value(true) .help("Redirect logging to the specified file, '-' for standard error"), ) + .arg( + Arg::with_name("wait_for_super_majority") + .long("wait-for-super-majority") + .takes_value(false) + .help("After processing the ledger, wait until a super majority of stake is visible on gossip before starting PoH"), + ) .get_matches(); let identity_keypair = Arc::new( @@ -582,6 +588,7 @@ pub fn main() { validator_config.dev_halt_at_slot = value_t!(matches, "dev_halt_at_slot", Slot).ok(); validator_config.rpc_config.enable_validator_exit = matches.is_present("enable_rpc_exit"); + validator_config.wait_for_super_majority = matches.is_present("wait_for_super_majority"); validator_config.rpc_config.faucet_addr = matches.value_of("rpc_faucet_addr").map(|address| { solana_net_utils::parse_host_port(address).expect("failed to parse faucet address")