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

Multi-version snapshot support (#9980)

Browse Source 
* Multi-version snapshot support

* rustfmt

* Remove CLI options and runtime support for selection output snapshot version.
Address some clippy complaints.

* Muzzle clippy type complexity warning.

Despite clippy's suggestion, it is not currently possible to create type aliases
for traits and so everything within the 'Box<...>' cannot be type aliased.

This then leaves creating full blown traits, and either implementing
said traits by closure (somehow) or moving the closures into new structs
implementing said traits which seems a bit of a palaver.

Alternatively it is possible to define and use the type alias 'type ResultBox<T> = Result<Box<T>>'
which does seems rather pointless and not a great reduction in complexity but is enough to keep clippy happy.

In the end I simply went with squelching the clippy warning.

* Remove now unused Serialize/Deserialize trait implementations for AccountStorageEntry and AppendVec

* refactor versioned de/serialisers

* rename serde_utils to serde_snapshot

* move call to accounts_db.generate_index() back down to context_accountsdb_from_stream()

* update version 1.1.1 to 1.2.0
remove nested use of serialize_bytes

* cleanups

* Add back measurement of account storage entry serialization.
Remove construction of Vec and HashMap temporaries during serialization.

* consolidate serialisation test cases into serde_snapshot.
clean up leakage of implementation details in serde_snapshot.

* move short term / legacy snapshot code into child module

* add serialize_iter_as_tuple

* preliminary integration of following commit

commit 6d58b73c47294bfb93465d5a83cd2175660b6e6d
Author: Ryo Onodera <ryoqun@gmail.com>
Date:   Wed May 20 14:02:02 2020 +0900

    Confine snapshot 1.1 relic to versioned codepath

* refactored serde_snapshot, rustfmt
legacy accounts_db format now "owns" both leading u64s, legacy bank_rc format has none

* reduce type complexity (clippy)
This commit is contained in:
Kristofer Peterson
2020-05-22 18:54:24 +01:00
committed by GitHub
parent 967320a091
commit b7a32f01c0
14 changed files with 1217 additions and 626 deletions
Download Patch File Download Diff File Expand all files Collapse all files

310
runtime/src/accounts_db.rs
View File

@@ -21,20 +21,13 @@
use crate::{
accounts_index::{AccountsIndex, Ancestors, SlotList, SlotSlice},
append_vec::{AppendVec, StoredAccount, StoredMeta},
bank::deserialize_from_snapshot,
};
use bincode::{deserialize_from, serialize_into};
use byteorder::{ByteOrder, LittleEndian};
use fs_extra::dir::CopyOptions;
use lazy_static::lazy_static;
use log::*;
use rand::{thread_rng, Rng};
use rayon::{prelude::*, ThreadPool};
use serde::{
de::{MapAccess, Visitor},
ser::{SerializeMap, Serializer},
Deserialize, Serialize,
};
use serde::{Deserialize, Serialize};
use solana_measure::measure::Measure;
use solana_rayon_threadlimit::get_thread_count;
use solana_sdk::{
@@ -45,8 +38,7 @@ use solana_sdk::{
};
use std::{
collections::{HashMap, HashSet},
fmt,
io::{BufReader, Cursor, Error as IOError, ErrorKind, Read, Result as IOResult},
io::{Error as IOError, Result as IOResult},
ops::RangeBounds,
path::{Path, PathBuf},
sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering},
@@ -100,35 +92,7 @@ pub type SnapshotStorage = Vec<Arc<AccountStorageEntry>>;
pub type SnapshotStorages = Vec<SnapshotStorage>;
// Each slot has a set of storage entries.
type SlotStores = HashMap<usize, Arc<AccountStorageEntry>>;
struct AccountStorageVisitor;
impl<'de> Visitor<'de> for AccountStorageVisitor {
type Value = AccountStorage;
fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result {
formatter.write_str("Expecting AccountStorage")
}
#[allow(clippy::mutex_atomic)]
fn visit_map<M>(self, mut access: M) -> Result<Self::Value, M::Error>
where
M: MapAccess<'de>,
{
let mut map = HashMap::new();
while let Some((slot, storage_entries)) = access.next_entry()? {
let storage_entries: Vec<AccountStorageEntry> = storage_entries;
let storage_slot_map = map.entry(slot).or_insert_with(HashMap::new);
for mut storage in storage_entries {
storage.slot = slot;
storage_slot_map.insert(storage.id, Arc::new(storage));
}
}
Ok(AccountStorage(map))
}
}
pub(crate) type SlotStores = HashMap<usize, Arc<AccountStorageEntry>>;
trait Versioned {
fn version(&self) -> u64;
@@ -146,32 +110,6 @@ impl Versioned for (u64, AccountInfo) {
}
}
struct AccountStorageSerialize<'a> {
account_storage_entries: &'a [SnapshotStorage],
}
impl<'a> Serialize for AccountStorageSerialize<'a> {
fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
where
S: Serializer,
{
let mut map = serializer.serialize_map(Some(self.account_storage_entries.len()))?;
let mut count = 0;
let mut serialize_account_storage_timer = Measure::start("serialize_account_storage_ms");
for storage_entries in self.account_storage_entries {
map.serialize_entry(&storage_entries.first().unwrap().slot, storage_entries)?;
count += storage_entries.len();
}
serialize_account_storage_timer.stop();
datapoint_info!(
"serialize_account_storage_ms",
("duration", serialize_account_storage_timer.as_ms(), i64),
("num_entries", count, i64),
);
map.end()
}
}
#[derive(Clone, Default, Debug)]
pub struct AccountStorage(pub HashMap<Slot, SlotStores>);
@@ -193,22 +131,19 @@ impl AccountStorage {
}
}
impl<'de> Deserialize<'de> for AccountStorage {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: serde::Deserializer<'de>,
{
deserializer.deserialize_map(AccountStorageVisitor)
}
}
#[derive(Debug, PartialEq, Copy, Clone, Deserialize, Serialize)]
#[derive(Debug, Eq, PartialEq, Copy, Clone, Deserialize, Serialize)]
pub enum AccountStorageStatus {
Available = 0,
Full = 1,
Candidate = 2,
}
impl Default for AccountStorageStatus {
fn default() -> Self {
Self::Available
}
}
#[derive(Debug)]
pub enum BankHashVerificationError {
MismatchedAccountHash,
@@ -217,15 +152,14 @@ pub enum BankHashVerificationError {
}
/// Persistent storage structure holding the accounts
#[derive(Debug, Serialize, Deserialize)]
#[derive(Debug)]
pub struct AccountStorageEntry {
id: AppendVecId,
pub(crate) id: AppendVecId,
#[serde(skip)]
slot: Slot,
pub(crate) slot: Slot,
/// storage holding the accounts
accounts: AppendVec,
pub(crate) accounts: AppendVec,
/// Keeps track of the number of accounts stored in a specific AppendVec.
/// This is periodically checked to reuse the stores that do not have
@@ -235,13 +169,24 @@ pub struct AccountStorageEntry {
count_and_status: RwLock<(usize, AccountStorageStatus)>,
}
impl Default for AccountStorageEntry {
fn default() -> Self {
Self {
id: 0,
slot: 0,
accounts: AppendVec::new_empty_map(0),
count_and_status: RwLock::new((0, AccountStorageStatus::Available)),
}
}
}
impl AccountStorageEntry {
pub fn new(path: &Path, slot: Slot, id: usize, file_size: u64) -> Self {
let tail = AppendVec::new_relative_path(slot, id);
let path = Path::new(path).join(&tail);
let accounts = AppendVec::new(&path, true, file_size as usize);
AccountStorageEntry {
Self {
id,
slot,
accounts,
@@ -249,6 +194,15 @@ impl AccountStorageEntry {
}
}
pub(crate) fn new_empty_map(id: AppendVecId, accounts_current_len: usize) -> Self {
Self {
id,
slot: 0,
accounts: AppendVec::new_empty_map(accounts_current_len),
count_and_status: RwLock::new((0, AccountStorageStatus::Available)),
}
}
pub fn set_status(&self, mut status: AccountStorageStatus) {
let mut count_and_status = self.count_and_status.write().unwrap();
@@ -365,55 +319,6 @@ pub fn get_temp_accounts_paths(count: u32) -> IOResult<(Vec<TempDir>, Vec<PathBu
Ok((temp_dirs, paths))
}
pub struct AccountsDBSerialize<'a, 'b> {
accounts_db: &'a AccountsDB,
slot: Slot,
account_storage_entries: &'b [SnapshotStorage],
}
impl<'a, 'b> AccountsDBSerialize<'a, 'b> {
pub fn new(
accounts_db: &'a AccountsDB,
slot: Slot,
account_storage_entries: &'b [SnapshotStorage],
) -> Self {
Self {
accounts_db,
slot,
account_storage_entries,
}
}
}
impl<'a, 'b> Serialize for AccountsDBSerialize<'a, 'b> {
fn serialize<S>(&self, serializer: S) -> std::result::Result<S::Ok, S::Error>
where
S: serde::ser::Serializer,
{
use serde::ser::Error;
let mut wr = Cursor::new(vec![]);
let version = self.accounts_db.write_version.load(Ordering::Relaxed);
let account_storage_serialize = AccountStorageSerialize {
account_storage_entries: self.account_storage_entries,
};
serialize_into(&mut wr, &account_storage_serialize).map_err(Error::custom)?;
serialize_into(&mut wr, &version).map_err(Error::custom)?;
let bank_hashes = self.accounts_db.bank_hashes.read().unwrap();
serialize_into(
&mut wr,
&(
self.slot,
&*bank_hashes
.get(&self.slot)
.unwrap_or_else(|| panic!("No bank_hashes entry for slot {}", self.slot)),
),
)
.map_err(Error::custom)?;
let len = wr.position() as usize;
serializer.serialize_bytes(&wr.into_inner()[..len])
}
}
#[derive(Clone, Default, Debug, Serialize, Deserialize, PartialEq)]
pub struct BankHashStats {
pub num_updated_accounts: u64,
@@ -473,10 +378,10 @@ pub struct AccountsDB {
pub next_id: AtomicUsize,
pub shrink_candidate_slots: Mutex<Vec<Slot>>,
write_version: AtomicU64,
pub(crate) write_version: AtomicU64,
/// Set of storage paths to pick from
paths: Vec<PathBuf>,
pub(crate) paths: Vec<PathBuf>,
/// Directory of paths this accounts_db needs to hold/remove
temp_paths: Option<Vec<TempDir>>,
@@ -592,102 +497,6 @@ impl AccountsDB {
}
}
pub fn accounts_from_stream<R: Read, P: AsRef<Path>>(
&self,
mut stream: &mut BufReader<R>,
stream_append_vecs_path: P,
) -> Result<(), IOError> {
let _len: usize =
deserialize_from(&mut stream).map_err(|e| AccountsDB::get_io_error(&e.to_string()))?;
let storage: AccountStorage = deserialize_from_snapshot(&mut stream)
.map_err(|e| AccountsDB::get_io_error(&e.to_string()))?;
// Remap the deserialized AppendVec paths to point to correct local paths
let new_storage_map: Result<HashMap<Slot, SlotStores>, IOError> = storage
.0
.into_iter()
.map(|(slot, mut slot_storage)| {
let mut new_slot_storage = HashMap::new();
for (id, storage_entry) in slot_storage.drain() {
let path_index = thread_rng().gen_range(0, self.paths.len());
let local_dir = &self.paths[path_index];
std::fs::create_dir_all(local_dir).expect("Create directory failed");
// Move the corresponding AppendVec from the snapshot into the directory pointed
// at by `local_dir`
let append_vec_relative_path =
AppendVec::new_relative_path(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());
if std::fs::rename(append_vec_abs_path.clone(), target).is_err() {
let mut copy_options = CopyOptions::new();
copy_options.overwrite = true;
let e = fs_extra::move_items(
&vec![&append_vec_abs_path],
&local_dir,
&copy_options,
)
.map_err(|e| {
AccountsDB::get_io_error(&format!(
"Unable to move {:?} to {:?}: {}",
append_vec_abs_path, local_dir, e
))
});
if e.is_err() {
info!("{:?}", e);
continue;
}
};
// Notify the AppendVec of the new file location
let local_path = local_dir.join(append_vec_relative_path);
let mut u_storage_entry = Arc::try_unwrap(storage_entry).unwrap();
u_storage_entry
.set_file(local_path)
.map_err(|e| AccountsDB::get_io_error(&e.to_string()))?;
new_slot_storage.insert(id, Arc::new(u_storage_entry));
}
Ok((slot, new_slot_storage))
})
.collect();
let new_storage_map = new_storage_map?;
let mut storage = AccountStorage(new_storage_map);
// discard any slots with no storage entries
// this can happen if a non-root slot was serialized
// but non-root stores should not be included in the snapshot
storage.0.retain(|_slot, stores| !stores.is_empty());
let version: u64 = deserialize_from(&mut stream)
.map_err(|_| AccountsDB::get_io_error("write version deserialize error"))?;
let (slot, bank_hash): (Slot, BankHashInfo) = deserialize_from(&mut stream)
.map_err(|_| AccountsDB::get_io_error("bank hashes deserialize error"))?;
self.bank_hashes.write().unwrap().insert(slot, bank_hash);
// Process deserialized data, set necessary fields in self
let max_id: usize = *storage
.0
.values()
.flat_map(HashMap::keys)
.max()
.expect("At least one storage entry must exist from deserializing stream");
{
let mut stores = self.storage.write().unwrap();
stores.0.extend(storage.0);
}
self.next_id.store(max_id + 1, Ordering::Relaxed);
self.write_version.fetch_add(version, Ordering::Relaxed);
self.generate_index();
Ok(())
}
fn new_storage_entry(&self, slot: Slot, path: &Path, size: u64) -> AccountStorageEntry {
AccountStorageEntry::new(
path,
@@ -1635,7 +1444,7 @@ impl AccountsDB {
let accounts_index = self.accounts_index.read().unwrap();
let storage = self.storage.read().unwrap();
let keys: Vec<_> = accounts_index.account_maps.keys().collect();
let mismatch_found = AtomicBool::new(false);
let mismatch_found = AtomicU64::new(0);
let hashes: Vec<_> = keys
.par_iter()
.filter_map(|pubkey| {
@@ -1652,9 +1461,7 @@ impl AccountsDB {
if check_hash {
let hash = Self::hash_stored_account(*slot, &account);
if hash != *account.hash {
mismatch_found.store(true, Ordering::Relaxed);
}
if mismatch_found.load(Ordering::Relaxed) {
mismatch_found.fetch_add(1, Ordering::Relaxed);
return None;
}
}
@@ -1669,7 +1476,11 @@ impl AccountsDB {
}
})
.collect();
if mismatch_found.load(Ordering::Relaxed) {
if mismatch_found.load(Ordering::Relaxed) > 0 {
warn!(
"{} mismatched account hash(es) found",
mismatch_found.load(Ordering::Relaxed)
);
return Err(MismatchedAccountHash);
}
@@ -2000,12 +1811,7 @@ impl AccountsDB {
}
}
fn get_io_error(error: &str) -> IOError {
warn!("AccountsDB error: {:?}", error);
IOError::new(ErrorKind::Other, error)
}
fn generate_index(&self) {
pub fn generate_index(&self) {
let storage = self.storage.read().unwrap();
let mut slots: Vec<Slot> = storage.0.keys().cloned().collect();
slots.sort();
@@ -2085,14 +1891,11 @@ impl AccountsDB {
pub mod tests {
// TODO: all the bank tests are bank specific, issue: 2194
use super::*;
use crate::accounts_index::RefCount;
use crate::append_vec::AccountMeta;
use crate::{accounts_index::RefCount, append_vec::AccountMeta};
use assert_matches::assert_matches;
use bincode::serialize_into;
use rand::{thread_rng, Rng};
use solana_sdk::{account::Account, hash::HASH_BYTES};
use std::{fs, str::FromStr};
use tempfile::TempDir;
fn linear_ancestors(end_slot: u64) -> Ancestors {
let mut ancestors: Ancestors = vec![(0, 0)].into_iter().collect();
@@ -2954,26 +2757,9 @@ pub mod tests {
}
fn reconstruct_accounts_db_via_serialization(accounts: &AccountsDB, slot: Slot) -> AccountsDB {
let mut writer = Cursor::new(vec![]);
let snapshot_storages = accounts.get_snapshot_storages(slot);
serialize_into(
&mut writer,
&AccountsDBSerialize::new(&accounts, slot, &snapshot_storages),
)
.unwrap();
let buf = writer.into_inner();
let mut reader = BufReader::new(&buf[..]);
let daccounts = AccountsDB::new(Vec::new());
let copied_accounts = TempDir::new().unwrap();
// Simulate obtaining a copy of the AppendVecs from a tarball
copy_append_vecs(&accounts, copied_accounts.path()).unwrap();
daccounts
.accounts_from_stream(&mut reader, copied_accounts.path())
.unwrap();
let daccounts =
crate::serde_snapshot::reconstruct_accounts_db_via_serialization(accounts, slot);
print_count_and_status("daccounts", &daccounts);
daccounts
}