gaspersic/solana
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Fix resetting PohRecorder to wrong bank (#3553) (#3574)

Browse Source 
* Check whether future slot already has transmission
This commit is contained in:
carllin
2019-03-30 02:06:49 -07:00
committed by GitHub
parent 9568a1da03
commit bf11d7ef43
9 changed files with 1173 additions and 842 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
benches/banking_stage.rs
View File

@ -1,10 +1,13 @@
#![feature(test)]
extern crate test;
#[macro_use]
extern crate solana;
use rand::{thread_rng, Rng};
use rayon::prelude::*;
use solana::banking_stage::{create_test_recorder, BankingStage};
use solana::blocktree::{get_tmp_ledger_path, Blocktree};
use solana::cluster_info::ClusterInfo;
use solana::cluster_info::Node;
use solana::packet::to_packets_chunked;
@ -104,7 +107,13 @@ fn bench_banking_stage_multi_accounts(bencher: &mut Bencher) {
(x, iter::repeat(1).take(len).collect())
})
.collect();
let (exit, poh_recorder, poh_service, signal_receiver) = create_test_recorder(&bank);
let ledger_path = get_tmp_ledger_path!();
{
let blocktree = Arc::new(
Blocktree::open(&ledger_path).expect("Expected to be able to open database ledger"),
);
let (exit, poh_recorder, poh_service, signal_receiver) =
create_test_recorder(&bank, &blocktree);
let cluster_info = ClusterInfo::new_with_invalid_keypair(Node::new_localhost().info);
let cluster_info = Arc::new(RwLock::new(cluster_info));
let _banking_stage = BankingStage::new(&cluster_info, &poh_recorder, verified_receiver);
@ -131,6 +140,8 @@ fn bench_banking_stage_multi_accounts(bencher: &mut Bencher) {
});
exit.store(true, Ordering::Relaxed);
poh_service.join().unwrap();
}
Blocktree::destroy(&ledger_path).unwrap();
}
#[bench]
@ -211,7 +222,14 @@ fn bench_banking_stage_multi_programs(bencher: &mut Bencher) {
(x, iter::repeat(1).take(len).collect())
})
.collect();
let (exit, poh_recorder, poh_service, signal_receiver) = create_test_recorder(&bank);
let ledger_path = get_tmp_ledger_path!();
{
let blocktree = Arc::new(
Blocktree::open(&ledger_path).expect("Expected to be able to open database ledger"),
);
let (exit, poh_recorder, poh_service, signal_receiver) =
create_test_recorder(&bank, &blocktree);
let cluster_info = ClusterInfo::new_with_invalid_keypair(Node::new_localhost().info);
let cluster_info = Arc::new(RwLock::new(cluster_info));
let _banking_stage = BankingStage::new(&cluster_info, &poh_recorder, verified_receiver);
@ -238,4 +256,6 @@ fn bench_banking_stage_multi_programs(bencher: &mut Bencher) {
});
exit.store(true, Ordering::Relaxed);
poh_service.join().unwrap();
}
Blocktree::destroy(&ledger_path).unwrap();
}

71
core/src/banking_stage.rs
View File

@ -1,7 +1,7 @@
//! The `banking_stage` processes Transaction messages. It is intended to be used
//! to contruct a software pipeline. The stage uses all available CPU cores and
//! can do its processing in parallel with signature verification on the GPU.
use crate::blocktree::Blocktree;
use crate::cluster_info::ClusterInfo;
use crate::entry::Entry;
use crate::leader_schedule_utils;
@ -437,6 +437,7 @@ impl Service for BankingStage {
pub fn create_test_recorder(
bank: &Arc<Bank>,
blocktree: &Arc<Blocktree>,
) -> (
Arc<AtomicBool>,
Arc<Mutex<PohRecorder>>,
@ -451,6 +452,7 @@ pub fn create_test_recorder(
Some(4),
bank.ticks_per_slot(),
&Pubkey::default(),
blocktree,
);
let poh_recorder = Arc::new(Mutex::new(poh_recorder));
let poh_service = PohService::new(poh_recorder.clone(), &PohServiceConfig::default(), &exit);
@ -460,10 +462,12 @@ pub fn create_test_recorder(
#[cfg(test)]
mod tests {
use super::*;
use crate::blocktree::get_tmp_ledger_path;
use crate::cluster_info::Node;
use crate::entry::EntrySlice;
use crate::packet::to_packets;
use crate::poh_recorder::WorkingBank;
use crate::{get_tmp_ledger_path, tmp_ledger_name};
use solana_sdk::genesis_block::GenesisBlock;
use solana_sdk::native_program::ProgramError;
use solana_sdk::signature::{Keypair, KeypairUtil};
@ -476,7 +480,13 @@ mod tests {
let (genesis_block, _mint_keypair) = GenesisBlock::new(2);
let bank = Arc::new(Bank::new(&genesis_block));
let (verified_sender, verified_receiver) = channel();
let (exit, poh_recorder, poh_service, _entry_receiever) = create_test_recorder(&bank);
let ledger_path = get_tmp_ledger_path!();
{
let blocktree = Arc::new(
Blocktree::open(&ledger_path).expect("Expected to be able to open database ledger"),
);
let (exit, poh_recorder, poh_service, _entry_receiever) =
create_test_recorder(&bank, &blocktree);
let cluster_info = ClusterInfo::new_with_invalid_keypair(Node::new_localhost().info);
let cluster_info = Arc::new(RwLock::new(cluster_info));
let banking_stage = BankingStage::new(&cluster_info, &poh_recorder, verified_receiver);
@ -485,6 +495,8 @@ mod tests {
banking_stage.join().unwrap();
poh_service.join().unwrap();
}
Blocktree::destroy(&ledger_path).unwrap();
}
#[test]
fn test_banking_stage_tick() {
@ -494,7 +506,13 @@ mod tests {
let bank = Arc::new(Bank::new(&genesis_block));
let start_hash = bank.last_blockhash();
let (verified_sender, verified_receiver) = channel();
let (exit, poh_recorder, poh_service, entry_receiver) = create_test_recorder(&bank);
let ledger_path = get_tmp_ledger_path!();
{
let blocktree = Arc::new(
Blocktree::open(&ledger_path).expect("Expected to be able to open database ledger"),
);
let (exit, poh_recorder, poh_service, entry_receiver) =
create_test_recorder(&bank, &blocktree);
let cluster_info = ClusterInfo::new_with_invalid_keypair(Node::new_localhost().info);
let cluster_info = Arc::new(RwLock::new(cluster_info));
poh_recorder.lock().unwrap().set_bank(&bank);
@ -517,6 +535,8 @@ mod tests {
assert_eq!(entries[entries.len() - 1].hash, bank.last_blockhash());
banking_stage.join().unwrap();
}
Blocktree::destroy(&ledger_path).unwrap();
}
#[test]
fn test_banking_stage_entries_only() {
@ -524,7 +544,13 @@ mod tests {
let bank = Arc::new(Bank::new(&genesis_block));
let start_hash = bank.last_blockhash();
let (verified_sender, verified_receiver) = channel();
let (exit, poh_recorder, poh_service, entry_receiver) = create_test_recorder(&bank);
let ledger_path = get_tmp_ledger_path!();
{
let blocktree = Arc::new(
Blocktree::open(&ledger_path).expect("Expected to be able to open database ledger"),
);
let (exit, poh_recorder, poh_service, entry_receiver) =
create_test_recorder(&bank, &blocktree);
let cluster_info = ClusterInfo::new_with_invalid_keypair(Node::new_localhost().info);
let cluster_info = Arc::new(RwLock::new(cluster_info));
poh_recorder.lock().unwrap().set_bank(&bank);
@ -540,7 +566,8 @@ mod tests {
// bad tx, AccountNotFound
let keypair = Keypair::new();
let tx_anf = SystemTransaction::new_account(&keypair, &keypair.pubkey(), 1, start_hash, 0);
let tx_anf =
SystemTransaction::new_account(&keypair, &keypair.pubkey(), 1, start_hash, 0);
// send 'em over
let packets = to_packets(&[tx, tx_no_ver, tx_anf]);
@ -573,6 +600,8 @@ mod tests {
drop(entry_receiver);
banking_stage.join().unwrap();
}
Blocktree::destroy(&ledger_path).unwrap();
}
#[test]
fn test_banking_stage_entryfication() {
@ -582,7 +611,13 @@ mod tests {
let (genesis_block, mint_keypair) = GenesisBlock::new(2);
let bank = Arc::new(Bank::new(&genesis_block));
let (verified_sender, verified_receiver) = channel();
let (exit, poh_recorder, poh_service, entry_receiver) = create_test_recorder(&bank);
let ledger_path = get_tmp_ledger_path!();
{
let blocktree = Arc::new(
Blocktree::open(&ledger_path).expect("Expected to be able to open database ledger"),
);
let (exit, poh_recorder, poh_service, entry_receiver) =
create_test_recorder(&bank, &blocktree);
let cluster_info = ClusterInfo::new_with_invalid_keypair(Node::new_localhost().info);
let cluster_info = Arc::new(RwLock::new(cluster_info));
poh_recorder.lock().unwrap().set_bank(&bank);
@ -642,11 +677,8 @@ mod tests {
sleep(Duration::from_millis(100));
}
// Assert the user holds one lamport, not two. If the stage only outputs one
// entry, then the second transaction will be rejected, because it drives
// the account balance below zero before the credit is added.
assert_eq!(bank.get_balance(&alice.pubkey()), 1);
}
Blocktree::destroy(&ledger_path).unwrap();
}
#[test]
@ -659,6 +691,10 @@ mod tests {
max_tick_height: std::u64::MAX,
};
let ledger_path = get_tmp_ledger_path!();
{
let blocktree =
Blocktree::open(&ledger_path).expect("Expected to be able to open database ledger");
let (poh_recorder, entry_receiver) = PohRecorder::new(
bank.tick_height(),
bank.last_blockhash(),
@ -666,6 +702,7 @@ mod tests {
None,
bank.ticks_per_slot(),
&Pubkey::default(),
&Arc::new(blocktree),
);
let poh_recorder = Arc::new(Mutex::new(poh_recorder));
@ -697,6 +734,8 @@ mod tests {
let (_, entries) = entry_receiver.recv().unwrap();
assert_eq!(entries[0].0.transactions.len(), transactions.len() - 1);
}
Blocktree::destroy(&ledger_path).unwrap();
}
#[test]
fn test_bank_process_and_record_transactions() {
@ -718,6 +757,10 @@ mod tests {
min_tick_height: bank.tick_height(),
max_tick_height: bank.tick_height() + 1,
};
let ledger_path = get_tmp_ledger_path!();
{
let blocktree =
Blocktree::open(&ledger_path).expect("Expected to be able to open database ledger");
let (poh_recorder, entry_receiver) = PohRecorder::new(
bank.tick_height(),
bank.last_blockhash(),
@ -725,12 +768,14 @@ mod tests {
Some(4),
bank.ticks_per_slot(),
&pubkey,
&Arc::new(blocktree),
);
let poh_recorder = Arc::new(Mutex::new(poh_recorder));
poh_recorder.lock().unwrap().set_working_bank(working_bank);
BankingStage::process_and_record_transactions(&bank, &transactions, &poh_recorder).unwrap();
BankingStage::process_and_record_transactions(&bank, &transactions, &poh_recorder)
.unwrap();
poh_recorder.lock().unwrap().tick();
let mut done = false;
@ -767,4 +812,6 @@ mod tests {
assert_eq!(bank.get_balance(&pubkey), 1);
}
Blocktree::destroy(&ledger_path).unwrap();
}
}

5
core/src/fullnode.rs
View File

@ -108,13 +108,15 @@ impl Fullnode {
bank.tick_height(),
bank.last_blockhash(),
);
let blocktree = Arc::new(blocktree);
let (poh_recorder, entry_receiver) = PohRecorder::new(
bank.tick_height(),
bank.last_blockhash(),
bank.slot(),
leader_schedule_utils::next_leader_slot(&id, bank.slot(), &bank),
leader_schedule_utils::next_leader_slot(&id, bank.slot(), &bank, Some(&blocktree)),
bank.ticks_per_slot(),
&id,
&blocktree,
);
let poh_recorder = Arc::new(Mutex::new(poh_recorder));
let poh_service = PohService::new(poh_recorder.clone(), &config.tick_config, &exit);
@ -133,7 +135,6 @@ impl Fullnode {
node.sockets.gossip.local_addr().unwrap()
);
let blocktree = Arc::new(blocktree);
let bank_forks = Arc::new(RwLock::new(bank_forks));
node.info.wallclock = timestamp();

151
core/src/leader_schedule_utils.rs
View File

@ -1,3 +1,4 @@
use crate::blocktree::Blocktree;
use crate::leader_schedule::LeaderSchedule;
use crate::staking_utils;
use solana_runtime::bank::Bank;
@ -44,7 +45,12 @@ pub fn slot_leader_at(slot: u64, bank: &Bank) -> Option<Pubkey> {
}
/// Return the next slot after the given current_slot that the given node will be leader
pub fn next_leader_slot(pubkey: &Pubkey, mut current_slot: u64, bank: &Bank) -> Option<u64> {
pub fn next_leader_slot(
pubkey: &Pubkey,
mut current_slot: u64,
bank: &Bank,
blocktree: Option<&Blocktree>,
) -> Option<u64> {
let (mut epoch, mut start_index) = bank.get_epoch_and_slot_index(current_slot + 1);
while let Some(leader_schedule) = leader_schedule(epoch, bank) {
// clippy thinks I should do this:
@ -59,6 +65,15 @@ pub fn next_leader_slot(pubkey: &Pubkey, mut current_slot: u64, bank: &Bank) ->
for i in start_index..bank.get_slots_in_epoch(epoch) {
current_slot += 1;
if *pubkey == leader_schedule[i] {
if let Some(blocktree) = blocktree {
if let Some(meta) = blocktree.meta(current_slot).unwrap() {
// We have already sent a blob for this slot, so skip it
if meta.received > 0 {
continue;
}
}
}
return Some(current_slot);
}
}
@ -82,9 +97,13 @@ pub fn tick_height_to_slot(ticks_per_slot: u64, tick_height: u64) -> u64 {
#[cfg(test)]
mod tests {
use super::*;
use crate::blocktree::get_tmp_ledger_path;
use crate::blocktree::tests::make_slot_entries;
use crate::staking_utils;
use crate::voting_keypair::tests::new_vote_account_with_delegate;
use solana_sdk::genesis_block::{GenesisBlock, BOOTSTRAP_LEADER_LAMPORTS};
use solana_sdk::signature::{Keypair, KeypairUtil};
use std::sync::Arc;
#[test]
fn test_next_leader_slot() {
@ -99,13 +118,14 @@ mod tests {
let bank = Bank::new(&genesis_block);
assert_eq!(slot_leader_at(bank.slot(), &bank).unwrap(), pubkey);
assert_eq!(next_leader_slot(&pubkey, 0, &bank), Some(1));
assert_eq!(next_leader_slot(&pubkey, 1, &bank), Some(2));
assert_eq!(next_leader_slot(&pubkey, 0, &bank, None), Some(1));
assert_eq!(next_leader_slot(&pubkey, 1, &bank, None), Some(2));
assert_eq!(
next_leader_slot(
&pubkey,
2 * genesis_block.slots_per_epoch - 1, // no schedule generated for epoch 2
&bank
&bank,
None
),
None
);
@ -114,12 +134,133 @@ mod tests {
next_leader_slot(
&Keypair::new().pubkey(), // not in leader_schedule
0,
&bank
&bank,
None
),
None
);
}
#[test]
fn test_next_leader_slot_blocktree() {
let pubkey = Keypair::new().pubkey();
let mut genesis_block = GenesisBlock::new_with_leader(
BOOTSTRAP_LEADER_LAMPORTS,
&pubkey,
BOOTSTRAP_LEADER_LAMPORTS,
)
.0;
genesis_block.epoch_warmup = false;
let bank = Bank::new(&genesis_block);
let ledger_path = get_tmp_ledger_path!();
{
let blocktree = Arc::new(
Blocktree::open(&ledger_path).expect("Expected to be able to open database ledger"),
);
assert_eq!(slot_leader_at(bank.slot(), &bank).unwrap(), pubkey);
// Check that the next leader slot after 0 is slot 1
assert_eq!(
next_leader_slot(&pubkey, 0, &bank, Some(&blocktree)),
Some(1)
);
// Write a blob into slot 2 that chains to slot 1,
// but slot 1 is empty so should not be skipped
let (blobs, _) = make_slot_entries(2, 1, 1);
blocktree.write_blobs(&blobs[..]).unwrap();
assert_eq!(
next_leader_slot(&pubkey, 0, &bank, Some(&blocktree)),
Some(1)
);
// Write a blob into slot 1
let (blobs, _) = make_slot_entries(1, 0, 1);
// Check that slot 1 and 2 are skipped
blocktree.write_blobs(&blobs[..]).unwrap();
assert_eq!(
next_leader_slot(&pubkey, 0, &bank, Some(&blocktree)),
Some(3)
);
// Integrity checks
assert_eq!(
next_leader_slot(
&pubkey,
2 * genesis_block.slots_per_epoch - 1, // no schedule generated for epoch 2
&bank,
Some(&blocktree)
),
None
);
assert_eq!(
next_leader_slot(
&Keypair::new().pubkey(), // not in leader_schedule
0,
&bank,
Some(&blocktree)
),
None
);
}
Blocktree::destroy(&ledger_path).unwrap();
}
#[test]
fn test_next_leader_slot_next_epoch() {
let pubkey = Keypair::new().pubkey();
let (mut genesis_block, mint_keypair) = GenesisBlock::new_with_leader(
2 * BOOTSTRAP_LEADER_LAMPORTS,
&pubkey,
BOOTSTRAP_LEADER_LAMPORTS,
);
genesis_block.epoch_warmup = false;
let bank = Bank::new(&genesis_block);
let delegate_id = Keypair::new().pubkey();
// Create new vote account
let new_voting_keypair = Keypair::new();
new_vote_account_with_delegate(
&mint_keypair,
&new_voting_keypair,
&delegate_id,
&bank,
BOOTSTRAP_LEADER_LAMPORTS,
);
// Have to wait until the epoch at after the epoch stakes generated at genesis
// for the new votes to take effect.
let mut target_slot = 1;
let epoch = bank.get_stakers_epoch(0);
while bank.get_stakers_epoch(target_slot) == epoch {
target_slot += 1;
}
let bank = Bank::new_from_parent(&Arc::new(bank), &Pubkey::default(), target_slot);
let mut expected_slot = 0;
let epoch = bank.get_stakers_epoch(target_slot);
for i in 0..epoch {
expected_slot += bank.get_slots_in_epoch(i);
}
let schedule = leader_schedule(epoch, &bank).unwrap();
let mut index = 0;
while schedule[index] != delegate_id {
index += 1
}
expected_slot += index;
assert_eq!(
next_leader_slot(&delegate_id, 0, &bank, None),
Some(expected_slot),
);
}
#[test]
fn test_leader_schedule_via_bank() {
let pubkey = Keypair::new().pubkey();

138
core/src/poh_recorder.rs
View File

@ -10,6 +10,7 @@
//! For Entries:
//! * recorded entry must be >= WorkingBank::min_tick_height && entry must be < WorkingBank::man_tick_height
//!
use crate::blocktree::Blocktree;
use crate::entry::Entry;
use crate::leader_schedule_utils;
use crate::poh::Poh;
@ -51,14 +52,19 @@ pub struct PohRecorder {
last_leader_tick: Option<u64>,
max_last_leader_grace_ticks: u64,
id: Pubkey,
blocktree: Arc<Blocktree>,
}
impl PohRecorder {
pub fn clear_bank(&mut self) {
if let Some(working_bank) = self.working_bank.take() {
let bank = working_bank.bank;
let next_leader_slot =
leader_schedule_utils::next_leader_slot(&self.id, bank.slot(), &bank);
let next_leader_slot = leader_schedule_utils::next_leader_slot(
&self.id,
bank.slot(),
&bank,
Some(&self.blocktree),
);
let (start_leader_at_tick, last_leader_tick) = Self::compute_leader_slot_ticks(
&next_leader_slot,
bank.ticks_per_slot(),
@ -272,6 +278,7 @@ impl PohRecorder {
my_leader_slot_index: Option<u64>,
ticks_per_slot: u64,
id: &Pubkey,
blocktree: &Arc<Blocktree>,
) -> (Self, Receiver<WorkingBankEntries>) {
let poh = Poh::new(last_entry_hash, tick_height);
let (sender, receiver) = channel();
@ -294,6 +301,7 @@ impl PohRecorder {
last_leader_tick,
max_last_leader_grace_ticks,
id: *id,
blocktree: blocktree.clone(),
},
receiver,
)
@ -336,6 +344,7 @@ impl PohRecorder {
#[cfg(test)]
mod tests {
use super::*;
use crate::blocktree::{get_tmp_ledger_path, Blocktree};
use crate::test_tx::test_tx;
use solana_sdk::genesis_block::GenesisBlock;
use solana_sdk::hash::hash;
@ -346,6 +355,11 @@ mod tests {
#[test]
fn test_poh_recorder_no_zero_tick() {
let prev_hash = Hash::default();
let ledger_path = get_tmp_ledger_path!();
{
let blocktree =
Blocktree::open(&ledger_path).expect("Expected to be able to open database ledger");
let (mut poh_recorder, _entry_receiver) = PohRecorder::new(
0,
prev_hash,
@ -353,16 +367,24 @@ mod tests {
Some(4),
DEFAULT_TICKS_PER_SLOT,
&Pubkey::default(),
&Arc::new(blocktree),
);
poh_recorder.tick();
assert_eq!(poh_recorder.tick_cache.len(), 1);
assert_eq!(poh_recorder.tick_cache[0].1, 1);
assert_eq!(poh_recorder.poh.tick_height, 1);
}
Blocktree::destroy(&ledger_path).unwrap();
}
#[test]
fn test_poh_recorder_tick_height_is_last_tick() {
let prev_hash = Hash::default();
let ledger_path = get_tmp_ledger_path!();
{
let blocktree =
Blocktree::open(&ledger_path).expect("Expected to be able to open database ledger");
let (mut poh_recorder, _entry_receiver) = PohRecorder::new(
0,
prev_hash,
@ -370,6 +392,7 @@ mod tests {
Some(4),
DEFAULT_TICKS_PER_SLOT,
&Pubkey::default(),
&Arc::new(blocktree),
);
poh_recorder.tick();
poh_recorder.tick();
@ -377,9 +400,15 @@ mod tests {
assert_eq!(poh_recorder.tick_cache[1].1, 2);
assert_eq!(poh_recorder.poh.tick_height, 2);
}
Blocktree::destroy(&ledger_path).unwrap();
}
#[test]
fn test_poh_recorder_reset_clears_cache() {
let ledger_path = get_tmp_ledger_path!();
{
let blocktree =
Blocktree::open(&ledger_path).expect("Expected to be able to open database ledger");
let (mut poh_recorder, _entry_receiver) = PohRecorder::new(
0,
Hash::default(),
@ -387,15 +416,22 @@ mod tests {
Some(4),
DEFAULT_TICKS_PER_SLOT,
&Pubkey::default(),
&Arc::new(blocktree),
);
poh_recorder.tick();
assert_eq!(poh_recorder.tick_cache.len(), 1);
poh_recorder.reset(0, Hash::default(), 0, Some(4), DEFAULT_TICKS_PER_SLOT);
assert_eq!(poh_recorder.tick_cache.len(), 0);
}
Blocktree::destroy(&ledger_path).unwrap();
}
#[test]
fn test_poh_recorder_clear() {
let ledger_path = get_tmp_ledger_path!();
{
let blocktree =
Blocktree::open(&ledger_path).expect("Expected to be able to open database ledger");
let (genesis_block, _mint_keypair) = GenesisBlock::new(2);
let bank = Arc::new(Bank::new(&genesis_block));
let prev_hash = bank.last_blockhash();
@ -406,6 +442,7 @@ mod tests {
Some(4),
bank.ticks_per_slot(),
&Pubkey::default(),
&Arc::new(blocktree),
);
let working_bank = WorkingBank {
@ -418,9 +455,15 @@ mod tests {
poh_recorder.clear_bank();
assert!(poh_recorder.working_bank.is_none());
}
Blocktree::destroy(&ledger_path).unwrap();
}
#[test]
fn test_poh_recorder_tick_sent_after_min() {
let ledger_path = get_tmp_ledger_path!();
{
let blocktree =
Blocktree::open(&ledger_path).expect("Expected to be able to open database ledger");
let (genesis_block, _mint_keypair) = GenesisBlock::new(2);
let bank = Arc::new(Bank::new(&genesis_block));
let prev_hash = bank.last_blockhash();
@ -431,6 +474,7 @@ mod tests {
Some(4),
bank.ticks_per_slot(),
&Pubkey::default(),
&Arc::new(blocktree),
);
let working_bank = WorkingBank {
@ -455,9 +499,15 @@ mod tests {
assert_eq!(bank_.slot(), bank.slot());
assert!(poh_recorder.working_bank.is_none());
}
Blocktree::destroy(&ledger_path).unwrap();
}
#[test]
fn test_poh_recorder_tick_sent_upto_and_including_max() {
let ledger_path = get_tmp_ledger_path!();
{
let blocktree =
Blocktree::open(&ledger_path).expect("Expected to be able to open database ledger");
let (genesis_block, _mint_keypair) = GenesisBlock::new(2);
let bank = Arc::new(Bank::new(&genesis_block));
let prev_hash = bank.last_blockhash();
@ -468,6 +518,7 @@ mod tests {
Some(4),
bank.ticks_per_slot(),
&Pubkey::default(),
&Arc::new(blocktree),
);
poh_recorder.tick();
@ -490,9 +541,15 @@ mod tests {
let (_, e) = entry_receiver.recv().expect("recv 1");
assert_eq!(e.len(), 3);
}
Blocktree::destroy(&ledger_path).unwrap();
}
#[test]
fn test_poh_recorder_record_to_early() {
let ledger_path = get_tmp_ledger_path!();
{
let blocktree =
Blocktree::open(&ledger_path).expect("Expected to be able to open database ledger");
let (genesis_block, _mint_keypair) = GenesisBlock::new(2);
let bank = Arc::new(Bank::new(&genesis_block));
let prev_hash = bank.last_blockhash();
@ -503,6 +560,7 @@ mod tests {
Some(4),
bank.ticks_per_slot(),
&Pubkey::default(),
&Arc::new(blocktree),
);
let working_bank = WorkingBank {
@ -517,9 +575,15 @@ mod tests {
assert!(poh_recorder.record(h1, vec![tx.clone()]).is_err());
assert!(entry_receiver.try_recv().is_err());
}
Blocktree::destroy(&ledger_path).unwrap();
}
#[test]
fn test_poh_recorder_record_at_min_passes() {
let ledger_path = get_tmp_ledger_path!();
{
let blocktree =
Blocktree::open(&ledger_path).expect("Expected to be able to open database ledger");
let (genesis_block, _mint_keypair) = GenesisBlock::new(2);
let bank = Arc::new(Bank::new(&genesis_block));
let prev_hash = bank.last_blockhash();
@ -530,6 +594,7 @@ mod tests {
Some(4),
bank.ticks_per_slot(),
&Pubkey::default(),
&Arc::new(blocktree),
);
let working_bank = WorkingBank {
@ -553,9 +618,15 @@ mod tests {
let (_b, e) = entry_receiver.recv().expect("recv 2");
assert!(!e[0].0.is_tick());
}
Blocktree::destroy(&ledger_path).unwrap();
}
#[test]
fn test_poh_recorder_record_at_max_fails() {
let ledger_path = get_tmp_ledger_path!();
{
let blocktree =
Blocktree::open(&ledger_path).expect("Expected to be able to open database ledger");
let (genesis_block, _mint_keypair) = GenesisBlock::new(2);
let bank = Arc::new(Bank::new(&genesis_block));
let prev_hash = bank.last_blockhash();
@ -566,6 +637,7 @@ mod tests {
Some(4),
bank.ticks_per_slot(),
&Pubkey::default(),
&Arc::new(blocktree),
);
let working_bank = WorkingBank {
@ -586,9 +658,15 @@ mod tests {
assert!(e[0].0.is_tick());
assert!(e[1].0.is_tick());
}
Blocktree::destroy(&ledger_path).unwrap();
}
#[test]
fn test_poh_cache_on_disconnect() {
let ledger_path = get_tmp_ledger_path!();
{
let blocktree =
Blocktree::open(&ledger_path).expect("Expected to be able to open database ledger");
let (genesis_block, _mint_keypair) = GenesisBlock::new(2);
let bank = Arc::new(Bank::new(&genesis_block));
let prev_hash = bank.last_blockhash();
@ -599,6 +677,7 @@ mod tests {
Some(4),
bank.ticks_per_slot(),
&Pubkey::default(),
&Arc::new(blocktree),
);
let working_bank = WorkingBank {
@ -615,9 +694,15 @@ mod tests {
assert!(poh_recorder.working_bank.is_none());
assert_eq!(poh_recorder.tick_cache.len(), 3);
}
Blocktree::destroy(&ledger_path).unwrap();
}
#[test]
fn test_reset_current() {
let ledger_path = get_tmp_ledger_path!();
{
let blocktree =
Blocktree::open(&ledger_path).expect("Expected to be able to open database ledger");
let (mut poh_recorder, _entry_receiver) = PohRecorder::new(
0,
Hash::default(),
@ -625,6 +710,7 @@ mod tests {
Some(4),
DEFAULT_TICKS_PER_SLOT,
&Pubkey::default(),
&Arc::new(blocktree),
);
poh_recorder.tick();
poh_recorder.tick();
@ -638,9 +724,15 @@ mod tests {
);
assert_eq!(poh_recorder.tick_cache.len(), 0);
}
Blocktree::destroy(&ledger_path).unwrap();
}
#[test]
fn test_reset_with_cached() {
let ledger_path = get_tmp_ledger_path!();
{
let blocktree =
Blocktree::open(&ledger_path).expect("Expected to be able to open database ledger");
let (mut poh_recorder, _entry_receiver) = PohRecorder::new(
0,
Hash::default(),
@ -648,6 +740,7 @@ mod tests {
Some(4),
DEFAULT_TICKS_PER_SLOT,
&Pubkey::default(),
&Arc::new(blocktree),
);
poh_recorder.tick();
poh_recorder.tick();
@ -661,9 +754,15 @@ mod tests {
);
assert_eq!(poh_recorder.tick_cache.len(), 0);
}
Blocktree::destroy(&ledger_path).unwrap();
}
#[test]
fn test_reset_to_new_value() {
let ledger_path = get_tmp_ledger_path!();
{
let blocktree =
Blocktree::open(&ledger_path).expect("Expected to be able to open database ledger");
let (mut poh_recorder, _entry_receiver) = PohRecorder::new(
0,
Hash::default(),
@ -671,6 +770,7 @@ mod tests {
Some(4),
DEFAULT_TICKS_PER_SLOT,
&Pubkey::default(),
&Arc::new(blocktree),
);
poh_recorder.tick();
poh_recorder.tick();
@ -682,9 +782,15 @@ mod tests {
poh_recorder.tick();
assert_eq!(poh_recorder.poh.tick_height, 2);
}
Blocktree::destroy(&ledger_path).unwrap();
}
#[test]
fn test_reset_clear_bank() {
let ledger_path = get_tmp_ledger_path!();
{
let blocktree =
Blocktree::open(&ledger_path).expect("Expected to be able to open database ledger");
let (genesis_block, _mint_keypair) = GenesisBlock::new(2);
let bank = Arc::new(Bank::new(&genesis_block));
let (mut poh_recorder, _entry_receiver) = PohRecorder::new(
@ -694,6 +800,7 @@ mod tests {
Some(4),
bank.ticks_per_slot(),
&Pubkey::default(),
&Arc::new(blocktree),
);
let ticks_per_slot = bank.ticks_per_slot();
let working_bank = WorkingBank {
@ -705,9 +812,15 @@ mod tests {
poh_recorder.reset(1, hash(b"hello"), 0, Some(4), ticks_per_slot);
assert!(poh_recorder.working_bank.is_none());
}
Blocktree::destroy(&ledger_path).unwrap();
}
#[test]
pub fn test_clear_signal() {
let ledger_path = get_tmp_ledger_path!();
{
let blocktree =
Blocktree::open(&ledger_path).expect("Expected to be able to open database ledger");
let (genesis_block, _mint_keypair) = GenesisBlock::new(2);
let bank = Arc::new(Bank::new(&genesis_block));
let (mut poh_recorder, _entry_receiver) = PohRecorder::new(
@ -717,6 +830,7 @@ mod tests {
None,
bank.ticks_per_slot(),
&Pubkey::default(),
&Arc::new(blocktree),
);
let (sender, receiver) = sync_channel(1);
poh_recorder.set_bank(&bank);
@ -724,9 +838,15 @@ mod tests {
poh_recorder.clear_bank();
assert!(receiver.try_recv().is_ok());
}
Blocktree::destroy(&ledger_path).unwrap();
}
#[test]
fn test_poh_recorder_reset_start_slot() {
let ledger_path = get_tmp_ledger_path!();
{
let blocktree =
Blocktree::open(&ledger_path).expect("Expected to be able to open database ledger");
let ticks_per_slot = 5;
let (mut genesis_block, _mint_keypair) = GenesisBlock::new(2);
genesis_block.ticks_per_slot = ticks_per_slot;
@ -740,6 +860,7 @@ mod tests {
Some(4),
bank.ticks_per_slot(),
&Pubkey::default(),
&Arc::new(blocktree),
);
let end_slot = 3;
@ -762,9 +883,15 @@ mod tests {
// Make sure the starting slot is updated
assert_eq!(poh_recorder.start_slot(), end_slot);
}
Blocktree::destroy(&ledger_path).unwrap();
}
#[test]
fn test_reached_leader_tick() {
let ledger_path = get_tmp_ledger_path!();
{
let blocktree =
Blocktree::open(&ledger_path).expect("Expected to be able to open database ledger");
let (genesis_block, _mint_keypair) = GenesisBlock::new(2);
let bank = Arc::new(Bank::new(&genesis_block));
let prev_hash = bank.last_blockhash();
@ -775,6 +902,7 @@ mod tests {
None,
bank.ticks_per_slot(),
&Pubkey::default(),
&Arc::new(blocktree),
);
// Test that with no leader slot, we don't reach the leader tick
@ -884,7 +1012,9 @@ mod tests {
);
// Send remaining ticks for the slot (remember we sent extra ticks in the previous part of the test)
for _ in bank.ticks_per_slot() / MAX_LAST_LEADER_GRACE_TICKS_FACTOR..bank.ticks_per_slot() {
for _ in
bank.ticks_per_slot() / MAX_LAST_LEADER_GRACE_TICKS_FACTOR..bank.ticks_per_slot()
{
poh_recorder.tick();
}
@ -923,4 +1053,6 @@ mod tests {
// We are not the leader, as expected
assert_eq!(poh_recorder.reached_leader_tick().0, false);
}
Blocktree::destroy(&ledger_path).unwrap();
}
}

9
core/src/poh_service.rs
View File

@ -1,6 +1,5 @@
//! The `poh_service` module implements a service that records the passing of
//! "ticks", a measure of time in the PoH stream
use crate::poh_recorder::PohRecorder;
use crate::service::Service;
use solana_sdk::timing::NUM_TICKS_PER_SECOND;
@ -98,6 +97,7 @@ impl Service for PohService {
#[cfg(test)]
mod tests {
use super::*;
use crate::blocktree::{get_tmp_ledger_path, Blocktree};
use crate::poh_recorder::WorkingBank;
use crate::result::Result;
use crate::test_tx::test_tx;
@ -111,6 +111,10 @@ mod tests {
let (genesis_block, _mint_keypair) = GenesisBlock::new(2);
let bank = Arc::new(Bank::new(&genesis_block));
let prev_hash = bank.last_blockhash();
let ledger_path = get_tmp_ledger_path!();
{
let blocktree =
Blocktree::open(&ledger_path).expect("Expected to be able to open database ledger");
let (poh_recorder, entry_receiver) = PohRecorder::new(
bank.tick_height(),
prev_hash,
@ -118,6 +122,7 @@ mod tests {
Some(4),
bank.ticks_per_slot(),
&Pubkey::default(),
&Arc::new(blocktree),
);
let poh_recorder = Arc::new(Mutex::new(poh_recorder));
let exit = Arc::new(AtomicBool::new(false));
@ -190,4 +195,6 @@ mod tests {
let _ = poh_service.join().unwrap();
let _ = entry_producer.join().unwrap();
}
Blocktree::destroy(&ledger_path).unwrap();
}
}

36
core/src/replay_stage.rs
View File

@ -158,8 +158,12 @@ impl ReplayStage {
locktower.update_epoch(&bank);
cluster_info.write().unwrap().push_vote(vote);
}
let next_leader_slot =
leader_schedule_utils::next_leader_slot(&my_id, bank.slot(), &bank);
let next_leader_slot = leader_schedule_utils::next_leader_slot(
&my_id,
bank.slot(),
&bank,
Some(&blocktree),
);
poh_recorder.lock().unwrap().reset(
bank.tick_height(),
bank.last_blockhash(),
@ -195,7 +199,6 @@ impl ReplayStage {
&bank_forks,
&poh_recorder,
&cluster_info,
&blocktree,
poh_slot,
reached_leader_tick,
grace_ticks,
@ -228,35 +231,11 @@ impl ReplayStage {
bank_forks: &Arc<RwLock<BankForks>>,
poh_recorder: &Arc<Mutex<PohRecorder>>,
cluster_info: &Arc<RwLock<ClusterInfo>>,
blocktree: &Blocktree,
poh_slot: u64,
reached_leader_tick: bool,
grace_ticks: u64,
) {
trace!("{} checking poh slot {}", my_id, poh_slot);
if blocktree.meta(poh_slot).unwrap().is_some() {
// We've already broadcasted entries for this slot, skip it
// Since we are skipping our leader slot, let's tell poh recorder when we should be
// leader again
if reached_leader_tick {
let _ = bank_forks.read().unwrap().get(poh_slot).map(|bank| {
let next_leader_slot =
leader_schedule_utils::next_leader_slot(&my_id, bank.slot(), &bank);
let mut poh = poh_recorder.lock().unwrap();
let start_slot = poh.start_slot();
poh.reset(
bank.tick_height(),
bank.last_blockhash(),
start_slot,
next_leader_slot,
bank.ticks_per_slot(),
);
});
}
return;
}
if bank_forks.read().unwrap().get(poh_slot).is_none() {
let frozen = bank_forks.read().unwrap().frozen_banks();
let parent_slot = poh_recorder.lock().unwrap().start_slot();
@ -579,7 +558,8 @@ mod test {
let bank = bank_forks.working_bank();
let blocktree = Arc::new(blocktree);
let (exit, poh_recorder, poh_service, _entry_receiver) = create_test_recorder(&bank);
let (exit, poh_recorder, poh_service, _entry_receiver) =
create_test_recorder(&bank, &blocktree);
let (replay_stage, _slot_full_receiver, ledger_writer_recv) = ReplayStage::new(
&my_keypair.pubkey(),
&voting_keypair.pubkey(),

6
core/src/tvu.rs
View File

@ -202,8 +202,10 @@ pub mod tests {
let blocktree_path = get_tmp_ledger_path!();
let (blocktree, l_receiver) = Blocktree::open_with_signal(&blocktree_path)
.expect("Expected to successfully open ledger");
let blocktree = Arc::new(blocktree);
let bank = bank_forks.working_bank();
let (exit, poh_recorder, poh_service, _entry_receiver) = create_test_recorder(&bank);
let (exit, poh_recorder, poh_service, _entry_receiver) =
create_test_recorder(&bank, &blocktree);
let voting_keypair = Keypair::new();
let tvu = Tvu::new(
&voting_keypair.pubkey(),
@ -218,7 +220,7 @@ pub mod tests {
fetch: target1.sockets.tvu,
}
},
Arc::new(blocktree),
blocktree,
STORAGE_ROTATE_TEST_COUNT,
&StorageState::default(),
None,

7
tests/tvu.rs
View File

@ -34,7 +34,6 @@ fn new_gossip(
GossipService::new(&cluster_info, None, None, gossip, exit)
}
/// Test that message sent from leader to target1 and replayed to target2
#[test]
fn test_replay() {
solana_logger::setup();
@ -99,8 +98,9 @@ fn test_replay() {
let dr_1 = new_gossip(cref1.clone(), target1.sockets.gossip, &exit);
let voting_keypair = Keypair::new();
let blocktree = Arc::new(blocktree);
let (poh_service_exit, poh_recorder, poh_service, _entry_receiver) =
create_test_recorder(&bank);
create_test_recorder(&bank, &blocktree);
let tvu = Tvu::new(
&voting_keypair.pubkey(),
Some(Arc::new(voting_keypair)),
@ -114,7 +114,7 @@ fn test_replay() {
fetch: target1.sockets.tvu,
}
},
Arc::new(blocktree),
blocktree,
STORAGE_ROTATE_TEST_COUNT,
&StorageState::default(),
None,
@ -185,6 +185,7 @@ fn test_replay() {
dr_1.join().unwrap();
t_receiver.join().unwrap();
t_responder.join().unwrap();
drop(poh_recorder);
Blocktree::destroy(&blocktree_path).expect("Expected successful database destruction");
let _ignored = remove_dir_all(&blocktree_path);
}