blockworks-foundation
/
solana-with-rpc-optimizations
mirror of https://github.com/blockworks-foundation/solana-with-rpc-optimizations.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Change BlockStore TryPrimaryThenSecondary to just Secondary (#23391)

This commit is contained in:
steviez 2022-04-29 20:05:39 -05:00 committed by GitHub
parent 5245eb4229
commit 428cf54c91
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
7 changed files with 239 additions and 192 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
ledger-tool/src/bigtable.rs
View File

@ -582,7 +582,7 @@ pub fn bigtable_process_command(ledger_path: &Path, matches: &ArgMatches<'_>) {
let force_reupload = arg_matches.is_present("force_reupload"); let force_reupload = arg_matches.is_present("force_reupload");
let blockstore = crate::open_blockstore( let blockstore = crate::open_blockstore(
&canonicalize_ledger_path(ledger_path), &canonicalize_ledger_path(ledger_path),
AccessType::TryPrimaryThenSecondary, AccessType::Secondary,
None, None,
); );
let config = solana_storage_bigtable::LedgerStorageConfig { let config = solana_storage_bigtable::LedgerStorageConfig {

142
ledger-tool/src/main.rs
View File

@ -763,7 +763,7 @@ fn load_bank_forks(
vec![blockstore.ledger_path().join("accounts")] vec![blockstore.ledger_path().join("accounts")]
} else { } else {
let non_primary_accounts_path = blockstore.ledger_path().join("accounts.ledger-tool"); let non_primary_accounts_path = blockstore.ledger_path().join("accounts.ledger-tool");
warn!( info!(
"Default accounts path is switched aligning with Blockstore's secondary access: {:?}", "Default accounts path is switched aligning with Blockstore's secondary access: {:?}",
non_primary_accounts_path non_primary_accounts_path
); );
@ -1714,11 +1714,7 @@ fn main() {
let allow_dead_slots = arg_matches.is_present("allow_dead_slots"); let allow_dead_slots = arg_matches.is_present("allow_dead_slots");
let only_rooted = arg_matches.is_present("only_rooted"); let only_rooted = arg_matches.is_present("only_rooted");
output_ledger( output_ledger(
open_blockstore( open_blockstore(&ledger_path, AccessType::Secondary, wal_recovery_mode),
&ledger_path,
AccessType::TryPrimaryThenSecondary,
wal_recovery_mode,
),
starting_slot, starting_slot,
ending_slot, ending_slot,
allow_dead_slots, allow_dead_slots,
@ -1732,9 +1728,8 @@ fn main() {
let starting_slot = value_t_or_exit!(arg_matches, "starting_slot", Slot); let starting_slot = value_t_or_exit!(arg_matches, "starting_slot", Slot);
let ending_slot = value_t_or_exit!(arg_matches, "ending_slot", Slot); let ending_slot = value_t_or_exit!(arg_matches, "ending_slot", Slot);
let target_db = PathBuf::from(value_t_or_exit!(arg_matches, "target_db", String)); let target_db = PathBuf::from(value_t_or_exit!(arg_matches, "target_db", String));
let source = let source = open_blockstore(&ledger_path, AccessType::Secondary, None);
open_blockstore(&ledger_path, AccessType::TryPrimaryThenSecondary, None); let target = open_blockstore(&target_db, AccessType::Primary, None);
let target = open_blockstore(&target_db, AccessType::PrimaryOnly, None);
for (slot, _meta) in source.slot_meta_iterator(starting_slot).unwrap() { for (slot, _meta) in source.slot_meta_iterator(starting_slot).unwrap() {
if slot > ending_slot { if slot > ending_slot {
break; break;
@ -1807,11 +1802,8 @@ fn main() {
..ProcessOptions::default() ..ProcessOptions::default()
}; };
let genesis_config = open_genesis_config_by(&ledger_path, arg_matches); let genesis_config = open_genesis_config_by(&ledger_path, arg_matches);
let blockstore = open_blockstore( let blockstore =
&ledger_path, open_blockstore(&ledger_path, AccessType::Secondary, wal_recovery_mode);
AccessType::TryPrimaryThenSecondary,
wal_recovery_mode,
);
match load_bank_forks( match load_bank_forks(
arg_matches, arg_matches,
&genesis_config, &genesis_config,
@ -1857,8 +1849,7 @@ fn main() {
} }
let starting_slot = value_t_or_exit!(arg_matches, "starting_slot", Slot); let starting_slot = value_t_or_exit!(arg_matches, "starting_slot", Slot);
let ending_slot = value_t!(arg_matches, "ending_slot", Slot).unwrap_or(Slot::MAX); let ending_slot = value_t!(arg_matches, "ending_slot", Slot).unwrap_or(Slot::MAX);
let ledger = let ledger = open_blockstore(&ledger_path, AccessType::Secondary, None);
open_blockstore(&ledger_path, AccessType::TryPrimaryThenSecondary, None);
for (slot, _meta) in ledger for (slot, _meta) in ledger
.slot_meta_iterator(starting_slot) .slot_meta_iterator(starting_slot)
.unwrap() .unwrap()
@ -1892,11 +1883,8 @@ fn main() {
..ProcessOptions::default() ..ProcessOptions::default()
}; };
let genesis_config = open_genesis_config_by(&ledger_path, arg_matches); let genesis_config = open_genesis_config_by(&ledger_path, arg_matches);
let blockstore = open_blockstore( let blockstore =
&ledger_path, open_blockstore(&ledger_path, AccessType::Secondary, wal_recovery_mode);
AccessType::TryPrimaryThenSecondary,
wal_recovery_mode,
);
match load_bank_forks( match load_bank_forks(
arg_matches, arg_matches,
&genesis_config, &genesis_config,
@ -1916,11 +1904,8 @@ fn main() {
("slot", Some(arg_matches)) => { ("slot", Some(arg_matches)) => {
let slots = values_t_or_exit!(arg_matches, "slots", Slot); let slots = values_t_or_exit!(arg_matches, "slots", Slot);
let allow_dead_slots = arg_matches.is_present("allow_dead_slots"); let allow_dead_slots = arg_matches.is_present("allow_dead_slots");
let blockstore = open_blockstore( let blockstore =
&ledger_path, open_blockstore(&ledger_path, AccessType::Secondary, wal_recovery_mode);
AccessType::TryPrimaryThenSecondary,
wal_recovery_mode,
);
for slot in slots { for slot in slots {
println!("Slot {}", slot); println!("Slot {}", slot);
if let Err(err) = output_slot( if let Err(err) = output_slot(
@ -1938,11 +1923,7 @@ fn main() {
let starting_slot = value_t_or_exit!(arg_matches, "starting_slot", Slot); let starting_slot = value_t_or_exit!(arg_matches, "starting_slot", Slot);
let allow_dead_slots = arg_matches.is_present("allow_dead_slots"); let allow_dead_slots = arg_matches.is_present("allow_dead_slots");
output_ledger( output_ledger(
open_blockstore( open_blockstore(&ledger_path, AccessType::Secondary, wal_recovery_mode),
&ledger_path,
AccessType::TryPrimaryThenSecondary,
wal_recovery_mode,
),
starting_slot, starting_slot,
Slot::MAX, Slot::MAX,
allow_dead_slots, allow_dead_slots,
@ -1953,22 +1934,16 @@ fn main() {
); );
} }
("dead-slots", Some(arg_matches)) => { ("dead-slots", Some(arg_matches)) => {
let blockstore = open_blockstore( let blockstore =
&ledger_path, open_blockstore(&ledger_path, AccessType::Secondary, wal_recovery_mode);
AccessType::TryPrimaryThenSecondary,
wal_recovery_mode,
);
let starting_slot = value_t_or_exit!(arg_matches, "starting_slot", Slot); let starting_slot = value_t_or_exit!(arg_matches, "starting_slot", Slot);
for slot in blockstore.dead_slots_iterator(starting_slot).unwrap() { for slot in blockstore.dead_slots_iterator(starting_slot).unwrap() {
println!("{}", slot); println!("{}", slot);
} }
} }
("duplicate-slots", Some(arg_matches)) => { ("duplicate-slots", Some(arg_matches)) => {
let blockstore = open_blockstore( let blockstore =
&ledger_path, open_blockstore(&ledger_path, AccessType::Secondary, wal_recovery_mode);
AccessType::TryPrimaryThenSecondary,
wal_recovery_mode,
);
let starting_slot = value_t_or_exit!(arg_matches, "starting_slot", Slot); let starting_slot = value_t_or_exit!(arg_matches, "starting_slot", Slot);
for slot in blockstore.duplicate_slots_iterator(starting_slot).unwrap() { for slot in blockstore.duplicate_slots_iterator(starting_slot).unwrap() {
println!("{}", slot); println!("{}", slot);
@ -1977,7 +1952,7 @@ fn main() {
("set-dead-slot", Some(arg_matches)) => { ("set-dead-slot", Some(arg_matches)) => {
let slots = values_t_or_exit!(arg_matches, "slots", Slot); let slots = values_t_or_exit!(arg_matches, "slots", Slot);
let blockstore = let blockstore =
open_blockstore(&ledger_path, AccessType::PrimaryOnly, wal_recovery_mode); open_blockstore(&ledger_path, AccessType::Primary, wal_recovery_mode);
for slot in slots { for slot in slots {
match blockstore.set_dead_slot(slot) { match blockstore.set_dead_slot(slot) {
Ok(_) => println!("Slot {} dead", slot), Ok(_) => println!("Slot {} dead", slot),
@ -1988,7 +1963,7 @@ fn main() {
("remove-dead-slot", Some(arg_matches)) => { ("remove-dead-slot", Some(arg_matches)) => {
let slots = values_t_or_exit!(arg_matches, "slots", Slot); let slots = values_t_or_exit!(arg_matches, "slots", Slot);
let blockstore = let blockstore =
open_blockstore(&ledger_path, AccessType::PrimaryOnly, wal_recovery_mode); open_blockstore(&ledger_path, AccessType::Primary, wal_recovery_mode);
for slot in slots { for slot in slots {
match blockstore.remove_dead_slot(slot) { match blockstore.remove_dead_slot(slot) {
Ok(_) => println!("Slot {} not longer marked dead", slot), Ok(_) => println!("Slot {} not longer marked dead", slot),
@ -2001,11 +1976,8 @@ fn main() {
("parse_full_frozen", Some(arg_matches)) => { ("parse_full_frozen", Some(arg_matches)) => {
let starting_slot = value_t_or_exit!(arg_matches, "starting_slot", Slot); let starting_slot = value_t_or_exit!(arg_matches, "starting_slot", Slot);
let ending_slot = value_t_or_exit!(arg_matches, "ending_slot", Slot); let ending_slot = value_t_or_exit!(arg_matches, "ending_slot", Slot);
let blockstore = open_blockstore( let blockstore =
&ledger_path, open_blockstore(&ledger_path, AccessType::Secondary, wal_recovery_mode);
AccessType::TryPrimaryThenSecondary,
wal_recovery_mode,
);
let mut ancestors = BTreeSet::new(); let mut ancestors = BTreeSet::new();
assert!( assert!(
blockstore.meta(ending_slot).unwrap().is_some(), blockstore.meta(ending_slot).unwrap().is_some(),
@ -2144,11 +2116,8 @@ fn main() {
open_genesis_config_by(&ledger_path, arg_matches).hash() open_genesis_config_by(&ledger_path, arg_matches).hash()
); );
let blockstore = open_blockstore( let blockstore =
&ledger_path, open_blockstore(&ledger_path, AccessType::Secondary, wal_recovery_mode);
AccessType::TryPrimaryThenSecondary,
wal_recovery_mode,
);
let (bank_forks, ..) = load_bank_forks( let (bank_forks, ..) = load_bank_forks(
arg_matches, arg_matches,
&open_genesis_config_by(&ledger_path, arg_matches), &open_genesis_config_by(&ledger_path, arg_matches),
@ -2178,11 +2147,8 @@ fn main() {
..ProcessOptions::default() ..ProcessOptions::default()
}; };
let blockstore = open_blockstore( let blockstore =
&ledger_path, open_blockstore(&ledger_path, AccessType::Secondary, wal_recovery_mode);
AccessType::TryPrimaryThenSecondary,
wal_recovery_mode,
);
match load_bank_forks( match load_bank_forks(
arg_matches, arg_matches,
&open_genesis_config_by(&ledger_path, arg_matches), &open_genesis_config_by(&ledger_path, arg_matches),
@ -2271,11 +2237,8 @@ fn main() {
usize usize
); );
let genesis_config = open_genesis_config_by(&ledger_path, arg_matches); let genesis_config = open_genesis_config_by(&ledger_path, arg_matches);
let blockstore = open_blockstore( let blockstore =
&ledger_path, open_blockstore(&ledger_path, AccessType::Secondary, wal_recovery_mode);
AccessType::TryPrimaryThenSecondary,
wal_recovery_mode,
);
let is_incremental = arg_matches.is_present("incremental"); let is_incremental = arg_matches.is_present("incremental");
let snapshot_slot = if Some("ROOT") == arg_matches.value_of("snapshot_slot") { let snapshot_slot = if Some("ROOT") == arg_matches.value_of("snapshot_slot") {
@ -2605,11 +2568,8 @@ fn main() {
}; };
let genesis_config = open_genesis_config_by(&ledger_path, arg_matches); let genesis_config = open_genesis_config_by(&ledger_path, arg_matches);
let include_sysvars = arg_matches.is_present("include_sysvars"); let include_sysvars = arg_matches.is_present("include_sysvars");
let blockstore = open_blockstore( let blockstore =
&ledger_path, open_blockstore(&ledger_path, AccessType::Secondary, wal_recovery_mode);
AccessType::TryPrimaryThenSecondary,
wal_recovery_mode,
);
let (bank_forks, ..) = load_bank_forks( let (bank_forks, ..) = load_bank_forks(
arg_matches, arg_matches,
&genesis_config, &genesis_config,
@ -2667,11 +2627,8 @@ fn main() {
..ProcessOptions::default() ..ProcessOptions::default()
}; };
let genesis_config = open_genesis_config_by(&ledger_path, arg_matches); let genesis_config = open_genesis_config_by(&ledger_path, arg_matches);
let blockstore = open_blockstore( let blockstore =
&ledger_path, open_blockstore(&ledger_path, AccessType::Secondary, wal_recovery_mode);
AccessType::TryPrimaryThenSecondary,
wal_recovery_mode,
);
match load_bank_forks( match load_bank_forks(
arg_matches, arg_matches,
&genesis_config, &genesis_config,
@ -3195,9 +3152,9 @@ fn main() {
let dead_slots_only = arg_matches.is_present("dead_slots_only"); let dead_slots_only = arg_matches.is_present("dead_slots_only");
let batch_size = value_t_or_exit!(arg_matches, "batch_size", usize); let batch_size = value_t_or_exit!(arg_matches, "batch_size", usize);
let access_type = if !no_compaction { let access_type = if !no_compaction {
AccessType::PrimaryOnly AccessType::Primary
} else { } else {
AccessType::PrimaryOnlyForMaintenance AccessType::PrimaryForMaintenance
}; };
let blockstore = open_blockstore(&ledger_path, access_type, wal_recovery_mode); let blockstore = open_blockstore(&ledger_path, access_type, wal_recovery_mode);
@ -3270,11 +3227,8 @@ fn main() {
} }
} }
("list-roots", Some(arg_matches)) => { ("list-roots", Some(arg_matches)) => {
let blockstore = open_blockstore( let blockstore =
&ledger_path, open_blockstore(&ledger_path, AccessType::Secondary, wal_recovery_mode);
AccessType::TryPrimaryThenSecondary,
wal_recovery_mode,
);
let max_height = if let Some(height) = arg_matches.value_of("max_height") { let max_height = if let Some(height) = arg_matches.value_of("max_height") {
usize::from_str(height).expect("Maximum height must be a number") usize::from_str(height).expect("Maximum height must be a number")
} else { } else {
@ -3336,11 +3290,8 @@ fn main() {
}); });
} }
("repair-roots", Some(arg_matches)) => { ("repair-roots", Some(arg_matches)) => {
let blockstore = open_blockstore( let blockstore =
&ledger_path, open_blockstore(&ledger_path, AccessType::Primary, wal_recovery_mode);
AccessType::TryPrimaryThenSecondary,
wal_recovery_mode,
);
let start_root = if let Some(root) = arg_matches.value_of("start_root") { let start_root = if let Some(root) = arg_matches.value_of("start_root") {
Slot::from_str(root).expect("Before root must be a number") Slot::from_str(root).expect("Before root must be a number")
} else { } else {
@ -3388,11 +3339,8 @@ fn main() {
} }
} }
("bounds", Some(arg_matches)) => { ("bounds", Some(arg_matches)) => {
let blockstore = open_blockstore( let blockstore =
&ledger_path, open_blockstore(&ledger_path, AccessType::Secondary, wal_recovery_mode);
AccessType::TryPrimaryThenSecondary,
wal_recovery_mode,
);
match blockstore.slot_meta_iterator(0) { match blockstore.slot_meta_iterator(0) {
Ok(metas) => { Ok(metas) => {
let all = arg_matches.is_present("all"); let all = arg_matches.is_present("all");
@ -3453,21 +3401,13 @@ fn main() {
} }
("analyze-storage", _) => { ("analyze-storage", _) => {
analyze_storage( analyze_storage(
&open_blockstore( &open_blockstore(&ledger_path, AccessType::Secondary, wal_recovery_mode).db(),
&ledger_path,
AccessType::TryPrimaryThenSecondary,
wal_recovery_mode,
)
.db(),
); );
println!("Ok."); println!("Ok.");
} }
("compute-slot-cost", Some(arg_matches)) => { ("compute-slot-cost", Some(arg_matches)) => {
let blockstore = open_blockstore( let blockstore =
&ledger_path, open_blockstore(&ledger_path, AccessType::Secondary, wal_recovery_mode);
AccessType::TryPrimaryThenSecondary,
wal_recovery_mode,
);
let mut slots: Vec<u64> = vec![]; let mut slots: Vec<u64> = vec![];
if !arg_matches.is_present("slots") { if !arg_matches.is_present("slots") {

5
ledger/src/blockstore.rs
View File

@ -515,7 +515,7 @@ impl Blockstore {
/// compaction. /// compaction.
/// ///
/// To disable RocksDB's background compaction, open the Blockstore /// To disable RocksDB's background compaction, open the Blockstore
/// with AccessType::PrimaryOnlyForMaintenance. /// with AccessType::PrimaryForMaintenance.
pub fn set_no_compaction(&mut self, no_compaction: bool) { pub fn set_no_compaction(&mut self, no_compaction: bool) {
self.no_compaction = no_compaction; self.no_compaction = no_compaction;
} }
@ -3198,6 +3198,7 @@ impl Blockstore {
shred_code_cf.get_int_property(RocksProperties::TOTAL_SST_FILES_SIZE) shred_code_cf.get_int_property(RocksProperties::TOTAL_SST_FILES_SIZE)
} }
/// Returns whether the blockstore has primary (read and write) access
pub fn is_primary_access(&self) -> bool { pub fn is_primary_access(&self) -> bool {
self.db.is_primary_access() self.db.is_primary_access()
} }
@ -3796,7 +3797,7 @@ pub fn create_new_ledger(
let blockstore = Blockstore::open_with_options( let blockstore = Blockstore::open_with_options(
ledger_path, ledger_path,
BlockstoreOptions { BlockstoreOptions {
access_type: AccessType::PrimaryOnly, access_type: AccessType::Primary,
recovery_mode: None, recovery_mode: None,
enforce_ulimit_nofile: false, enforce_ulimit_nofile: false,
column_options: column_options.clone(), column_options: column_options.clone(),

106
ledger/src/blockstore_db.rs
View File

@ -440,15 +440,15 @@ pub mod columns {
// - Account for column in `analyze_storage()` in ledger-tool/src/main.rs // - Account for column in `analyze_storage()` in ledger-tool/src/main.rs
} }
#[derive(Clone, Debug, PartialEq)]
pub enum AccessType { pub enum AccessType {
PrimaryOnly, /// Primary (read/write) access; only one process can have Primary access.
PrimaryOnlyForMaintenance, // this indicates no compaction
TryPrimaryThenSecondary,
}
#[derive(Debug, PartialEq)]
pub enum ActualAccessType {
Primary, Primary,
/// Primary (read/write) access with RocksDB automatic compaction disabled.
PrimaryForMaintenance,
/// Secondary (read) access; multiple processes can have Secondary access.
/// Additionally, Secondary access can be obtained while another process
/// already has Primary access.
Secondary, Secondary,
} }
@ -514,73 +514,66 @@ impl OldestSlot {
#[derive(Debug)] #[derive(Debug)]
struct Rocks { struct Rocks {
db: rocksdb::DB, db: rocksdb::DB,
actual_access_type: ActualAccessType, access_type: AccessType,
oldest_slot: OldestSlot, oldest_slot: OldestSlot,
column_options: LedgerColumnOptions, column_options: LedgerColumnOptions,
} }
impl Rocks { impl Rocks {
fn open(path: &Path, options: BlockstoreOptions) -> Result<Rocks> { fn open(path: &Path, options: BlockstoreOptions) -> Result<Rocks> {
let access_type = &options.access_type; let access_type = options.access_type.clone();
let recovery_mode = options.recovery_mode.clone(); let recovery_mode = options.recovery_mode.clone();
fs::create_dir_all(&path)?; fs::create_dir_all(&path)?;
// Use default database options // Use default database options
if should_disable_auto_compactions(access_type) { if should_disable_auto_compactions(&access_type) {
warn!("Disabling rocksdb's auto compaction for maintenance bulk ledger update..."); info!("Disabling rocksdb's automatic compactions...");
} }
let mut db_options = get_db_options(access_type); let mut db_options = get_db_options(&access_type);
if let Some(recovery_mode) = recovery_mode { if let Some(recovery_mode) = recovery_mode {
db_options.set_wal_recovery_mode(recovery_mode.into()); db_options.set_wal_recovery_mode(recovery_mode.into());
} }
let oldest_slot = OldestSlot::default(); let oldest_slot = OldestSlot::default();
let cf_descriptors = Self::cf_descriptors(&options, &oldest_slot);
let cf_names = Self::columns();
let column_options = options.column_options.clone(); let column_options = options.column_options.clone();
// Open the database // Open the database
let db = match access_type { let db = match access_type {
AccessType::PrimaryOnly | AccessType::PrimaryOnlyForMaintenance => Rocks { AccessType::Primary | AccessType::PrimaryForMaintenance => Rocks {
db: DB::open_cf_descriptors(&db_options, path, cf_descriptors)?, db: DB::open_cf_descriptors(
actual_access_type: ActualAccessType::Primary, &db_options,
path,
Self::cf_descriptors(&options, &oldest_slot),
)?,
access_type: access_type.clone(),
oldest_slot, oldest_slot,
column_options, column_options,
}, },
AccessType::TryPrimaryThenSecondary => { AccessType::Secondary => {
match DB::open_cf_descriptors(&db_options, path, cf_descriptors) { let secondary_path = path.join("solana-secondary");
Ok(db) => Rocks {
db,
actual_access_type: ActualAccessType::Primary,
oldest_slot,
column_options,
},
Err(err) => {
let secondary_path = path.join("solana-secondary");
warn!("Error when opening as primary: {}", err); info!(
warn!("Trying as secondary at : {:?}", secondary_path); "Opening Rocks with secondary (read only) access at: {:?}",
warn!("This active secondary db use may temporarily cause the performance of another db use (like by validator) to degrade"); secondary_path
);
info!("This secondary access could temporarily degrade other accesses, such as by solana-validator");
Rocks { Rocks {
db: DB::open_cf_as_secondary( db: DB::open_cf_descriptors_as_secondary(
&db_options, &db_options,
path, path,
&secondary_path, &secondary_path,
cf_names.clone(), Self::cf_descriptors(&options, &oldest_slot),
)?, )?,
actual_access_type: ActualAccessType::Secondary, access_type: access_type.clone(),
oldest_slot, oldest_slot,
column_options, column_options,
}
}
} }
} }
}; };
// this is only needed for LedgerCleanupService. so guard with PrimaryOnly (i.e. running solana-validator) // This is only needed by solana-validator for LedgerCleanupService so guard with AccessType::Primary
if matches!(access_type, AccessType::PrimaryOnly) { if matches!(access_type, AccessType::Primary) {
for cf_name in cf_names { for cf_name in Self::columns() {
// these special column families must be excluded from LedgerCleanupService's rocksdb // these special column families must be excluded from LedgerCleanupService's rocksdb
// compactions // compactions
if should_exclude_from_compaction(cf_name) { if should_exclude_from_compaction(cf_name) {
@ -767,7 +760,8 @@ impl Rocks {
} }
fn is_primary_access(&self) -> bool { fn is_primary_access(&self) -> bool {
self.actual_access_type == ActualAccessType::Primary self.access_type == AccessType::Primary
|| self.access_type == AccessType::PrimaryForMaintenance
} }
/// Retrieves the specified RocksDB integer property of the current /// Retrieves the specified RocksDB integer property of the current
@ -2045,7 +2039,7 @@ impl LedgerColumnOptions {
} }
pub struct BlockstoreOptions { pub struct BlockstoreOptions {
// The access type of blockstore. Default: PrimaryOnly // The access type of blockstore. Default: Primary
pub access_type: AccessType, pub access_type: AccessType,
// Whether to open a blockstore under a recovery mode. Default: None. // Whether to open a blockstore under a recovery mode. Default: None.
pub recovery_mode: Option<BlockstoreRecoveryMode>, pub recovery_mode: Option<BlockstoreRecoveryMode>,
@ -2058,7 +2052,7 @@ impl Default for BlockstoreOptions {
/// The default options are the values used by [`Blockstore::open`]. /// The default options are the values used by [`Blockstore::open`].
fn default() -> Self { fn default() -> Self {
Self { Self {
access_type: AccessType::PrimaryOnly, access_type: AccessType::Primary,
recovery_mode: None, recovery_mode: None,
enforce_ulimit_nofile: true, enforce_ulimit_nofile: true,
column_options: LedgerColumnOptions::default(), column_options: LedgerColumnOptions::default(),
@ -2989,8 +2983,9 @@ fn get_db_options(access_type: &AccessType) -> Options {
// Returns whether automatic compactions should be disabled based upon access type // Returns whether automatic compactions should be disabled based upon access type
fn should_disable_auto_compactions(access_type: &AccessType) -> bool { fn should_disable_auto_compactions(access_type: &AccessType) -> bool {
// Disable automatic compactions in maintenance mode to prevent accidental cleaning // Leave automatic compactions enabled (do not disable) in Primary mode;
matches!(access_type, AccessType::PrimaryOnlyForMaintenance) // disable in all other modes to prevent accidental cleaning
!matches!(access_type, AccessType::Primary)
} }
// Returns whether the supplied column (name) should be excluded from compaction // Returns whether the supplied column (name) should be excluded from compaction
@ -3077,6 +3072,15 @@ pub mod tests {
); );
} }
#[test]
fn test_should_disable_auto_compactions() {
assert!(!should_disable_auto_compactions(&AccessType::Primary));
assert!(should_disable_auto_compactions(
&AccessType::PrimaryForMaintenance
));
assert!(should_disable_auto_compactions(&AccessType::Secondary));
}
#[test] #[test]
fn test_should_exclude_from_compaction() { fn test_should_exclude_from_compaction() {
// currently there are three CFs excluded from compaction: // currently there are three CFs excluded from compaction:

139
ledger/src/blockstore_processor.rs
View File

@ -659,9 +659,9 @@ pub fn process_blockstore_from_root(
.set_roots(std::iter::once(&start_slot)) .set_roots(std::iter::once(&start_slot))
.expect("Couldn't set root slot on startup"); .expect("Couldn't set root slot on startup");
} else { } else {
assert!( info!(
blockstore.is_root(start_slot), "Starting slot {} isn't root and won't be updated due to being secondary blockstore access",
"starting slot isn't root and can't update due to being secondary blockstore access: {}", start_slot start_slot
); );
} }
@ -1083,7 +1083,9 @@ fn process_bank_0(
) )
.expect("processing for bank 0 must succeed"); .expect("processing for bank 0 must succeed");
bank0.freeze(); bank0.freeze();
blockstore.insert_bank_hash(bank0.slot(), bank0.hash(), false); if blockstore.is_primary_access() {
blockstore.insert_bank_hash(bank0.slot(), bank0.hash(), false);
}
cache_block_meta(bank0, cache_block_meta_sender); cache_block_meta(bank0, cache_block_meta_sender);
} }
@ -1255,8 +1257,12 @@ fn load_frozen_forks(
new_root_bank = new_root_bank.parent().unwrap(); new_root_bank = new_root_bank.parent().unwrap();
} }
inc_new_counter_info!("load_frozen_forks-cluster-confirmed-root", rooted_slots.len()); inc_new_counter_info!("load_frozen_forks-cluster-confirmed-root", rooted_slots.len());
blockstore.set_roots(rooted_slots.iter().map(|(slot, _hash)| slot)).expect("Blockstore::set_roots should succeed"); if blockstore.is_primary_access() {
blockstore.set_duplicate_confirmed_slots_and_hashes(rooted_slots.into_iter()).expect("Blockstore::set_duplicate_confirmed should succeed"); blockstore.set_roots(rooted_slots.iter().map(|(slot, _hash)| slot))
.expect("Blockstore::set_roots should succeed");
blockstore.set_duplicate_confirmed_slots_and_hashes(rooted_slots.into_iter())
.expect("Blockstore::set_duplicate_confirmed should succeed");
}
Some(cluster_root_bank) Some(cluster_root_bank)
} else { } else {
None None
@ -1380,7 +1386,17 @@ fn process_single_slot(
) -> result::Result<(), BlockstoreProcessorError> { ) -> result::Result<(), BlockstoreProcessorError> {
// Mark corrupt slots as dead so validators don't replay this slot and // Mark corrupt slots as dead so validators don't replay this slot and
// see AlreadyProcessed errors later in ReplayStage // see AlreadyProcessed errors later in ReplayStage
confirm_full_slot(blockstore, bank, opts, recyclers, progress, transaction_status_sender, replay_vote_sender, timing).map_err(|err| { confirm_full_slot(
blockstore,
bank,
opts,
recyclers,
progress,
transaction_status_sender,
replay_vote_sender,
timing,
)
.map_err(|err| {
let slot = bank.slot(); let slot = bank.slot();
warn!("slot {} failed to verify: {}", slot, err); warn!("slot {} failed to verify: {}", slot, err);
if blockstore.is_primary_access() { if blockstore.is_primary_access() {
@ -1388,13 +1404,18 @@ fn process_single_slot(
.set_dead_slot(slot) .set_dead_slot(slot)
.expect("Failed to mark slot as dead in blockstore"); .expect("Failed to mark slot as dead in blockstore");
} else { } else {
assert!(blockstore.is_dead(slot), "Failed slot isn't dead and can't update due to being secondary blockstore access: {}", slot); info!(
"Failed slot {} won't be marked dead due to being secondary blockstore access",
slot
);
} }
err err
})?; })?;
bank.freeze(); // all banks handled by this routine are created from complete slots bank.freeze(); // all banks handled by this routine are created from complete slots
blockstore.insert_bank_hash(bank.slot(), bank.hash(), false); if blockstore.is_primary_access() {
blockstore.insert_bank_hash(bank.slot(), bank.hash(), false);
}
cache_block_meta(bank, cache_block_meta_sender); cache_block_meta(bank, cache_block_meta_sender);
Ok(()) Ok(())
@ -1538,8 +1559,11 @@ fn check_accounts_data_size<'a>(
pub mod tests { pub mod tests {
use { use {
super::*, super::*,
crate::genesis_utils::{ crate::{
create_genesis_config, create_genesis_config_with_leader, GenesisConfigInfo, blockstore_db::{AccessType, BlockstoreOptions},
genesis_utils::{
create_genesis_config, create_genesis_config_with_leader, GenesisConfigInfo,
},
}, },
matches::assert_matches, matches::assert_matches,
rand::{thread_rng, Rng}, rand::{thread_rng, Rng},
@ -1569,8 +1593,50 @@ pub mod tests {
trees::tr, trees::tr,
}; };
// Convenience wrapper to optionally process blockstore with Secondary access.
//
// Setting up the ledger for a test requires Primary access as items will need to be inserted.
// However, once a Secondary access has been opened, it won't automaticaly see updates made by
// the Primary access. So, open (and close) the Secondary access within this function to ensure
// that "stale" Secondary accesses don't propagate.
fn test_process_blockstore_with_custom_options(
genesis_config: &GenesisConfig,
blockstore: &Blockstore,
opts: ProcessOptions,
access_type: AccessType,
) -> (Arc<RwLock<BankForks>>, LeaderScheduleCache) {
match access_type {
AccessType::Primary | AccessType::PrimaryForMaintenance => {
// Attempting to open a second Primary access would fail, so
// just pass the original session if it is a Primary variant
test_process_blockstore(genesis_config, blockstore, opts)
}
AccessType::Secondary => {
let secondary_blockstore = Blockstore::open_with_options(
blockstore.ledger_path(),
BlockstoreOptions {
access_type,
..BlockstoreOptions::default()
},
)
.expect("Unable to open access to blockstore");
test_process_blockstore(genesis_config, &secondary_blockstore, opts)
}
}
}
#[test] #[test]
fn test_process_blockstore_with_missing_hashes() { fn test_process_blockstore_with_missing_hashes() {
do_test_process_blockstore_with_missing_hashes(AccessType::Primary);
}
#[test]
fn test_process_blockstore_with_missing_hashes_secondary_access() {
do_test_process_blockstore_with_missing_hashes(AccessType::Secondary);
}
// Intentionally make slot 1 faulty and ensure that processing sees it as dead
fn do_test_process_blockstore_with_missing_hashes(blockstore_access_type: AccessType) {
solana_logger::setup(); solana_logger::setup();
let hashes_per_tick = 2; let hashes_per_tick = 2;
@ -1601,15 +1667,28 @@ pub mod tests {
Ok(_) Ok(_)
); );
let (bank_forks, ..) = test_process_blockstore( let (bank_forks, ..) = test_process_blockstore_with_custom_options(
&genesis_config, &genesis_config,
&blockstore, &blockstore,
ProcessOptions { ProcessOptions {
poh_verify: true, poh_verify: true,
..ProcessOptions::default() ..ProcessOptions::default()
}, },
blockstore_access_type.clone(),
); );
assert_eq!(frozen_bank_slots(&bank_forks.read().unwrap()), vec![0]); assert_eq!(frozen_bank_slots(&bank_forks.read().unwrap()), vec![0]);
let dead_slots: Vec<Slot> = blockstore.dead_slots_iterator(0).unwrap().collect();
match blockstore_access_type {
// Secondary access is immutable so even though a dead slot
// will be identified, it won't actually be marked dead.
AccessType::Secondary => {
assert_eq!(dead_slots.len(), 0);
}
AccessType::Primary | AccessType::PrimaryForMaintenance => {
assert_eq!(&dead_slots, &[1]);
}
}
} }
#[test] #[test]
@ -3520,7 +3599,10 @@ pub mod tests {
.unwrap(); .unwrap();
} }
fn run_test_process_blockstore_with_supermajority_root(blockstore_root: Option<Slot>) { fn run_test_process_blockstore_with_supermajority_root(
blockstore_root: Option<Slot>,
blockstore_access_type: AccessType,
) {
solana_logger::setup(); solana_logger::setup();
/* /*
Build fork structure: Build fork structure:
@ -3587,7 +3669,13 @@ pub mod tests {
accounts_db_test_hash_calculation: true, accounts_db_test_hash_calculation: true,
..ProcessOptions::default() ..ProcessOptions::default()
}; };
let (bank_forks, ..) = test_process_blockstore(&genesis_config, &blockstore, opts.clone());
let (bank_forks, ..) = test_process_blockstore_with_custom_options(
&genesis_config,
&blockstore,
opts.clone(),
blockstore_access_type.clone(),
);
let bank_forks = bank_forks.read().unwrap(); let bank_forks = bank_forks.read().unwrap();
// prepare to add votes // prepare to add votes
@ -3619,7 +3707,12 @@ pub mod tests {
&leader_keypair, &leader_keypair,
); );
let (bank_forks, ..) = test_process_blockstore(&genesis_config, &blockstore, opts.clone()); let (bank_forks, ..) = test_process_blockstore_with_custom_options(
&genesis_config,
&blockstore,
opts.clone(),
blockstore_access_type.clone(),
);
let bank_forks = bank_forks.read().unwrap(); let bank_forks = bank_forks.read().unwrap();
assert_eq!(bank_forks.root(), expected_root_slot); assert_eq!(bank_forks.root(), expected_root_slot);
@ -3674,7 +3767,12 @@ pub mod tests {
&leader_keypair, &leader_keypair,
); );
let (bank_forks, ..) = test_process_blockstore(&genesis_config, &blockstore, opts); let (bank_forks, ..) = test_process_blockstore_with_custom_options(
&genesis_config,
&blockstore,
opts,
blockstore_access_type,
);
let bank_forks = bank_forks.read().unwrap(); let bank_forks = bank_forks.read().unwrap();
assert_eq!(bank_forks.root(), really_expected_root_slot); assert_eq!(bank_forks.root(), really_expected_root_slot);
@ -3682,12 +3780,17 @@ pub mod tests {
#[test] #[test]
fn test_process_blockstore_with_supermajority_root_without_blockstore_root() { fn test_process_blockstore_with_supermajority_root_without_blockstore_root() {
run_test_process_blockstore_with_supermajority_root(None); run_test_process_blockstore_with_supermajority_root(None, AccessType::Primary);
}
#[test]
fn test_process_blockstore_with_supermajority_root_without_blockstore_root_secondary_access() {
run_test_process_blockstore_with_supermajority_root(None, AccessType::Secondary);
} }
#[test] #[test]
fn test_process_blockstore_with_supermajority_root_with_blockstore_root() { fn test_process_blockstore_with_supermajority_root_with_blockstore_root() {
run_test_process_blockstore_with_supermajority_root(Some(1)) run_test_process_blockstore_with_supermajority_root(Some(1), AccessType::Primary)
} }
#[test] #[test]

19
local-cluster/tests/common.rs
View File

@ -73,14 +73,27 @@ pub fn open_blockstore(ledger_path: &Path) -> Blockstore {
Blockstore::open_with_options( Blockstore::open_with_options(
ledger_path, ledger_path,
BlockstoreOptions { BlockstoreOptions {
access_type: AccessType::TryPrimaryThenSecondary, access_type: AccessType::Primary,
recovery_mode: None, recovery_mode: None,
enforce_ulimit_nofile: true, enforce_ulimit_nofile: true,
..BlockstoreOptions::default() ..BlockstoreOptions::default()
}, },
) )
.unwrap_or_else(|e| { // Fall back on Secondary if Primary fails; Primary will fail if
panic!("Failed to open ledger at {:?}, err: {}", ledger_path, e); // a handle to Blockstore is being held somewhere else
.unwrap_or_else(|_| {
Blockstore::open_with_options(
ledger_path,
BlockstoreOptions {
access_type: AccessType::Secondary,
recovery_mode: None,
enforce_ulimit_nofile: true,
..BlockstoreOptions::default()
},
)
.unwrap_or_else(|e| {
panic!("Failed to open ledger at {:?}, err: {}", ledger_path, e);
})
}) })
} }

18
local-cluster/tests/local_cluster_flakey.rs
View File

@ -9,12 +9,7 @@ use {
log::*, log::*,
serial_test::serial, serial_test::serial,
solana_core::validator::ValidatorConfig, solana_core::validator::ValidatorConfig,
solana_ledger::{ solana_ledger::{ancestor_iterator::AncestorIterator, leader_schedule::FixedSchedule},
ancestor_iterator::AncestorIterator,
blockstore::Blockstore,
blockstore_db::{AccessType, BlockstoreOptions},
leader_schedule::FixedSchedule,
},
solana_local_cluster::{ solana_local_cluster::{
cluster::Cluster, cluster::Cluster,
local_cluster::{ClusterConfig, LocalCluster}, local_cluster::{ClusterConfig, LocalCluster},
@ -334,16 +329,7 @@ fn do_test_optimistic_confirmation_violation_with_or_without_tower(with_tower: b
if let Some((last_vote, _)) = last_vote_in_tower(&val_a_ledger_path, &validator_a_pubkey) { if let Some((last_vote, _)) = last_vote_in_tower(&val_a_ledger_path, &validator_a_pubkey) {
a_votes.push(last_vote); a_votes.push(last_vote);
let blockstore = Blockstore::open_with_options( let blockstore = open_blockstore(&val_a_ledger_path);
&val_a_ledger_path,
BlockstoreOptions {
access_type: AccessType::TryPrimaryThenSecondary,
recovery_mode: None,
enforce_ulimit_nofile: true,
..BlockstoreOptions::default()
},
)
.unwrap();
let mut ancestors = AncestorIterator::new(last_vote, &blockstore); let mut ancestors = AncestorIterator::new(last_vote, &blockstore);
if ancestors.any(|a| votes_on_c_fork.contains(&a)) { if ancestors.any(|a| votes_on_c_fork.contains(&a)) {
bad_vote_detected = true; bad_vote_detected = true;