solana/core/src/bank_forks.rs

808 lines
29 KiB
Rust
Raw Normal View History

//! The `bank_forks` module implments BankForks a DAG of checkpointed Banks
use crate::result::{Error, Result};
use crate::snapshot_package::SnapshotPackageSender;
use crate::snapshot_package::{TAR_ACCOUNTS_DIR, TAR_SNAPSHOTS_DIR};
use crate::snapshot_utils;
use crate::snapshot_utils::untar_snapshot_in;
use fs_extra::dir::CopyOptions;
use solana_measure::measure::Measure;
2019-05-10 08:33:58 -07:00
use solana_metrics::inc_new_counter_info;
use solana_runtime::bank::Bank;
use solana_runtime::status_cache::MAX_CACHE_ENTRIES;
2019-03-29 21:21:59 -07:00
use solana_sdk::timing;
use std::collections::{HashMap, HashSet};
use std::fs;
use std::io::{Error as IOError, ErrorKind};
use std::ops::Index;
use std::path::{Path, PathBuf};
use std::sync::Arc;
2019-03-29 21:21:59 -07:00
use std::time::Instant;
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct SnapshotConfig {
snapshot_path: PathBuf,
snapshot_package_output_path: PathBuf,
snapshot_interval_slots: usize,
}
impl SnapshotConfig {
pub fn new(
snapshot_path: PathBuf,
snapshot_package_output_path: PathBuf,
snapshot_interval_slots: usize,
) -> Self {
Self {
snapshot_path,
snapshot_package_output_path,
snapshot_interval_slots,
}
}
pub fn snapshot_path(&self) -> &Path {
self.snapshot_path.as_path()
}
pub fn snapshot_package_output_path(&self) -> &Path {
&self.snapshot_package_output_path.as_path()
}
pub fn snapshot_interval_slots(&self) -> usize {
self.snapshot_interval_slots
}
}
pub struct BankForks {
banks: HashMap<u64, Arc<Bank>>,
working_bank: Arc<Bank>,
2019-05-07 23:34:10 -07:00
root: u64,
snapshot_config: Option<SnapshotConfig>,
slots_since_snapshot: Vec<u64>,
confidence: HashMap<u64, Confidence>,
}
#[derive(Debug, Default, PartialEq)]
pub struct Confidence {
fork_stakes: u64,
epoch_stakes: u64,
lockouts: u64,
stake_weighted_lockouts: u128,
}
impl Confidence {
pub fn new(fork_stakes: u64, epoch_stakes: u64, lockouts: u64) -> Self {
Self {
fork_stakes,
epoch_stakes,
lockouts,
stake_weighted_lockouts: 0,
}
}
pub fn new_with_stake_weighted(
fork_stakes: u64,
epoch_stakes: u64,
lockouts: u64,
stake_weighted_lockouts: u128,
) -> Self {
Self {
fork_stakes,
epoch_stakes,
lockouts,
stake_weighted_lockouts,
}
}
}
impl Index<u64> for BankForks {
type Output = Arc<Bank>;
2019-03-04 16:40:28 -08:00
fn index(&self, bank_slot: u64) -> &Arc<Bank> {
&self.banks[&bank_slot]
}
}
impl BankForks {
2019-03-04 16:40:28 -08:00
pub fn new(bank_slot: u64, bank: Bank) -> Self {
let mut banks = HashMap::new();
let working_bank = Arc::new(bank);
2019-03-04 16:40:28 -08:00
banks.insert(bank_slot, working_bank.clone());
Self {
banks,
working_bank,
2019-05-07 23:34:10 -07:00
root: 0,
snapshot_config: None,
slots_since_snapshot: vec![],
confidence: HashMap::new(),
}
}
/// Create a map of bank slot id to the set of ancestors for the bank slot.
pub fn ancestors(&self) -> HashMap<u64, HashSet<u64>> {
let mut ancestors = HashMap::new();
for bank in self.banks.values() {
2019-04-28 10:27:09 -07:00
let mut set: HashSet<u64> = bank.ancestors.keys().cloned().collect();
set.remove(&bank.slot());
ancestors.insert(bank.slot(), set);
}
ancestors
}
/// Create a map of bank slot id to the set of all of its descendants
#[allow(clippy::or_fun_call)]
2019-03-18 15:20:04 -07:00
pub fn descendants(&self) -> HashMap<u64, HashSet<u64>> {
let mut descendants = HashMap::new();
for bank in self.banks.values() {
2019-03-18 15:20:04 -07:00
let _ = descendants.entry(bank.slot()).or_insert(HashSet::new());
2019-04-28 10:27:09 -07:00
let mut set: HashSet<u64> = bank.ancestors.keys().cloned().collect();
set.remove(&bank.slot());
for parent in set {
2019-03-18 15:20:04 -07:00
descendants
2019-04-28 10:27:09 -07:00
.entry(parent)
.or_insert(HashSet::new())
.insert(bank.slot());
}
}
2019-03-18 15:20:04 -07:00
descendants
}
2019-02-26 21:57:45 -08:00
pub fn frozen_banks(&self) -> HashMap<u64, Arc<Bank>> {
self.banks
.iter()
.filter(|(_, b)| b.is_frozen())
.map(|(k, b)| (*k, b.clone()))
.collect()
2019-02-26 21:57:45 -08:00
}
2019-02-26 21:57:45 -08:00
pub fn active_banks(&self) -> Vec<u64> {
self.banks
.iter()
.filter(|(_, v)| !v.is_frozen())
.map(|(k, _v)| *k)
.collect()
2019-02-26 21:57:45 -08:00
}
2019-03-19 17:30:36 -07:00
2019-03-04 16:40:28 -08:00
pub fn get(&self, bank_slot: u64) -> Option<&Arc<Bank>> {
self.banks.get(&bank_slot)
2019-02-26 21:57:45 -08:00
}
2019-05-07 23:34:10 -07:00
pub fn new_from_banks(initial_banks: &[Arc<Bank>], root: u64) -> Self {
let mut banks = HashMap::new();
let working_bank = initial_banks[0].clone();
for bank in initial_banks {
banks.insert(bank.slot(), bank.clone());
}
Self {
2019-05-07 23:34:10 -07:00
root,
banks,
working_bank,
snapshot_config: None,
slots_since_snapshot: vec![],
confidence: HashMap::new(),
}
}
pub fn insert(&mut self, bank: Bank) -> Arc<Bank> {
let bank = Arc::new(bank);
2019-03-18 20:23:34 -07:00
let prev = self.banks.insert(bank.slot(), bank.clone());
assert!(prev.is_none());
self.working_bank = bank.clone();
bank
}
// TODO: really want to kill this...
pub fn working_bank(&self) -> Arc<Bank> {
self.working_bank.clone()
}
pub fn set_root(&mut self, root: u64, snapshot_package_sender: &Option<SnapshotPackageSender>) {
2019-05-07 23:34:10 -07:00
self.root = root;
2019-03-29 21:21:59 -07:00
let set_root_start = Instant::now();
let root_bank = self
.banks
.get(&root)
.expect("root bank didn't exist in bank_forks");
let root_tx_count = root_bank
.parents()
.last()
.map(|bank| bank.transaction_count())
.unwrap_or(0);
root_bank.squash();
let new_tx_count = root_bank.transaction_count();
self.slots_since_snapshot.push(root);
2019-05-07 23:34:10 -07:00
// Generate a snapshot if snapshots are configured and it's been an appropriate number
// of banks since the last snapshot
if self.snapshot_config.is_some() && snapshot_package_sender.is_some() {
let config = self
.snapshot_config
.as_ref()
.expect("Called package_snapshot without a snapshot configuration");
if root - self.slots_since_snapshot[0] >= config.snapshot_interval_slots as u64 {
let mut snapshot_time = Measure::start("total-snapshot-ms");
let r = self.generate_snapshot(
root,
&self.slots_since_snapshot[1..],
snapshot_package_sender.as_ref().unwrap(),
snapshot_utils::get_snapshot_tar_path(&config.snapshot_package_output_path),
);
if r.is_err() {
warn!("Error generating snapshot for bank: {}, err: {:?}", root, r);
} else {
self.slots_since_snapshot = vec![root];
}
// Cleanup outdated snapshots
self.purge_old_snapshots();
snapshot_time.stop();
inc_new_counter_info!("total-snapshot-setup-ms", snapshot_time.as_ms() as usize);
}
}
self.prune_non_root(root);
2019-03-29 21:21:59 -07:00
inc_new_counter_info!(
"bank-forks_set_root_ms",
timing::duration_as_ms(&set_root_start.elapsed()) as usize
);
inc_new_counter_info!(
"bank-forks_set_root_tx_count",
(new_tx_count - root_tx_count) as usize
);
}
2019-05-07 23:34:10 -07:00
pub fn root(&self) -> u64 {
self.root
}
fn purge_old_snapshots(&self) {
// Remove outdated snapshots
let config = self.snapshot_config.as_ref().unwrap();
let slot_snapshot_paths = snapshot_utils::get_snapshot_paths(&config.snapshot_path);
let num_to_remove = slot_snapshot_paths.len().saturating_sub(MAX_CACHE_ENTRIES);
for slot_files in &slot_snapshot_paths[..num_to_remove] {
let r = snapshot_utils::remove_snapshot(slot_files.slot, &config.snapshot_path);
if r.is_err() {
warn!("Couldn't remove snapshot at: {:?}", config.snapshot_path);
}
}
}
fn generate_snapshot<P: AsRef<Path>>(
&self,
root: u64,
slots_since_snapshot: &[u64],
snapshot_package_sender: &SnapshotPackageSender,
tar_output_file: P,
) -> Result<()> {
let config = self.snapshot_config.as_ref().unwrap();
// Add a snapshot for the new root
let bank = self
.get(root)
.cloned()
.expect("root must exist in BankForks");
snapshot_utils::add_snapshot(&config.snapshot_path, &bank, slots_since_snapshot)?;
// Package the relevant snapshots
let slot_snapshot_paths = snapshot_utils::get_snapshot_paths(&config.snapshot_path);
// We only care about the last MAX_CACHE_ENTRIES snapshots of roots because
// the status cache of anything older is thrown away by the bank in
// status_cache.prune_roots()
let start = slot_snapshot_paths.len().saturating_sub(MAX_CACHE_ENTRIES);
let package = snapshot_utils::package_snapshot(
&bank,
&slot_snapshot_paths[start..],
tar_output_file,
)?;
// Send the package to the packaging thread
snapshot_package_sender.send(package)?;
Ok(())
}
fn prune_non_root(&mut self, root: u64) {
2019-04-28 10:27:09 -07:00
let descendants = self.descendants();
2019-03-18 20:23:34 -07:00
self.banks
.retain(|slot, _| slot == &root || descendants[&root].contains(slot));
self.confidence
.retain(|slot, _| slot == &root || descendants[&root].contains(slot));
}
pub fn cache_fork_confidence(
&mut self,
fork: u64,
fork_stakes: u64,
epoch_stakes: u64,
lockouts: u64,
) {
self.confidence
.entry(fork)
.and_modify(|entry| {
entry.fork_stakes = fork_stakes;
entry.epoch_stakes = epoch_stakes;
entry.lockouts = lockouts;
})
.or_insert_with(|| Confidence::new(fork_stakes, epoch_stakes, lockouts));
}
pub fn cache_stake_weighted_lockouts(&mut self, fork: u64, stake_weighted_lockouts: u128) {
self.confidence
.entry(fork)
.and_modify(|entry| {
entry.stake_weighted_lockouts = stake_weighted_lockouts;
})
.or_insert(Confidence {
fork_stakes: 0,
epoch_stakes: 0,
lockouts: 0,
stake_weighted_lockouts,
});
}
pub fn get_fork_confidence(&self, fork: u64) -> Option<&Confidence> {
self.confidence.get(&fork)
}
pub fn set_snapshot_config(&mut self, snapshot_config: SnapshotConfig) {
self.snapshot_config = Some(snapshot_config);
}
pub fn snapshot_config(&self) -> &Option<SnapshotConfig> {
&self.snapshot_config
}
pub fn load_from_snapshot<P: AsRef<Path>>(
account_paths: String,
snapshot_config: &SnapshotConfig,
snapshot_tar: P,
) -> Result<Self> {
// Untar the snapshot into a temp directory under `snapshot_config.snapshot_path()`
let unpack_dir = tempfile::tempdir_in(snapshot_config.snapshot_path())?;
untar_snapshot_in(&snapshot_tar, &unpack_dir)?;
let unpacked_accounts_dir = unpack_dir.as_ref().join(TAR_ACCOUNTS_DIR);
let unpacked_snapshots_dir = unpack_dir.as_ref().join(TAR_SNAPSHOTS_DIR);
let snapshot_paths = snapshot_utils::get_snapshot_paths(&unpacked_snapshots_dir);
let bank = snapshot_utils::bank_from_snapshots(
account_paths,
&snapshot_paths,
unpacked_accounts_dir,
)?;
let bank = Arc::new(bank);
// Move the unpacked snapshots into `snapshot_config.snapshot_path()`
let dir_files = fs::read_dir(unpacked_snapshots_dir).expect("Invalid snapshot path");
let paths: Vec<PathBuf> = dir_files
.filter_map(|entry| entry.ok().map(|e| e.path()))
.collect();
let mut copy_options = CopyOptions::new();
copy_options.overwrite = true;
fs_extra::move_items(&paths, snapshot_config.snapshot_path(), &copy_options)?;
let mut banks = HashMap::new();
banks.insert(bank.slot(), bank.clone());
let root = bank.slot();
let snapshot_paths = snapshot_utils::get_snapshot_paths(&snapshot_config.snapshot_path);
if snapshot_paths.is_empty() {
return Err(Error::IO(IOError::new(
ErrorKind::Other,
"no snapshots found",
)));
}
Ok(BankForks {
banks,
working_bank: bank,
root,
snapshot_config: None,
slots_since_snapshot: vec![snapshot_paths.last().unwrap().slot],
confidence: HashMap::new(),
})
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::genesis_utils::{create_genesis_block, GenesisBlockInfo};
use crate::service::Service;
use crate::snapshot_package::SnapshotPackagerService;
use fs_extra::dir::CopyOptions;
use itertools::Itertools;
use solana_sdk::hash::{hashv, Hash};
use solana_sdk::pubkey::Pubkey;
use solana_sdk::signature::{Keypair, KeypairUtil};
use solana_sdk::system_transaction;
use std::sync::atomic::AtomicBool;
use std::sync::mpsc::channel;
use tempfile::TempDir;
#[test]
fn test_bank_forks() {
let GenesisBlockInfo { genesis_block, .. } = create_genesis_block(10_000);
2019-02-20 16:14:58 -08:00
let bank = Bank::new(&genesis_block);
2019-02-18 14:35:02 -08:00
let mut bank_forks = BankForks::new(0, bank);
let child_bank = Bank::new_from_parent(&bank_forks[0u64], &Pubkey::default(), 1);
child_bank.register_tick(&Hash::default());
2019-03-18 20:23:34 -07:00
bank_forks.insert(child_bank);
assert_eq!(bank_forks[1u64].tick_height(), 1);
assert_eq!(bank_forks.working_bank().tick_height(), 1);
}
2019-02-26 21:57:45 -08:00
#[test]
2019-03-18 15:20:04 -07:00
fn test_bank_forks_descendants() {
let GenesisBlockInfo { genesis_block, .. } = create_genesis_block(10_000);
let bank = Bank::new(&genesis_block);
let mut bank_forks = BankForks::new(0, bank);
let bank0 = bank_forks[0].clone();
let bank = Bank::new_from_parent(&bank0, &Pubkey::default(), 1);
2019-03-18 20:23:34 -07:00
bank_forks.insert(bank);
let bank = Bank::new_from_parent(&bank0, &Pubkey::default(), 2);
2019-03-18 20:23:34 -07:00
bank_forks.insert(bank);
2019-03-18 15:20:04 -07:00
let descendants = bank_forks.descendants();
let children: HashSet<u64> = [1u64, 2u64].to_vec().into_iter().collect();
assert_eq!(children, *descendants.get(&0).unwrap());
2019-03-18 15:20:04 -07:00
assert!(descendants[&1].is_empty());
assert!(descendants[&2].is_empty());
}
#[test]
fn test_bank_forks_ancestors() {
let GenesisBlockInfo { genesis_block, .. } = create_genesis_block(10_000);
let bank = Bank::new(&genesis_block);
let mut bank_forks = BankForks::new(0, bank);
let bank0 = bank_forks[0].clone();
let bank = Bank::new_from_parent(&bank0, &Pubkey::default(), 1);
2019-03-18 20:23:34 -07:00
bank_forks.insert(bank);
let bank = Bank::new_from_parent(&bank0, &Pubkey::default(), 2);
2019-03-18 20:23:34 -07:00
bank_forks.insert(bank);
let ancestors = bank_forks.ancestors();
assert!(ancestors[&0].is_empty());
let parents: Vec<u64> = ancestors[&1].iter().cloned().collect();
assert_eq!(parents, vec![0]);
let parents: Vec<u64> = ancestors[&2].iter().cloned().collect();
assert_eq!(parents, vec![0]);
}
2019-02-26 21:57:45 -08:00
#[test]
fn test_bank_forks_frozen_banks() {
let GenesisBlockInfo { genesis_block, .. } = create_genesis_block(10_000);
2019-02-26 21:57:45 -08:00
let bank = Bank::new(&genesis_block);
let mut bank_forks = BankForks::new(0, bank);
let child_bank = Bank::new_from_parent(&bank_forks[0u64], &Pubkey::default(), 1);
2019-03-18 20:23:34 -07:00
bank_forks.insert(child_bank);
2019-02-26 21:57:45 -08:00
assert!(bank_forks.frozen_banks().get(&0).is_some());
assert!(bank_forks.frozen_banks().get(&1).is_none());
}
#[test]
fn test_bank_forks_active_banks() {
let GenesisBlockInfo { genesis_block, .. } = create_genesis_block(10_000);
2019-02-26 21:57:45 -08:00
let bank = Bank::new(&genesis_block);
let mut bank_forks = BankForks::new(0, bank);
let child_bank = Bank::new_from_parent(&bank_forks[0u64], &Pubkey::default(), 1);
2019-03-18 20:23:34 -07:00
bank_forks.insert(child_bank);
2019-02-26 21:57:45 -08:00
assert_eq!(bank_forks.active_banks(), vec![1]);
}
#[test]
fn test_bank_forks_confidence_cache() {
let GenesisBlockInfo { genesis_block, .. } = create_genesis_block(10_000);
let bank = Bank::new(&genesis_block);
let fork = bank.slot();
let mut bank_forks = BankForks::new(0, bank);
assert!(bank_forks.confidence.get(&fork).is_none());
bank_forks.cache_fork_confidence(fork, 11, 12, 13);
assert_eq!(
bank_forks.confidence.get(&fork).unwrap(),
&Confidence {
fork_stakes: 11,
epoch_stakes: 12,
lockouts: 13,
stake_weighted_lockouts: 0,
}
);
// Ensure that {fork_stakes, epoch_stakes, lockouts} and stake_weighted_lockouts
// can be updated separately
bank_forks.cache_stake_weighted_lockouts(fork, 20);
assert_eq!(
bank_forks.confidence.get(&fork).unwrap(),
&Confidence {
fork_stakes: 11,
epoch_stakes: 12,
lockouts: 13,
stake_weighted_lockouts: 20,
}
);
bank_forks.cache_fork_confidence(fork, 21, 22, 23);
assert_eq!(
bank_forks
.confidence
.get(&fork)
.unwrap()
.stake_weighted_lockouts,
20,
);
}
fn restore_from_snapshot(bank_forks: BankForks, account_paths: String, last_slot: u64) {
let (snapshot_path, snapshot_package_output_path) = bank_forks
.snapshot_config
.as_ref()
.map(|c| (&c.snapshot_path, &c.snapshot_package_output_path))
.unwrap();
let new = BankForks::load_from_snapshot(
account_paths,
bank_forks.snapshot_config.as_ref().unwrap(),
snapshot_utils::get_snapshot_tar_path(snapshot_package_output_path),
)
.unwrap();
for (slot, _) in new.banks.iter() {
2019-06-02 14:05:32 -07:00
if *slot > 0 {
let bank = bank_forks.banks.get(slot).unwrap().clone();
let new_bank = new.banks.get(slot).unwrap();
bank.compare_bank(&new_bank);
}
}
assert_eq!(new.working_bank().slot(), last_slot);
for (slot, _) in new.banks.iter() {
snapshot_utils::remove_snapshot(*slot, snapshot_path).unwrap();
}
}
// creates banks upto "last_slot" and runs the input function `f` on each bank created
// also marks each bank as root and generates snapshots
// finally tries to restore from the last bank's snapshot and compares the restored bank to the
// `last_slot` bank
fn run_bank_forks_snapshot_n<F>(last_slot: u64, f: F)
where
F: Fn(&mut Bank, &Keypair),
{
solana_logger::setup();
let accounts_dir = TempDir::new().unwrap();
let snapshot_dir = TempDir::new().unwrap();
let snapshot_output_path = TempDir::new().unwrap();
let GenesisBlockInfo {
genesis_block,
mint_keypair,
..
} = create_genesis_block(10_000);
let bank0 = Bank::new_with_paths(
&genesis_block,
Some(accounts_dir.path().to_str().unwrap().to_string()),
);
bank0.freeze();
let mut bank_forks = BankForks::new(0, bank0);
let snapshot_config = SnapshotConfig::new(
PathBuf::from(snapshot_dir.path()),
PathBuf::from(snapshot_output_path.path()),
100,
);
bank_forks.set_snapshot_config(snapshot_config.clone());
let bank0 = bank_forks.get(0).unwrap();
snapshot_utils::add_snapshot(&snapshot_config.snapshot_path, bank0, &vec![]).unwrap();
for slot in 0..last_slot {
let mut bank = Bank::new_from_parent(&bank_forks[slot], &Pubkey::default(), slot + 1);
f(&mut bank, &mint_keypair);
let bank = bank_forks.insert(bank);
// set root to make sure we don't end up with too many account storage entries
// and to allow purging logic on status_cache to kick in
bank_forks.set_root(bank.slot(), &None);
snapshot_utils::add_snapshot(
&snapshot_config.snapshot_path,
&bank,
&(vec![bank.slot()]),
)
.unwrap();
}
// Generate a snapshot package for last bank
let last_bank = bank_forks.get(last_slot).unwrap();
let slot_snapshot_paths =
snapshot_utils::get_snapshot_paths(&snapshot_config.snapshot_path);
let snapshot_package = snapshot_utils::package_snapshot(
last_bank,
&slot_snapshot_paths,
snapshot_utils::get_snapshot_tar_path(&snapshot_config.snapshot_package_output_path),
)
.unwrap();
SnapshotPackagerService::package_snapshots(&snapshot_package).unwrap();
restore_from_snapshot(
bank_forks,
accounts_dir.path().to_str().unwrap().to_string(),
last_slot,
);
}
#[test]
fn test_bank_forks_snapshot_n() {
// create banks upto slot 4 and create 1 new account in each bank. test that bank 4 snapshots
// and restores correctly
run_bank_forks_snapshot_n(4, |bank, mint_keypair| {
let key1 = Keypair::new().pubkey();
let tx = system_transaction::create_user_account(
&mint_keypair,
&key1,
1,
bank.last_blockhash(),
);
assert_eq!(bank.process_transaction(&tx), Ok(()));
bank.freeze();
});
}
fn goto_end_of_slot(bank: &mut Bank) {
let mut tick_hash = bank.last_blockhash();
loop {
tick_hash = hashv(&[&tick_hash.as_ref(), &[42]]);
bank.register_tick(&tick_hash);
if tick_hash == bank.last_blockhash() {
bank.freeze();
return;
}
}
}
#[test]
fn test_bank_forks_status_cache_snapshot_n() {
// create banks upto slot 256 while transferring 1 lamport into 2 different accounts each time
// this is done to ensure the AccountStorageEntries keep getting cleaned up as the root moves
// ahead. Also tests the status_cache purge and status cache snapshotting.
// Makes sure that the last bank is restored correctly
let key1 = Keypair::new().pubkey();
let key2 = Keypair::new().pubkey();
run_bank_forks_snapshot_n(256, |bank, mint_keypair| {
let tx = system_transaction::transfer(
&mint_keypair,
&key1,
1,
bank.parent().unwrap().last_blockhash(),
);
assert_eq!(bank.process_transaction(&tx), Ok(()));
let tx = system_transaction::transfer(
&mint_keypair,
&key2,
1,
bank.parent().unwrap().last_blockhash(),
);
assert_eq!(bank.process_transaction(&tx), Ok(()));
goto_end_of_slot(bank);
});
}
#[test]
fn test_concurrent_snapshot_packaging() {
solana_logger::setup();
let accounts_dir = TempDir::new().unwrap();
let snapshots_dir = TempDir::new().unwrap();
let snapshot_output_path = TempDir::new().unwrap();
let GenesisBlockInfo {
genesis_block,
mint_keypair,
..
} = create_genesis_block(10_000);
let (sender, receiver) = channel();
let (fake_sender, _fake_receiver) = channel();
let bank0 = Bank::new_with_paths(
&genesis_block,
Some(accounts_dir.path().to_str().unwrap().to_string()),
);
bank0.freeze();
// Set up bank forks
let mut bank_forks = BankForks::new(0, bank0);
let snapshot_config = SnapshotConfig::new(
PathBuf::from(snapshots_dir.path()),
PathBuf::from(snapshot_output_path.path()),
1,
);
bank_forks.set_snapshot_config(snapshot_config.clone());
// Take snapshot of zeroth bank
let bank0 = bank_forks.get(0).unwrap();
snapshot_utils::add_snapshot(&snapshot_config.snapshot_path, bank0, &vec![]).unwrap();
// Create next MAX_CACHE_ENTRIES + 2 banks and snapshots. Every bank will get snapshotted
// and the snapshot purging logic will run on every snapshot taken. This means the three
// (including snapshot for bank0 created above) earliest snapshots will get purged by the
// time this loop is done.
// Also, make a saved copy of the state of the snapshot for a bank with
// bank.slot == saved_slot, so we can use it for a correctness check later.
let saved_snapshots_dir = TempDir::new().unwrap();
let saved_accounts_dir = TempDir::new().unwrap();
let saved_slot = 4;
let saved_tar = snapshot_config
.snapshot_package_output_path
.join(saved_slot.to_string());
for forks in 0..MAX_CACHE_ENTRIES + 2 {
let bank = Bank::new_from_parent(
&bank_forks[forks as u64],
&Pubkey::default(),
(forks + 1) as u64,
);
let slot = bank.slot();
let key1 = Keypair::new().pubkey();
let tx = system_transaction::create_user_account(
&mint_keypair,
&key1,
1,
genesis_block.hash(),
);
assert_eq!(bank.process_transaction(&tx), Ok(()));
bank.freeze();
bank_forks.insert(bank);
let package_sender = {
if slot == saved_slot as u64 {
// Only send one package on the real sende so that the packaging service
// doesn't take forever to run the packaging logic on all MAX_CACHE_ENTRIES
// later
&sender
} else {
&fake_sender
}
};
bank_forks
.generate_snapshot(
slot,
&vec![],
&package_sender,
snapshot_config
.snapshot_package_output_path
.join(slot.to_string()),
)
.unwrap();
if slot == saved_slot as u64 {
let options = CopyOptions::new();
fs_extra::dir::copy(&accounts_dir, &saved_accounts_dir, &options).unwrap();
fs_extra::dir::copy(&snapshots_dir, &saved_snapshots_dir, &options).unwrap();
}
}
// Purge all the outdated snapshots, including the ones needed to generate the package
// currently sitting in the channel
bank_forks.purge_old_snapshots();
let mut snapshot_paths = snapshot_utils::get_snapshot_paths(&snapshots_dir);
snapshot_paths.sort();
assert_eq!(
snapshot_paths.iter().map(|path| path.slot).collect_vec(),
(3..=MAX_CACHE_ENTRIES as u64 + 2).collect_vec()
);
// Create a SnapshotPackagerService to create tarballs from all the pending
// SnapshotPackage's on the channel. By the time this service starts, we have already
// purged the first two snapshots, which are needed by every snapshot other than
// the last two snapshots. However, the packaging service should still be able to
// correctly construct the earlier snapshots because the SnapshotPackage's on the
// channel hold hard links to these deleted snapshots. We verify this is the case below.
let exit = Arc::new(AtomicBool::new(false));
let snapshot_packager_service = SnapshotPackagerService::new(receiver, &exit);
// Close the channel so that the package service will exit after reading all the
// packages off the channel
drop(sender);
// Wait for service to finish
snapshot_packager_service
.join()
.expect("SnapshotPackagerService exited with error");
// Check the tar we cached the state for earlier was generated correctly
snapshot_utils::tests::verify_snapshot_tar(
saved_tar,
saved_snapshots_dir
.path()
.join(snapshots_dir.path().file_name().unwrap()),
saved_accounts_dir
.path()
.join(accounts_dir.path().file_name().unwrap()),
);
}
}