Do periodic inbound cleaning for rooted slots (#8436) (#8582)

automerge

ledger/src/snapshot_utils.rs

@@ -337,7 +337,7 @@ pub fn add_snapshot<P: AsRef<Path>>(
     bank: &Bank,
     snapshot_storages: &[SnapshotStorage],
 ) -> Result<SlotSnapshotPaths> {
-    bank.purge_zero_lamport_accounts();
+    bank.clean_accounts();
     let slot = bank.slot();
     // snapshot_path/slot
     let slot_snapshot_dir = get_bank_snapshot_dir(snapshot_path, slot);

runtime/src/accounts_db.rs

@@ -614,18 +614,67 @@ impl AccountsDB {
         false
     }
 
-    // Purge zero lamport accounts for garbage collection purposes
+    // Reclaim older states of rooted non-zero lamport accounts as a general
+    // AccountsDB bloat mitigation and preprocess for better zero-lamport purging.
+    fn clean_old_rooted_accounts(&self, purges_in_root: Vec<Pubkey>) {
+        // This number isn't carefully chosen; just guessed randomly such that
+        // the hot loop will be the order of ~Xms.
+        const INDEX_CLEAN_BULK_COUNT: usize = 4096;
+
+        let mut measure = Measure::start("clean_old_root-ms");
+        let reclaim_vecs =
+            purges_in_root
+                .par_chunks(INDEX_CLEAN_BULK_COUNT)
+                .map(|pubkeys: &[Pubkey]| {
+                    let mut reclaims = Vec::new();
+                    let accounts_index = self.accounts_index.read().unwrap();
+                    for pubkey in pubkeys {
+                        accounts_index.clean_rooted_entries(&pubkey, &mut reclaims);
+                    }
+                    reclaims
+                });
+        let reclaims: Vec<_> = reclaim_vecs.flatten().collect();
+        measure.stop();
+        inc_new_counter_info!("clean-old-root-par-clean-ms", measure.as_ms() as usize);
+
+        let mut measure = Measure::start("clean_old_root-ms");
+        self.handle_reclaims(&reclaims);
+        measure.stop();
+        inc_new_counter_info!("clean-old-root-reclaim-ms", measure.as_ms() as usize);
+    }
+
+    fn clear_uncleaned_roots(&self) {
+        let mut accounts_index = self.accounts_index.write().unwrap();
+        accounts_index.uncleaned_roots.clear();
+        drop(accounts_index);
+    }
+
+    // Purge zero lamport accounts and older rooted account states as garbage
+    // collection
     // Only remove those accounts where the entire rooted history of the account
     // can be purged because there are no live append vecs in the ancestors
-    pub fn purge_zero_lamport_accounts(&self, ancestors: &HashMap<u64, usize>) {
+    pub fn clean_accounts(&self) {
         self.report_store_stats();
         let mut purges = HashMap::new();
+        let mut purges_in_root = Vec::new();
+        let no_ancestors = HashMap::new();
         let accounts_index = self.accounts_index.read().unwrap();
-        accounts_index.scan_accounts(ancestors, |pubkey, (account_info, slot)| {
-            if account_info.lamports == 0 && accounts_index.is_root(slot) {
+
+        accounts_index.scan_accounts(&no_ancestors, |pubkey, (account_info, slot)| {
+            if account_info.lamports == 0 {
                 purges.insert(*pubkey, accounts_index.would_purge(pubkey));
+            } else if accounts_index.uncleaned_roots.contains(&slot) {
+                purges_in_root.push(*pubkey);
             }
         });
+        drop(accounts_index);
+
+        if !purges_in_root.is_empty() {
+            self.clean_old_rooted_accounts(purges_in_root);
+        }
+        self.clear_uncleaned_roots();
+
+        let accounts_index = self.accounts_index.read().unwrap();
 
         // Calculate store counts as if everything was purged
         // Then purge if we can
@@ -701,9 +750,9 @@ impl AccountsDB {
         let mut dead_slots = self.remove_dead_accounts(reclaims);
         dead_accounts.stop();
 
-        let mut cleanup_dead_slots = Measure::start("reclaims::purge_slots");
-        self.cleanup_dead_slots(&mut dead_slots);
-        cleanup_dead_slots.stop();
+        let mut clean_dead_slots = Measure::start("reclaims::purge_slots");
+        self.clean_dead_slots(&mut dead_slots);
+        clean_dead_slots.stop();
 
         let mut purge_slots = Measure::start("reclaims::purge_slots");
         for slot in dead_slots {
@@ -1165,7 +1214,7 @@ impl AccountsDB {
         dead_slots
     }
 
-    fn cleanup_dead_slots(&self, dead_slots: &mut HashSet<Slot>) {
+    fn clean_dead_slots(&self, dead_slots: &mut HashSet<Slot>) {
         if self.dont_cleanup_dead_slots.load(Ordering::Relaxed) {
             return;
         }
@@ -1174,7 +1223,7 @@ impl AccountsDB {
             {
                 let mut index = self.accounts_index.write().unwrap();
                 for slot in dead_slots.iter() {
-                    index.cleanup_dead_slot(*slot);
+                    index.clean_dead_slot(*slot);
                 }
             }
             {
@@ -1802,7 +1851,7 @@ pub mod tests {
         //slot is still there, since gc is lazy
         assert!(accounts.storage.read().unwrap().0[&0].get(&id).is_some());
 
-        //store causes cleanup
+        //store causes clean
         accounts.store(1, &[(&pubkey, &account)]);
 
         //slot is gone
@@ -1813,6 +1862,149 @@ pub mod tests {
         assert_eq!(accounts.load_slow(&ancestors, &pubkey), Some((account, 1)));
     }
 
+    impl AccountsDB {
+        fn store_count_for_slot(&self, slot: Slot) -> usize {
+            let storage = self.storage.read().unwrap();
+
+            let slot_storage = storage.0.get(&slot);
+            if let Some(slot_storage) = slot_storage {
+                slot_storage.values().nth(0).unwrap().count()
+            } else {
+                0
+            }
+        }
+
+        fn uncleaned_root_count(&self) -> usize {
+            self.accounts_index.read().unwrap().uncleaned_roots.len()
+        }
+    }
+
+    #[test]
+    fn test_clean_old_with_normal_account() {
+        solana_logger::setup();
+
+        let accounts = AccountsDB::new(Vec::new());
+        let pubkey = Pubkey::new_rand();
+        let account = Account::new(1, 0, &Account::default().owner);
+        //store an account
+        accounts.store(0, &[(&pubkey, &account)]);
+        accounts.store(1, &[(&pubkey, &account)]);
+
+        // simulate slots are rooted after while
+        accounts.add_root(0);
+        accounts.add_root(1);
+
+        //even if rooted, old state isn't cleaned up
+        assert_eq!(accounts.store_count_for_slot(0), 1);
+        assert_eq!(accounts.store_count_for_slot(1), 1);
+
+        accounts.clean_accounts();
+
+        //now old state is cleaned up
+        assert_eq!(accounts.store_count_for_slot(0), 0);
+        assert_eq!(accounts.store_count_for_slot(1), 1);
+    }
+
+    #[test]
+    fn test_clean_old_with_zero_lamport_account() {
+        solana_logger::setup();
+
+        let accounts = AccountsDB::new(Vec::new());
+        let pubkey1 = Pubkey::new_rand();
+        let pubkey2 = Pubkey::new_rand();
+        let normal_account = Account::new(1, 0, &Account::default().owner);
+        let zero_account = Account::new(0, 0, &Account::default().owner);
+        //store an account
+        accounts.store(0, &[(&pubkey1, &normal_account)]);
+        accounts.store(1, &[(&pubkey1, &zero_account)]);
+        accounts.store(0, &[(&pubkey2, &normal_account)]);
+        accounts.store(1, &[(&pubkey2, &normal_account)]);
+
+        //simulate slots are rooted after while
+        accounts.add_root(0);
+        accounts.add_root(1);
+
+        //even if rooted, old state isn't cleaned up
+        assert_eq!(accounts.store_count_for_slot(0), 2);
+        assert_eq!(accounts.store_count_for_slot(1), 2);
+
+        accounts.clean_accounts();
+
+        //still old state behind zero-lamport account isn't cleaned up
+        assert_eq!(accounts.store_count_for_slot(0), 1);
+        assert_eq!(accounts.store_count_for_slot(1), 2);
+    }
+
+    #[test]
+    fn test_clean_old_with_both_normal_and_zero_lamport_accounts() {
+        solana_logger::setup();
+
+        let accounts = AccountsDB::new(Vec::new());
+        let pubkey1 = Pubkey::new_rand();
+        let pubkey2 = Pubkey::new_rand();
+        let normal_account = Account::new(1, 0, &Account::default().owner);
+        let zero_account = Account::new(0, 0, &Account::default().owner);
+        //store an account
+        accounts.store(0, &[(&pubkey1, &normal_account)]);
+        accounts.store(1, &[(&pubkey1, &zero_account)]);
+        accounts.store(0, &[(&pubkey2, &normal_account)]);
+        accounts.store(2, &[(&pubkey2, &normal_account)]);
+
+        //simulate slots are rooted after while
+        accounts.add_root(0);
+        accounts.add_root(1);
+        accounts.add_root(2);
+
+        //even if rooted, old state isn't cleaned up
+        assert_eq!(accounts.store_count_for_slot(0), 2);
+        assert_eq!(accounts.store_count_for_slot(1), 1);
+        assert_eq!(accounts.store_count_for_slot(2), 1);
+
+        accounts.clean_accounts();
+
+        //both zero lamport and normal accounts are cleaned up
+        assert_eq!(accounts.store_count_for_slot(0), 0);
+        assert_eq!(accounts.store_count_for_slot(1), 0);
+        assert_eq!(accounts.store_count_for_slot(2), 1);
+    }
+
+    #[test]
+    fn test_uncleaned_roots_with_account() {
+        solana_logger::setup();
+
+        let accounts = AccountsDB::new(Vec::new());
+        let pubkey = Pubkey::new_rand();
+        let account = Account::new(1, 0, &Account::default().owner);
+        //store an account
+        accounts.store(0, &[(&pubkey, &account)]);
+        assert_eq!(accounts.uncleaned_root_count(), 0);
+
+        // simulate slots are rooted after while
+        accounts.add_root(0);
+        assert_eq!(accounts.uncleaned_root_count(), 1);
+
+        //now uncleaned roots are cleaned up
+        accounts.clean_accounts();
+        assert_eq!(accounts.uncleaned_root_count(), 0);
+    }
+
+    #[test]
+    fn test_uncleaned_roots_with_no_account() {
+        solana_logger::setup();
+
+        let accounts = AccountsDB::new(Vec::new());
+
+        assert_eq!(accounts.uncleaned_root_count(), 0);
+
+        // simulate slots are rooted after while
+        accounts.add_root(0);
+        assert_eq!(accounts.uncleaned_root_count(), 1);
+
+        //now uncleaned roots are cleaned up
+        accounts.clean_accounts();
+        assert_eq!(accounts.uncleaned_root_count(), 0);
+    }
+
     fn print_accounts(label: &'static str, accounts: &AccountsDB) {
         print_index(label, accounts);
         print_count_and_status(label, accounts);
@@ -1966,12 +2158,6 @@ pub mod tests {
         daccounts
     }
 
-    fn purge_zero_lamport_accounts(accounts: &AccountsDB, slot: Slot) {
-        let ancestors = vec![(slot as Slot, 0)].into_iter().collect();
-        info!("ancestors: {:?}", ancestors);
-        accounts.purge_zero_lamport_accounts(&ancestors);
-    }
-
     fn assert_no_stores(accounts: &AccountsDB, slot: Slot) {
         let stores = accounts.storage.read().unwrap();
         info!("{:?}", stores.0.get(&slot));
@@ -2014,7 +2200,7 @@ pub mod tests {
         current_slot += 1;
         accounts.add_root(current_slot);
 
-        purge_zero_lamport_accounts(&accounts, current_slot);
+        accounts.clean_accounts();
 
         print_accounts("post_purge", &accounts);
 
@@ -2068,7 +2254,7 @@ pub mod tests {
         current_slot += 1;
         accounts.add_root(current_slot);
 
-        purge_zero_lamport_accounts(&accounts, current_slot);
+        accounts.clean_accounts();
 
         print_accounts("post_purge", &accounts);
 
@@ -2134,7 +2320,7 @@ pub mod tests {
 
         print_accounts("accounts", &accounts);
 
-        purge_zero_lamport_accounts(&accounts, current_slot);
+        accounts.clean_accounts();
 
         print_accounts("accounts_post_purge", &accounts);
         let accounts = reconstruct_accounts_db_via_serialization(&accounts, current_slot);
@@ -2203,7 +2389,7 @@ pub mod tests {
     fn test_accounts_purge_chained_purge_before_snapshot_restore() {
         solana_logger::setup();
         with_chained_zero_lamport_accounts(|accounts, current_slot| {
-            purge_zero_lamport_accounts(&accounts, current_slot);
+            accounts.clean_accounts();
             let accounts = reconstruct_accounts_db_via_serialization(&accounts, current_slot);
             (accounts, false)
         });
@@ -2217,7 +2403,7 @@ pub mod tests {
             accounts
                 .dont_cleanup_dead_slots
                 .store(true, Ordering::Relaxed);
-            purge_zero_lamport_accounts(&accounts, current_slot);
+            accounts.clean_accounts();
             (accounts, true)
         });
     }

runtime/src/accounts_index.rs

@@ -10,6 +10,7 @@ pub struct AccountsIndex<T> {
     pub account_maps: HashMap<Pubkey, RwLock<SlotList<T>>>,
 
     pub roots: HashSet<Slot>,
+    pub uncleaned_roots: HashSet<Slot>,
 }
 
 impl<T: Clone> AccountsIndex<T> {
@@ -53,7 +54,7 @@ impl<T: Clone> AccountsIndex<T> {
         let mut max = 0;
         let mut rv = None;
         for (i, (slot, _t)) in list.iter().rev().enumerate() {
-            if *slot >= max && (ancestors.get(slot).is_some() || self.is_root(*slot)) {
+            if *slot >= max && (ancestors.contains_key(slot) || self.is_root(*slot)) {
                 rv = Some((list.len() - 1) - i);
                 max = *slot;
             }
@@ -113,31 +114,47 @@ impl<T: Clone> AccountsIndex<T> {
         account_info: T,
         reclaims: &mut Vec<(Slot, T)>,
     ) -> Option<T> {
-        let roots = &self.roots;
         if let Some(lock) = self.account_maps.get(pubkey) {
             let mut slot_vec = lock.write().unwrap();
-            // filter out old entries
+            // filter out other dirty entries
             reclaims.extend(slot_vec.iter().filter(|(f, _)| *f == slot).cloned());
             slot_vec.retain(|(f, _)| *f != slot);
 
-            // add the new entry
             slot_vec.push((slot, account_info));
+            // now, do lazy clean
+            self.purge_older_root_entries(&mut slot_vec, reclaims);
 
-            let max_root = Self::get_max_root(roots, &slot_vec);
-
-            reclaims.extend(
-                slot_vec
-                    .iter()
-                    .filter(|(slot, _)| Self::can_purge(max_root, *slot))
-                    .cloned(),
-            );
-            slot_vec.retain(|(slot, _)| !Self::can_purge(max_root, *slot));
             None
         } else {
             Some(account_info)
         }
     }
 
+    fn purge_older_root_entries(
+        &self,
+        slot_vec: &mut Vec<(Slot, T)>,
+        reclaims: &mut Vec<(Slot, T)>,
+    ) {
+        let roots = &self.roots;
+
+        let max_root = Self::get_max_root(roots, &slot_vec);
+
+        reclaims.extend(
+            slot_vec
+                .iter()
+                .filter(|(slot, _)| Self::can_purge(max_root, *slot))
+                .cloned(),
+        );
+        slot_vec.retain(|(slot, _)| !Self::can_purge(max_root, *slot));
+    }
+
+    pub fn clean_rooted_entries(&self, pubkey: &Pubkey, reclaims: &mut Vec<(Slot, T)>) {
+        if let Some(lock) = self.account_maps.get(pubkey) {
+            let mut slot_vec = lock.write().unwrap();
+            self.purge_older_root_entries(&mut slot_vec, reclaims);
+        }
+    }
+
     pub fn add_index(&mut self, slot: Slot, pubkey: &Pubkey, account_info: T) {
         let entry = self
             .account_maps
@@ -156,11 +173,13 @@ impl<T: Clone> AccountsIndex<T> {
 
     pub fn add_root(&mut self, slot: Slot) {
         self.roots.insert(slot);
+        self.uncleaned_roots.insert(slot);
     }
     /// Remove the slot when the storage for the slot is freed
     /// Accounts no longer reference this slot.
-    pub fn cleanup_dead_slot(&mut self, slot: Slot) {
+    pub fn clean_dead_slot(&mut self, slot: Slot) {
         self.roots.remove(&slot);
+        self.uncleaned_roots.remove(&slot);
     }
 }
 
@@ -260,26 +279,36 @@ mod tests {
     }
 
     #[test]
-    fn test_cleanup_first() {
+    fn test_clean_first() {
         let mut index = AccountsIndex::<bool>::default();
         index.add_root(0);
         index.add_root(1);
-        index.cleanup_dead_slot(0);
+        index.clean_dead_slot(0);
         assert!(index.is_root(1));
         assert!(!index.is_root(0));
     }
 
     #[test]
-    fn test_cleanup_last() {
+    fn test_clean_last() {
         //this behavior might be undefined, clean up should only occur on older slots
         let mut index = AccountsIndex::<bool>::default();
         index.add_root(0);
         index.add_root(1);
-        index.cleanup_dead_slot(1);
+        index.clean_dead_slot(1);
         assert!(!index.is_root(1));
         assert!(index.is_root(0));
     }
 
+    #[test]
+    fn test_clean_and_unclean_slot() {
+        let mut index = AccountsIndex::<bool>::default();
+        assert_eq!(0, index.uncleaned_roots.len());
+        index.add_root(1);
+        assert_eq!(1, index.uncleaned_roots.len());
+        index.clean_dead_slot(1);
+        assert_eq!(0, index.uncleaned_roots.len());
+    }
+
     #[test]
     fn test_update_last_wins() {
         let key = Keypair::new();

runtime/src/bank.rs

@@ -1892,7 +1892,7 @@ impl Bank {
     /// A snapshot bank should be purged of 0 lamport accounts which are not part of the hash
     /// calculation and could shield other real accounts.
     pub fn verify_snapshot_bank(&self) -> bool {
-        self.purge_zero_lamport_accounts();
+        self.clean_accounts();
         self.rc
             .accounts
             .verify_bank_hash(self.slot(), &self.ancestors)
@@ -2086,11 +2086,8 @@ impl Bank {
         );
     }
 
-    pub fn purge_zero_lamport_accounts(&self) {
-        self.rc
-            .accounts
-            .accounts_db
-            .purge_zero_lamport_accounts(&self.ancestors);
+    pub fn clean_accounts(&self) {
+        self.rc.accounts.accounts_db.clean_accounts();
     }
 }
 
@@ -3130,7 +3127,7 @@ mod tests {
             bank = Arc::new(new_from_parent(&bank));
         }
 
-        bank.purge_zero_lamport_accounts();
+        bank.clean_accounts();
 
         let bank0 = Arc::new(new_from_parent(&bank));
         let blockhash = bank.last_blockhash();
@@ -3146,11 +3143,11 @@ mod tests {
 
         assert_eq!(bank0.get_account(&keypair.pubkey()).unwrap().lamports, 10);
         assert_eq!(bank1.get_account(&keypair.pubkey()), None);
-        bank0.purge_zero_lamport_accounts();
+        bank0.clean_accounts();
 
         assert_eq!(bank0.get_account(&keypair.pubkey()).unwrap().lamports, 10);
         assert_eq!(bank1.get_account(&keypair.pubkey()), None);
-        bank1.purge_zero_lamport_accounts();
+        bank1.clean_accounts();
 
         assert_eq!(bank0.get_account(&keypair.pubkey()).unwrap().lamports, 10);
         assert_eq!(bank1.get_account(&keypair.pubkey()), None);
@@ -3167,7 +3164,7 @@ mod tests {
         // keypair should have 0 tokens on both forks
         assert_eq!(bank0.get_account(&keypair.pubkey()), None);
         assert_eq!(bank1.get_account(&keypair.pubkey()), None);
-        bank1.purge_zero_lamport_accounts();
+        bank1.clean_accounts();
 
         assert!(bank1.verify_hash_internal_state());
     }