Unpack snapshot AppendVecs directly into account paths
This commit is contained in:
parent
c078e01fa9
commit
1061d021c9
|
@ -5126,7 +5126,6 @@ dependencies = [
|
|||
"dir-diff",
|
||||
"flate2",
|
||||
"fnv",
|
||||
"fs_extra",
|
||||
"itertools",
|
||||
"lazy_static",
|
||||
"libc",
|
||||
|
|
|
@ -3214,7 +3214,6 @@ dependencies = [
|
|||
"dir-diff",
|
||||
"flate2",
|
||||
"fnv",
|
||||
"fs_extra",
|
||||
"itertools 0.9.0",
|
||||
"lazy_static",
|
||||
"libc",
|
||||
|
|
|
@ -20,7 +20,6 @@ crossbeam-channel = "0.4"
|
|||
dir-diff = "0.3.2"
|
||||
flate2 = "1.0.14"
|
||||
fnv = "1.0.7"
|
||||
fs_extra = "1.2.0"
|
||||
itertools = "0.9.0"
|
||||
lazy_static = "1.4.0"
|
||||
libc = "0.2.81"
|
||||
|
|
|
@ -710,7 +710,7 @@ pub struct AccountsDb {
|
|||
pub shrink_paths: RwLock<Option<Vec<PathBuf>>>,
|
||||
|
||||
/// Directory of paths this accounts_db needs to hold/remove
|
||||
temp_paths: Option<Vec<TempDir>>,
|
||||
pub(crate) temp_paths: Option<Vec<TempDir>>,
|
||||
|
||||
/// Starting file size of appendvecs
|
||||
file_size: u64,
|
||||
|
|
|
@ -1,20 +1,24 @@
|
|||
use bzip2::bufread::BzDecoder;
|
||||
use log::*;
|
||||
use solana_sdk::genesis_config::GenesisConfig;
|
||||
use std::{
|
||||
fs::{self, File},
|
||||
io::{BufReader, Read},
|
||||
path::{
|
||||
Component::{CurDir, Normal},
|
||||
Path,
|
||||
use {
|
||||
bzip2::bufread::BzDecoder,
|
||||
log::*,
|
||||
rand::{thread_rng, Rng},
|
||||
solana_sdk::genesis_config::GenesisConfig,
|
||||
std::{
|
||||
collections::HashMap,
|
||||
fs::{self, File},
|
||||
io::{BufReader, Read},
|
||||
path::{
|
||||
Component::{CurDir, Normal},
|
||||
Path, PathBuf,
|
||||
},
|
||||
time::Instant,
|
||||
},
|
||||
time::Instant,
|
||||
tar::{
|
||||
Archive,
|
||||
EntryType::{Directory, GNUSparse, Regular},
|
||||
},
|
||||
thiserror::Error,
|
||||
};
|
||||
use tar::{
|
||||
Archive,
|
||||
EntryType::{Directory, GNUSparse, Regular},
|
||||
};
|
||||
use thiserror::Error;
|
||||
|
||||
#[derive(Error, Debug)]
|
||||
pub enum UnpackError {
|
||||
|
@ -79,16 +83,15 @@ fn check_unpack_result(unpack_result: bool, path: String) -> Result<()> {
|
|||
Ok(())
|
||||
}
|
||||
|
||||
fn unpack_archive<A: Read, P: AsRef<Path>, C>(
|
||||
fn unpack_archive<'a, A: Read, C>(
|
||||
archive: &mut Archive<A>,
|
||||
unpack_dir: P,
|
||||
apparent_limit_size: u64,
|
||||
actual_limit_size: u64,
|
||||
limit_count: u64,
|
||||
entry_checker: C,
|
||||
mut entry_checker: C,
|
||||
) -> Result<()>
|
||||
where
|
||||
C: Fn(&[&str], tar::EntryType) -> bool,
|
||||
C: FnMut(&[&str], tar::EntryType) -> Option<&'a Path>,
|
||||
{
|
||||
let mut apparent_total_size: u64 = 0;
|
||||
let mut actual_total_size: u64 = 0;
|
||||
|
@ -118,13 +121,17 @@ where
|
|||
}
|
||||
|
||||
let parts: Vec<_> = parts.map(|p| p.unwrap()).collect();
|
||||
if !entry_checker(parts.as_slice(), entry.header().entry_type()) {
|
||||
return Err(UnpackError::Archive(format!(
|
||||
"extra entry found: {:?} {:?}",
|
||||
path_str,
|
||||
entry.header().entry_type(),
|
||||
)));
|
||||
}
|
||||
let unpack_dir = match entry_checker(parts.as_slice(), entry.header().entry_type()) {
|
||||
None => {
|
||||
return Err(UnpackError::Archive(format!(
|
||||
"extra entry found: {:?} {:?}",
|
||||
path_str,
|
||||
entry.header().entry_type(),
|
||||
)));
|
||||
}
|
||||
Some(unpack_dir) => unpack_dir,
|
||||
};
|
||||
|
||||
apparent_total_size = checked_total_size_sum(
|
||||
apparent_total_size,
|
||||
entry.header().size()?,
|
||||
|
@ -139,7 +146,7 @@ where
|
|||
|
||||
// unpack_in does its own sanitization
|
||||
// ref: https://docs.rs/tar/*/tar/struct.Entry.html#method.unpack_in
|
||||
check_unpack_result(entry.unpack_in(&unpack_dir)?, path_str)?;
|
||||
check_unpack_result(entry.unpack_in(unpack_dir)?, path_str)?;
|
||||
total_entries += 1;
|
||||
let now = Instant::now();
|
||||
if now.duration_since(last_log_update).as_secs() >= 10 {
|
||||
|
@ -152,18 +159,41 @@ where
|
|||
Ok(())
|
||||
}
|
||||
|
||||
pub fn unpack_snapshot<A: Read, P: AsRef<Path>>(
|
||||
// Map from AppendVec file name to unpacked file system location
|
||||
pub type UnpackedAppendVecMap = HashMap<String, PathBuf>;
|
||||
|
||||
pub fn unpack_snapshot<A: Read>(
|
||||
archive: &mut Archive<A>,
|
||||
unpack_dir: P,
|
||||
) -> Result<()> {
|
||||
ledger_dir: &Path,
|
||||
account_paths: &[PathBuf],
|
||||
) -> Result<UnpackedAppendVecMap> {
|
||||
assert!(!account_paths.is_empty());
|
||||
let mut unpacked_append_vec_map = UnpackedAppendVecMap::new();
|
||||
|
||||
unpack_archive(
|
||||
archive,
|
||||
unpack_dir,
|
||||
MAX_SNAPSHOT_ARCHIVE_UNPACKED_APPARENT_SIZE,
|
||||
MAX_SNAPSHOT_ARCHIVE_UNPACKED_ACTUAL_SIZE,
|
||||
MAX_SNAPSHOT_ARCHIVE_UNPACKED_COUNT,
|
||||
is_valid_snapshot_archive_entry,
|
||||
|parts, kind| {
|
||||
if is_valid_snapshot_archive_entry(parts, kind) {
|
||||
if let ["accounts", file] = parts {
|
||||
// Randomly distribute the accounts files about the available `account_paths`,
|
||||
let path_index = thread_rng().gen_range(0, account_paths.len());
|
||||
account_paths.get(path_index).map(|path_buf| {
|
||||
unpacked_append_vec_map
|
||||
.insert(file.to_string(), path_buf.join("accounts").join(file));
|
||||
path_buf.as_path()
|
||||
})
|
||||
} else {
|
||||
Some(ledger_dir)
|
||||
}
|
||||
} else {
|
||||
None
|
||||
}
|
||||
},
|
||||
)
|
||||
.map(|_| unpacked_append_vec_map)
|
||||
}
|
||||
|
||||
fn all_digits(v: &str) -> bool {
|
||||
|
@ -264,18 +294,23 @@ pub fn unpack_genesis_archive(
|
|||
Ok(())
|
||||
}
|
||||
|
||||
fn unpack_genesis<A: Read, P: AsRef<Path>>(
|
||||
fn unpack_genesis<A: Read>(
|
||||
archive: &mut Archive<A>,
|
||||
unpack_dir: P,
|
||||
unpack_dir: &Path,
|
||||
max_genesis_archive_unpacked_size: u64,
|
||||
) -> Result<()> {
|
||||
unpack_archive(
|
||||
archive,
|
||||
unpack_dir,
|
||||
max_genesis_archive_unpacked_size,
|
||||
max_genesis_archive_unpacked_size,
|
||||
MAX_GENESIS_ARCHIVE_UNPACKED_COUNT,
|
||||
is_valid_genesis_archive_entry,
|
||||
|p, k| {
|
||||
if is_valid_genesis_archive_entry(p, k) {
|
||||
Some(unpack_dir)
|
||||
} else {
|
||||
None
|
||||
}
|
||||
},
|
||||
)
|
||||
}
|
||||
|
||||
|
@ -427,7 +462,9 @@ mod tests {
|
|||
}
|
||||
|
||||
fn finalize_and_unpack_snapshot(archive: tar::Builder<Vec<u8>>) -> Result<()> {
|
||||
with_finalize_and_unpack(archive, |a, b| unpack_snapshot(a, b))
|
||||
with_finalize_and_unpack(archive, |a, b| {
|
||||
unpack_snapshot(a, b, &[PathBuf::new()]).map(|_| ())
|
||||
})
|
||||
}
|
||||
|
||||
fn finalize_and_unpack_genesis(archive: tar::Builder<Vec<u8>>) -> Result<()> {
|
||||
|
|
|
@ -7,6 +7,7 @@ use {
|
|||
bank::{Bank, BankFieldsToDeserialize, BankRc, Builtins},
|
||||
blockhash_queue::BlockhashQueue,
|
||||
epoch_stakes::EpochStakes,
|
||||
hardened_unpack::UnpackedAppendVecMap,
|
||||
message_processor::MessageProcessor,
|
||||
rent_collector::RentCollector,
|
||||
serde_snapshot::future::SerializableStorage,
|
||||
|
@ -14,9 +15,7 @@ use {
|
|||
},
|
||||
bincode,
|
||||
bincode::{config::Options, Error},
|
||||
fs_extra::dir::CopyOptions,
|
||||
log::{info, warn},
|
||||
rand::{thread_rng, Rng},
|
||||
log::*,
|
||||
serde::{de::DeserializeOwned, Deserialize, Serialize},
|
||||
solana_sdk::{
|
||||
clock::{Epoch, Slot, UnixTimestamp},
|
||||
|
@ -31,8 +30,8 @@ use {
|
|||
},
|
||||
std::{
|
||||
collections::{HashMap, HashSet},
|
||||
io::{BufReader, BufWriter, Read, Write},
|
||||
path::{Path, PathBuf},
|
||||
io::{self, BufReader, BufWriter, Read, Write},
|
||||
path::PathBuf,
|
||||
result::Result,
|
||||
sync::{atomic::Ordering, Arc, RwLock},
|
||||
time::Instant,
|
||||
|
@ -119,11 +118,11 @@ where
|
|||
}
|
||||
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
pub(crate) fn bank_from_stream<R, P>(
|
||||
pub(crate) fn bank_from_stream<R>(
|
||||
serde_style: SerdeStyle,
|
||||
stream: &mut BufReader<R>,
|
||||
append_vecs_path: P,
|
||||
account_paths: &[PathBuf],
|
||||
unpacked_append_vec_map: UnpackedAppendVecMap,
|
||||
genesis_config: &GenesisConfig,
|
||||
frozen_account_pubkeys: &[Pubkey],
|
||||
debug_keys: Option<Arc<HashSet<Pubkey>>>,
|
||||
|
@ -133,7 +132,6 @@ pub(crate) fn bank_from_stream<R, P>(
|
|||
) -> std::result::Result<Bank, Error>
|
||||
where
|
||||
R: Read,
|
||||
P: AsRef<Path>,
|
||||
{
|
||||
macro_rules! INTO {
|
||||
($x:ident) => {{
|
||||
|
@ -145,7 +143,7 @@ where
|
|||
genesis_config,
|
||||
frozen_account_pubkeys,
|
||||
account_paths,
|
||||
append_vecs_path,
|
||||
unpacked_append_vec_map,
|
||||
debug_keys,
|
||||
additional_builtins,
|
||||
account_indexes,
|
||||
|
@ -228,13 +226,13 @@ impl<'a, C: TypeContext<'a>> Serialize for SerializableAccountsDb<'a, C> {
|
|||
impl<'a, C> IgnoreAsHelper for SerializableAccountsDb<'a, C> {}
|
||||
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
fn reconstruct_bank_from_fields<E, P>(
|
||||
fn reconstruct_bank_from_fields<E>(
|
||||
bank_fields: BankFieldsToDeserialize,
|
||||
accounts_db_fields: AccountsDbFields<E>,
|
||||
genesis_config: &GenesisConfig,
|
||||
frozen_account_pubkeys: &[Pubkey],
|
||||
account_paths: &[PathBuf],
|
||||
append_vecs_path: P,
|
||||
unpacked_append_vec_map: UnpackedAppendVecMap,
|
||||
debug_keys: Option<Arc<HashSet<Pubkey>>>,
|
||||
additional_builtins: Option<&Builtins>,
|
||||
account_indexes: HashSet<AccountIndex>,
|
||||
|
@ -242,12 +240,11 @@ fn reconstruct_bank_from_fields<E, P>(
|
|||
) -> Result<Bank, Error>
|
||||
where
|
||||
E: SerializableStorage,
|
||||
P: AsRef<Path>,
|
||||
{
|
||||
let mut accounts_db = reconstruct_accountsdb_from_fields(
|
||||
accounts_db_fields,
|
||||
account_paths,
|
||||
append_vecs_path,
|
||||
unpacked_append_vec_map,
|
||||
&genesis_config.cluster_type,
|
||||
account_indexes,
|
||||
caching_enabled,
|
||||
|
@ -266,17 +263,16 @@ where
|
|||
Ok(bank)
|
||||
}
|
||||
|
||||
fn reconstruct_accountsdb_from_fields<E, P>(
|
||||
fn reconstruct_accountsdb_from_fields<E>(
|
||||
accounts_db_fields: AccountsDbFields<E>,
|
||||
account_paths: &[PathBuf],
|
||||
stream_append_vecs_path: P,
|
||||
unpacked_append_vec_map: UnpackedAppendVecMap,
|
||||
cluster_type: &ClusterType,
|
||||
account_indexes: HashSet<AccountIndex>,
|
||||
caching_enabled: bool,
|
||||
) -> Result<AccountsDb, Error>
|
||||
where
|
||||
E: SerializableStorage,
|
||||
P: AsRef<Path>,
|
||||
{
|
||||
let mut accounts_db = AccountsDb::new_with_config(
|
||||
account_paths.to_vec(),
|
||||
|
@ -308,29 +304,17 @@ where
|
|||
|
||||
let mut new_slot_storage = HashMap::new();
|
||||
for storage_entry in slot_storage.drain(..) {
|
||||
let path_index = thread_rng().gen_range(0, accounts_db.paths.len());
|
||||
let local_dir = &accounts_db.paths[path_index];
|
||||
let file_name = AppendVec::file_name(slot, storage_entry.id());
|
||||
|
||||
// Move the corresponding AppendVec from the snapshot into the directory pointed
|
||||
// at by `local_dir`
|
||||
let append_vec_relative_path = AppendVec::file_name(slot, storage_entry.id());
|
||||
let append_vec_abs_path = stream_append_vecs_path
|
||||
.as_ref()
|
||||
.join(&append_vec_relative_path);
|
||||
let target = local_dir.join(append_vec_abs_path.file_name().unwrap());
|
||||
std::fs::rename(append_vec_abs_path.clone(), target).or_else(|_| {
|
||||
let mut copy_options = CopyOptions::new();
|
||||
copy_options.overwrite = true;
|
||||
fs_extra::move_items(&[&append_vec_abs_path], &local_dir, ©_options)
|
||||
.map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e))
|
||||
.and(Ok(()))
|
||||
let append_vec_path = unpacked_append_vec_map.get(&file_name).ok_or_else(|| {
|
||||
io::Error::new(
|
||||
io::ErrorKind::NotFound,
|
||||
format!("{} not found in unpacked append vecs", file_name),
|
||||
)
|
||||
})?;
|
||||
|
||||
// Notify the AppendVec of the new file location
|
||||
let local_path = local_dir.join(append_vec_relative_path);
|
||||
|
||||
let (accounts, num_accounts) =
|
||||
AppendVec::new_from_file(&local_path, storage_entry.current_len())?;
|
||||
AppendVec::new_from_file(append_vec_path, storage_entry.current_len())?;
|
||||
let u_storage_entry = AccountStorageEntry::new_existing(
|
||||
slot,
|
||||
storage_entry.id(),
|
||||
|
|
|
@ -5,6 +5,7 @@ use {
|
|||
accounts::{create_test_accounts, Accounts},
|
||||
accounts_db::get_temp_accounts_paths,
|
||||
bank::{Bank, StatusCacheRc},
|
||||
hardened_unpack::UnpackedAppendVecMap,
|
||||
},
|
||||
bincode::serialize_into,
|
||||
rand::{thread_rng, Rng},
|
||||
|
@ -15,7 +16,10 @@ use {
|
|||
pubkey::Pubkey,
|
||||
signature::{Keypair, Signer},
|
||||
},
|
||||
std::io::{BufReader, Cursor},
|
||||
std::{
|
||||
io::{BufReader, Cursor},
|
||||
path::Path,
|
||||
},
|
||||
tempfile::TempDir,
|
||||
};
|
||||
|
||||
|
@ -23,19 +27,18 @@ use {
|
|||
fn copy_append_vecs<P: AsRef<Path>>(
|
||||
accounts_db: &AccountsDb,
|
||||
output_dir: P,
|
||||
) -> std::io::Result<()> {
|
||||
) -> std::io::Result<UnpackedAppendVecMap> {
|
||||
let storage_entries = accounts_db.get_snapshot_storages(Slot::max_value());
|
||||
let mut unpacked_append_vec_map = UnpackedAppendVecMap::new();
|
||||
for storage in storage_entries.iter().flatten() {
|
||||
let storage_path = storage.get_path();
|
||||
let output_path = output_dir.as_ref().join(AppendVec::file_name(
|
||||
storage.slot(),
|
||||
storage.append_vec_id(),
|
||||
));
|
||||
|
||||
std::fs::copy(storage_path, output_path)?;
|
||||
let file_name = AppendVec::file_name(storage.slot(), storage.append_vec_id());
|
||||
let output_path = output_dir.as_ref().join(&file_name);
|
||||
std::fs::copy(&storage_path, &output_path)?;
|
||||
unpacked_append_vec_map.insert(file_name, output_path);
|
||||
}
|
||||
|
||||
Ok(())
|
||||
Ok(unpacked_append_vec_map)
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
|
@ -53,21 +56,20 @@ fn check_accounts(accounts: &Accounts, pubkeys: &[Pubkey], num: usize) {
|
|||
}
|
||||
|
||||
#[cfg(test)]
|
||||
fn context_accountsdb_from_stream<'a, C, R, P>(
|
||||
fn context_accountsdb_from_stream<'a, C, R>(
|
||||
stream: &mut BufReader<R>,
|
||||
account_paths: &[PathBuf],
|
||||
stream_append_vecs_path: P,
|
||||
unpacked_append_vec_map: UnpackedAppendVecMap,
|
||||
) -> Result<AccountsDb, Error>
|
||||
where
|
||||
C: TypeContext<'a>,
|
||||
R: Read,
|
||||
P: AsRef<Path>,
|
||||
{
|
||||
// read and deserialise the accounts database directly from the stream
|
||||
reconstruct_accountsdb_from_fields(
|
||||
C::deserialize_accounts_db_fields(stream)?,
|
||||
account_paths,
|
||||
stream_append_vecs_path,
|
||||
unpacked_append_vec_map,
|
||||
&ClusterType::Development,
|
||||
HashSet::new(),
|
||||
false,
|
||||
|
@ -75,21 +77,20 @@ where
|
|||
}
|
||||
|
||||
#[cfg(test)]
|
||||
fn accountsdb_from_stream<R, P>(
|
||||
fn accountsdb_from_stream<R>(
|
||||
serde_style: SerdeStyle,
|
||||
stream: &mut BufReader<R>,
|
||||
account_paths: &[PathBuf],
|
||||
stream_append_vecs_path: P,
|
||||
unpacked_append_vec_map: UnpackedAppendVecMap,
|
||||
) -> Result<AccountsDb, Error>
|
||||
where
|
||||
R: Read,
|
||||
P: AsRef<Path>,
|
||||
{
|
||||
match serde_style {
|
||||
SerdeStyle::Newer => context_accountsdb_from_stream::<TypeContextFuture, R, P>(
|
||||
SerdeStyle::Newer => context_accountsdb_from_stream::<TypeContextFuture, R>(
|
||||
stream,
|
||||
account_paths,
|
||||
stream_append_vecs_path,
|
||||
unpacked_append_vec_map,
|
||||
),
|
||||
}
|
||||
}
|
||||
|
@ -143,7 +144,8 @@ fn test_accounts_serialize_style(serde_style: SerdeStyle) {
|
|||
let copied_accounts = TempDir::new().unwrap();
|
||||
|
||||
// Simulate obtaining a copy of the AppendVecs from a tarball
|
||||
copy_append_vecs(&accounts.accounts_db, copied_accounts.path()).unwrap();
|
||||
let unpacked_append_vec_map =
|
||||
copy_append_vecs(&accounts.accounts_db, copied_accounts.path()).unwrap();
|
||||
|
||||
let buf = writer.into_inner();
|
||||
let mut reader = BufReader::new(&buf[..]);
|
||||
|
@ -153,7 +155,7 @@ fn test_accounts_serialize_style(serde_style: SerdeStyle) {
|
|||
serde_style,
|
||||
&mut reader,
|
||||
&daccounts_paths,
|
||||
copied_accounts.path(),
|
||||
unpacked_append_vec_map,
|
||||
)
|
||||
.unwrap(),
|
||||
);
|
||||
|
@ -207,12 +209,13 @@ fn test_bank_serialize_style(serde_style: SerdeStyle) {
|
|||
ref_sc.status_cache.write().unwrap().add_root(2);
|
||||
// Create a directory to simulate AppendVecs unpackaged from a snapshot tar
|
||||
let copied_accounts = TempDir::new().unwrap();
|
||||
copy_append_vecs(&bank2.rc.accounts.accounts_db, copied_accounts.path()).unwrap();
|
||||
let unpacked_append_vec_map =
|
||||
copy_append_vecs(&bank2.rc.accounts.accounts_db, copied_accounts.path()).unwrap();
|
||||
let mut dbank = crate::serde_snapshot::bank_from_stream(
|
||||
serde_style,
|
||||
&mut reader,
|
||||
copied_accounts.path(),
|
||||
&dbank_paths,
|
||||
unpacked_append_vec_map,
|
||||
&genesis_config,
|
||||
&[],
|
||||
None,
|
||||
|
@ -247,9 +250,22 @@ pub(crate) fn reconstruct_accounts_db_via_serialization(
|
|||
let buf = writer.into_inner();
|
||||
let mut reader = BufReader::new(&buf[..]);
|
||||
let copied_accounts = TempDir::new().unwrap();
|
||||
|
||||
// Simulate obtaining a copy of the AppendVecs from a tarball
|
||||
copy_append_vecs(&accounts, copied_accounts.path()).unwrap();
|
||||
accountsdb_from_stream(SerdeStyle::Newer, &mut reader, &[], copied_accounts.path()).unwrap()
|
||||
let unpacked_append_vec_map = copy_append_vecs(&accounts, copied_accounts.path()).unwrap();
|
||||
let mut accounts_db =
|
||||
accountsdb_from_stream(SerdeStyle::Newer, &mut reader, &[], unpacked_append_vec_map)
|
||||
.unwrap();
|
||||
|
||||
// The append vecs will be used from `copied_accounts` directly by the new AccountsDb so keep
|
||||
// its TempDir alive
|
||||
accounts_db
|
||||
.temp_paths
|
||||
.as_mut()
|
||||
.unwrap()
|
||||
.push(copied_accounts);
|
||||
|
||||
accounts_db
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
|
|
@ -1,42 +1,43 @@
|
|||
use crate::{
|
||||
accounts_db::AccountsDb,
|
||||
accounts_index::AccountIndex,
|
||||
bank::{Bank, BankSlotDelta, Builtins},
|
||||
bank_forks::ArchiveFormat,
|
||||
hardened_unpack::{unpack_snapshot, UnpackError},
|
||||
serde_snapshot::{
|
||||
bank_from_stream, bank_to_stream, SerdeStyle, SnapshotStorage, SnapshotStorages,
|
||||
use {
|
||||
crate::{
|
||||
accounts_db::AccountsDb,
|
||||
accounts_index::AccountIndex,
|
||||
bank::{Bank, BankSlotDelta, Builtins},
|
||||
bank_forks::ArchiveFormat,
|
||||
hardened_unpack::{unpack_snapshot, UnpackError, UnpackedAppendVecMap},
|
||||
serde_snapshot::{
|
||||
bank_from_stream, bank_to_stream, SerdeStyle, SnapshotStorage, SnapshotStorages,
|
||||
},
|
||||
snapshot_package::{
|
||||
AccountsPackage, AccountsPackagePre, AccountsPackageSendError, AccountsPackageSender,
|
||||
},
|
||||
},
|
||||
snapshot_package::{
|
||||
AccountsPackage, AccountsPackagePre, AccountsPackageSendError, AccountsPackageSender,
|
||||
bincode::{config::Options, serialize_into},
|
||||
bzip2::bufread::BzDecoder,
|
||||
flate2::read::GzDecoder,
|
||||
log::*,
|
||||
rayon::ThreadPool,
|
||||
regex::Regex,
|
||||
solana_measure::measure::Measure,
|
||||
solana_sdk::{clock::Slot, genesis_config::GenesisConfig, hash::Hash, pubkey::Pubkey},
|
||||
std::{
|
||||
cmp::Ordering,
|
||||
collections::HashSet,
|
||||
fmt,
|
||||
fs::{self, File},
|
||||
io::{
|
||||
self, BufReader, BufWriter, Error as IoError, ErrorKind, Read, Seek, SeekFrom, Write,
|
||||
},
|
||||
path::{Path, PathBuf},
|
||||
process::{self, ExitStatus},
|
||||
str::FromStr,
|
||||
sync::Arc,
|
||||
},
|
||||
tar::Archive,
|
||||
thiserror::Error,
|
||||
};
|
||||
use bincode::{config::Options, serialize_into};
|
||||
use bzip2::bufread::BzDecoder;
|
||||
use flate2::read::GzDecoder;
|
||||
use log::*;
|
||||
use rayon::ThreadPool;
|
||||
use regex::Regex;
|
||||
use solana_measure::measure::Measure;
|
||||
use solana_sdk::{clock::Slot, genesis_config::GenesisConfig, hash::Hash, pubkey::Pubkey};
|
||||
use std::collections::HashSet;
|
||||
use std::sync::Arc;
|
||||
use std::{
|
||||
cmp::Ordering,
|
||||
fmt,
|
||||
fs::{self, File},
|
||||
io::{self, BufReader, BufWriter, Error as IoError, ErrorKind, Read, Seek, SeekFrom, Write},
|
||||
path::{Path, PathBuf},
|
||||
process::{self, ExitStatus},
|
||||
str::FromStr,
|
||||
};
|
||||
use tar::Archive;
|
||||
use thiserror::Error;
|
||||
|
||||
pub const SNAPSHOT_STATUS_CACHE_FILE_NAME: &str = "status_cache";
|
||||
pub const TAR_SNAPSHOTS_DIR: &str = "snapshots";
|
||||
pub const TAR_ACCOUNTS_DIR: &str = "accounts";
|
||||
pub const TAR_VERSION_FILE: &str = "version";
|
||||
|
||||
pub const MAX_SNAPSHOTS: usize = 8; // Save some snapshots but not too many
|
||||
const MAX_SNAPSHOT_DATA_FILE_SIZE: u64 = 32 * 1024 * 1024 * 1024; // 32 GiB
|
||||
|
@ -256,9 +257,9 @@ pub fn archive_snapshot_package(snapshot_package: &AccountsPackage) -> Result<()
|
|||
))
|
||||
.tempdir_in(tar_dir)?;
|
||||
|
||||
let staging_accounts_dir = staging_dir.path().join(TAR_ACCOUNTS_DIR);
|
||||
let staging_snapshots_dir = staging_dir.path().join(TAR_SNAPSHOTS_DIR);
|
||||
let staging_version_file = staging_dir.path().join(TAR_VERSION_FILE);
|
||||
let staging_accounts_dir = staging_dir.path().join("accounts");
|
||||
let staging_snapshots_dir = staging_dir.path().join("snapshots");
|
||||
let staging_version_file = staging_dir.path().join("version");
|
||||
fs::create_dir_all(&staging_accounts_dir)?;
|
||||
|
||||
// Add the snapshots to the staging directory
|
||||
|
@ -307,9 +308,9 @@ pub fn archive_snapshot_package(snapshot_package: &AccountsPackage) -> Result<()
|
|||
"chS",
|
||||
"-C",
|
||||
staging_dir.path().to_str().unwrap(),
|
||||
TAR_ACCOUNTS_DIR,
|
||||
TAR_SNAPSHOTS_DIR,
|
||||
TAR_VERSION_FILE,
|
||||
"accounts",
|
||||
"snapshots",
|
||||
"version",
|
||||
])
|
||||
.stdin(process::Stdio::null())
|
||||
.stdout(process::Stdio::piped())
|
||||
|
@ -595,26 +596,30 @@ pub fn bank_from_archive<P: AsRef<Path>>(
|
|||
account_indexes: HashSet<AccountIndex>,
|
||||
accounts_db_caching_enabled: bool,
|
||||
) -> Result<Bank> {
|
||||
// Untar the snapshot into a temporary directory
|
||||
let unpack_dir = tempfile::Builder::new()
|
||||
.prefix(TMP_SNAPSHOT_PREFIX)
|
||||
.tempdir_in(snapshot_path)?;
|
||||
untar_snapshot_in(&snapshot_tar, &unpack_dir, archive_format)?;
|
||||
|
||||
let unpacked_append_vec_map = untar_snapshot_in(
|
||||
&snapshot_tar,
|
||||
&unpack_dir.as_ref(),
|
||||
account_paths,
|
||||
archive_format,
|
||||
)?;
|
||||
|
||||
let mut measure = Measure::start("bank rebuild from snapshot");
|
||||
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 unpacked_version_file = unpack_dir.as_ref().join(TAR_VERSION_FILE);
|
||||
let unpacked_snapshots_dir = unpack_dir.as_ref().join("snapshots");
|
||||
let unpacked_version_file = unpack_dir.as_ref().join("version");
|
||||
|
||||
let mut snapshot_version = String::new();
|
||||
File::open(unpacked_version_file).and_then(|mut f| f.read_to_string(&mut snapshot_version))?;
|
||||
|
||||
let bank = rebuild_bank_from_snapshots(
|
||||
snapshot_version.trim(),
|
||||
account_paths,
|
||||
frozen_account_pubkeys,
|
||||
&unpacked_snapshots_dir,
|
||||
unpacked_accounts_dir,
|
||||
account_paths,
|
||||
unpacked_append_vec_map,
|
||||
genesis_config,
|
||||
debug_keys,
|
||||
additional_builtins,
|
||||
|
@ -722,56 +727,54 @@ pub fn purge_old_snapshot_archives<P: AsRef<Path>>(snapshot_output_dir: P) {
|
|||
}
|
||||
}
|
||||
|
||||
pub fn untar_snapshot_in<P: AsRef<Path>, Q: AsRef<Path>>(
|
||||
fn untar_snapshot_in<P: AsRef<Path>>(
|
||||
snapshot_tar: P,
|
||||
unpack_dir: Q,
|
||||
unpack_dir: &Path,
|
||||
account_paths: &[PathBuf],
|
||||
archive_format: ArchiveFormat,
|
||||
) -> Result<()> {
|
||||
) -> Result<UnpackedAppendVecMap> {
|
||||
let mut measure = Measure::start("snapshot untar");
|
||||
let tar_name = File::open(&snapshot_tar)?;
|
||||
match archive_format {
|
||||
let account_paths_map = match archive_format {
|
||||
ArchiveFormat::TarBzip2 => {
|
||||
let tar = BzDecoder::new(BufReader::new(tar_name));
|
||||
let mut archive = Archive::new(tar);
|
||||
unpack_snapshot(&mut archive, unpack_dir)?;
|
||||
unpack_snapshot(&mut archive, unpack_dir, account_paths)?
|
||||
}
|
||||
ArchiveFormat::TarGzip => {
|
||||
let tar = GzDecoder::new(BufReader::new(tar_name));
|
||||
let mut archive = Archive::new(tar);
|
||||
unpack_snapshot(&mut archive, unpack_dir)?;
|
||||
unpack_snapshot(&mut archive, unpack_dir, account_paths)?
|
||||
}
|
||||
ArchiveFormat::TarZstd => {
|
||||
let tar = zstd::stream::read::Decoder::new(BufReader::new(tar_name))?;
|
||||
let mut archive = Archive::new(tar);
|
||||
unpack_snapshot(&mut archive, unpack_dir)?;
|
||||
unpack_snapshot(&mut archive, unpack_dir, account_paths)?
|
||||
}
|
||||
ArchiveFormat::Tar => {
|
||||
let tar = BufReader::new(tar_name);
|
||||
let mut archive = Archive::new(tar);
|
||||
unpack_snapshot(&mut archive, unpack_dir)?;
|
||||
unpack_snapshot(&mut archive, unpack_dir, account_paths)?
|
||||
}
|
||||
};
|
||||
measure.stop();
|
||||
info!("{}", measure);
|
||||
Ok(())
|
||||
Ok(account_paths_map)
|
||||
}
|
||||
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
fn rebuild_bank_from_snapshots<P>(
|
||||
fn rebuild_bank_from_snapshots(
|
||||
snapshot_version: &str,
|
||||
account_paths: &[PathBuf],
|
||||
frozen_account_pubkeys: &[Pubkey],
|
||||
unpacked_snapshots_dir: &Path,
|
||||
append_vecs_path: P,
|
||||
account_paths: &[PathBuf],
|
||||
unpacked_append_vec_map: UnpackedAppendVecMap,
|
||||
genesis_config: &GenesisConfig,
|
||||
debug_keys: Option<Arc<HashSet<Pubkey>>>,
|
||||
additional_builtins: Option<&Builtins>,
|
||||
account_indexes: HashSet<AccountIndex>,
|
||||
accounts_db_caching_enabled: bool,
|
||||
) -> Result<Bank>
|
||||
where
|
||||
P: AsRef<Path>,
|
||||
{
|
||||
) -> Result<Bank> {
|
||||
info!("snapshot version: {}", snapshot_version);
|
||||
|
||||
let snapshot_version_enum =
|
||||
|
@ -798,8 +801,8 @@ where
|
|||
SnapshotVersion::V1_2_0 => bank_from_stream(
|
||||
SerdeStyle::Newer,
|
||||
&mut stream,
|
||||
&append_vecs_path,
|
||||
account_paths,
|
||||
unpacked_append_vec_map,
|
||||
genesis_config,
|
||||
frozen_account_pubkeys,
|
||||
debug_keys,
|
||||
|
@ -855,14 +858,20 @@ pub fn verify_snapshot_archive<P, Q, R>(
|
|||
{
|
||||
let temp_dir = tempfile::TempDir::new().unwrap();
|
||||
let unpack_dir = temp_dir.path();
|
||||
untar_snapshot_in(snapshot_archive, &unpack_dir, archive_format).unwrap();
|
||||
untar_snapshot_in(
|
||||
snapshot_archive,
|
||||
&unpack_dir,
|
||||
&[unpack_dir.to_path_buf()],
|
||||
archive_format,
|
||||
)
|
||||
.unwrap();
|
||||
|
||||
// Check snapshots are the same
|
||||
let unpacked_snapshots = unpack_dir.join(&TAR_SNAPSHOTS_DIR);
|
||||
let unpacked_snapshots = unpack_dir.join("snapshots");
|
||||
assert!(!dir_diff::is_different(&snapshots_to_verify, unpacked_snapshots).unwrap());
|
||||
|
||||
// Check the account entries are the same
|
||||
let unpacked_accounts = unpack_dir.join(&TAR_ACCOUNTS_DIR);
|
||||
let unpacked_accounts = unpack_dir.join("accounts");
|
||||
assert!(!dir_diff::is_different(&storages_to_verify, unpacked_accounts).unwrap());
|
||||
}
|
||||
|
||||
|
|
Loading…
Reference in New Issue