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

Make background services aware of incremental snapshots (#19401)

Browse Source 
AccountsBackgroundService now knows about incremental snapshots.  It is
now also in charge of deciding if an AccountsPackage is destined to be a
SnapshotPackage or not (or just used by AccountsHashVerifier).

!!! New behavior changes !!!

Taking snapshots (both bank and archive) **MUST** succeed.

This is required because of how the last full snapshot slot is
calculated, which is used by AccountsBackgroundService when calling
`clean_accounts()`.

File system calls are now unwrapped and will result in a crash. As Trent told me:

>Well I think if a snapshot fails due to some IO error, it's very likely that the operator is going to have to intervene before it works.  We should exit error in this case, otherwise the validator might happily spin for several more hours, never successfully writing a complete snapshot, before something else brings it down.  This would leave the validator's last local snapshot many more slots behind than it would be had we exited outright and potentially force the operator to abandon ledger continuity in favor of a quick catchup

Other errors will set the `exit` flag to `true`, and the node will gracefully shutdown.

Fixes #19167 
Fixes #19168
This commit is contained in:
Brooks Prumo
2021-08-31 18:33:27 -05:00
committed by GitHub
parent 718ab7c12e
commit fe9ee9134a
9 changed files with 408 additions and 425 deletions
Download Patch File Download Diff File Expand all files Collapse all files

160
core/src/accounts_hash_verifier.rs
View File

@ -6,14 +6,16 @@
use rayon::ThreadPool;
use solana_gossip::cluster_info::{ClusterInfo, MAX_SNAPSHOT_HASHES};
use solana_measure::measure::Measure;
use solana_runtime::{
accounts_db,
snapshot_archive_info::SnapshotArchiveInfoGetter,
accounts_db::{self, AccountsDb},
accounts_hash::HashStats,
snapshot_config::SnapshotConfig,
snapshot_package::{
AccountsPackage, AccountsPackageReceiver, PendingSnapshotPackage, SnapshotPackage,
SnapshotType,
},
snapshot_utils,
sorted_storages::SortedStorages,
};
use solana_sdk::{clock::Slot, hash::Hash, pubkey::Pubkey};
use std::collections::{HashMap, HashSet};
@ -48,7 +50,7 @@ impl AccountsHashVerifier {
.name("solana-hash-accounts".to_string())
.spawn(move || {
let mut hashes = vec![];
let mut thread_pool_storage = None;
let mut thread_pool = None;
loop {
if exit.load(Ordering::Relaxed) {
break;
@ -56,11 +58,9 @@ impl AccountsHashVerifier {
match accounts_package_receiver.recv_timeout(Duration::from_secs(1)) {
Ok(accounts_package) => {
if accounts_package.hash_for_testing.is_some()
&& thread_pool_storage.is_none()
if accounts_package.hash_for_testing.is_some() && thread_pool.is_none()
{
thread_pool_storage =
Some(accounts_db::make_min_priority_thread_pool());
thread_pool = Some(accounts_db::make_min_priority_thread_pool());
}
Self::process_accounts_package(
@ -73,7 +73,7 @@ impl AccountsHashVerifier {
&exit,
fault_injection_rate_slots,
snapshot_config.as_ref(),
thread_pool_storage.as_ref(),
thread_pool.as_ref(),
);
}
Err(RecvTimeoutError::Disconnected) => break,
@ -100,45 +100,69 @@ impl AccountsHashVerifier {
snapshot_config: Option<&SnapshotConfig>,
thread_pool: Option<&ThreadPool>,
) {
let snapshot_package =
snapshot_utils::process_accounts_package(accounts_package, thread_pool, None);
Self::process_snapshot_package(
snapshot_package,
Self::verify_accounts_package_hash(&accounts_package, thread_pool);
Self::push_accounts_hashes_to_cluster(
&accounts_package,
cluster_info,
trusted_validators,
halt_on_trusted_validator_accounts_hash_mismatch,
pending_snapshot_package,
hashes,
exit,
fault_injection_rate_slots,
snapshot_config,
);
Self::submit_for_packaging(accounts_package, pending_snapshot_package, snapshot_config);
}
fn verify_accounts_package_hash(
accounts_package: &AccountsPackage,
thread_pool: Option<&ThreadPool>,
) {
let mut measure_hash = Measure::start("hash");
if let Some(expected_hash) = accounts_package.hash_for_testing {
let sorted_storages = SortedStorages::new(&accounts_package.snapshot_storages);
let (hash, lamports) = AccountsDb::calculate_accounts_hash_without_index(
&sorted_storages,
thread_pool,
HashStats::default(),
false,
None,
)
.unwrap();
assert_eq!(accounts_package.expected_capitalization, lamports);
assert_eq!(expected_hash, hash);
};
measure_hash.stop();
datapoint_info!(
"accounts_hash_verifier",
("calculate_hash", measure_hash.as_us(), i64),
);
}
fn process_snapshot_package(
snapshot_package: SnapshotPackage,
fn push_accounts_hashes_to_cluster(
accounts_package: &AccountsPackage,
cluster_info: &ClusterInfo,
trusted_validators: Option<&HashSet<Pubkey>>,
halt_on_trusted_validator_accounts_hash_mismatch: bool,
pending_snapshot_package: Option<&PendingSnapshotPackage>,
hashes: &mut Vec<(Slot, Hash)>,
exit: &Arc<AtomicBool>,
fault_injection_rate_slots: u64,
snapshot_config: Option<&SnapshotConfig>,
) {
let hash = *snapshot_package.hash();
let hash = accounts_package.hash;
if fault_injection_rate_slots != 0
&& snapshot_package.slot() % fault_injection_rate_slots == 0
&& accounts_package.slot % fault_injection_rate_slots == 0
{
// For testing, publish an invalid hash to gossip.
use rand::{thread_rng, Rng};
use solana_sdk::hash::extend_and_hash;
warn!("inserting fault at slot: {}", snapshot_package.slot());
warn!("inserting fault at slot: {}", accounts_package.slot);
let rand = thread_rng().gen_range(0, 10);
let hash = extend_and_hash(&hash, &[rand]);
hashes.push((snapshot_package.slot(), hash));
hashes.push((accounts_package.slot, hash));
} else {
hashes.push((snapshot_package.slot(), hash));
hashes.push((accounts_package.slot, hash));
}
while hashes.len() > MAX_SNAPSHOT_HASHES {
@ -155,19 +179,43 @@ impl AccountsHashVerifier {
}
}
if let Some(snapshot_config) = snapshot_config {
if snapshot_package.block_height % snapshot_config.full_snapshot_archive_interval_slots
== 0
{
if let Some(pending_snapshot_package) = pending_snapshot_package {
*pending_snapshot_package.lock().unwrap() = Some(snapshot_package);
}
}
}
cluster_info.push_accounts_hashes(hashes.clone());
}
fn submit_for_packaging(
accounts_package: AccountsPackage,
pending_snapshot_package: Option<&PendingSnapshotPackage>,
snapshot_config: Option<&SnapshotConfig>,
) {
if accounts_package.snapshot_type.is_none()
|| pending_snapshot_package.is_none()
|| snapshot_config.is_none()
{
return;
};
let snapshot_package = SnapshotPackage::from(accounts_package);
let pending_snapshot_package = pending_snapshot_package.unwrap();
let _snapshot_config = snapshot_config.unwrap();
// If the snapshot package is an Incremental Snapshot, do not submit it if there's already
// a pending Full Snapshot.
let can_submit = match snapshot_package.snapshot_type {
SnapshotType::FullSnapshot => true,
SnapshotType::IncrementalSnapshot(_) => pending_snapshot_package
.lock()
.unwrap()
.as_ref()
.map_or(true, |snapshot_package| {
snapshot_package.snapshot_type.is_incremental_snapshot()
}),
};
if can_submit {
*pending_snapshot_package.lock().unwrap() = Some(snapshot_package);
}
}
fn should_halt(
cluster_info: &ClusterInfo,
trusted_validators: Option<&HashSet<Pubkey>>,
@ -225,10 +273,10 @@ mod tests {
use solana_gossip::{cluster_info::make_accounts_hashes_message, contact_info::ContactInfo};
use solana_runtime::{
snapshot_config::LastFullSnapshotSlot,
snapshot_package::SnapshotType,
snapshot_utils::{ArchiveFormat, SnapshotVersion},
};
use solana_sdk::{
genesis_config::ClusterType,
hash::hash,
signature::{Keypair, Signer},
};
@ -301,30 +349,24 @@ mod tests {
last_full_snapshot_slot: LastFullSnapshotSlot::default(),
};
for i in 0..MAX_SNAPSHOT_HASHES + 1 {
let slot = full_snapshot_archive_interval_slots + i as u64;
let block_height = full_snapshot_archive_interval_slots + i as u64;
let slot_deltas = vec![];
let snapshot_links = TempDir::new().unwrap();
let storages = vec![];
let snapshot_archive_path = PathBuf::from(".");
let hash = hash(&[i as u8]);
let archive_format = ArchiveFormat::TarBzip2;
let snapshot_version = SnapshotVersion::default();
let snapshot_package = SnapshotPackage::new(
slot,
block_height,
slot_deltas,
snapshot_links,
storages,
snapshot_archive_path,
hash,
archive_format,
snapshot_version,
SnapshotType::FullSnapshot,
);
let accounts_package = AccountsPackage {
slot: full_snapshot_archive_interval_slots + i as u64,
block_height: full_snapshot_archive_interval_slots + i as u64,
slot_deltas: vec![],
snapshot_links: TempDir::new().unwrap(),
snapshot_storages: vec![],
hash: hash(&[i as u8]),
archive_format: ArchiveFormat::TarBzip2,
snapshot_version: SnapshotVersion::default(),
snapshot_archives_dir: PathBuf::default(),
expected_capitalization: 0,
hash_for_testing: None,
cluster_type: ClusterType::MainnetBeta,
snapshot_type: None,
};
AccountsHashVerifier::process_snapshot_package(
snapshot_package,
AccountsHashVerifier::process_accounts_package(
accounts_package,
&cluster_info,
Some(&trusted_validators),
false,
@ -333,7 +375,9 @@ mod tests {
&exit,
0,
Some(&snapshot_config),
None,
);
// sleep for 1ms to create a newer timestmap for gossip entry
// otherwise the timestamp won't be newer.
std::thread::sleep(Duration::from_millis(1));

73
core/src/snapshot_packager_service.rs
View File

@ -42,25 +42,26 @@ impl SnapshotPackagerService {
}
let snapshot_package = pending_snapshot_package.lock().unwrap().take();
if let Some(snapshot_package) = snapshot_package {
match snapshot_utils::archive_snapshot_package(
&snapshot_package,
maximum_snapshots_to_retain,
) {
Ok(_) => {
hashes.push((snapshot_package.slot(), *snapshot_package.hash()));
while hashes.len() > MAX_SNAPSHOT_HASHES {
hashes.remove(0);
}
cluster_info.push_snapshot_hashes(hashes.clone());
}
Err(err) => {
warn!("Failed to create snapshot archive: {}", err);
}
};
} else {
if snapshot_package.is_none() {
std::thread::sleep(Duration::from_millis(100));
continue;
}
let snapshot_package = snapshot_package.unwrap();
// Archiving the snapshot package is not allowed to fail.
// AccountsBackgroundService calls `clean_accounts()` with a value for
// last_full_snapshot_slot that requires this archive call to succeed.
snapshot_utils::archive_snapshot_package(
&snapshot_package,
maximum_snapshots_to_retain,
)
.expect("failed to archive snapshot package");
hashes.push((snapshot_package.slot(), *snapshot_package.hash()));
while hashes.len() > MAX_SNAPSHOT_HASHES {
hashes.remove(0);
}
cluster_info.push_snapshot_hashes(hashes.clone());
}
})
.unwrap();
@ -82,6 +83,7 @@ mod tests {
use solana_runtime::{
accounts_db::AccountStorageEntry,
bank::BankSlotDelta,
snapshot_archive_info::SnapshotArchiveInfo,
snapshot_package::{SnapshotPackage, SnapshotType},
snapshot_utils::{self, ArchiveFormat, SnapshotVersion, SNAPSHOT_STATUS_CACHE_FILE_NAME},
};
@ -160,24 +162,29 @@ mod tests {
}
// Create a packageable snapshot
let slot = 42;
let hash = Hash::default();
let archive_format = ArchiveFormat::TarBzip2;
let output_tar_path = snapshot_utils::build_full_snapshot_archive_path(
snapshot_archives_dir,
42,
&Hash::default(),
ArchiveFormat::TarBzip2,
);
let snapshot_package = SnapshotPackage::new(
5,
5,
vec![],
link_snapshots_dir,
vec![storage_entries],
output_tar_path.clone(),
Hash::default(),
ArchiveFormat::TarBzip2,
SnapshotVersion::default(),
SnapshotType::FullSnapshot,
slot,
&hash,
archive_format,
);
let snapshot_package = SnapshotPackage {
snapshot_archive_info: SnapshotArchiveInfo {
path: output_tar_path.clone(),
slot,
hash,
archive_format,
},
block_height: slot,
slot_deltas: vec![],
snapshot_links: link_snapshots_dir,
snapshot_storages: vec![storage_entries],
snapshot_version: SnapshotVersion::default(),
snapshot_type: SnapshotType::FullSnapshot,
};
// Make tarball from packageable snapshot
snapshot_utils::archive_snapshot_package(
@ -204,7 +211,7 @@ mod tests {
output_tar_path,
snapshots_dir,
accounts_dir,
ArchiveFormat::TarBzip2,
archive_format,
);
}
}

1
core/src/tvu.rs
View File

@ -340,6 +340,7 @@ impl Tvu {
tvu_config.accounts_db_caching_enabled,
tvu_config.test_hash_calculation,
tvu_config.use_index_hash_calculation,
None,
);
Tvu {

1
core/src/validator.rs
View File

@ -1212,7 +1212,6 @@ fn new_banks_from_ledger(
None,
&snapshot_config.snapshot_archives_dir,
snapshot_config.archive_format,
Some(bank_forks.root_bank().get_thread_pool()),
snapshot_config.maximum_snapshots_to_retain,
)
.unwrap_or_else(|err| {

65
core/tests/snapshots.rs
View File

@ -66,7 +66,9 @@ mod tests {
genesis_utils::{create_genesis_config, GenesisConfigInfo},
snapshot_archive_info::FullSnapshotArchiveInfo,
snapshot_config::{LastFullSnapshotSlot, SnapshotConfig},
snapshot_package::{AccountsPackage, PendingSnapshotPackage},
snapshot_package::{
AccountsPackage, PendingSnapshotPackage, SnapshotPackage, SnapshotType,
},
snapshot_utils::{
self, ArchiveFormat, SnapshotVersion, DEFAULT_MAX_FULL_SNAPSHOT_ARCHIVES_TO_RETAIN,
},
@ -255,18 +257,19 @@ mod tests {
snapshot_request_receiver,
accounts_package_sender,
};
for slot in 0..last_slot {
let mut bank = Bank::new_from_parent(&bank_forks[slot], &Pubkey::default(), slot + 1);
for slot in 1..=last_slot {
let mut bank = Bank::new_from_parent(&bank_forks[slot - 1], &Pubkey::default(), slot);
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 snapshotting of bank and the purging logic on status_cache to
// kick in
if slot % set_root_interval == 0 || slot == last_slot - 1 {
if slot % set_root_interval == 0 || slot == last_slot {
// set_root should send a snapshot request
bank_forks.set_root(bank.slot(), &request_sender, None);
bank.update_accounts_hash();
snapshot_request_handler.handle_snapshot_requests(false, false, false, 0);
snapshot_request_handler
.handle_snapshot_requests(false, false, false, 0, &mut None);
}
}
@ -277,7 +280,7 @@ mod tests {
let last_bank_snapshot_info =
snapshot_utils::get_highest_bank_snapshot_info(bank_snapshots_dir)
.expect("no bank snapshots found in path");
let accounts_package = AccountsPackage::new_for_full_snapshot(
let accounts_package = AccountsPackage::new(
last_bank,
&last_bank_snapshot_info,
bank_snapshots_dir,
@ -287,13 +290,10 @@ mod tests {
ArchiveFormat::TarBzip2,
snapshot_version,
None,
Some(SnapshotType::FullSnapshot),
)
.unwrap();
let snapshot_package = snapshot_utils::process_accounts_package(
accounts_package,
Some(last_bank.get_thread_pool()),
None,
);
let snapshot_package = SnapshotPackage::from(accounts_package);
snapshot_utils::archive_snapshot_package(
&snapshot_package,
DEFAULT_MAX_FULL_SNAPSHOT_ARCHIVES_TO_RETAIN,
@ -415,8 +415,9 @@ mod tests {
bank_snapshots_dir,
snapshot_archives_dir,
snapshot_config.snapshot_version,
&snapshot_config.archive_format,
snapshot_config.archive_format,
None,
Some(SnapshotType::FullSnapshot),
)
.unwrap();
@ -504,8 +505,6 @@ mod tests {
DEFAULT_MAX_FULL_SNAPSHOT_ARCHIVES_TO_RETAIN,
);
let thread_pool = accounts_db::make_min_priority_thread_pool();
let _package_receiver = std::thread::Builder::new()
.name("package-receiver".to_string())
.spawn(move || {
@ -515,11 +514,7 @@ mod tests {
accounts_package = new_accounts_package;
}
let snapshot_package = solana_runtime::snapshot_utils::process_accounts_package(
accounts_package,
Some(&thread_pool),
None,
);
let snapshot_package = SnapshotPackage::from(accounts_package);
*pending_snapshot_package.lock().unwrap() = Some(snapshot_package);
}
@ -625,7 +620,7 @@ mod tests {
run_bank_forks_snapshot_n(
snapshot_version,
cluster_type,
(MAX_CACHE_ENTRIES * 2 + 1) as u64,
(MAX_CACHE_ENTRIES * 2) as u64,
|bank, mint_keypair| {
let tx = system_transaction::transfer(
mint_keypair,
@ -712,14 +707,19 @@ mod tests {
// set_root sends a snapshot request
bank_forks.set_root(bank.slot(), &request_sender, None);
bank.update_accounts_hash();
snapshot_request_handler.handle_snapshot_requests(false, false, false, 0);
snapshot_request_handler.handle_snapshot_requests(
false,
false,
false,
0,
&mut last_full_snapshot_slot,
);
}
// Since AccountsBackgroundService isn't running, manually make a full snapshot archive
// at the right interval
if slot % FULL_SNAPSHOT_ARCHIVE_INTERVAL_SLOTS == 0 {
make_full_snapshot_archive(&bank, &snapshot_test_config.snapshot_config).unwrap();
last_full_snapshot_slot = Some(slot);
}
// Similarly, make an incremental snapshot archive at the right interval, but only if
// there's been at least one full snapshot first, and a full snapshot wasn't already
@ -764,7 +764,7 @@ mod tests {
"did not find bank snapshot with this path",
)
})?;
snapshot_utils::package_process_and_archive_full_snapshot(
snapshot_utils::package_and_archive_full_snapshot(
bank,
&bank_snapshot_info,
&snapshot_config.bank_snapshots_dir,
@ -772,7 +772,6 @@ mod tests {
bank.get_snapshot_storages(None),
snapshot_config.archive_format,
snapshot_config.snapshot_version,
None,
snapshot_config.maximum_snapshots_to_retain,
)?;
@ -800,7 +799,7 @@ mod tests {
)
})?;
let storages = bank.get_snapshot_storages(Some(incremental_snapshot_base_slot));
snapshot_utils::package_process_and_archive_incremental_snapshot(
snapshot_utils::package_and_archive_incremental_snapshot(
bank,
incremental_snapshot_base_slot,
&bank_snapshot_info,
@ -809,7 +808,6 @@ mod tests {
storages,
snapshot_config.archive_format,
snapshot_config.snapshot_version,
None,
snapshot_config.maximum_snapshots_to_retain,
)?;
@ -857,9 +855,8 @@ mod tests {
const INCREMENTAL_SNAPSHOT_ARCHIVE_INTERVAL_SLOTS: Slot = BANK_SNAPSHOT_INTERVAL_SLOTS * 3;
const FULL_SNAPSHOT_ARCHIVE_INTERVAL_SLOTS: Slot =
INCREMENTAL_SNAPSHOT_ARCHIVE_INTERVAL_SLOTS * 5;
const LAST_SLOT: Slot = FULL_SNAPSHOT_ARCHIVE_INTERVAL_SLOTS * 3 - 1;
const EXPECTED_SLOT_FOR_LAST_SNAPSHOT_ARCHIVE: Slot =
LAST_SLOT + 1 - FULL_SNAPSHOT_ARCHIVE_INTERVAL_SLOTS;
const LAST_SLOT: Slot = FULL_SNAPSHOT_ARCHIVE_INTERVAL_SLOTS * 3
+ INCREMENTAL_SNAPSHOT_ARCHIVE_INTERVAL_SLOTS * 2;
info!("Running snapshots with background services test...");
trace!(
@ -949,6 +946,7 @@ mod tests {
false,
false,
true,
None,
);
let mint_keypair = &snapshot_test_config.genesis_config_info.mint_keypair;
@ -1019,9 +1017,14 @@ mod tests {
)
.unwrap();
assert_eq!(deserialized_bank.slot(), LAST_SLOT,);
assert_eq!(
deserialized_bank.slot(),
EXPECTED_SLOT_FOR_LAST_SNAPSHOT_ARCHIVE
deserialized_bank,
**bank_forks
.read()
.unwrap()
.get(deserialized_bank.slot())
.unwrap()
);
// Stop the background services