add ledger-tool verify verify-accounts-index option (#18375)

* add ledger-tool verify verify-accounts-index option

* comment, merge, respond to feedback, cleanup

core/tests/snapshots.rs

@@ -171,6 +171,7 @@ mod tests {
             None,
             accounts_db::AccountShrinkThreshold::default(),
             check_hash_calculation,
+            false,
         )
         .unwrap();
 

ledger-tool/src/main.rs

@@ -845,6 +845,10 @@ fn main() {
         .validator(is_slot)
         .takes_value(true)
         .help("Halt processing at the given slot");
+    let verify_index_arg = Arg::with_name("verify_accounts_index")
+        .long("verify-accounts-index")
+        .takes_value(false)
+        .help("For debugging and tests on accounts index.");
     let limit_load_slot_count_from_snapshot_arg = Arg::with_name("limit_load_slot_count_from_snapshot")
         .long("limit-load-slot-count-from-snapshot")
         .value_name("SLOT")
@@ -1121,6 +1125,7 @@ fn main() {
             .arg(&account_paths_arg)
             .arg(&halt_at_slot_arg)
             .arg(&limit_load_slot_count_from_snapshot_arg)
+            .arg(&verify_index_arg)
             .arg(&hard_forks_arg)
             .arg(&no_accounts_db_caching_arg)
             .arg(&accounts_db_test_hash_calculation_arg)
@@ -1845,6 +1850,7 @@ fn main() {
                     usize
                 )
                 .ok(),
+                verify_index: arg_matches.is_present("verify_accounts_index"),
                 allow_dead_slots: arg_matches.is_present("allow_dead_slots"),
                 accounts_db_test_hash_calculation: arg_matches
                     .is_present("accounts_db_test_hash_calculation"),

ledger/src/bank_forks_utils.rs

@@ -140,6 +140,7 @@ fn load_from_snapshot(
         process_options.limit_load_slot_count_from_snapshot,
         process_options.shrink_ratio,
         process_options.accounts_db_test_hash_calculation,
+        process_options.verify_index,
     )
     .expect("Load from snapshot failed");
     if let Some(shrink_paths) = shrink_paths {

ledger/src/blockstore_processor.rs

@@ -377,6 +377,7 @@ pub struct ProcessOptions {
     pub limit_load_slot_count_from_snapshot: Option<usize>,
     pub allow_dead_slots: bool,
     pub accounts_db_test_hash_calculation: bool,
+    pub verify_index: bool,
     pub shrink_ratio: AccountShrinkThreshold,
 }
 

runtime/src/accounts_db.rs

@@ -5912,69 +5912,104 @@ impl AccountsDb {
     }
 
     #[allow(clippy::needless_collect)]
-    pub fn generate_index(&self, limit_load_slot_count_from_snapshot: Option<usize>) {
+    pub fn generate_index(&self, limit_load_slot_count_from_snapshot: Option<usize>, verify: bool) {
         let mut slots = self.storage.all_slots();
         #[allow(clippy::stable_sort_primitive)]
         slots.sort();
         if let Some(limit) = limit_load_slot_count_from_snapshot {
             slots.truncate(limit); // get rid of the newer slots and keep just the older
         }
-        let total_processed_slots_across_all_threads = AtomicU64::new(0);
+        // pass == 0 always runs and generates the index
-        let outer_slots_len = slots.len();
+        // pass == 1 only runs if verify == true.
-        let chunk_size = (outer_slots_len / 7) + 1; // approximately 400k slots in a snapshot
+        // verify checks that all the expected items are in the accounts index and measures how long it takes to look them all up
-        let mut index_time = Measure::start("index");
+        let passes = if verify { 2 } else { 1 };
-        let insertion_time_us = AtomicU64::new(0);
+        for pass in 0..passes {
-        let scan_time: u64 = slots
+            let total_processed_slots_across_all_threads = AtomicU64::new(0);
-            .par_chunks(chunk_size)
+            let outer_slots_len = slots.len();
-            .map(|slots| {
+            let chunk_size = (outer_slots_len / 7) + 1; // approximately 400k slots in a snapshot
-                let mut log_status = MultiThreadProgress::new(
+            let mut index_time = Measure::start("index");
-                    &total_processed_slots_across_all_threads,
+            let insertion_time_us = AtomicU64::new(0);
-                    2,
+            let scan_time: u64 = slots
-                    outer_slots_len as u64,
+                .par_chunks(chunk_size)
-                );
+                .map(|slots| {
-                let mut scan_time_sum = 0;
+                    let mut log_status = MultiThreadProgress::new(
-                for (index, slot) in slots.iter().enumerate() {
+                        &total_processed_slots_across_all_threads,
-                    let mut scan_time = Measure::start("scan");
+                        2,
-                    log_status.report(index as u64);
+                        outer_slots_len as u64,
-                    let storage_maps: Vec<Arc<AccountStorageEntry>> = self
+                    );
-                        .storage
+                    let mut scan_time_sum = 0;
-                        .get_slot_storage_entries(*slot)
+                    for (index, slot) in slots.iter().enumerate() {
-                        .unwrap_or_default();
+                        let mut scan_time = Measure::start("scan");
-                    let accounts_map = Self::process_storage_slot(&storage_maps);
+                        log_status.report(index as u64);
-                    scan_time.stop();
+                        let storage_maps: Vec<Arc<AccountStorageEntry>> = self
-                    scan_time_sum += scan_time.as_us();
+                            .storage
+                            .get_slot_storage_entries(*slot)
+                            .unwrap_or_default();
+                        let accounts_map = Self::process_storage_slot(&storage_maps);
+                        scan_time.stop();
+                        scan_time_sum += scan_time.as_us();
 
-                    let insert_us = self.generate_index_for_slot(accounts_map, slot);
+                        let insert_us = if pass == 0 {
-                    insertion_time_us.fetch_add(insert_us, Ordering::Relaxed);
+                            // generate index
-                }
+                            self.generate_index_for_slot(accounts_map, slot)
-                scan_time_sum
+                        } else {
-            })
+                            // verify index matches expected and measure the time to get all items
-            .sum();
+                            assert!(verify);
-        index_time.stop();
+                            let mut lookup_time = Measure::start("lookup_time");
+                            for account in accounts_map.into_iter() {
+                                let (key, account_info) = account;
+                                let lock = self.accounts_index.get_account_maps_read_lock(&key);
+                                let x = lock.get(&key).unwrap();
+                                let sl = x.slot_list.read().unwrap();
+                                let mut count = 0;
+                                for (slot2, account_info2) in sl.iter() {
+                                    if slot2 == slot {
+                                        count += 1;
+                                        let ai = AccountInfo {
+                                            store_id: account_info.1,
+                                            offset: account_info.2.offset,
+                                            stored_size: account_info.2.stored_size,
+                                            lamports: account_info.2.account_meta.lamports,
+                                        };
+                                        assert_eq!(&ai, account_info2);
+                                    }
+                                }
+                                assert_eq!(1, count);
+                            }
+                            lookup_time.stop();
+                            lookup_time.as_us()
+                        };
+                        insertion_time_us.fetch_add(insert_us, Ordering::Relaxed);
+                    }
+                    scan_time_sum
+                })
+                .sum();
+            index_time.stop();
 
-        let mut min_bin_size = usize::MAX;
+            let mut min_bin_size = usize::MAX;
-        let mut max_bin_size = usize::MIN;
+            let mut max_bin_size = usize::MIN;
-        let total_items = self
+            let total_items = self
-            .accounts_index
+                .accounts_index
-            .account_maps
+                .account_maps
-            .iter()
+                .iter()
-            .map(|i| {
+                .map(|i| {
-                let len = i.read().unwrap().len();
+                    let len = i.read().unwrap().len();
-                min_bin_size = std::cmp::min(min_bin_size, len);
+                    min_bin_size = std::cmp::min(min_bin_size, len);
-                max_bin_size = std::cmp::max(max_bin_size, len);
+                    max_bin_size = std::cmp::max(max_bin_size, len);
-                len
+                    len
-            })
+                })
-            .sum();
+                .sum();
 
-        let timings = GenerateIndexTimings {
+            let timings = GenerateIndexTimings {
-            scan_time,
+                scan_time,
-            index_time: index_time.as_us(),
+                index_time: index_time.as_us(),
-            insertion_time_us: insertion_time_us.load(Ordering::Relaxed),
+                insertion_time_us: insertion_time_us.load(Ordering::Relaxed),
-            min_bin_size,
+                min_bin_size,
-            max_bin_size,
+                max_bin_size,
-            total_items,
+                total_items,
-        };
+            };
-        timings.report();
+            timings.report();
+        }
 
         // Need to add these last, otherwise older updates will be cleaned
         for slot in slots {

runtime/src/serde_snapshot.rs

@@ -140,6 +140,7 @@ pub(crate) fn bank_from_stream<R>(
     caching_enabled: bool,
     limit_load_slot_count_from_snapshot: Option<usize>,
     shrink_ratio: AccountShrinkThreshold,
+    verify_index: bool,
 ) -> std::result::Result<Bank, Error>
 where
     R: Read,
@@ -161,6 +162,7 @@ where
                 caching_enabled,
                 limit_load_slot_count_from_snapshot,
                 shrink_ratio,
+                verify_index,
             )?;
             Ok(bank)
         }};
@@ -252,6 +254,7 @@ fn reconstruct_bank_from_fields<E>(
     caching_enabled: bool,
     limit_load_slot_count_from_snapshot: Option<usize>,
     shrink_ratio: AccountShrinkThreshold,
+    verify_index: bool,
 ) -> Result<Bank, Error>
 where
     E: SerializableStorage + std::marker::Sync,
@@ -265,6 +268,7 @@ where
         caching_enabled,
         limit_load_slot_count_from_snapshot,
         shrink_ratio,
+        verify_index,
     )?;
     accounts_db.freeze_accounts(
         &Ancestors::from(&bank_fields.ancestors),
@@ -314,6 +318,7 @@ fn reconstruct_accountsdb_from_fields<E>(
     caching_enabled: bool,
     limit_load_slot_count_from_snapshot: Option<usize>,
     shrink_ratio: AccountShrinkThreshold,
+    verify_index: bool,
 ) -> Result<AccountsDb, Error>
 where
     E: SerializableStorage + std::marker::Sync,
@@ -396,6 +401,6 @@ where
     accounts_db
         .write_version
         .fetch_add(version, Ordering::Relaxed);
-    accounts_db.generate_index(limit_load_slot_count_from_snapshot);
+    accounts_db.generate_index(limit_load_slot_count_from_snapshot, verify_index);
     Ok(accounts_db)
 }

runtime/src/serde_snapshot/tests.rs

@@ -75,6 +75,7 @@ where
         false,
         None,
         AccountShrinkThreshold::default(),
+        false,
     )
 }
 
@@ -231,6 +232,7 @@ fn test_bank_serialize_style(serde_style: SerdeStyle) {
         false,
         None,
         AccountShrinkThreshold::default(),
+        false,
     )
     .unwrap();
     dbank.src = ref_sc;

runtime/src/snapshot_utils.rs

@@ -645,6 +645,7 @@ pub fn bank_from_snapshot_archive<P>(
     limit_load_slot_count_from_snapshot: Option<usize>,
     shrink_ratio: AccountShrinkThreshold,
     test_hash_calculation: bool,
+    verify_index: bool,
 ) -> Result<(Bank, BankFromArchiveTimings)>
 where
     P: AsRef<Path> + std::marker::Sync,
@@ -688,6 +689,7 @@ where
         accounts_db_caching_enabled,
         limit_load_slot_count_from_snapshot,
         shrink_ratio,
+        verify_index,
     )?;
     measure.stop();
 
@@ -945,6 +947,7 @@ fn rebuild_bank_from_snapshots(
     accounts_db_caching_enabled: bool,
     limit_load_slot_count_from_snapshot: Option<usize>,
     shrink_ratio: AccountShrinkThreshold,
+    verify_index: bool,
 ) -> Result<Bank> {
     let (snapshot_version_enum, root_paths) =
         verify_snapshot_version_and_folder(snapshot_version, unpacked_snapshots_dir)?;
@@ -967,6 +970,7 @@ fn rebuild_bank_from_snapshots(
                 accounts_db_caching_enabled,
                 limit_load_slot_count_from_snapshot,
                 shrink_ratio,
+                verify_index,
             ),
         }?)
     })?;
@@ -1500,6 +1504,7 @@ mod tests {
             None,
             AccountShrinkThreshold::default(),
             false,
+            false,
         )
         .unwrap();
 
@@ -1588,6 +1593,7 @@ mod tests {
             None,
             AccountShrinkThreshold::default(),
             false,
+            false,
         )
         .unwrap();