blockworks-foundation
/
solana
mirror of https://github.com/blockworks-foundation/solana.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

leader_scheduler: remove bootstrap_height

This commit is contained in:
Michael Vines 2019-02-05 08:03:52 -08:00
parent 73979d8f5a
commit c5a74ada05
27 changed files with 1591 additions and 1789 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
benches/banking_stage.rs
View File

@ -12,7 +12,7 @@ use solana::last_id_queue::MAX_ENTRY_IDS;
use solana::packet::to_packets_chunked;
use solana_sdk::hash::hash;
use solana_sdk::pubkey::Pubkey;
use solana_sdk::signature::{Keypair, KeypairUtil, Signature};
use solana_sdk::signature::{KeypairUtil, Signature};
use solana_sdk::system_transaction::SystemTransaction;
use std::iter;
use std::sync::mpsc::{channel, Receiver};
@ -48,7 +48,6 @@ fn bench_banking_stage_multi_accounts(bencher: &mut Bencher) {
let (verified_sender, verified_receiver) = channel();
let bank = Arc::new(Bank::new(&genesis_block));
let dummy_leader_id = Keypair::new().pubkey();
let dummy = SystemTransaction::new_move(
&mint_keypair,
mint_keypair.pubkey(),
@ -106,8 +105,8 @@ fn bench_banking_stage_multi_accounts(bencher: &mut Bencher) {
verified_receiver,
Default::default(),
&genesis_block.last_id(),
None,
dummy_leader_id,
std::u64::MAX,
genesis_block.bootstrap_leader_id,
&to_leader_sender,
);
@ -143,7 +142,6 @@ fn bench_banking_stage_multi_programs(bencher: &mut Bencher) {
let (verified_sender, verified_receiver) = channel();
let bank = Arc::new(Bank::new(&genesis_block));
let dummy_leader_id = Keypair::new().pubkey();
let dummy = SystemTransaction::new_move(
&mint_keypair,
mint_keypair.pubkey(),
@ -216,8 +214,8 @@ fn bench_banking_stage_multi_programs(bencher: &mut Bencher) {
verified_receiver,
Default::default(),
&genesis_block.last_id(),
None,
dummy_leader_id,
std::u64::MAX,
genesis_block.bootstrap_leader_id,
&to_leader_sender,
);

17
fullnode/src/main.rs
View File

@ -3,7 +3,7 @@ use log::*;
use solana::client::mk_client;
use solana::cluster_info::{Node, NodeInfo, FULLNODE_PORT_RANGE};
use solana::fullnode::{Fullnode, FullnodeConfig};
use solana::leader_scheduler::LeaderScheduler;
use solana::genesis_block::GenesisBlock;
use solana::local_vote_signer_service::LocalVoteSignerService;
use solana::socketaddr;
use solana::thin_client::{poll_gossip_for_leader, ThinClient};
@ -19,9 +19,6 @@ use std::net::{Ipv4Addr, SocketAddr};
use std::process::exit;
use std::sync::mpsc::channel;
use std::sync::Arc;
use std::sync::RwLock;
use std::thread::sleep;
use std::time::Duration;
fn parse_identity(matches: &ArgMatches<'_>) -> (Keypair, SocketAddr) {
if let Some(i) = matches.value_of("identity") {
@ -248,13 +245,16 @@ fn main() {
node.info.rpc.set_port(rpc_port);
node.info.rpc_pubsub.set_port(rpc_pubsub_port);
let mut leader_scheduler = LeaderScheduler::default();
leader_scheduler.use_only_bootstrap_leader = use_only_bootstrap_leader;
let genesis_block = GenesisBlock::load(ledger_path).expect("Unable to load genesis block");
if use_only_bootstrap_leader && node.info.id != genesis_block.bootstrap_leader_id {
fullnode_config.voting_disabled = true;
}
let vote_signer: Box<dyn VoteSigner + Sync + Send> = if !no_signer {
info!("Signer service address: {:?}", signer_addr);
info!("Vote signer service address: {:?}", signer_addr);
Box::new(RemoteVoteSigner::new(signer_addr))
} else {
info!("Node will not vote");
Box::new(LocalVoteSigner::default())
};
let vote_signer = VotingKeypair::new_with_signer(&keypair, vote_signer);
@ -266,7 +266,6 @@ fn main() {
node,
&keypair,
ledger_path,
Arc::new(RwLock::new(leader_scheduler)),
vote_signer,
cluster_entrypoint
.map(|i| NodeInfo::new_entry_point(&i))
@ -277,7 +276,7 @@ fn main() {
let (rotation_sender, rotation_receiver) = channel();
fullnode.run(Some(rotation_sender));
if !no_signer {
if !fullnode_config.voting_disabled {
let leader_node_info = loop {
info!("Looking for leader...");
match poll_gossip_for_leader(gossip_addr, Some(10)) {

4
genesis/src/main.rs
View File

@ -3,7 +3,7 @@
use clap::{crate_version, value_t_or_exit, App, Arg};
use solana::db_ledger::create_new_ledger;
use solana::genesis_block::GenesisBlock;
use solana_sdk::signature::{read_keypair, KeypairUtil};
use solana_sdk::signature::{read_keypair, Keypair, KeypairUtil};
use std::error;
/**
@ -66,11 +66,13 @@ fn main() -> Result<(), Box<dyn error::Error>> {
let bootstrap_leader_keypair = read_keypair(bootstrap_leader_keypair_file)?;
let mint_keypair = read_keypair(mint_keypair_file)?;
let bootstrap_leader_vote_account_keypair = Keypair::new();
let genesis_block = GenesisBlock {
mint_id: mint_keypair.pubkey(),
tokens: num_tokens,
bootstrap_leader_id: bootstrap_leader_keypair.pubkey(),
bootstrap_leader_tokens: BOOTSTRAP_LEADER_TOKENS,
bootstrap_leader_vote_account_id: bootstrap_leader_vote_account_keypair.pubkey(),
};
create_new_ledger(ledger_path, &genesis_block)?;

221
src/bank.rs
View File

@ -9,7 +9,7 @@ use crate::entry::Entry;
use crate::entry::EntrySlice;
use crate::genesis_block::GenesisBlock;
use crate::last_id_queue::{LastIdQueue, MAX_ENTRY_IDS};
use crate::leader_scheduler::LeaderScheduler;
use crate::leader_scheduler::{LeaderScheduler, LeaderSchedulerConfig};
use crate::poh_recorder::{PohRecorder, PohRecorderError};
use crate::result::Error;
use crate::runtime::{self, RuntimeError};
@ -130,19 +130,30 @@ impl Default for Bank {
last_id_queue: RwLock::new(LastIdQueue::default()),
status_cache: RwLock::new(BankStatusCache::default()),
confirmation_time: AtomicUsize::new(std::usize::MAX),
leader_scheduler: Arc::new(RwLock::new(LeaderScheduler::default())),
leader_scheduler: Arc::new(RwLock::new(Default::default())),
subscriptions: RwLock::new(Box::new(Arc::new(LocalSubscriptions::default()))),
}
}
}
impl Bank {
pub fn new(genesis_block: &GenesisBlock) -> Self {
let bank = Self::default();
pub fn new_with_leader_scheduler_config(
genesis_block: &GenesisBlock,
leader_scheduler_config_option: Option<&LeaderSchedulerConfig>,
) -> Self {
let mut bank = Self::default();
if let Some(leader_scheduler_config) = leader_scheduler_config_option {
bank.leader_scheduler =
Arc::new(RwLock::new(LeaderScheduler::new(leader_scheduler_config)));
}
bank.process_genesis_block(genesis_block);
bank.add_builtin_programs();
bank
}
pub fn new(genesis_block: &GenesisBlock) -> Self {
Self::new_with_leader_scheduler_config(genesis_block, None)
}
pub fn set_subscriptions(&self, subscriptions: Box<Arc<BankSubscriptions + Send + Sync>>) {
let mut sub = self.subscriptions.write().unwrap();
*sub = subscriptions
@ -163,32 +174,60 @@ impl Bank {
fn process_genesis_block(&self, genesis_block: &GenesisBlock) {
assert!(genesis_block.mint_id != Pubkey::default());
assert!(genesis_block.bootstrap_leader_id != Pubkey::default());
assert!(genesis_block.bootstrap_leader_vote_account_id != Pubkey::default());
assert!(genesis_block.tokens >= genesis_block.bootstrap_leader_tokens);
assert!(genesis_block.bootstrap_leader_tokens >= 2);
let mut mint_account = Account::default();
let mut bootstrap_leader_account = Account::default();
mint_account.tokens += genesis_block.tokens;
if genesis_block.bootstrap_leader_id != Pubkey::default() {
mint_account.tokens -= genesis_block.bootstrap_leader_tokens;
bootstrap_leader_account.tokens += genesis_block.bootstrap_leader_tokens;
self.accounts.store_slow(
true,
&genesis_block.bootstrap_leader_id,
&bootstrap_leader_account,
);
};
mint_account.tokens = genesis_block.tokens - genesis_block.bootstrap_leader_tokens;
self.accounts
.store_slow(true, &genesis_block.mint_id, &mint_account);
let mut bootstrap_leader_account = Account::default();
bootstrap_leader_account.tokens = genesis_block.bootstrap_leader_tokens - 1;
self.accounts.store_slow(
true,
&genesis_block.bootstrap_leader_id,
&bootstrap_leader_account,
);
// Construct a vote account for the bootstrap_leader such that the leader_scheduler
// will be forced to select it as the leader for height 0
let mut bootstrap_leader_vote_account = Account {
tokens: 1,
userdata: vec![0; vote_program::get_max_size() as usize],
owner: vote_program::id(),
executable: false,
loader: Pubkey::default(),
};
let mut vote_state = vote_program::VoteState::new(
genesis_block.bootstrap_leader_id,
genesis_block.bootstrap_leader_id,
);
vote_state.votes.push_back(vote_program::Vote::new(0));
vote_state
.serialize(&mut bootstrap_leader_vote_account.userdata)
.unwrap();
self.accounts.store_slow(
true,
&genesis_block.bootstrap_leader_vote_account_id,
&bootstrap_leader_vote_account,
);
self.leader_scheduler
.write()
.unwrap()
.update_tick_height(0, self);
self.last_id_queue
.write()
.unwrap()
.genesis_last_id(&genesis_block.last_id());
}
fn add_system_program(&self) {
fn add_builtin_programs(&self) {
let system_program_account = Account {
tokens: 1,
owner: system_program::id(),
@ -198,10 +237,6 @@ impl Bank {
};
self.accounts
.store_slow(true, &system_program::id(), &system_program_account);
}
fn add_builtin_programs(&self) {
self.add_system_program();
// Vote program
let vote_program_account = Account {
@ -352,9 +387,16 @@ impl Bank {
/// the oldest ones once its internal cache is full. Once boot, the
/// bank will reject transactions using that `last_id`.
pub fn register_tick(&self, last_id: &Hash) {
let mut last_id_queue = self.last_id_queue.write().unwrap();
inc_new_counter_info!("bank-register_tick-registered", 1);
last_id_queue.register_tick(last_id)
let current_tick_height = {
let mut last_id_queue = self.last_id_queue.write().unwrap();
inc_new_counter_info!("bank-register_tick-registered", 1);
last_id_queue.register_tick(last_id);
last_id_queue.tick_height
};
self.leader_scheduler
.write()
.unwrap()
.update_tick_height(current_tick_height, self);
}
/// Process a Transaction. This is used for unit tests and simply calls the vector Bank::process_transactions method.
@ -664,10 +706,6 @@ impl Bank {
}
} else {
self.register_tick(&entry.id);
self.leader_scheduler
.write()
.unwrap()
.update_height(self.tick_height(), self);
}
Ok(())
@ -731,10 +769,6 @@ impl Bank {
// if its a tick, execute the group and register the tick
self.par_execute_entries(&mt_group)?;
self.register_tick(&entry.id);
self.leader_scheduler
.write()
.unwrap()
.update_height(self.tick_height(), self);
mt_group = vec![];
continue;
}
@ -882,11 +916,12 @@ impl Bank {
}
}
pub fn get_current_leader(&self) -> Option<(Pubkey, u64)> {
self.leader_scheduler
.read()
.unwrap()
.get_scheduled_leader(self.tick_height() + 1)
#[cfg(test)]
fn get_current_leader(&self) -> Option<Pubkey> {
let tick_height = self.tick_height();
let leader_scheduler = self.leader_scheduler.read().unwrap();
let slot = leader_scheduler.tick_height_to_slot(tick_height);
leader_scheduler.get_leader_for_slot(slot)
}
pub fn tick_height(&self) -> u64 {
@ -904,6 +939,7 @@ mod tests {
use super::*;
use crate::entry::{next_entries, next_entry, Entry};
use crate::gen_keys::GenKeys;
use crate::genesis_block::BOOTSTRAP_LEADER_TOKENS;
use bincode::serialize;
use hashbrown::HashSet;
use solana_sdk::hash::hash;
@ -921,18 +957,28 @@ mod tests {
fn test_bank_new() {
let (genesis_block, _) = GenesisBlock::new(10_000);
let bank = Bank::new(&genesis_block);
assert_eq!(bank.get_balance(&genesis_block.mint_id), 10_000);
assert_eq!(
bank.get_balance(&genesis_block.mint_id),
10_000 - genesis_block.bootstrap_leader_tokens
);
}
#[test]
fn test_bank_new_with_leader() {
let dummy_leader_id = Keypair::new().pubkey();
let dummy_leader_tokens = 1;
let dummy_leader_tokens = BOOTSTRAP_LEADER_TOKENS;
let (genesis_block, _) =
GenesisBlock::new_with_leader(10_000, dummy_leader_id, dummy_leader_tokens);
assert_eq!(genesis_block.bootstrap_leader_tokens, dummy_leader_tokens);
let bank = Bank::new(&genesis_block);
assert_eq!(bank.get_balance(&genesis_block.mint_id), 9999);
assert_eq!(bank.get_balance(&dummy_leader_id), 1);
assert_eq!(
bank.get_balance(&genesis_block.mint_id),
10_000 - dummy_leader_tokens
);
assert_eq!(
bank.get_balance(&dummy_leader_id),
dummy_leader_tokens - 1 /* 1 token goes to the vote account associated with dummy_leader_tokens */
);
}
#[test]
@ -954,7 +1000,7 @@ mod tests {
#[test]
fn test_one_source_two_tx_one_batch() {
let (genesis_block, mint_keypair) = GenesisBlock::new(1);
let (genesis_block, mint_keypair) = GenesisBlock::new(1 + BOOTSTRAP_LEADER_TOKENS);
let key1 = Keypair::new().pubkey();
let key2 = Keypair::new().pubkey();
let bank = Bank::new(&genesis_block);
@ -979,7 +1025,7 @@ mod tests {
#[test]
fn test_one_tx_two_out_atomic_fail() {
let (genesis_block, mint_keypair) = GenesisBlock::new(1);
let (genesis_block, mint_keypair) = GenesisBlock::new(1 + BOOTSTRAP_LEADER_TOKENS);
let key1 = Keypair::new().pubkey();
let key2 = Keypair::new().pubkey();
let bank = Bank::new(&genesis_block);
@ -1028,7 +1074,7 @@ mod tests {
#[test]
fn test_one_tx_two_out_atomic_pass() {
let (genesis_block, mint_keypair) = GenesisBlock::new(2);
let (genesis_block, mint_keypair) = GenesisBlock::new(2 + BOOTSTRAP_LEADER_TOKENS);
let key1 = Keypair::new().pubkey();
let key2 = Keypair::new().pubkey();
let bank = Bank::new(&genesis_block);
@ -1051,7 +1097,7 @@ mod tests {
// See github issue 1157 (https://github.com/solana-labs/solana/issues/1157)
#[test]
fn test_detect_failed_duplicate_transactions_issue_1157() {
let (genesis_block, mint_keypair) = GenesisBlock::new(1);
let (genesis_block, mint_keypair) = GenesisBlock::new(2 + BOOTSTRAP_LEADER_TOKENS);
let bank = Bank::new(&genesis_block);
let dest = Keypair::new();
@ -1087,7 +1133,7 @@ mod tests {
#[test]
fn test_account_not_found() {
let (genesis_block, mint_keypair) = GenesisBlock::new(1);
let (genesis_block, mint_keypair) = GenesisBlock::new(BOOTSTRAP_LEADER_TOKENS);
let bank = Bank::new(&genesis_block);
let keypair = Keypair::new();
assert_eq!(
@ -1099,7 +1145,7 @@ mod tests {
#[test]
fn test_insufficient_funds() {
let (genesis_block, mint_keypair) = GenesisBlock::new(11_000);
let (genesis_block, mint_keypair) = GenesisBlock::new(11_000 + BOOTSTRAP_LEADER_TOKENS);
let bank = Bank::new(&genesis_block);
let pubkey = Keypair::new().pubkey();
bank.transfer(1_000, &mint_keypair, pubkey, genesis_block.last_id())
@ -1132,7 +1178,7 @@ mod tests {
#[test]
fn test_debits_before_credits() {
let (genesis_block, mint_keypair) = GenesisBlock::new(2);
let (genesis_block, mint_keypair) = GenesisBlock::new(2 + BOOTSTRAP_LEADER_TOKENS);
let bank = Bank::new(&genesis_block);
let keypair = Keypair::new();
let tx0 = SystemTransaction::new_account(
@ -1159,7 +1205,7 @@ mod tests {
#[test]
fn test_process_empty_entry_is_registered() {
let (genesis_block, mint_keypair) = GenesisBlock::new(1);
let (genesis_block, mint_keypair) = GenesisBlock::new(2 + BOOTSTRAP_LEADER_TOKENS);
let bank = Bank::new(&genesis_block);
let keypair = Keypair::new();
let entry = next_entry(&genesis_block.last_id(), 1, vec![]);
@ -1178,22 +1224,15 @@ mod tests {
#[test]
fn test_process_genesis() {
solana_logger::setup();
let dummy_leader_id = Keypair::new().pubkey();
let dummy_leader_tokens = 1;
let dummy_leader_tokens = 2;
let (genesis_block, _) =
GenesisBlock::new_with_leader(5, dummy_leader_id, dummy_leader_tokens);
let bank = Bank::default();
bank.process_genesis_block(&genesis_block);
assert_eq!(bank.get_balance(&genesis_block.mint_id), 4);
let bank = Bank::new(&genesis_block);
assert_eq!(bank.get_balance(&genesis_block.mint_id), 3);
assert_eq!(bank.get_balance(&dummy_leader_id), 1);
// TODO: Restore next assert_eq() once leader scheduler configuration is stored in the
// genesis block
/*
assert_eq!(
bank.leader_scheduler.read().unwrap().bootstrap_leader,
dummy_leader_id
);
*/
assert_eq!(bank.get_current_leader(), Some(dummy_leader_id));
}
fn create_sample_block_with_next_entries_using_keypairs(
@ -1228,11 +1267,10 @@ mod tests {
fn create_sample_block_with_ticks(
genesis_block: &GenesisBlock,
mint_keypair: &Keypair,
num_entries: usize,
num_one_token_transfers: usize,
tick_interval: usize,
) -> impl Iterator<Item = Entry> {
assert!(num_entries > 0);
let mut entries = Vec::with_capacity(num_entries);
let mut entries = vec![];
// Start off the ledger with a tick linked to the genesis block
let tick = Entry::new(&genesis_block.last_id(), 0, 1, vec![]);
@ -1240,11 +1278,11 @@ mod tests {
let mut last_id = tick.id;
entries.push(tick);
let num_hashes = 1;
for i in 1..num_entries {
for i in 0..num_one_token_transfers {
// Transfer one token from the mint to a random account
let keypair = Keypair::new();
let tx = SystemTransaction::new_account(mint_keypair, keypair.pubkey(), 1, last_id, 0);
let entry = Entry::new(&hash, 0, num_hashes, vec![tx]);
let entry = Entry::new(&hash, 0, 1, vec![tx]);
hash = entry.id;
entries.push(entry);
@ -1252,12 +1290,12 @@ mod tests {
// ProgramError<0, ResultWithNegativeTokens> error when processed
let keypair2 = Keypair::new();
let tx = SystemTransaction::new_account(&keypair, keypair2.pubkey(), 42, last_id, 0);
let entry = Entry::new(&hash, 0, num_hashes, vec![tx]);
let entry = Entry::new(&hash, 0, 1, vec![tx]);
hash = entry.id;
entries.push(entry);
if (i + 1) % tick_interval == 0 {
let tick = Entry::new(&hash, 0, num_hashes, vec![]);
let tick = Entry::new(&hash, 0, 1, vec![]);
hash = tick.id;
last_id = hash;
entries.push(tick);
@ -1268,30 +1306,42 @@ mod tests {
fn create_sample_ledger(
tokens: u64,
num_entries: usize,
num_one_token_transfers: usize,
) -> (GenesisBlock, Keypair, impl Iterator<Item = Entry>) {
let mint_keypair = Keypair::new();
let bootstrap_leader_vote_account_keypair = Keypair::new();
let genesis_block = GenesisBlock {
bootstrap_leader_id: Keypair::new().pubkey(),
bootstrap_leader_tokens: 1,
bootstrap_leader_tokens: BOOTSTRAP_LEADER_TOKENS,
bootstrap_leader_vote_account_id: bootstrap_leader_vote_account_keypair.pubkey(),
mint_id: mint_keypair.pubkey(),
tokens,
};
let block =
create_sample_block_with_ticks(&genesis_block, &mint_keypair, num_entries, num_entries);
let block = create_sample_block_with_ticks(
&genesis_block,
&mint_keypair,
num_one_token_transfers,
num_one_token_transfers,
);
(genesis_block, mint_keypair, block)
}
#[test]
fn test_process_ledger_simple() {
let (genesis_block, mint_keypair, ledger) = create_sample_ledger(100, 2);
let (genesis_block, mint_keypair, ledger) =
create_sample_ledger(100 + BOOTSTRAP_LEADER_TOKENS, 3);
let mut bank = Bank::default();
bank.add_builtin_programs();
bank.process_genesis_block(&genesis_block);
assert_eq!(bank.tick_height(), 0);
bank.add_system_program();
assert_eq!(bank.get_balance(&mint_keypair.pubkey()), 100);
assert_eq!(
bank.get_current_leader(),
Some(genesis_block.bootstrap_leader_id)
);
let (ledger_height, last_id) = bank.process_ledger(ledger).unwrap();
assert_eq!(bank.get_balance(&mint_keypair.pubkey()), 98);
assert_eq!(ledger_height, 4);
assert_eq!(bank.get_balance(&mint_keypair.pubkey()), 100 - 3);
assert_eq!(ledger_height, 8);
assert_eq!(bank.tick_height(), 2);
assert_eq!(bank.last_id(), last_id);
}
@ -1299,9 +1349,11 @@ mod tests {
#[test]
fn test_hash_internal_state() {
let mint_keypair = Keypair::new();
let bootstrap_leader_vote_account_keypair = Keypair::new();
let genesis_block = GenesisBlock {
bootstrap_leader_id: Keypair::new().pubkey(),
bootstrap_leader_tokens: 1,
bootstrap_leader_tokens: BOOTSTRAP_LEADER_TOKENS,
bootstrap_leader_vote_account_id: bootstrap_leader_vote_account_keypair.pubkey(),
mint_id: mint_keypair.pubkey(),
tokens: 2_000,
};
@ -1320,11 +1372,11 @@ mod tests {
);
let mut bank0 = Bank::default();
bank0.add_system_program();
bank0.add_builtin_programs();
bank0.process_genesis_block(&genesis_block);
bank0.process_ledger(ledger0).unwrap();
let mut bank1 = Bank::default();
bank1.add_system_program();
bank1.add_builtin_programs();
bank1.process_genesis_block(&genesis_block);
bank1.process_ledger(ledger1).unwrap();
@ -1351,7 +1403,7 @@ mod tests {
}
#[test]
fn test_interleaving_locks() {
let (genesis_block, mint_keypair) = GenesisBlock::new(3);
let (genesis_block, mint_keypair) = GenesisBlock::new(3 + BOOTSTRAP_LEADER_TOKENS);
let bank = Bank::new(&genesis_block);
let alice = Keypair::new();
let bob = Keypair::new();
@ -1643,7 +1695,8 @@ mod tests {
let (genesis_block, mint_keypair) = GenesisBlock::new(10_000);
let bank = Arc::new(Bank::new(&genesis_block));
let (entry_sender, entry_receiver) = channel();
let poh_recorder = PohRecorder::new(bank.clone(), entry_sender, bank.last_id(), None);
let poh_recorder =
PohRecorder::new(bank.clone(), entry_sender, bank.last_id(), std::u64::MAX);
let pubkey = Keypair::new().pubkey();
let transactions = vec![
@ -1764,7 +1817,7 @@ mod tests {
bank.clone(),
entry_sender,
bank.last_id(),
Some(bank.tick_height() + 1),
bank.tick_height() + 1,
);
bank.process_and_record_transactions(&transactions, &poh_recorder)

54
src/banking_stage.rs
View File

@ -45,7 +45,7 @@ pub struct BankingStage {
bank_thread_hdls: Vec<JoinHandle<Option<BankingStageReturnType>>>,
poh_service: PohService,
compute_confirmation_service: ComputeLeaderConfirmationService,
max_tick_height: Option<u64>,
max_tick_height: u64,
}
impl BankingStage {
@ -56,7 +56,7 @@ impl BankingStage {
verified_receiver: Receiver<VerifiedPackets>,
config: Config,
last_entry_id: &Hash,
max_tick_height: Option<u64>,
max_tick_height: u64,
leader_id: Pubkey,
to_validator_sender: &TpuRotationSender,
) -> (Self, Receiver<Vec<Entry>>) {
@ -116,8 +116,6 @@ impl BankingStage {
break Some(BankingStageReturnType::RecordFailure);
}
Error::PohRecorderError(PohRecorderError::MaxHeightReached) => {
assert!(max_tick_height.is_some());
let max_tick_height = max_tick_height.unwrap();
if !thread_did_notify_rotation.load(Ordering::Relaxed) {
// Leader rotation should only happen if a max_tick_height was specified
let _ = thread_sender.send(
@ -279,9 +277,7 @@ impl Service for BankingStage {
match poh_return_value {
Ok(_) => (),
Err(Error::PohRecorderError(PohRecorderError::MaxHeightReached)) => {
return_value = Some(BankingStageReturnType::LeaderRotation(
self.max_tick_height.unwrap(),
));
return_value = Some(BankingStageReturnType::LeaderRotation(self.max_tick_height));
}
Err(Error::SendError) => {
return_value = Some(BankingStageReturnType::ChannelDisconnected);
@ -299,7 +295,7 @@ mod tests {
use crate::bank::Bank;
use crate::banking_stage::BankingStageReturnType;
use crate::entry::EntrySlice;
use crate::genesis_block::GenesisBlock;
use crate::genesis_block::{GenesisBlock, BOOTSTRAP_LEADER_TOKENS};
use crate::packet::to_packets;
use solana_sdk::signature::{Keypair, KeypairUtil};
use solana_sdk::system_transaction::SystemTransaction;
@ -307,9 +303,8 @@ mod tests {
#[test]
fn test_banking_stage_shutdown1() {
let (genesis_block, _mint_keypair) = GenesisBlock::new(2);
let (genesis_block, _mint_keypair) = GenesisBlock::new(2 + BOOTSTRAP_LEADER_TOKENS);
let bank = Arc::new(Bank::new(&genesis_block));
let dummy_leader_id = Keypair::new().pubkey();
let (verified_sender, verified_receiver) = channel();
let (to_validator_sender, _) = channel();
let (banking_stage, _entry_receiver) = BankingStage::new(
@ -317,8 +312,8 @@ mod tests {
verified_receiver,
Default::default(),
&bank.last_id(),
None,
dummy_leader_id,
std::u64::MAX,
genesis_block.bootstrap_leader_id,
&to_validator_sender,
);
drop(verified_sender);
@ -330,9 +325,8 @@ mod tests {
#[test]
fn test_banking_stage_shutdown2() {
let (genesis_block, _mint_keypair) = GenesisBlock::new(2);
let (genesis_block, _mint_keypair) = GenesisBlock::new(2 + BOOTSTRAP_LEADER_TOKENS);
let bank = Arc::new(Bank::new(&genesis_block));
let dummy_leader_id = Keypair::new().pubkey();
let (_verified_sender, verified_receiver) = channel();
let (to_validator_sender, _) = channel();
let (banking_stage, entry_receiver) = BankingStage::new(
@ -340,8 +334,8 @@ mod tests {
verified_receiver,
Default::default(),
&bank.last_id(),
None,
dummy_leader_id,
std::u64::MAX,
genesis_block.bootstrap_leader_id,
&to_validator_sender,
);
drop(entry_receiver);
@ -353,9 +347,8 @@ mod tests {
#[test]
fn test_banking_stage_tick() {
let (genesis_block, _mint_keypair) = GenesisBlock::new(2);
let (genesis_block, _mint_keypair) = GenesisBlock::new(2 + BOOTSTRAP_LEADER_TOKENS);
let bank = Arc::new(Bank::new(&genesis_block));
let dummy_leader_id = Keypair::new().pubkey();
let start_hash = bank.last_id();
let (verified_sender, verified_receiver) = channel();
let (to_validator_sender, _) = channel();
@ -364,8 +357,8 @@ mod tests {
verified_receiver,
Config::Sleep(Duration::from_millis(1)),
&bank.last_id(),
None,
dummy_leader_id,
std::u64::MAX,
genesis_block.bootstrap_leader_id,
&to_validator_sender,
);
sleep(Duration::from_millis(500));
@ -383,9 +376,8 @@ mod tests {
#[test]
fn test_banking_stage_entries_only() {
let (genesis_block, mint_keypair) = GenesisBlock::new(2);
let (genesis_block, mint_keypair) = GenesisBlock::new(2 + BOOTSTRAP_LEADER_TOKENS);
let bank = Arc::new(Bank::new(&genesis_block));
let dummy_leader_id = Keypair::new().pubkey();
let start_hash = bank.last_id();
let (verified_sender, verified_receiver) = channel();
let (to_validator_sender, _) = channel();
@ -394,8 +386,8 @@ mod tests {
verified_receiver,
Default::default(),
&bank.last_id(),
None,
dummy_leader_id,
std::u64::MAX,
genesis_block.bootstrap_leader_id,
&to_validator_sender,
);
@ -443,9 +435,8 @@ mod tests {
// In this attack we'll demonstrate that a verifier can interpret the ledger
// differently if either the server doesn't signal the ledger to add an
// Entry OR if the verifier tries to parallelize across multiple Entries.
let (genesis_block, mint_keypair) = GenesisBlock::new(2);
let (genesis_block, mint_keypair) = GenesisBlock::new(2 + BOOTSTRAP_LEADER_TOKENS);
let bank = Arc::new(Bank::new(&genesis_block));
let dummy_leader_id = Keypair::new().pubkey();
let (verified_sender, verified_receiver) = channel();
let (to_validator_sender, _) = channel();
let (banking_stage, entry_receiver) = BankingStage::new(
@ -453,8 +444,8 @@ mod tests {
verified_receiver,
Default::default(),
&bank.last_id(),
None,
dummy_leader_id,
std::u64::MAX,
genesis_block.bootstrap_leader_id,
&to_validator_sender,
);
@ -512,9 +503,8 @@ mod tests {
// with reason BankingStageReturnType::LeaderRotation
#[test]
fn test_max_tick_height_shutdown() {
let (genesis_block, _mint_keypair) = GenesisBlock::new(2);
let (genesis_block, _mint_keypair) = GenesisBlock::new(2 + BOOTSTRAP_LEADER_TOKENS);
let bank = Arc::new(Bank::new(&genesis_block));
let dummy_leader_id = Keypair::new().pubkey();
let (_verified_sender_, verified_receiver) = channel();
let (to_validator_sender, _to_validator_receiver) = channel();
let max_tick_height = 10;
@ -523,8 +513,8 @@ mod tests {
verified_receiver,
Default::default(),
&bank.last_id(),
Some(max_tick_height),
dummy_leader_id,
max_tick_height,
genesis_block.bootstrap_leader_id,
&to_validator_sender,
);
assert_eq!(

40
src/broadcast_service.rs
View File

@ -33,7 +33,7 @@ pub enum BroadcastServiceReturnType {
struct Broadcast {
id: Pubkey,
max_tick_height: Option<u64>,
max_tick_height: u64,
blob_index: u64,
#[cfg(feature = "erasure")]
@ -60,15 +60,12 @@ impl Broadcast {
num_entries += entries.len();
ventries.push(entries);
}
let last_tick = match self.max_tick_height {
Some(max_tick_height) => {
if let Some(Some(last)) = ventries.last().map(|entries| entries.last()) {
last.tick_height == max_tick_height
} else {
false
}
let last_tick = {
if let Some(Some(last)) = ventries.last().map(|entries| entries.last()) {
last.tick_height == self.max_tick_height
} else {
false
}
None => false,
};
inc_new_counter_info!("broadcast_service-entries_received", num_entries);
@ -151,10 +148,7 @@ fn generate_slots(
} else {
e.tick_height + 1
};
let (_, slot) = r_leader_scheduler
.get_scheduled_leader(tick_height)
.expect("Leader schedule should never be unknown while indexing blobs");
slot
r_leader_scheduler.tick_height_to_slot(tick_height)
})
.collect();
@ -193,7 +187,7 @@ impl BroadcastService {
entry_height: u64,
leader_scheduler: &Arc<RwLock<LeaderScheduler>>,
receiver: &Receiver<Vec<Entry>>,
max_tick_height: Option<u64>,
max_tick_height: u64,
exit_signal: &Arc<AtomicBool>,
blob_sender: &BlobSender,
) -> BroadcastServiceReturnType {
@ -259,7 +253,7 @@ impl BroadcastService {
entry_height: u64,
leader_scheduler: Arc<RwLock<LeaderScheduler>>,
receiver: Receiver<Vec<Entry>>,
max_tick_height: Option<u64>,
max_tick_height: u64,
exit_sender: Arc<AtomicBool>,
blob_sender: &BlobSender,
) -> Self {
@ -352,7 +346,7 @@ mod test {
entry_height,
leader_scheduler,
entry_receiver,
Some(max_tick_height),
max_tick_height,
exit_sender,
&blob_fetch_sender,
);
@ -371,15 +365,12 @@ mod test {
{
// Create the leader scheduler
let leader_keypair = Keypair::new();
let mut leader_scheduler =
LeaderScheduler::from_bootstrap_leader(leader_keypair.pubkey());
let mut leader_scheduler = LeaderScheduler::default();
// Mock the tick height to look like the tick height right after a leader transition
leader_scheduler.last_seed_height = Some(leader_scheduler.bootstrap_height);
leader_scheduler.set_leader_schedule(vec![leader_keypair.pubkey()]);
leader_scheduler.use_only_bootstrap_leader = false;
let start_tick_height = leader_scheduler.bootstrap_height;
let max_tick_height = start_tick_height + leader_scheduler.last_seed_height.unwrap();
let start_tick_height = 0;
let max_tick_height = start_tick_height + leader_scheduler.seed_rotation_interval;
let entry_height = 2 * start_tick_height;
let leader_scheduler = Arc::new(RwLock::new(leader_scheduler));
@ -405,11 +396,10 @@ mod test {
sleep(Duration::from_millis(2000));
let db_ledger = broadcast_service.db_ledger;
for i in 0..max_tick_height - start_tick_height {
let (_, slot) = leader_scheduler
let slot = leader_scheduler
.read()
.unwrap()
.get_scheduled_leader(start_tick_height + i + 1)
.expect("Leader should exist");
.tick_height_to_slot(start_tick_height + i + 1);
let result = db_ledger.get_data_blob(slot, entry_height + i).unwrap();
assert!(result.is_some());

5
src/compute_leader_confirmation_service.rs
View File

@ -176,7 +176,6 @@ pub mod tests {
solana_logger::setup();
let (genesis_block, mint_keypair) = GenesisBlock::new(1234);
let dummy_leader_id = Keypair::new().pubkey();
let bank = Arc::new(Bank::new(&genesis_block));
// generate 10 validators, but only vote for the first 6 validators
let ids: Vec<_> = (0..10)
@ -216,7 +215,7 @@ pub mod tests {
let mut last_confirmation_time = 0;
ComputeLeaderConfirmationService::compute_confirmation(
&bank,
dummy_leader_id,
genesis_block.bootstrap_leader_id,
&mut last_confirmation_time,
);
assert_eq!(bank.confirmation_time(), std::usize::MAX);
@ -228,7 +227,7 @@ pub mod tests {
ComputeLeaderConfirmationService::compute_confirmation(
&bank,
dummy_leader_id,
genesis_block.bootstrap_leader_id,
&mut last_confirmation_time,
);
assert!(bank.confirmation_time() != std::usize::MAX);

72
src/db_window.rs
View File

@ -31,7 +31,6 @@ pub fn repair(
leader_scheduler_option: &Arc<RwLock<LeaderScheduler>>,
) -> Result<Vec<(SocketAddr, Vec<u8>)>> {
let rcluster_info = cluster_info.read().unwrap();
let mut is_next_leader = false;
let meta = db_ledger.meta()?;
if meta.is_none() {
return Ok(vec![]);
@ -43,35 +42,33 @@ pub fn repair(
// Repair should only be called when received > consumed, enforced in window_service
assert!(received > consumed);
{
let ls_lock = leader_scheduler_option.read().unwrap();
if !ls_lock.use_only_bootstrap_leader {
// Calculate the next leader rotation height and check if we are the leader
if let Some(next_leader_rotation_height) = ls_lock.max_height_for_leader(tick_height) {
match ls_lock.get_scheduled_leader(next_leader_rotation_height) {
Some((leader_id, _)) if leader_id == *id => is_next_leader = true,
// In the case that we are not in the current scope of the leader schedule
// window then either:
//
// 1) The replay stage hasn't caught up to the "consumed" entries we sent,
// in which case it will eventually catch up
//
// 2) We are on the border between seed_rotation_intervals, so the
// schedule won't be known until the entry on that cusp is received
// by the replay stage (which comes after this stage). Hence, the next
// leader at the beginning of that next epoch will not know they are the
// leader until they receive that last "cusp" entry. The leader also won't ask for repairs
// for that entry because "is_next_leader" won't be set here. In this case,
// everybody will be blocking waiting for that "cusp" entry instead of repairing,
// until the leader hits "times" >= the max times in calculate_max_repair_entry_height().
// The impact of this, along with the similar problem from broadcast for the transitioning
// leader, can be observed in the multinode test, test_full_leader_validator_network(),
None => (),
_ => (),
}
}
// Check if we are the next next slot leader
let is_next_leader = {
let leader_scheduler = leader_scheduler_option.read().unwrap();
let next_slot = leader_scheduler.tick_height_to_slot(tick_height) + 1;
match leader_scheduler.get_leader_for_slot(next_slot) {
Some(leader_id) if leader_id == *id => true,
// In the case that we are not in the current scope of the leader schedule
// window then either:
//
// 1) The replay stage hasn't caught up to the "consumed" entries we sent,
// in which case it will eventually catch up
//
// 2) We are on the border between seed_rotation_intervals, so the
// schedule won't be known until the entry on that cusp is received
// by the replay stage (which comes after this stage). Hence, the next
// leader at the beginning of that next epoch will not know they are the
// leader until they receive that last "cusp" entry. The leader also won't ask for repairs
// for that entry because "is_next_leader" won't be set here. In this case,
// everybody will be blocking waiting for that "cusp" entry instead of repairing,
// until the leader hits "times" >= the max times in calculate_max_repair_entry_height().
// The impact of this, along with the similar problem from broadcast for the transitioning
// leader, can be observed in the multinode test, test_full_leader_validator_network(),
None => false,
_ => false,
}
}
};
let num_peers = rcluster_info.repair_peers().len() as u64;
@ -195,7 +192,8 @@ pub fn process_blob(
// TODO: Once the original leader signature is added to the blob, make sure that
// the blob was originally generated by the expected leader for this slot
if leader.is_none() {
return Ok(());
warn!("No leader for slot {}, blob dropped", slot);
return Ok(()); // Occurs as a leader is rotating into a validator
}
// Insert the new blob into the window
@ -393,8 +391,9 @@ mod test {
pub fn test_retransmit() {
let leader = Keypair::new().pubkey();
let nonleader = Keypair::new().pubkey();
let leader_scheduler =
Arc::new(RwLock::new(LeaderScheduler::from_bootstrap_leader(leader)));
let mut leader_scheduler = LeaderScheduler::default();
leader_scheduler.set_leader_schedule(vec![leader]);
let leader_scheduler = Arc::new(RwLock::new(leader_scheduler));
let blob = SharedBlob::default();
let (blob_sender, blob_receiver) = channel();
@ -655,17 +654,12 @@ mod test {
#[test]
fn test_process_blob() {
// Create the leader scheduler
let leader_keypair = Keypair::new();
let mut leader_scheduler = LeaderScheduler::from_bootstrap_leader(leader_keypair.pubkey());
let mut leader_scheduler = LeaderScheduler::default();
leader_scheduler.set_leader_schedule(vec![Keypair::new().pubkey()]);
let db_ledger_path = get_tmp_ledger_path("test_process_blob");
let db_ledger = Arc::new(DbLedger::open(&db_ledger_path).unwrap());
// Mock the tick height to look like the tick height right after a leader transition
leader_scheduler.last_seed_height = None;
leader_scheduler.use_only_bootstrap_leader = false;
let leader_scheduler = Arc::new(RwLock::new(leader_scheduler));
let num_entries = 10;
let original_entries = make_tiny_test_entries(num_entries);

394
src/fullnode.rs
View File

@ -6,7 +6,7 @@ use crate::counter::Counter;
use crate::db_ledger::DbLedger;
use crate::genesis_block::GenesisBlock;
use crate::gossip_service::GossipService;
use crate::leader_scheduler::LeaderScheduler;
use crate::leader_scheduler::{LeaderScheduler, LeaderSchedulerConfig};
use crate::rpc::JsonRpcService;
use crate::rpc_pubsub::PubSubService;
use crate::service::Service;
@ -25,8 +25,7 @@ use std::net::{IpAddr, Ipv4Addr, SocketAddr};
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::sync::mpsc::{channel, Receiver, Sender, SyncSender};
use std::sync::{Arc, RwLock};
use std::thread::{sleep, spawn, Result};
use std::time::Duration;
use std::thread::{spawn, Result};
use std::time::Instant;
pub type TvuRotationSender = Sender<TvuReturnType>;
@ -73,6 +72,7 @@ pub struct FullnodeConfig {
pub voting_disabled: bool,
pub entry_stream: Option<String>,
pub storage_rotate_count: u64,
pub leader_scheduler_config: LeaderSchedulerConfig,
}
impl Default for FullnodeConfig {
fn default() -> Self {
@ -85,6 +85,7 @@ impl Default for FullnodeConfig {
voting_disabled: false,
entry_stream: None,
storage_rotate_count: NUM_HASHES_FOR_STORAGE_ROTATE,
leader_scheduler_config: Default::default(),
}
}
}
@ -110,7 +111,6 @@ impl Fullnode {
mut node: Node,
keypair: &Arc<Keypair>,
ledger_path: &str,
leader_scheduler: Arc<RwLock<LeaderScheduler>>,
voting_keypair: VotingKeypair,
entrypoint_info_option: Option<&NodeInfo>,
config: &FullnodeConfig,
@ -118,9 +118,11 @@ impl Fullnode {
let id = keypair.pubkey();
let (genesis_block, db_ledger, ledger_signal_sender, ledger_signal_receiver) =
Self::make_db_ledger(ledger_path);
let (bank, entry_height, last_entry_id) =
Self::new_bank_from_db_ledger(&genesis_block, &db_ledger, leader_scheduler);
let (bank, entry_height, last_entry_id) = Self::new_bank_from_db_ledger(
&genesis_block,
&db_ledger,
Some(&config.leader_scheduler_config),
);
info!("node info: {:?}", node.info);
info!("node entrypoint_info: {:?}", entrypoint_info_option);
info!(
@ -186,10 +188,24 @@ impl Fullnode {
}
// Get the scheduled leader
let (scheduled_leader, _) = bank
.get_current_leader()
.expect("Leader not known after processing bank");
let (scheduled_leader, max_tpu_tick_height) = {
let tick_height = bank.tick_height();
let leader_scheduler = bank.leader_scheduler.read().unwrap();
let slot = leader_scheduler.tick_height_to_slot(tick_height);
(
leader_scheduler
.get_leader_for_slot(slot)
.expect("Leader not known after processing bank"),
tick_height + leader_scheduler.num_ticks_left_in_slot(tick_height),
)
};
trace!(
"scheduled_leader: {} until tick_height {}",
scheduled_leader,
max_tpu_tick_height
);
cluster_info.write().unwrap().set_leader(scheduled_leader);
// TODO: always start leader and validator, keep leader side switching between tpu
@ -238,11 +254,6 @@ impl Fullnode {
ledger_signal_sender,
ledger_signal_receiver,
);
let max_tick_height = {
let ls_lock = bank.leader_scheduler.read().unwrap();
ls_lock.max_height_for_leader(bank.tick_height() + 1)
};
let tpu = Tpu::new(
&Arc::new(bank.copy_for_tpu()),
Default::default(),
@ -258,7 +269,7 @@ impl Fullnode {
cluster_info.clone(),
entry_height,
config.sigverify_disabled,
max_tick_height,
max_tpu_tick_height,
&last_entry_id,
id,
scheduled_leader == id,
@ -286,30 +297,35 @@ impl Fullnode {
}
pub fn leader_to_validator(&mut self, tick_height: u64) -> FullnodeReturnType {
trace!("leader_to_validator: tick_height={}", tick_height);
trace!(
"leader_to_validator({:?}): tick_height={}",
self.id,
tick_height,
);
while self.bank.tick_height() < tick_height {
sleep(Duration::from_millis(10));
}
let scheduled_leader = {
let mut leader_scheduler = self.bank.leader_scheduler.write().unwrap();
let (scheduled_leader, _) = self
.bank
.leader_scheduler
.read()
.unwrap()
.get_scheduled_leader(tick_height + 1)
.unwrap();
// A transition is only permitted on the final tick of a slot
assert_eq!(leader_scheduler.num_ticks_left_in_slot(tick_height), 0);
let first_tick_of_next_slot = tick_height + 1;
leader_scheduler.update_tick_height(first_tick_of_next_slot, &self.bank);
let slot = leader_scheduler.tick_height_to_slot(first_tick_of_next_slot);
leader_scheduler.get_leader_for_slot(slot).unwrap()
};
self.cluster_info
.write()
.unwrap()
.set_leader(scheduled_leader);
if scheduled_leader == self.id {
debug!("node is still the leader");
let (last_entry_id, entry_height) = self.node_services.tvu.get_state();
self.validator_to_leader(tick_height, entry_height, last_entry_id);
FullnodeReturnType::LeaderToLeaderRotation
} else {
debug!("new leader is {}", scheduled_leader);
self.node_services.tpu.switch_to_forwarder(
self.tpu_sockets
.iter()
@ -321,14 +337,43 @@ impl Fullnode {
}
}
pub fn validator_to_leader(&mut self, tick_height: u64, entry_height: u64, last_id: Hash) {
trace!("validator_to_leader");
pub fn validator_to_leader(
&mut self,
tick_height: u64,
entry_height: u64,
last_entry_id: Hash,
) {
trace!(
"validator_to_leader({:?}): tick_height={} entry_height={} last_entry_id={}",
self.id,
tick_height,
entry_height,
last_entry_id,
);
let (scheduled_leader, max_tick_height) = {
let mut leader_scheduler = self.bank.leader_scheduler.write().unwrap();
// A transition is only permitted on the final tick of a slot
assert_eq!(leader_scheduler.num_ticks_left_in_slot(tick_height), 0);
let first_tick_of_next_slot = tick_height + 1;
leader_scheduler.update_tick_height(first_tick_of_next_slot, &self.bank);
let slot = leader_scheduler.tick_height_to_slot(first_tick_of_next_slot);
(
leader_scheduler.get_leader_for_slot(slot).unwrap(),
first_tick_of_next_slot
+ leader_scheduler.num_ticks_left_in_slot(first_tick_of_next_slot),
)
};
assert_eq!(scheduled_leader, self.id, "node is not the leader");
self.cluster_info.write().unwrap().set_leader(self.id);
let max_tick_height = {
let ls_lock = self.bank.leader_scheduler.read().unwrap();
ls_lock.max_height_for_leader(tick_height + 1)
};
debug!(
"node scheduled as leader for ticks [{}, {})",
tick_height + 1,
max_tick_height
);
let (to_validator_sender, to_validator_receiver) = channel();
self.role_notifiers.1 = to_validator_receiver;
@ -346,14 +391,14 @@ impl Fullnode {
self.sigverify_disabled,
max_tick_height,
entry_height,
&last_id,
&last_entry_id,
self.id,
&to_validator_sender,
&self.blob_sender,
)
}
pub fn handle_role_transition(&mut self) -> Option<FullnodeReturnType> {
fn handle_role_transition(&mut self) -> Option<(FullnodeReturnType, u64)> {
loop {
if self.exit.load(Ordering::Relaxed) {
return None;
@ -363,11 +408,14 @@ impl Fullnode {
match should_be_leader {
Ok(TvuReturnType::LeaderRotation(tick_height, entry_height, last_entry_id)) => {
self.validator_to_leader(tick_height, entry_height, last_entry_id);
return Some(FullnodeReturnType::ValidatorToLeaderRotation);
return Some((
FullnodeReturnType::ValidatorToLeaderRotation,
tick_height + 1,
));
}
_ => match should_be_forwarder {
Ok(TpuReturnType::LeaderRotation(tick_height)) => {
return Some(self.leader_to_validator(tick_height))
return Some((self.leader_to_validator(tick_height), tick_height + 1))
}
_ => {
continue;
@ -379,7 +427,10 @@ impl Fullnode {
// Runs a thread to manage node role transitions. The returned closure can be used to signal the
// node to exit.
pub fn run(mut self, rotation_notifier: Option<Sender<FullnodeReturnType>>) -> impl FnOnce() {
pub fn run(
mut self,
rotation_notifier: Option<Sender<(FullnodeReturnType, u64)>>,
) -> impl FnOnce() {
let (sender, receiver) = channel();
let exit = self.exit.clone();
spawn(move || loop {
@ -426,11 +477,10 @@ impl Fullnode {
pub fn new_bank_from_db_ledger(
genesis_block: &GenesisBlock,
db_ledger: &DbLedger,
leader_scheduler: Arc<RwLock<LeaderScheduler>>,
leader_scheduler_config: Option<&LeaderSchedulerConfig>,
) -> (Bank, u64, Hash) {
let mut bank = Bank::new(genesis_block);
leader_scheduler.write().unwrap().bootstrap_leader = genesis_block.bootstrap_leader_id;
bank.leader_scheduler = leader_scheduler;
let mut bank =
Bank::new_with_leader_scheduler_config(genesis_block, leader_scheduler_config);
let now = Instant::now();
let entries = db_ledger.read_ledger().expect("opening ledger");
@ -440,19 +490,20 @@ impl Fullnode {
// entry_height is the network-wide agreed height of the ledger.
// initialize it from the input ledger
info!(
"processed {} ledger entries in {}ms...",
"processed {} ledger entries in {}ms, tick_height={}...",
entry_height,
duration_as_ms(&now.elapsed())
duration_as_ms(&now.elapsed()),
bank.tick_height()
);
(bank, entry_height, last_entry_id)
}
pub fn new_bank_from_ledger(
ledger_path: &str,
leader_scheduler: Arc<RwLock<LeaderScheduler>>,
leader_scheduler_config: Option<&LeaderSchedulerConfig>,
) -> (Bank, u64, Hash) {
let (genesis_block, db_ledger, _, _) = Self::make_db_ledger(ledger_path);
Self::new_bank_from_db_ledger(&genesis_block, &db_ledger, leader_scheduler)
Self::new_bank_from_db_ledger(&genesis_block, &db_ledger, leader_scheduler_config)
}
pub fn get_leader_scheduler(&self) -> &Arc<RwLock<LeaderScheduler>> {
@ -498,22 +549,16 @@ mod tests {
use crate::cluster_info::Node;
use crate::db_ledger::*;
use crate::entry::make_consecutive_blobs;
use crate::leader_scheduler::{
make_active_set_entries, LeaderScheduler, LeaderSchedulerConfig,
};
use crate::poh_service::NUM_TICKS_PER_SECOND;
use crate::leader_scheduler::{make_active_set_entries, LeaderSchedulerConfig};
use crate::service::Service;
use crate::streamer::responder;
use crate::tpu::TpuReturnType;
use crate::tvu::TvuReturnType;
use crate::voting_keypair::VotingKeypair;
use solana_sdk::hash::Hash;
use solana_sdk::signature::{Keypair, KeypairUtil};
use std::cmp;
use std::fs::remove_dir_all;
use std::net::UdpSocket;
use std::sync::mpsc::channel;
use std::sync::{Arc, RwLock};
use std::sync::Arc;
#[test]
fn validator_exit() {
@ -529,7 +574,6 @@ mod tests {
validator_node,
&Arc::new(validator_keypair),
&validator_ledger_path,
Arc::new(RwLock::new(LeaderScheduler::new(&Default::default()))),
VotingKeypair::new(),
Some(&leader_node.info),
&FullnodeConfig::default(),
@ -556,12 +600,10 @@ mod tests {
1000,
);
ledger_paths.push(validator_ledger_path.clone());
Fullnode::new(
validator_node,
&Arc::new(validator_keypair),
&validator_ledger_path,
Arc::new(RwLock::new(LeaderScheduler::new(&Default::default()))),
VotingKeypair::new(),
Some(&leader_node.info),
&FullnodeConfig::default(),
@ -584,13 +626,12 @@ mod tests {
#[test]
fn test_leader_to_leader_transition() {
// Create the leader node information
solana_logger::setup();
let bootstrap_leader_keypair = Keypair::new();
let bootstrap_leader_node =
Node::new_localhost_with_pubkey(bootstrap_leader_keypair.pubkey());
let bootstrap_leader_info = bootstrap_leader_node.info.clone();
// Make a mint and a genesis entries for leader ledger
let (_mint_keypair, bootstrap_leader_ledger_path, _genesis_entry_height, _last_id) =
create_tmp_sample_ledger(
"test_leader_to_leader_transition",
@ -600,19 +641,14 @@ mod tests {
500,
);
// Create the common leader scheduling configuration
let num_slots_per_epoch = 3;
// Once the bootstrap leader hits the second epoch, because there are no other choices in
// the active set, this leader will remain the leader in the second epoch. In the second
// epoch, check that the same leader knows to shut down and restart as a leader again.
let leader_rotation_interval = 5;
let num_slots_per_epoch = 2;
let seed_rotation_interval = num_slots_per_epoch * leader_rotation_interval;
let active_window_length = 5;
// Set the bootstrap height to be bigger than the initial tick height.
// Once the leader hits the bootstrap height ticks, because there are no other
// choices in the active set, this leader will remain the leader in the next
// epoch. In the next epoch, check that the same leader knows to shut down and
// restart as a leader again.
let active_window_length = 10 * seed_rotation_interval;
let leader_scheduler_config = LeaderSchedulerConfig::new(
2,
leader_rotation_interval,
seed_rotation_interval,
active_window_length,
@ -620,15 +656,16 @@ mod tests {
let bootstrap_leader_keypair = Arc::new(bootstrap_leader_keypair);
let voting_keypair = VotingKeypair::new_local(&bootstrap_leader_keypair);
// Start up the leader
// Start the bootstrap leader
let mut fullnode_config = FullnodeConfig::default();
fullnode_config.leader_scheduler_config = leader_scheduler_config;
let bootstrap_leader = Fullnode::new(
bootstrap_leader_node,
&bootstrap_leader_keypair,
&bootstrap_leader_ledger_path,
Arc::new(RwLock::new(LeaderScheduler::new(&leader_scheduler_config))),
voting_keypair,
Some(&bootstrap_leader_info),
&FullnodeConfig::default(),
None,
&fullnode_config,
);
let (rotation_sender, rotation_receiver) = channel();
@ -638,7 +675,10 @@ mod tests {
// cluster it will continue to be the leader
assert_eq!(
rotation_receiver.recv().unwrap(),
FullnodeReturnType::LeaderToLeaderRotation
(
FullnodeReturnType::LeaderToLeaderRotation,
leader_rotation_interval
)
);
bootstrap_leader_exit();
}
@ -647,6 +687,14 @@ mod tests {
fn test_wrong_role_transition() {
solana_logger::setup();
let mut fullnode_config = FullnodeConfig::default();
let leader_rotation_interval = 16;
fullnode_config.leader_scheduler_config = LeaderSchedulerConfig::new(
leader_rotation_interval,
leader_rotation_interval * 2,
leader_rotation_interval * 2,
);
// Create the leader and validator nodes
let bootstrap_leader_keypair = Arc::new(Keypair::new());
let validator_keypair = Arc::new(Keypair::new());
@ -655,7 +703,9 @@ mod tests {
&bootstrap_leader_keypair,
&validator_keypair,
0,
10,
// Generate enough ticks for two epochs to flush the bootstrap_leader's vote at
// tick_height = 0 from the leader scheduler's active window
leader_rotation_interval * 4,
"test_wrong_role_transition",
);
let bootstrap_leader_info = bootstrap_leader_node.info.clone();
@ -668,29 +718,15 @@ mod tests {
validator_ledger_path.clone(),
];
// Create the common leader scheduling configuration
let leader_rotation_interval = 3;
// Set the bootstrap height exactly the current tick height, so that we can
// test if the bootstrap leader knows to immediately transition to a validator
// after parsing the ledger during startup
let leader_scheduler_config = LeaderSchedulerConfig::new(
1,
leader_rotation_interval,
leader_rotation_interval,
leader_rotation_interval * 10,
);
{
// Test that a node knows to transition to a validator based on parsing the ledger
let bootstrap_leader = Fullnode::new(
bootstrap_leader_node,
&bootstrap_leader_keypair,
&bootstrap_leader_ledger_path,
Arc::new(RwLock::new(LeaderScheduler::new(&leader_scheduler_config))),
VotingKeypair::new(),
Some(&bootstrap_leader_info),
&FullnodeConfig::default(),
&fullnode_config,
);
assert!(!bootstrap_leader.node_services.tpu.is_leader());
@ -700,10 +736,9 @@ mod tests {
validator_node,
&validator_keypair,
&validator_ledger_path,
Arc::new(RwLock::new(LeaderScheduler::new(&leader_scheduler_config))),
VotingKeypair::new(),
Some(&bootstrap_leader_info),
&FullnodeConfig::default(),
&fullnode_config,
);
assert!(validator.node_services.tpu.is_leader());
@ -720,15 +755,15 @@ mod tests {
#[test]
fn test_validator_to_leader_transition() {
solana_logger::setup();
// Make leader and validator node
let leader_keypair = Arc::new(Keypair::new());
let validator_keypair = Arc::new(Keypair::new());
let num_genesis_ticks = 1;
let (leader_node, validator_node, validator_ledger_path, ledger_initial_len, last_id) =
setup_leader_validator(
&leader_keypair,
&validator_keypair,
num_genesis_ticks,
0,
0,
"test_validator_to_leader_transition",
);
@ -736,16 +771,17 @@ mod tests {
let leader_id = leader_keypair.pubkey();
let validator_info = validator_node.info.clone();
// Set the leader scheduler for the validator
let leader_rotation_interval = 16;
let num_bootstrap_slots = 2;
let bootstrap_height = num_bootstrap_slots * leader_rotation_interval;
info!("leader: {:?}", leader_id);
info!("validator: {:?}", validator_info.id);
let leader_scheduler_config = LeaderSchedulerConfig::new(
bootstrap_height,
// Set the leader scheduler for the validator
let leader_rotation_interval = 10;
let mut fullnode_config = FullnodeConfig::default();
fullnode_config.leader_scheduler_config = LeaderSchedulerConfig::new(
leader_rotation_interval,
leader_rotation_interval * 2,
bootstrap_height,
leader_rotation_interval * 4,
leader_rotation_interval * 4,
);
let voting_keypair = VotingKeypair::new_local(&validator_keypair);
@ -754,12 +790,18 @@ mod tests {
validator_node,
&validator_keypair,
&validator_ledger_path,
Arc::new(RwLock::new(LeaderScheduler::new(&leader_scheduler_config))),
voting_keypair,
Some(&leader_node.info),
&FullnodeConfig::default(),
&fullnode_config,
);
let blobs_to_send = fullnode_config
.leader_scheduler_config
.seed_rotation_interval
+ fullnode_config
.leader_scheduler_config
.leader_rotation_interval;
// Send blobs to the validator from our mock leader
let t_responder = {
let (s_responder, r_responder) = channel();
@ -771,14 +813,11 @@ mod tests {
r_responder,
);
// Send the blobs out of order, in reverse. Also send an extra
// "extra_blobs" number of blobs to make sure the window stops in the right place.
let extra_blobs = cmp::max(leader_rotation_interval / 3, 1);
let total_blobs_to_send = bootstrap_height + extra_blobs;
let tvu_address = &validator_info.tvu;
let msgs = make_consecutive_blobs(
&leader_id,
total_blobs_to_send,
blobs_to_send,
ledger_initial_len,
last_id,
&tvu_address,
@ -790,37 +829,28 @@ mod tests {
t_responder
};
assert_ne!(validator.bank.get_current_leader().unwrap().0, validator.id);
loop {
let should_be_forwarder = validator.role_notifiers.1.try_recv();
let should_be_leader = validator.role_notifiers.0.try_recv();
match should_be_leader {
Ok(TvuReturnType::LeaderRotation(tick_height, entry_height, _)) => {
assert_eq!(validator.node_services.tvu.get_state().1, entry_height);
assert_eq!(validator.bank.tick_height(), tick_height);
assert_eq!(tick_height, bootstrap_height);
break;
}
_ => match should_be_forwarder {
Ok(TpuReturnType::LeaderRotation(_)) => {
panic!("shouldn't be rotating to forwarder")
}
_ => continue,
},
}
}
//close the validator so that rocksdb has locks available
validator.close().unwrap();
let (bank, entry_height, _) = Fullnode::new_bank_from_ledger(
&validator_ledger_path,
Arc::new(RwLock::new(LeaderScheduler::new(&leader_scheduler_config))),
info!("waiting for validator to rotate into the leader role");
let (rotation_sender, rotation_receiver) = channel();
let validator_exit = validator.run(Some(rotation_sender));
let rotation = rotation_receiver.recv().unwrap();
assert_eq!(
rotation,
(FullnodeReturnType::ValidatorToLeaderRotation, blobs_to_send)
);
assert!(bank.tick_height() >= bootstrap_height);
// Only the first genesis entry has num_hashes = 0, every other entry
// had num_hashes = 1
assert!(entry_height >= bootstrap_height + ledger_initial_len - num_genesis_ticks);
// Close the validator so that rocksdb has locks available
// validator.close().unwrap();
validator_exit();
let (bank, entry_height, _) = Fullnode::new_bank_from_ledger(&validator_ledger_path, None);
assert!(
bank.tick_height()
>= fullnode_config
.leader_scheduler_config
.seed_rotation_interval
);
assert!(entry_height >= ledger_initial_len);
// Shut down
t_responder.join().expect("responder thread join");
@ -830,44 +860,48 @@ mod tests {
}
#[test]
#[ignore] // TODO: Make this test less hacky
fn test_tvu_behind() {
solana_logger::setup();
// Make leader node
let leader_keypair = Arc::new(Keypair::new());
let validator_keypair = Arc::new(Keypair::new());
info!("leader: {:?}", leader_keypair.pubkey());
info!("validator: {:?}", validator_keypair.pubkey());
let (leader_node, _, leader_ledger_path, _, _) =
setup_leader_validator(&leader_keypair, &validator_keypair, 1, 0, "test_tvu_behind");
let leader_node_info = leader_node.info.clone();
// Set the leader scheduler for the validator
let leader_rotation_interval = NUM_TICKS_PER_SECOND as u64 * 5;
let bootstrap_height = leader_rotation_interval;
let leader_rotation_interval = 5;
let leader_scheduler_config = LeaderSchedulerConfig::new(
bootstrap_height,
let mut fullnode_config = FullnodeConfig::default();
fullnode_config.leader_scheduler_config = LeaderSchedulerConfig::new(
leader_rotation_interval,
leader_rotation_interval * 2,
bootstrap_height,
leader_rotation_interval * 2,
);
let voting_keypair = VotingKeypair::new_local(&leader_keypair);
// Start the bootstrap leader
info!("Start the bootstrap leader");
let mut leader = Fullnode::new(
leader_node,
&leader_keypair,
&leader_ledger_path,
Arc::new(RwLock::new(LeaderScheduler::new(&leader_scheduler_config))),
voting_keypair,
Some(&leader_node_info),
&FullnodeConfig::default(),
&fullnode_config,
);
// Hold Tvu bank lock to prevent tvu from making progress
info!("Hold Tvu bank lock to prevent tvu from making progress");
{
let w_last_ids = leader.bank.last_ids().write().unwrap();
// Wait for leader -> validator transition
info!("Wait for leader -> validator transition");
let signal = leader
.role_notifiers
.1
@ -877,26 +911,42 @@ mod tests {
rn_sender.send(signal).expect("send");
leader.role_notifiers = (leader.role_notifiers.0, rn_receiver);
// Make sure the tvu bank is behind
assert!(w_last_ids.tick_height < bootstrap_height);
info!("Make sure the tvu bank is behind");
assert_eq!(w_last_ids.tick_height, 2);
}
// Release tvu bank lock, tvu should start making progress again and
// handle_role_transition should successfully rotate the leader to a validator
assert_eq!(
leader.handle_role_transition().unwrap(),
FullnodeReturnType::LeaderToValidatorRotation
);
assert_eq!(
leader.cluster_info.read().unwrap().leader_id(),
validator_keypair.pubkey(),
);
assert!(!leader.node_services.tpu.is_leader());
// Confirm the bank actually made progress
assert_eq!(leader.bank.tick_height(), bootstrap_height);
// Release tvu bank lock, tvu should start making progress again and should signal a
// rotate. After rotation it will still be the slot leader as a new leader schedule has
// not been computed yet (still in epoch 0)
info!("Release tvu bank lock");
let (rotation_sender, rotation_receiver) = channel();
let leader_exit = leader.run(Some(rotation_sender));
let expected_rotations = vec![
(
FullnodeReturnType::LeaderToLeaderRotation,
leader_rotation_interval,
),
(
FullnodeReturnType::LeaderToLeaderRotation,
2 * leader_rotation_interval,
),
(
FullnodeReturnType::LeaderToValidatorRotation,
3 * leader_rotation_interval,
),
];
// Shut down
leader.close().expect("leader shutdown");
for expected_rotation in expected_rotations {
loop {
let transition = rotation_receiver.recv().unwrap();
info!("leader transition: {:?}", transition);
assert_eq!(expected_rotation, transition);
break;
}
}
info!("Shut down");
leader_exit();
DbLedger::destroy(&leader_ledger_path).expect("Expected successful database destruction");
let _ignored = remove_dir_all(&leader_ledger_path).unwrap();
}
@ -910,11 +960,15 @@ mod tests {
) -> (Node, Node, String, u64, Hash) {
// Make a leader identity
let leader_node = Node::new_localhost_with_pubkey(leader_keypair.pubkey());
let leader_id = leader_node.info.id;
// Create validator identity
let (mint_keypair, ledger_path, genesis_entry_height, last_id) =
create_tmp_sample_ledger(test_name, 10_000, num_genesis_ticks, leader_id, 500);
let (mint_keypair, ledger_path, genesis_entry_height, last_id) = create_tmp_sample_ledger(
test_name,
10_000,
num_genesis_ticks,
leader_node.info.id,
500,
);
let validator_node = Node::new_localhost_with_pubkey(validator_keypair.pubkey());
@ -928,6 +982,8 @@ mod tests {
let (active_set_entries, _) = make_active_set_entries(
validator_keypair,
&mint_keypair,
10,
1,
&last_id,
&last_id,
num_ending_ticks,

24
src/genesis_block.rs
View File

@ -7,10 +7,16 @@ use std::fs::File;
use std::io::Write;
use std::path::Path;
// The default (and minimal) amount of tokens given to the bootstrap leader:
// * 1 token for the bootstrap leader ID account
// * 1 token for the bootstrap leader vote account
pub const BOOTSTRAP_LEADER_TOKENS: u64 = 2;
#[derive(Serialize, Deserialize, Debug)]
pub struct GenesisBlock {
pub bootstrap_leader_id: Pubkey,
pub bootstrap_leader_tokens: u64,
pub bootstrap_leader_vote_account_id: Pubkey,
pub mint_id: Pubkey,
pub tokens: u64,
}
@ -18,11 +24,15 @@ pub struct GenesisBlock {
impl GenesisBlock {
#[allow(clippy::new_ret_no_self)]
pub fn new(tokens: u64) -> (Self, Keypair) {
assert!(tokens >= 2);
let mint_keypair = Keypair::new();
let bootstrap_leader_keypair = Keypair::new();
let bootstrap_leader_vote_account_keypair = Keypair::new();
(
Self {
bootstrap_leader_id: Pubkey::default(),
bootstrap_leader_tokens: 0,
bootstrap_leader_id: bootstrap_leader_keypair.pubkey(),
bootstrap_leader_tokens: BOOTSTRAP_LEADER_TOKENS,
bootstrap_leader_vote_account_id: bootstrap_leader_vote_account_keypair.pubkey(),
mint_id: mint_keypair.pubkey(),
tokens,
},
@ -36,10 +46,12 @@ impl GenesisBlock {
bootstrap_leader_tokens: u64,
) -> (Self, Keypair) {
let mint_keypair = Keypair::new();
let bootstrap_leader_vote_account_keypair = Keypair::new();
(
Self {
bootstrap_leader_id,
bootstrap_leader_tokens,
bootstrap_leader_vote_account_id: bootstrap_leader_vote_account_keypair.pubkey(),
mint_id: mint_keypair.pubkey(),
tokens,
},
@ -74,8 +86,12 @@ mod tests {
let (genesis_block, mint) = GenesisBlock::new(10_000);
assert_eq!(genesis_block.tokens, 10_000);
assert_eq!(genesis_block.mint_id, mint.pubkey());
assert_eq!(genesis_block.bootstrap_leader_id, Pubkey::default());
assert_eq!(genesis_block.bootstrap_leader_tokens, 0);
assert!(genesis_block.bootstrap_leader_id != Pubkey::default());
assert!(genesis_block.bootstrap_leader_vote_account_id != Pubkey::default());
assert_eq!(
genesis_block.bootstrap_leader_tokens,
BOOTSTRAP_LEADER_TOKENS
);
}
#[test]

1185
src/leader_scheduler.rs
View File

File diff suppressed because it is too large Load Diff

19
src/poh_recorder.rs
View File

@ -21,11 +21,11 @@ pub struct PohRecorder {
poh: Arc<Mutex<Poh>>,
bank: Arc<Bank>,
sender: Sender<Vec<Entry>>,
max_tick_height: Option<u64>,
max_tick_height: u64,
}
impl PohRecorder {
pub fn max_tick_height(&self) -> Option<u64> {
pub fn max_tick_height(&self) -> u64 {
self.max_tick_height
}
@ -69,7 +69,7 @@ impl PohRecorder {
bank: Arc<Bank>,
sender: Sender<Vec<Entry>>,
last_entry_id: Hash,
max_tick_height: Option<u64>,
max_tick_height: u64,
) -> Self {
let poh = Arc::new(Mutex::new(Poh::new(last_entry_id, bank.tick_height())));
PohRecorder {
@ -81,11 +81,10 @@ impl PohRecorder {
}
fn check_tick_height(&self, poh: &Poh) -> Result<()> {
match self.max_tick_height {
Some(max_tick_height) if poh.tick_height >= max_tick_height => {
Err(Error::PohRecorderError(PohRecorderError::MaxHeightReached))
}
_ => Ok(()),
if poh.tick_height >= self.max_tick_height {
Err(Error::PohRecorderError(PohRecorderError::MaxHeightReached))
} else {
Ok(())
}
}
@ -127,11 +126,11 @@ mod tests {
#[test]
fn test_poh_recorder() {
let (genesis_block, _mint_keypair) = GenesisBlock::new(1);
let (genesis_block, _mint_keypair) = GenesisBlock::new(2);
let bank = Arc::new(Bank::new(&genesis_block));
let prev_id = bank.last_id();
let (entry_sender, entry_receiver) = channel();
let mut poh_recorder = PohRecorder::new(bank, entry_sender, prev_id, Some(2));
let mut poh_recorder = PohRecorder::new(bank, entry_sender, prev_id, 2);
//send some data
let h1 = hash(b"hello world!");

15
src/poh_service.rs
View File

@ -91,11 +91,8 @@ impl PohService {
let res = poh.hash();
if let Err(e) = res {
if let Error::PohRecorderError(PohRecorderError::MaxHeightReached) = e {
// Leader rotation should only happen if a max_tick_height was specified
assert!(max_tick_height.is_some());
to_validator_sender.send(TpuReturnType::LeaderRotation(
max_tick_height.unwrap(),
))?;
to_validator_sender
.send(TpuReturnType::LeaderRotation(max_tick_height))?;
}
return Err(e);
}
@ -109,9 +106,7 @@ impl PohService {
if let Err(e) = res {
if let Error::PohRecorderError(PohRecorderError::MaxHeightReached) = e {
// Leader rotation should only happen if a max_tick_height was specified
assert!(max_tick_height.is_some());
to_validator_sender
.send(TpuReturnType::LeaderRotation(max_tick_height.unwrap()))?;
to_validator_sender.send(TpuReturnType::LeaderRotation(max_tick_height))?;
}
return Err(e);
}
@ -147,11 +142,11 @@ mod tests {
#[test]
fn test_poh_service() {
let (genesis_block, _mint_keypair) = GenesisBlock::new(1);
let (genesis_block, _mint_keypair) = GenesisBlock::new(2);
let bank = Arc::new(Bank::new(&genesis_block));
let prev_id = bank.last_id();
let (entry_sender, entry_receiver) = channel();
let poh_recorder = PohRecorder::new(bank, entry_sender, prev_id, None);
let poh_recorder = PohRecorder::new(bank, entry_sender, prev_id, std::u64::MAX);
let exit = Arc::new(AtomicBool::new(false));
let entry_producer: JoinHandle<Result<()>> = {

227
src/replay_stage.rs
View File

@ -11,7 +11,6 @@ use crate::entry_stream::EntryStreamHandler;
#[cfg(test)]
use crate::entry_stream::MockEntryStream as EntryStream;
use crate::fullnode::TvuRotationSender;
use crate::leader_scheduler::DEFAULT_TICKS_PER_SLOT;
use crate::packet::BlobError;
use crate::result::{Error, Result};
use crate::service::Service;
@ -73,7 +72,7 @@ impl ReplayStage {
error!("Entry Stream error: {:?}, {:?}", e, stream.socket);
});
}
//coalesce all the available entries into a single vote
// Coalesce all the available entries into a single vote
submit(
influxdb::Point::new("replicate-stage")
.add_field("count", influxdb::Value::Integer(entries.len() as i64))
@ -92,17 +91,22 @@ impl ReplayStage {
duration_as_ms(&now.elapsed()) as usize
);
let (current_leader, _) = bank
.get_current_leader()
.expect("Scheduled leader should be calculated by this point");
let mut num_ticks_to_next_vote = bank
.leader_scheduler
.read()
.unwrap()
.num_ticks_left_in_slot(bank.tick_height());
// Next vote tick is ceiling of (current tick/ticks per block)
let mut num_ticks_to_next_vote =
DEFAULT_TICKS_PER_SLOT - (bank.tick_height() % DEFAULT_TICKS_PER_SLOT);
let mut start_entry_index = 0;
for (i, entry) in entries.iter().enumerate() {
inc_new_counter_info!("replicate-stage_bank-tick", bank.tick_height() as usize);
if entry.is_tick() {
if num_ticks_to_next_vote == 0 {
num_ticks_to_next_vote = bank
.leader_scheduler
.read()
.unwrap()
.leader_rotation_interval;
}
num_ticks_to_next_vote -= 1;
}
inc_new_counter_info!(
@ -113,7 +117,7 @@ impl ReplayStage {
// If we don't process the entry now, the for loop will exit and the entry
// will be dropped.
if 0 == num_ticks_to_next_vote || (i + 1) == entries.len() {
res = bank.process_entries(&entries[start_entry_index..=i]);
res = bank.process_entries(&entries[0..=i]);
if res.is_err() {
// TODO: This will return early from the first entry that has an erroneous
@ -122,11 +126,7 @@ impl ReplayStage {
// bank.process_entries() was used to process the entries, but doesn't solve the
// issue that the bank state was still changed, leading to inconsistencies with the
// leader as the leader currently should not be publishing erroneous transactions
inc_new_counter_info!(
"replicate-stage_failed_process_entries",
(i - start_entry_index)
);
inc_new_counter_info!("replicate-stage_failed_process_entries", i);
break;
}
@ -142,19 +142,8 @@ impl ReplayStage {
cluster_info.write().unwrap().push_vote(vote);
}
}
let (scheduled_leader, _) = bank
.get_current_leader()
.expect("Scheduled leader should be calculated by this point");
// TODO: Remove this soon once we boot the leader from ClusterInfo
if scheduled_leader != current_leader {
cluster_info.write().unwrap().set_leader(scheduled_leader);
num_entries_to_write = i + 1;
break;
}
start_entry_index = i + 1;
num_ticks_to_next_vote = DEFAULT_TICKS_PER_SLOT;
num_entries_to_write = i + 1;
break;
}
}
@ -206,24 +195,26 @@ impl ReplayStage {
let (ledger_entry_sender, ledger_entry_receiver) = channel();
let mut entry_stream = entry_stream.cloned().map(EntryStream::new);
let (_, mut current_slot) = bank
.get_current_leader()
.expect("Scheduled leader should be calculated by this point");
let mut max_tick_height_for_slot = bank
.leader_scheduler
.read()
.unwrap()
.max_tick_height_for_slot(current_slot);
let exit_ = exit.clone();
let t_replay = Builder::new()
.name("solana-replay-stage".to_string())
.spawn(move || {
let _exit = Finalizer::new(exit_.clone());
let (mut last_leader_id, _) = bank
.get_current_leader()
.expect("Scheduled leader should be calculated by this point");
let mut last_leader_id = Self::get_leader_for_next_tick(&bank);
let (mut current_slot, mut max_tick_height_for_slot) = {
let tick_height = bank.tick_height();
let leader_scheduler = bank.leader_scheduler.read().unwrap();
let current_slot = leader_scheduler.tick_height_to_slot(tick_height + 1);
let first_tick_in_current_slot =
current_slot * leader_scheduler.leader_rotation_interval;
(
current_slot,
first_tick_in_current_slot
+ leader_scheduler.num_ticks_left_in_slot(first_tick_in_current_slot),
)
};
// Loop through db_ledger MAX_ENTRY_RECV_PER_ITER entries at a time for each
// relevant slot to see if there are any available updates
loop {
@ -259,7 +250,7 @@ impl ReplayStage {
&last_entry_id,
entry_stream.as_mut(),
) {
error!("{:?}", e);
error!("process_entries failed: {:?}", e);
}
let current_tick_height = bank.tick_height();
@ -268,11 +259,15 @@ impl ReplayStage {
// for leader rotation
if max_tick_height_for_slot == current_tick_height {
// Check for leader rotation
let leader_id = Self::get_leader(&bank, &cluster_info);
let leader_id = Self::get_leader_for_next_tick(&bank);
// TODO: Remove this soon once we boot the leader from ClusterInfo
cluster_info.write().unwrap().set_leader(leader_id);
if leader_id != last_leader_id && my_id == leader_id {
to_leader_sender
.send(TvuReturnType::LeaderRotation(
bank.tick_height(),
current_tick_height,
*entry_height.read().unwrap(),
*last_entry_id.read().unwrap(),
))
@ -280,11 +275,11 @@ impl ReplayStage {
}
current_slot += 1;
max_tick_height_for_slot = bank
max_tick_height_for_slot += bank
.leader_scheduler
.read()
.unwrap()
.max_tick_height_for_slot(current_slot);
.leader_rotation_interval;
last_leader_id = leader_id;
}
}
@ -319,15 +314,13 @@ impl ReplayStage {
let _ = self.ledger_signal_sender.send(true);
}
fn get_leader(bank: &Bank, cluster_info: &Arc<RwLock<ClusterInfo>>) -> Pubkey {
let (scheduled_leader, _) = bank
.get_current_leader()
.expect("Scheduled leader should be calculated by this point");
// TODO: Remove this soon once we boot the leader from ClusterInfo
cluster_info.write().unwrap().set_leader(scheduled_leader);
scheduled_leader
fn get_leader_for_next_tick(bank: &Bank) -> Pubkey {
let tick_height = bank.tick_height();
let leader_scheduler = bank.leader_scheduler.read().unwrap();
let slot = leader_scheduler.tick_height_to_slot(tick_height + 1);
leader_scheduler
.get_leader_for_slot(slot)
.expect("Scheduled leader should be calculated by this point")
}
}
@ -350,9 +343,7 @@ mod test {
use crate::entry::Entry;
use crate::fullnode::Fullnode;
use crate::genesis_block::GenesisBlock;
use crate::leader_scheduler::{
make_active_set_entries, LeaderScheduler, LeaderSchedulerConfig,
};
use crate::leader_scheduler::{make_active_set_entries, LeaderSchedulerConfig};
use crate::replay_stage::ReplayStage;
use crate::tvu::TvuReturnType;
use crate::voting_keypair::VotingKeypair;
@ -366,6 +357,7 @@ mod test {
use std::sync::{Arc, RwLock};
#[test]
#[ignore] // TODO: Fix this test to not send all entries in slot 0
pub fn test_replay_stage_leader_rotation_exit() {
solana_logger::setup();
@ -379,43 +371,38 @@ mod test {
let old_leader_id = Keypair::new().pubkey();
// Create a ledger
let num_ending_ticks = 3;
let (mint_keypair, my_ledger_path, genesis_entry_height, mut last_id) =
create_tmp_sample_ledger(
"test_replay_stage_leader_rotation_exit",
10_000,
num_ending_ticks,
0,
old_leader_id,
500,
);
info!("my_id: {:?}", my_id);
info!("old_leader_id: {:?}", old_leader_id);
// Set up the LeaderScheduler so that my_id becomes the leader for epoch 1
let leader_rotation_interval = 16;
let leader_scheduler_config = LeaderSchedulerConfig::new(
leader_rotation_interval,
leader_rotation_interval,
leader_rotation_interval,
);
let my_keypair = Arc::new(my_keypair);
// Write two entries to the ledger so that the validator is in the active set:
// 1) Give the validator a nonzero number of tokens 2) A vote from the validator.
// This will cause leader rotation after the bootstrap height
let (active_set_entries, voting_keypair) =
make_active_set_entries(&my_keypair, &mint_keypair, &last_id, &last_id, 0);
let (active_set_entries, voting_keypair) = make_active_set_entries(
&my_keypair,
&mint_keypair,
100,
leader_rotation_interval, // add a vote for tick_height = leader_rotation_interval
&last_id,
&last_id,
0,
);
last_id = active_set_entries.last().unwrap().id;
let initial_tick_height = genesis_entry_height;
let active_set_entries_len = active_set_entries.len() as u64;
let initial_non_tick_height = genesis_entry_height - initial_tick_height;
{
// Set up the LeaderScheduler so that this this node becomes the leader at
// bootstrap_height = num_bootstrap_slots * leader_rotation_interval
let leader_rotation_interval = 16;
let bootstrap_height = 2 * leader_rotation_interval;
assert!((num_ending_ticks as u64) < bootstrap_height);
let leader_scheduler_config = LeaderSchedulerConfig::new(
bootstrap_height,
leader_rotation_interval,
leader_rotation_interval * 2,
bootstrap_height,
);
let leader_scheduler =
Arc::new(RwLock::new(LeaderScheduler::new(&leader_scheduler_config)));
let (db_ledger, l_sender, l_receiver) =
DbLedger::open_with_signal(&my_ledger_path).unwrap();
let db_ledger = Arc::new(db_ledger);
@ -431,18 +418,22 @@ mod test {
.expect("Expected to successfully open genesis block");
// Set up the bank
let (bank, _, last_entry_id) =
Fullnode::new_bank_from_db_ledger(&genesis_block, &db_ledger, leader_scheduler);
let (bank, _, last_entry_id) = Fullnode::new_bank_from_db_ledger(
&genesis_block,
&db_ledger,
Some(&leader_scheduler_config),
);
// Set up the replay stage
let (rotation_sender, rotation_receiver) = channel();
let meta = db_ledger.meta().unwrap().unwrap();
let exit = Arc::new(AtomicBool::new(false));
let bank = Arc::new(bank);
let (replay_stage, ledger_writer_recv) = ReplayStage::new(
my_id,
Some(Arc::new(voting_keypair)),
db_ledger.clone(),
Arc::new(bank),
bank.clone(),
Arc::new(RwLock::new(cluster_info_me)),
exit.clone(),
Arc::new(RwLock::new(meta.consumed)),
@ -453,36 +444,26 @@ mod test {
l_receiver,
);
// Send enough ticks to trigger leader rotation
let extra_entries = leader_rotation_interval;
let total_entries_to_send = (bootstrap_height + extra_entries) as usize;
let num_hashes = 1;
let total_entries_to_send = 2 * leader_rotation_interval as usize - 2;
let mut entries_to_send = vec![];
while entries_to_send.len() < total_entries_to_send {
let entry = Entry::new(&mut last_id, 0, num_hashes, vec![]);
let entry = Entry::new(&mut last_id, 0, 1, vec![]);
last_id = entry.id;
entries_to_send.push(entry);
}
assert!((num_ending_ticks as u64) < bootstrap_height);
// Add on the only entries that weren't ticks to the bootstrap height to get the
// total expected entry length
let leader_rotation_index = (bootstrap_height - initial_tick_height) as usize;
let expected_entry_height =
bootstrap_height + initial_non_tick_height + active_set_entries_len;
let expected_last_id = entries_to_send[leader_rotation_index - 1].id;
let expected_entry_height = (active_set_entries.len() + total_entries_to_send) as u64;
let expected_last_id = entries_to_send.last().unwrap().id;
// Write the entries to the ledger, replay_stage should get notified of changes
db_ledger
.write_entries(DEFAULT_SLOT_HEIGHT, meta.consumed, &entries_to_send)
.unwrap();
// Wait for replay_stage to exit and check return value is correct
info!("Wait for replay_stage to exit and check return value is correct");
assert_eq!(
Some(TvuReturnType::LeaderRotation(
bootstrap_height,
2 * leader_rotation_interval - 1,
expected_entry_height,
expected_last_id,
)),
@ -495,24 +476,20 @@ mod test {
}
);
// Check that the entries on the ledger writer channel are correct
info!("Check that the entries on the ledger writer channel are correct");
let mut received_ticks = ledger_writer_recv
.recv()
.expect("Expected to recieve an entry on the ledger writer receiver");
.expect("Expected to receive an entry on the ledger writer receiver");
while let Ok(entries) = ledger_writer_recv.try_recv() {
received_ticks.extend(entries);
}
assert_eq!(&received_ticks[..], &entries_to_send[..]);
assert_eq!(
&received_ticks[..],
&entries_to_send[..leader_rotation_index]
);
//replay stage should continue running even after rotation has happened (tvu never goes down)
// Replay stage should continue running even after rotation has happened (tvu never goes down)
assert_eq!(exit.load(Ordering::Relaxed), false);
//force exit
info!("Close replay_stage");
replay_stage
.close()
.expect("Expect successful ReplayStage exit");
@ -529,13 +506,11 @@ mod test {
// Create keypair for the leader
let leader_id = Keypair::new().pubkey();
let leader_scheduler = Arc::new(RwLock::new(LeaderScheduler::default()));
let num_ending_ticks = 1;
let (_, my_ledger_path, _, _) = create_tmp_sample_ledger(
"test_vote_error_replay_stage_correctness",
10_000,
num_ending_ticks,
1,
leader_id,
500,
);
@ -556,7 +531,7 @@ mod test {
let genesis_block = GenesisBlock::load(&my_ledger_path)
.expect("Expected to successfully open genesis block");
let (bank, entry_height, last_entry_id) =
Fullnode::new_bank_from_db_ledger(&genesis_block, &db_ledger, leader_scheduler);
Fullnode::new_bank_from_db_ledger(&genesis_block, &db_ledger, None);
let bank = Arc::new(bank);
let (replay_stage, ledger_writer_recv) = ReplayStage::new(
my_keypair.pubkey(),
@ -625,7 +600,7 @@ mod test {
// 1) Give the validator a nonzero number of tokens 2) A vote from the validator.
// This will cause leader rotation after the bootstrap height
let (active_set_entries, voting_keypair) =
make_active_set_entries(&my_keypair, &mint_keypair, &last_id, &last_id, 0);
make_active_set_entries(&my_keypair, &mint_keypair, 100, 1, &last_id, &last_id, 0);
let mut last_id = active_set_entries.last().unwrap().id;
let initial_tick_height = genesis_entry_height;
let active_set_entries_len = active_set_entries.len() as u64;
@ -639,15 +614,11 @@ mod test {
let num_bootstrap_slots = 2;
let bootstrap_height = num_bootstrap_slots * leader_rotation_interval;
let leader_scheduler_config = LeaderSchedulerConfig::new(
bootstrap_height,
leader_rotation_interval,
leader_rotation_interval * 2,
bootstrap_height,
);
let leader_scheduler =
Arc::new(RwLock::new(LeaderScheduler::new(&leader_scheduler_config)));
// Set up the cluster info
let cluster_info_me = Arc::new(RwLock::new(ClusterInfo::new(my_node.info.clone())));
@ -673,8 +644,11 @@ mod test {
// Set up the bank
let genesis_block = GenesisBlock::load(&my_ledger_path)
.expect("Expected to successfully open genesis block");
let (bank, _, last_entry_id) =
Fullnode::new_bank_from_db_ledger(&genesis_block, &db_ledger, leader_scheduler);
let (bank, _, last_entry_id) = Fullnode::new_bank_from_db_ledger(
&genesis_block,
&db_ledger,
Some(&leader_scheduler_config),
);
let voting_keypair = Arc::new(voting_keypair);
let bank = Arc::new(bank);
@ -776,7 +750,7 @@ mod test {
let voting_keypair = Arc::new(VotingKeypair::new_local(&my_keypair));
let res = ReplayStage::process_entries(
entries.clone(),
&Arc::new(Bank::default()),
&Arc::new(Bank::new(&GenesisBlock::new(10_000).0)),
&cluster_info_me,
Some(&voting_keypair),
&ledger_entry_sender,
@ -845,9 +819,10 @@ mod test {
let my_keypair = Arc::new(my_keypair);
let voting_keypair = Arc::new(VotingKeypair::new_local(&my_keypair));
let bank = Bank::new(&GenesisBlock::new(123).0);
ReplayStage::process_entries(
entries.clone(),
&Arc::new(Bank::default()),
&Arc::new(bank),
&cluster_info_me,
Some(&voting_keypair),
&ledger_entry_sender,

6
src/replicator.rs
View File

@ -128,7 +128,7 @@ impl Replicator {
{
let mut cluster_info_w = cluster_info.write().unwrap();
cluster_info_w.insert_info(leader_info.clone());
cluster_info_w.set_leader(leader_info.id);
cluster_info_w.set_leader(leader_pubkey);
}
// Create DbLedger, eventually will simply repurpose the input
@ -182,9 +182,7 @@ impl Replicator {
blob_fetch_receiver,
retransmit_sender,
repair_socket,
Arc::new(RwLock::new(LeaderScheduler::from_bootstrap_leader(
leader_pubkey,
))),
Arc::new(RwLock::new(LeaderScheduler::default())),
done.clone(),
exit.clone(),
);

3
src/rpc.rs
View File

@ -493,7 +493,8 @@ mod tests {
#[test]
fn test_rpc_new() {
let (genesis_block, alice) = GenesisBlock::new(10_000);
let (genesis_block, alice) =
GenesisBlock::new(10_000 + crate::genesis_block::BOOTSTRAP_LEADER_TOKENS);
let bank = Bank::new(&genesis_block);
let cluster_info = Arc::new(RwLock::new(ClusterInfo::new(NodeInfo::default())));
let rpc_addr = SocketAddr::new(

75
src/thin_client.rs
View File

@ -20,7 +20,7 @@ use solana_metrics::influxdb;
use solana_sdk::account::Account;
use solana_sdk::hash::Hash;
use solana_sdk::pubkey::Pubkey;
use solana_sdk::signature::{Keypair, Signature};
use solana_sdk::signature::{Keypair, KeypairUtil, Signature};
use solana_sdk::system_transaction::SystemTransaction;
use solana_sdk::timing;
use solana_sdk::transaction::Transaction;
@ -136,6 +136,13 @@ impl ThinClient {
to: Pubkey,
last_id: &Hash,
) -> io::Result<Signature> {
debug!(
"transfer: n={} from={:?} to={:?} last_id={:?}",
n,
keypair.pubkey(),
to,
last_id
);
let now = Instant::now();
let tx = SystemTransaction::new_account(keypair, to, n, *last_id, 0);
let result = self.transfer_signed(&tx);
@ -439,7 +446,7 @@ pub fn retry_get_balance(
pub fn new_fullnode(ledger_name: &'static str) -> (Fullnode, NodeInfo, Keypair, String) {
use crate::cluster_info::Node;
use crate::db_ledger::create_tmp_sample_ledger;
use crate::leader_scheduler::LeaderScheduler;
use crate::fullnode::Fullnode;
use crate::voting_keypair::VotingKeypair;
use solana_sdk::signature::KeypairUtil;
@ -450,14 +457,12 @@ pub fn new_fullnode(ledger_name: &'static str) -> (Fullnode, NodeInfo, Keypair,
let (mint_keypair, ledger_path, _, _) =
create_tmp_sample_ledger(ledger_name, 10_000, 0, node_info.id, 42);
let leader_scheduler = LeaderScheduler::from_bootstrap_leader(node_info.id);
let vote_account_keypair = Arc::new(Keypair::new());
let voting_keypair = VotingKeypair::new_local(&vote_account_keypair);
let node = Fullnode::new(
node,
&node_keypair,
&ledger_path,
Arc::new(RwLock::new(leader_scheduler)),
voting_keypair,
None,
&FullnodeConfig::default(),
@ -478,41 +483,52 @@ mod tests {
use std::fs::remove_dir_all;
#[test]
fn test_thin_client() {
fn test_thin_client_basic() {
solana_logger::setup();
let (server, leader_data, alice, ledger_path) = new_fullnode("thin_client");
let server_exit = server.run(None);
let bob_pubkey = Keypair::new().pubkey();
sleep(Duration::from_millis(900));
info!(
"found leader: {:?}",
poll_gossip_for_leader(leader_data.gossip, Some(5)).unwrap()
);
let mut client = mk_client(&leader_data);
let transaction_count = client.transaction_count();
assert_eq!(transaction_count, 0);
let confirmation = client.get_confirmation_time();
assert_eq!(confirmation, 18446744073709551615);
let last_id = client.get_last_id();
info!("test_thin_client last_id: {:?}", last_id);
let signature = client.transfer(500, &alice, bob_pubkey, &last_id).unwrap();
info!("test_thin_client signature: {:?}", signature);
client.poll_for_signature(&signature).unwrap();
let balance = client.get_balance(&bob_pubkey);
assert_eq!(balance.unwrap(), 500);
let transaction_count = client.transaction_count();
assert_eq!(transaction_count, 1);
server.close().unwrap();
server_exit();
remove_dir_all(ledger_path).unwrap();
}
// sleep(Duration::from_millis(300)); is unstable
#[test]
#[ignore]
fn test_bad_sig() {
solana_logger::setup();
let (server, leader_data, alice, ledger_path) = new_fullnode("bad_sig");
let server_exit = server.run(None);
let bob_pubkey = Keypair::new().pubkey();
//TODO: remove this sleep, or add a retry so CI is stable
sleep(Duration::from_millis(300));
info!(
"found leader: {:?}",
poll_gossip_for_leader(leader_data.gossip, Some(5)).unwrap()
);
let mut client = mk_client(&leader_data);
@ -534,24 +550,8 @@ mod tests {
client.poll_for_signature(&signature).unwrap();
let balance = client.get_balance(&bob_pubkey);
assert_eq!(balance.unwrap(), 500);
server.close().unwrap();
remove_dir_all(ledger_path).unwrap();
}
#[test]
fn test_client_check_signature() {
solana_logger::setup();
let (server, leader_data, alice, ledger_path) = new_fullnode("thin_client");
let bob_pubkey = Keypair::new().pubkey();
let mut client = mk_client(&leader_data);
let last_id = client.get_last_id();
let signature = client.transfer(500, &alice, bob_pubkey, &last_id).unwrap();
client.poll_for_signature(&signature).unwrap();
server.close().unwrap();
assert_eq!(balance.unwrap(), 1001);
server_exit();
remove_dir_all(ledger_path).unwrap();
}
@ -559,6 +559,11 @@ mod tests {
fn test_register_vote_account() {
solana_logger::setup();
let (server, leader_data, alice, ledger_path) = new_fullnode("thin_client");
let server_exit = server.run(None);
info!(
"found leader: {:?}",
poll_gossip_for_leader(leader_data.gossip, Some(5)).unwrap()
);
let mut client = mk_client(&leader_data);
@ -604,7 +609,7 @@ mod tests {
sleep(Duration::from_millis(900));
}
server.close().unwrap();
server_exit();
remove_dir_all(ledger_path).unwrap();
}
@ -623,8 +628,14 @@ mod tests {
fn test_zero_balance_after_nonzero() {
solana_logger::setup();
let (server, leader_data, alice, ledger_path) = new_fullnode("thin_client");
let server_exit = server.run(None);
let bob_keypair = Keypair::new();
info!(
"found leader: {:?}",
poll_gossip_for_leader(leader_data.gossip, Some(5)).unwrap()
);
let mut client = mk_client(&leader_data);
let last_id = client.get_last_id();
@ -651,7 +662,7 @@ mod tests {
let balance = client.poll_get_balance(&bob_keypair.pubkey());
assert!(balance.is_err());
server.close().unwrap();
server_exit();
remove_dir_all(ledger_path).unwrap();
}
}

4
src/tpu.rs
View File

@ -82,7 +82,7 @@ impl Tpu {
cluster_info: Arc<RwLock<ClusterInfo>>,
entry_height: u64,
sigverify_disabled: bool,
max_tick_height: Option<u64>,
max_tick_height: u64,
last_entry_id: &Hash,
leader_id: Pubkey,
is_leader: bool,
@ -171,7 +171,7 @@ impl Tpu {
broadcast_socket: UdpSocket,
cluster_info: Arc<RwLock<ClusterInfo>>,
sigverify_disabled: bool,
max_tick_height: Option<u64>,
max_tick_height: u64,
entry_height: u64,
last_entry_id: &Hash,
leader_id: Pubkey,

19
src/tvu.rs
View File

@ -203,10 +203,8 @@ pub mod tests {
use crate::db_ledger::get_tmp_ledger_path;
use crate::db_ledger::DbLedger;
use crate::entry::Entry;
use crate::gossip_service::GossipService;
use crate::leader_scheduler::LeaderScheduler;
use crate::genesis_block::GenesisBlock;
use crate::gossip_service::GossipService;
use crate::packet::SharedBlob;
use crate::service::Service;
use crate::storage_stage::{StorageState, STORAGE_ROTATE_TEST_COUNT};
@ -241,13 +239,7 @@ pub mod tests {
let starting_balance = 10_000;
let (genesis_block, _mint_keypair) = GenesisBlock::new(starting_balance);
let leader_id = leader.info.id;
let leader_scheduler = Arc::new(RwLock::new(LeaderScheduler::from_bootstrap_leader(
leader_id,
)));
let mut bank = Bank::new(&genesis_block);
bank.leader_scheduler = leader_scheduler;
let bank = Arc::new(bank);
let bank = Arc::new(Bank::new(&genesis_block));
//start cluster_info1
let mut cluster_info1 = ClusterInfo::new(target1.info.clone());
@ -332,12 +324,7 @@ pub mod tests {
let starting_balance = 10_000;
let (genesis_block, mint_keypair) = GenesisBlock::new(starting_balance);
let tvu_addr = target1.info.tvu;
let leader_scheduler = Arc::new(RwLock::new(LeaderScheduler::from_bootstrap_leader(
leader_id,
)));
let mut bank = Bank::new(&genesis_block);
bank.leader_scheduler = leader_scheduler;
let bank = Arc::new(bank);
let bank = Arc::new(Bank::new(&genesis_block));
//start cluster_info1
let mut cluster_info1 = ClusterInfo::new(target1.info.clone());

56
src/window.rs
View File

@ -128,38 +128,32 @@ impl WindowUtil for Window {
leader_scheduler_option: &Arc<RwLock<LeaderScheduler>>,
) -> Vec<(SocketAddr, Vec<u8>)> {
let rcluster_info = cluster_info.read().unwrap();
let mut is_next_leader = false;
{
let ls_lock = leader_scheduler_option.read().unwrap();
if !ls_lock.use_only_bootstrap_leader {
// Calculate the next leader rotation height and check if we are the leader
if let Some(next_leader_rotation_height) =
ls_lock.max_height_for_leader(tick_height)
{
match ls_lock.get_scheduled_leader(next_leader_rotation_height) {
Some((leader_id, _)) if leader_id == *id => is_next_leader = true,
// In the case that we are not in the current scope of the leader schedule
// window then either:
//
// 1) The replay stage hasn't caught up to the "consumed" entries we sent,
// in which case it will eventually catch up
//
// 2) We are on the border between seed_rotation_intervals, so the
// schedule won't be known until the entry on that cusp is received
// by the replay stage (which comes after this stage). Hence, the next
// leader at the beginning of that next epoch will not know they are the
// leader until they receive that last "cusp" entry. The leader also won't ask for repairs
// for that entry because "is_next_leader" won't be set here. In this case,
// everybody will be blocking waiting for that "cusp" entry instead of repairing,
// until the leader hits "times" >= the max times in calculate_max_repair().
// The impact of this, along with the similar problem from broadcast for the transitioning
// leader, can be observed in the multinode test, test_full_leader_validator_network(),
None => (),
_ => (),
}
}
// Check if we are the next next slot leader
let is_next_leader = {
let leader_scheduler = leader_scheduler_option.read().unwrap();
let next_slot = leader_scheduler.tick_height_to_slot(tick_height) + 1;
match leader_scheduler.get_leader_for_slot(next_slot) {
Some(leader_id) if leader_id == *id => true,
// In the case that we are not in the current scope of the leader schedule
// window then either:
//
// 1) The replay stage hasn't caught up to the "consumed" entries we sent,
// in which case it will eventually catch up
//
// 2) We are on the border between seed_rotation_intervals, so the
// schedule won't be known until the entry on that cusp is received
// by the replay stage (which comes after this stage). Hence, the next
// leader at the beginning of that next epoch will not know they are the
// leader until they receive that last "cusp" entry. The leader also won't ask for repairs
// for that entry because "is_next_leader" won't be set here. In this case,
// everybody will be blocking waiting for that "cusp" entry instead of repairing,
// until the leader hits "times" >= the max times in calculate_max_repair().
// The impact of this, along with the similar problem from broadcast for the transitioning
// leader, can be observed in the multinode test, test_full_leader_validator_network(),
None => false,
_ => false,
}
}
};
let num_peers = rcluster_info.repair_peers().len() as u64;
let max_repair = if max_entry_height == 0 {

8
src/window_service.rs
View File

@ -244,6 +244,8 @@ mod test {
let db_ledger = Arc::new(
DbLedger::open(&db_ledger_path).expect("Expected to be able to open database ledger"),
);
let mut leader_schedule = LeaderScheduler::default();
leader_schedule.set_leader_schedule(vec![me_id]);
let t_window = window_service(
db_ledger,
subs,
@ -253,7 +255,7 @@ mod test {
r_reader,
s_retransmit,
Arc::new(leader_node.sockets.repair),
Arc::new(RwLock::new(LeaderScheduler::from_bootstrap_leader(me_id))),
Arc::new(RwLock::new(leader_schedule)),
done,
exit.clone(),
);
@ -324,6 +326,8 @@ mod test {
let db_ledger = Arc::new(
DbLedger::open(&db_ledger_path).expect("Expected to be able to open database ledger"),
);
let mut leader_schedule = LeaderScheduler::default();
leader_schedule.set_leader_schedule(vec![me_id]);
let t_window = window_service(
db_ledger,
subs.clone(),
@ -333,7 +337,7 @@ mod test {
r_reader,
s_retransmit,
Arc::new(leader_node.sockets.repair),
Arc::new(RwLock::new(LeaderScheduler::from_bootstrap_leader(me_id))),
Arc::new(RwLock::new(leader_schedule)),
done,
exit.clone(),
);

844
tests/multinode.rs
View File

File diff suppressed because it is too large Load Diff

55
tests/replicator.rs
View File

@ -12,7 +12,6 @@ use solana::db_ledger::DbLedger;
use solana::db_ledger::{create_tmp_sample_ledger, get_tmp_ledger_path, tmp_copy_ledger};
use solana::entry::Entry;
use solana::fullnode::{Fullnode, FullnodeConfig};
use solana::leader_scheduler::LeaderScheduler;
use solana::replicator::Replicator;
use solana::storage_stage::STORAGE_ROTATE_TEST_COUNT;
use solana::streamer::blob_receiver;
@ -23,13 +22,12 @@ use solana_sdk::system_transaction::SystemTransaction;
use std::fs::remove_dir_all;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::mpsc::channel;
use std::sync::{Arc, RwLock};
use std::thread::sleep;
use std::sync::Arc;
use std::time::Duration;
#[test]
#[ignore]
fn test_replicator_startup() {
fn test_replicator_startup_basic() {
solana_logger::setup();
info!("starting replicator test");
let replicator_ledger_path = &get_tmp_ledger_path("replicator_test_replicator_ledger");
@ -41,7 +39,7 @@ fn test_replicator_startup() {
let leader_ledger_path = "replicator_test_leader_ledger";
let (mint_keypair, leader_ledger_path, _last_entry_height, _last_entry_id) =
create_tmp_sample_ledger(leader_ledger_path, 1_000_000_000, 0, leader_info.id, 1);
create_tmp_sample_ledger(leader_ledger_path, 1_000_000_000, 0, leader_info.id, 42);
let validator_ledger_path =
tmp_copy_ledger(&leader_ledger_path, "replicator_test_validator_ledger");
@ -55,21 +53,23 @@ fn test_replicator_startup() {
leader_node,
&leader_keypair,
&leader_ledger_path,
Arc::new(RwLock::new(LeaderScheduler::from_bootstrap_leader(
leader_info.id.clone(),
))),
voting_keypair,
None,
&fullnode_config,
);
let leader_exit = leader.run(None);
debug!(
"leader: {:?}",
solana::thin_client::poll_gossip_for_leader(leader_info.gossip, Some(5)).unwrap()
);
let validator_keypair = Arc::new(Keypair::new());
let voting_keypair = VotingKeypair::new_local(&validator_keypair);
let mut leader_client = mk_client(&leader_info);
let last_id = leader_client.get_last_id();
let mut leader_client = mk_client(&leader_info);
debug!("last_id: {:?}", last_id);
leader_client
.transfer(10, &mint_keypair, validator_keypair.pubkey(), &last_id)
@ -83,24 +83,29 @@ fn test_replicator_startup() {
validator_node,
&validator_keypair,
&validator_ledger_path,
Arc::new(RwLock::new(LeaderScheduler::from_bootstrap_leader(
leader_info.id,
))),
voting_keypair,
Some(&leader_info),
&fullnode_config,
);
let validator_exit = validator.run(None);
let bob = Keypair::new();
info!("starting transfers..");
for _ in 0..64 {
for i in 0..64 {
debug!("transfer {}", i);
let last_id = leader_client.get_last_id();
let mut transaction =
SystemTransaction::new_account(&mint_keypair, bob.pubkey(), 1, last_id, 0);
leader_client
.transfer(1, &mint_keypair, bob.pubkey(), &last_id)
.retry_transfer(&mint_keypair, &mut transaction, 5)
.unwrap();
sleep(Duration::from_millis(200));
debug!(
"transfer {}: mint balance={:?}, bob balance={:?}",
i,
leader_client.get_balance(&mint_keypair.pubkey()),
leader_client.get_balance(&bob.pubkey()),
);
}
let replicator_keypair = Keypair::new();
@ -109,11 +114,10 @@ fn test_replicator_startup() {
let last_id = leader_client.get_last_id();
// Give the replicator some tokens
let amount = 1;
let mut tx = SystemTransaction::new_account(
&mint_keypair,
replicator_keypair.pubkey(),
amount,
1,
last_id,
0,
);
@ -209,10 +213,11 @@ fn test_replicator_startup() {
}
replicator.close();
validator.exit();
leader.close().expect("Expected successful node closure");
validator_exit();
leader_exit();
}
info!("cleanup");
DbLedger::destroy(&leader_ledger_path).expect("Expected successful database destruction");
DbLedger::destroy(&replicator_ledger_path).expect("Expected successful database destruction");
let _ignored = remove_dir_all(&leader_ledger_path);
@ -271,7 +276,7 @@ fn test_replicator_startup_ledger_hang() {
let leader_ledger_path = "replicator_test_leader_ledger";
let (_mint_keypair, leader_ledger_path, _last_entry_height, _last_entry_id) =
create_tmp_sample_ledger(leader_ledger_path, 100, 0, leader_info.id, 1);
create_tmp_sample_ledger(leader_ledger_path, 100, 0, leader_info.id, 42);
let validator_ledger_path =
tmp_copy_ledger(&leader_ledger_path, "replicator_test_validator_ledger");
@ -283,9 +288,6 @@ fn test_replicator_startup_ledger_hang() {
leader_node,
&leader_keypair,
&leader_ledger_path,
Arc::new(RwLock::new(LeaderScheduler::from_bootstrap_leader(
leader_info.id.clone(),
))),
voting_keypair,
None,
&FullnodeConfig::default(),
@ -299,9 +301,6 @@ fn test_replicator_startup_ledger_hang() {
validator_node,
&validator_keypair,
&validator_ledger_path,
Arc::new(RwLock::new(LeaderScheduler::from_bootstrap_leader(
leader_info.id,
))),
voting_keypair,
Some(&leader_info),
&FullnodeConfig::default(),

3
tests/rpc.rs
View File

@ -17,6 +17,7 @@ fn test_rpc_send_tx() {
solana_logger::setup();
let (server, leader_data, alice, ledger_path) = new_fullnode("test_rpc_send_tx");
let server_exit = server.run(None);
let bob_pubkey = Keypair::new().pubkey();
let client = reqwest::Client::new();
@ -92,6 +93,6 @@ fn test_rpc_send_tx() {
assert_eq!(confirmed_tx, true);
server.close().unwrap();
server_exit();
remove_dir_all(ledger_path).unwrap();
}

5
wallet/tests/deploy.rs
View File

@ -11,6 +11,8 @@ use std::sync::mpsc::channel;
#[test]
fn test_wallet_deploy_program() {
solana_logger::setup();
let mut pathbuf = PathBuf::from(env!("CARGO_MANIFEST_DIR"));
pathbuf.push("tests");
pathbuf.push("fixtures");
@ -18,6 +20,7 @@ fn test_wallet_deploy_program() {
pathbuf.set_extension("so");
let (server, leader_data, alice, ledger_path) = new_fullnode("test_wallet_deploy_program");
let server_exit = server.run(None);
let (sender, receiver) = channel();
run_local_drone(alice, sender);
@ -73,6 +76,6 @@ fn test_wallet_deploy_program() {
&elf
);
server.close().unwrap();
server_exit();
remove_dir_all(ledger_path).unwrap();
}

9
wallet/tests/pay.rs
View File

@ -25,6 +25,7 @@ fn check_balance(expected_balance: u64, client: &RpcClient, params: Value) {
#[test]
fn test_wallet_timestamp_tx() {
let (server, leader_data, alice, ledger_path) = new_fullnode("test_wallet_timestamp_tx");
let server_exit = server.run(None);
let bob_pubkey = Keypair::new().pubkey();
let (sender, receiver) = channel();
@ -85,13 +86,14 @@ fn test_wallet_timestamp_tx() {
let params = json!([format!("{}", bob_pubkey)]);
check_balance(10, &rpc_client, params); // recipient balance
server.close().unwrap();
server_exit();
remove_dir_all(ledger_path).unwrap();
}
#[test]
fn test_wallet_witness_tx() {
let (server, leader_data, alice, ledger_path) = new_fullnode("test_wallet_witness_tx");
let server_exit = server.run(None);
let bob_pubkey = Keypair::new().pubkey();
let (sender, receiver) = channel();
@ -148,13 +150,14 @@ fn test_wallet_witness_tx() {
let params = json!([format!("{}", bob_pubkey)]);
check_balance(10, &rpc_client, params); // recipient balance
server.close().unwrap();
server_exit();
remove_dir_all(ledger_path).unwrap();
}
#[test]
fn test_wallet_cancel_tx() {
let (server, leader_data, alice, ledger_path) = new_fullnode("test_wallet_cancel_tx");
let server_exit = server.run(None);
let bob_pubkey = Keypair::new().pubkey();
let (sender, receiver) = channel();
@ -211,6 +214,6 @@ fn test_wallet_cancel_tx() {
let params = json!([format!("{}", bob_pubkey)]);
check_balance(0, &rpc_client, params); // recipient balance
server.close().unwrap();
server_exit();
remove_dir_all(ledger_path).unwrap();
}

4
wallet/tests/request_airdrop.rs
View File

@ -10,7 +10,7 @@ use std::sync::mpsc::channel;
#[test]
fn test_wallet_request_airdrop() {
let (server, leader_data, alice, ledger_path) = new_fullnode("test_wallet_request_airdrop");
let server_exit = server.run(None);
let (sender, receiver) = channel();
run_local_drone(alice, sender);
let drone_addr = receiver.recv().unwrap();
@ -33,6 +33,6 @@ fn test_wallet_request_airdrop() {
.unwrap();
assert_eq!(balance, 50);
server.close().unwrap();
server_exit();
remove_dir_all(ledger_path).unwrap();
}