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

Deserialize snapshots (#5417)

Browse Source 
* Deserialize snapshots
This commit is contained in:
carllin
2019-08-05 22:53:19 -07:00
committed by GitHub
parent 06d8c06119
commit 1dbb5c8647
14 changed files with 433 additions and 266 deletions
Download Patch File Download Diff File Expand all files Collapse all files

135
core/src/bank_forks.rs
View File

@ -2,12 +2,14 @@
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;
use solana_metrics::inc_new_counter_info;
use solana_runtime::bank::{Bank, BankRc, StatusCacheRc};
use solana_runtime::bank::Bank;
use solana_runtime::status_cache::MAX_CACHE_ENTRIES;
use solana_sdk::genesis_block::GenesisBlock;
use solana_sdk::timing;
use std::collections::{HashMap, HashSet};
use std::fs;
@ -275,7 +277,7 @@ impl BankForks {
.get(root)
.cloned()
.expect("root must exist in BankForks");
snapshot_utils::add_snapshot(&config.snapshot_path, &bank, root)?;
snapshot_utils::add_snapshot(&config.snapshot_path, &bank)?;
// Package the relevant snapshots
let names = snapshot_utils::get_snapshot_names(&config.snapshot_path);
@ -348,50 +350,35 @@ impl BankForks {
&self.snapshot_config
}
fn setup_banks(
bank_maps: &mut Vec<(u64, u64, Bank)>,
bank_rc: &BankRc,
status_cache_rc: &StatusCacheRc,
) -> (HashMap<u64, Arc<Bank>>, u64) {
let mut banks = HashMap::new();
let (last_slot, last_parent_slot, mut last_bank) = bank_maps.remove(0);
last_bank.set_bank_rc(&bank_rc, &status_cache_rc);
while let Some((slot, parent_slot, mut bank)) = bank_maps.pop() {
bank.set_bank_rc(&bank_rc, &status_cache_rc);
if parent_slot != 0 {
if let Some(parent) = banks.get(&parent_slot) {
bank.set_parent(parent);
}
}
if slot > 0 {
banks.insert(slot, Arc::new(bank));
}
}
if last_parent_slot != 0 {
if let Some(parent) = banks.get(&last_parent_slot) {
last_bank.set_parent(parent);
}
}
banks.insert(last_slot, Arc::new(last_bank));
(banks, last_slot)
}
pub fn load_from_snapshot(
genesis_block: &GenesisBlock,
account_paths: Option<String>,
pub fn load_from_snapshot<P: AsRef<Path>>(
account_paths: String,
snapshot_config: &SnapshotConfig,
snapshot_tar: P,
) -> Result<Self> {
fs::create_dir_all(&snapshot_config.snapshot_path)?;
let names = snapshot_utils::get_snapshot_names(&snapshot_config.snapshot_path);
if names.is_empty() {
return Err(Error::IO(IOError::new(
ErrorKind::Other,
"no snapshots found",
)));
}
let mut bank_maps = vec![];
// 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 bank = snapshot_utils::bank_from_snapshots(
account_paths,
&unpacked_snapshots_dir,
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 bank_maps = vec![];
let status_cache_rc = StatusCacheRc::default();
let id = (names[names.len() - 1] + 1) as usize;
let mut bank0 =
@ -411,17 +398,27 @@ impl BankForks {
)));
}
let root = bank_root.unwrap();
let (banks, last_slot) =
BankForks::setup_banks(&mut bank_maps, &bank0.rc, &status_cache_rc);
let working_bank = banks[&last_slot].clone();
let (banks, last_slot) = BankForks::setup_banks(&mut bank_maps, &bank.rc, &status_cache_rc);
let working_bank = banks[&last_slot].clone();*/
let mut banks = HashMap::new();
banks.insert(bank.slot(), bank.clone());
let root = bank.slot();
let names = snapshot_utils::get_snapshot_names(&snapshot_config.snapshot_path);
if names.is_empty() {
return Err(Error::IO(IOError::new(
ErrorKind::Other,
"no snapshots found",
)));
}
Ok(BankForks {
banks,
working_bank,
working_bank: bank,
root,
snapshot_config: None,
last_snapshot: *names.last().unwrap(),
last_snapshot: *names
.last()
.expect("untarred snapshot should have at least one snapshot"),
confidence: HashMap::new(),
})
}
@ -551,22 +548,17 @@ mod tests {
);
}
fn restore_from_snapshot(
genesis_block: &GenesisBlock,
bank_forks: BankForks,
account_paths: Option<String>,
last_slot: u64,
) {
let snapshot_path = bank_forks
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)
.map(|c| (&c.snapshot_path, &c.snapshot_package_output_path))
.unwrap();
let new = BankForks::load_from_snapshot(
&genesis_block,
account_paths,
bank_forks.snapshot_config.as_ref().unwrap(),
snapshot_utils::get_snapshot_tar_path(snapshot_package_output_path),
)
.unwrap();
@ -595,7 +587,7 @@ mod tests {
mint_keypair,
..
} = create_genesis_block(10_000);
for index in 0..10 {
for index in 0..4 {
let bank0 = Bank::new_with_paths(
&genesis_block,
Some(accounts_dir.path().to_str().unwrap().to_string()),
@ -609,7 +601,7 @@ mod tests {
);
bank_forks.set_snapshot_config(snapshot_config.clone());
let bank0 = bank_forks.get(0).unwrap();
snapshot_utils::add_snapshot(&snapshot_config.snapshot_path, bank0, 0).unwrap();
snapshot_utils::add_snapshot(&snapshot_config.snapshot_path, bank0).unwrap();
for forks in 0..index {
let bank = Bank::new_from_parent(&bank_forks[forks], &Pubkey::default(), forks + 1);
let key1 = Keypair::new().pubkey();
@ -621,13 +613,26 @@ mod tests {
);
assert_eq!(bank.process_transaction(&tx), Ok(()));
bank.freeze();
snapshot_utils::add_snapshot(&snapshot_config.snapshot_path, &bank, 0).unwrap();
snapshot_utils::add_snapshot(&snapshot_config.snapshot_path, &bank).unwrap();
bank_forks.insert(bank);
}
// Generate a snapshot package for last bank
let last_bank = bank_forks.get(index.saturating_sub(1)).unwrap();
let names: Vec<_> = (0..=index).collect();
let snapshot_package = snapshot_utils::package_snapshot(
last_bank,
&names,
&snapshot_config.snapshot_path,
snapshot_utils::get_snapshot_tar_path(
&snapshot_config.snapshot_package_output_path,
),
)
.unwrap();
SnapshotPackagerService::package_snapshots(&snapshot_package).unwrap();
restore_from_snapshot(
&genesis_block,
bank_forks,
Some(accounts_dir.path().to_str().unwrap().to_string()),
accounts_dir.path().to_str().unwrap().to_string(),
index,
);
}
@ -663,7 +668,7 @@ mod tests {
// Take snapshot of zeroth bank
let bank0 = bank_forks.get(0).unwrap();
snapshot_utils::add_snapshot(&snapshot_config.snapshot_path, bank0, 0).unwrap();
snapshot_utils::add_snapshot(&snapshot_config.snapshot_path, bank0).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

2
core/src/replicator.rs
View File

@ -582,7 +582,7 @@ impl Replicator {
return Err(Error::IO(<io::Error>::new(
io::ErrorKind::Other,
"unable to get recent blockhash, can't submit proof",
)))
)));
}
};

21
core/src/snapshot_package.rs
View File

@ -15,7 +15,7 @@ use std::time::Duration;
pub type SnapshotPackageSender = Sender<SnapshotPackage>;
pub type SnapshotPackageReceiver = Receiver<SnapshotPackage>;
pub const TAR_SNAPSHOT_DIR: &str = "snapshots";
pub const TAR_SNAPSHOTS_DIR: &str = "snapshots";
pub const TAR_ACCOUNTS_DIR: &str = "accounts";
pub struct SnapshotPackage {
@ -57,7 +57,7 @@ impl SnapshotPackagerService {
if exit.load(Ordering::Relaxed) {
break;
}
if let Err(e) = Self::package_snapshots(&snapshot_package_receiver) {
if let Err(e) = Self::run(&snapshot_package_receiver) {
match e {
Error::RecvTimeoutError(RecvTimeoutError::Disconnected) => break,
Error::RecvTimeoutError(RecvTimeoutError::Timeout) => (),
@ -71,9 +71,7 @@ impl SnapshotPackagerService {
}
}
pub fn package_snapshots(snapshot_receiver: &SnapshotPackageReceiver) -> Result<()> {
let snapshot_package = snapshot_receiver.recv_timeout(Duration::from_secs(1))?;
pub fn package_snapshots(snapshot_package: &SnapshotPackage) -> Result<()> {
// Create the tar builder
let tar_gz = tempfile::Builder::new()
.prefix("new_state")
@ -84,7 +82,7 @@ impl SnapshotPackagerService {
let mut tar = tar::Builder::new(enc);
// Create the list of paths to compress, starting with the snapshots
let tar_output_snapshots_dir = Path::new(&TAR_SNAPSHOT_DIR);
let tar_output_snapshots_dir = Path::new(&TAR_SNAPSHOTS_DIR);
// Add the snapshots to the tarball and delete the directory of hardlinks to the snapshots
// that was created to persist those snapshots while this package was being created
@ -117,6 +115,12 @@ impl SnapshotPackagerService {
fs::hard_link(&temp_tar_path, &snapshot_package.tar_output_file)?;
Ok(())
}
fn run(snapshot_receiver: &SnapshotPackageReceiver) -> Result<()> {
let snapshot_package = snapshot_receiver.recv_timeout(Duration::from_secs(1))?;
Self::package_snapshots(&snapshot_package)?;
Ok(())
}
}
impl Service for SnapshotPackagerService {
@ -133,14 +137,12 @@ mod tests {
use crate::snapshot_utils;
use std::fs::OpenOptions;
use std::io::Write;
use std::sync::mpsc::channel;
use tempfile::TempDir;
#[test]
fn test_package_snapshots() {
// Create temprorary placeholder directory for all test files
let temp_dir = TempDir::new().unwrap();
let (sender, receiver) = channel();
let accounts_dir = temp_dir.path().join("accounts");
let snapshots_dir = temp_dir.path().join("snapshots");
let snapshot_package_output_path = temp_dir.path().join("snapshots_output");
@ -184,10 +186,9 @@ mod tests {
storage_entries.clone(),
output_tar_path.clone(),
);
sender.send(snapshot_package).unwrap();
// Make tarball from packageable snapshot
SnapshotPackagerService::package_snapshots(&receiver).unwrap();
SnapshotPackagerService::package_snapshots(&snapshot_package).unwrap();
// Check tarball is correct
snapshot_utils::tests::verify_snapshot_tar(output_tar_path, snapshots_dir, accounts_dir);

99
core/src/snapshot_utils.rs
View File

@ -2,7 +2,7 @@ use crate::result::{Error, Result};
use crate::snapshot_package::SnapshotPackage;
use bincode::{deserialize_from, serialize_into};
use flate2::read::GzDecoder;
use solana_runtime::bank::{Bank, StatusCacheRc};
use solana_runtime::bank::Bank;
use std::fs;
use std::fs::File;
use std::io::{BufReader, BufWriter, Error as IOError, ErrorKind};
@ -33,7 +33,7 @@ pub fn package_snapshot<P: AsRef<Path>, Q: AsRef<Path>>(
let account_storage_entries = bank.rc.get_storage_entries();
// Create a snapshot package
trace!(
info!(
"Snapshot for bank: {} has {} account storage entries",
slot,
account_storage_entries.len()
@ -60,7 +60,8 @@ pub fn get_snapshot_names<P: AsRef<Path>>(snapshot_path: P) -> Vec<u64> {
entry.ok().and_then(|e| {
e.path()
.file_name()
.and_then(|n| n.to_str().map(|s| s.parse::<u64>().unwrap()))
.and_then(|n| n.to_str().map(|s| s.parse::<u64>().ok()))
.unwrap_or(None)
})
})
.collect::<Vec<u64>>();
@ -69,13 +70,13 @@ pub fn get_snapshot_names<P: AsRef<Path>>(snapshot_path: P) -> Vec<u64> {
names
}
pub fn add_snapshot<P: AsRef<Path>>(snapshot_path: P, bank: &Bank, root: u64) -> Result<()> {
pub fn add_snapshot<P: AsRef<Path>>(snapshot_path: P, bank: &Bank) -> Result<()> {
let slot = bank.slot();
let slot_snapshot_dir = get_bank_snapshot_dir(snapshot_path, slot);
fs::create_dir_all(slot_snapshot_dir.clone()).map_err(Error::from)?;
let snapshot_file_path = slot_snapshot_dir.join(get_snapshot_file_name(slot));
trace!(
info!(
"creating snapshot {}, path: {:?}",
bank.slot(),
snapshot_file_path
@ -84,20 +85,12 @@ pub fn add_snapshot<P: AsRef<Path>>(snapshot_path: P, bank: &Bank, root: u64) ->
let mut stream = BufWriter::new(file);
// Create the snapshot
serialize_into(&mut stream, &*bank).map_err(|_| get_io_error("serialize bank error"))?;
let mut parent_slot: u64 = 0;
if let Some(parent_bank) = bank.parent() {
parent_slot = parent_bank.slot();
}
serialize_into(&mut stream, &parent_slot)
.map_err(|_| get_io_error("serialize bank parent error"))?;
serialize_into(&mut stream, &root).map_err(|_| get_io_error("serialize root error"))?;
serialize_into(&mut stream, &bank.src)
.map_err(|_| get_io_error("serialize bank status cache error"))?;
serialize_into(&mut stream, &bank.rc)
.map_err(|_| get_io_error("serialize bank accounts error"))?;
serialize_into(&mut stream, &*bank).map_err(|e| get_io_error(&e.to_string()))?;
serialize_into(&mut stream, &bank.rc).map_err(|e| get_io_error(&e.to_string()))?;
// TODO: Add status cache serialization code
/*serialize_into(&mut stream, &bank.src).map_err(|e| get_io_error(&e.to_string()))?;*/
trace!(
info!(
"successfully created snapshot {}, path: {:?}",
bank.slot(),
snapshot_file_path
@ -112,44 +105,50 @@ pub fn remove_snapshot<P: AsRef<Path>>(slot: u64, snapshot_path: P) -> Result<()
Ok(())
}
pub fn load_snapshots<P: AsRef<Path>>(
names: &[u64],
bank0: &mut Bank,
bank_maps: &mut Vec<(u64, u64, Bank)>,
status_cache_rc: &StatusCacheRc,
pub fn bank_from_snapshots<P, Q>(
local_account_paths: String,
snapshot_path: P,
) -> Option<u64> {
let mut bank_root: Option<u64> = None;
append_vecs_path: Q,
) -> Result<Bank>
where
P: AsRef<Path>,
Q: AsRef<Path>,
{
// Rebuild the last root bank
let names = get_snapshot_names(&snapshot_path);
let last_root = names
.last()
.ok_or_else(|| get_io_error("No snapshots found in snapshots directory"))?;
let snapshot_file_name = get_snapshot_file_name(*last_root);
let snapshot_dir = get_bank_snapshot_dir(&snapshot_path, *last_root);
let snapshot_file_path = snapshot_dir.join(&snapshot_file_name);
info!("Load from {:?}", snapshot_file_path);
let file = File::open(snapshot_file_path)?;
let mut stream = BufReader::new(file);
let bank: Bank = deserialize_from(&mut stream).map_err(|e| get_io_error(&e.to_string()))?;
for (i, bank_slot) in names.iter().rev().enumerate() {
// Rebuild accounts
bank.rc
.accounts_from_stream(&mut stream, local_account_paths, append_vecs_path)?;
for bank_slot in names.iter().rev() {
let snapshot_file_name = get_snapshot_file_name(*bank_slot);
let snapshot_dir = get_bank_snapshot_dir(&snapshot_path, *bank_slot);
let snapshot_file_path = snapshot_dir.join(snapshot_file_name.clone());
trace!("Load from {:?}", snapshot_file_path);
let file = File::open(snapshot_file_path);
if file.is_err() {
warn!("Snapshot file open failed for {}", bank_slot);
continue;
}
let file = file.unwrap();
let file = File::open(snapshot_file_path)?;
let mut stream = BufReader::new(file);
let bank: Result<Bank> =
deserialize_from(&mut stream).map_err(|_| get_io_error("deserialize bank error"));
let slot: Result<u64> = deserialize_from(&mut stream)
.map_err(|_| get_io_error("deserialize bank parent error"));
let parent_slot = if slot.is_ok() { slot.unwrap() } else { 0 };
let root: Result<u64> =
deserialize_from(&mut stream).map_err(|_| get_io_error("deserialize root error"));
let status_cache: Result<StatusCacheRc> = deserialize_from(&mut stream)
.map_err(|_| get_io_error("deserialize bank status cache error"));
if bank_root.is_none() && bank0.rc.update_from_stream(&mut stream).is_ok() {
bank_root = Some(root.unwrap());
}
let _bank: Result<Bank> =
deserialize_from(&mut stream).map_err(|e| get_io_error(&e.to_string()));
// TODO: Uncomment and deserialize status cache here
/*let status_cache: Result<StatusCacheRc> = deserialize_from(&mut stream)
.map_err(|e| get_io_error(&e.to_string()));
if bank_root.is_some() {
match bank {
Ok(v) => {
if status_cache.is_ok() {
let status_cache = status_cache.unwrap();
let status_cache = status_cache?;
status_cache_rc.append(&status_cache);
// On the last snapshot, purge all outdated status cache
// entries
@ -164,10 +163,10 @@ pub fn load_snapshots<P: AsRef<Path>>(
}
} else {
warn!("Load snapshot rc failed for {}", bank_slot);
}
}*/
}
bank_root
Ok(bank)
}
pub fn get_snapshot_tar_path<P: AsRef<Path>>(snapshot_output_dir: P) -> PathBuf {
@ -228,7 +227,7 @@ fn get_io_error(error: &str) -> Error {
#[cfg(test)]
pub mod tests {
use super::*;
use crate::snapshot_package::{TAR_ACCOUNTS_DIR, TAR_SNAPSHOT_DIR};
use crate::snapshot_package::{TAR_ACCOUNTS_DIR, TAR_SNAPSHOTS_DIR};
use tempfile::TempDir;
pub fn verify_snapshot_tar<P, Q, R>(
@ -245,7 +244,7 @@ pub mod tests {
untar_snapshot_in(snapshot_tar, &unpack_dir).unwrap();
// Check snapshots are the same
let unpacked_snapshots = unpack_dir.join(&TAR_SNAPSHOT_DIR);
let unpacked_snapshots = unpack_dir.join(&TAR_SNAPSHOTS_DIR);
assert!(!dir_diff::is_different(&snapshots_to_verify, unpacked_snapshots).unwrap());
// Check the account entries are the same

2
core/src/storage_stage.rs
View File

@ -205,7 +205,7 @@ impl StorageStage {
) {
match e {
Error::RecvTimeoutError(RecvTimeoutError::Disconnected) => {
break
break;
}
Error::RecvTimeoutError(RecvTimeoutError::Timeout) => (),
_ => info!("Error from process_entries: {:?}", e),

44
core/src/validator.rs
View File

@ -16,6 +16,7 @@ use crate::rpc_pubsub_service::PubSubService;
use crate::rpc_service::JsonRpcService;
use crate::rpc_subscriptions::RpcSubscriptions;
use crate::service::Service;
use crate::snapshot_utils;
use crate::storage_stage::StorageState;
use crate::tpu::Tpu;
use crate::tvu::{Sockets, Tvu};
@ -314,21 +315,36 @@ fn get_bank_forks(
let (mut bank_forks, bank_forks_info, leader_schedule_cache) = {
let mut result = None;
if snapshot_config.is_some() {
let bank_forks = BankForks::load_from_snapshot(
&genesis_block,
account_paths.clone(),
snapshot_config.as_ref().unwrap(),
let snapshot_config = snapshot_config.as_ref().unwrap();
// Get the path to the tar
let tar = snapshot_utils::get_snapshot_tar_path(
&snapshot_config.snapshot_package_output_path(),
);
match bank_forks {
Ok(v) => {
let bank = &v.working_bank();
let fork_info = BankForksInfo {
bank_slot: bank.slot(),
entry_height: bank.tick_height(),
};
result = Some((v, vec![fork_info], LeaderScheduleCache::new_from_bank(bank)));
}
Err(_) => warn!("Failed to load from snapshot, fallback to load from ledger"),
// Check that the snapshot tar exists, try to load the snapshot if it does
if tar.exists() {
// Fail hard here if snapshot fails to load, don't silently continue
let bank_forks = BankForks::load_from_snapshot(
//&genesis_block,
account_paths
.clone()
.expect("Account paths not present when booting from snapshot"),
snapshot_config,
tar,
)
.expect("Load from snapshot failed");
let bank = &bank_forks.working_bank();
let fork_info = BankForksInfo {
bank_slot: bank.slot(),
entry_height: bank.tick_height(),
};
result = Some((
bank_forks,
vec![fork_info],
LeaderScheduleCache::new_from_bank(bank),
));
}
}