Skip to content

Commit

Permalink
runtime: remove Default req on account scan interfaces (solana-labs…
Browse files Browse the repository at this point in the history 
…#28533)
  • Loading branch information
@t-nelson
t-nelson committed Oct 26, 2022
1 parent 6c16037 commit 3d40915
Show file tree
Hide file tree
Showing 2 changed files with 85 additions and 81 deletions.
47 changes: 29 additions & 18 deletions runtime/src/accounts.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -614,10 +614,11 @@ impl Accounts {
if num == 0 {
return Ok(vec![]);
}
let account_balances = self.accounts_db.scan_accounts(
let mut account_balances = BinaryHeap::new();
self.accounts_db.scan_accounts(
ancestors,
bank_id,
|collector: &mut BinaryHeap<Reverse<(u64, Pubkey)>>, option| {
|option| {
if let Some((pubkey, account, _slot)) = option {
if account.lamports() == 0 {
return;
Expand All @@ -630,16 +631,16 @@ impl Accounts {
if !collect {
return;
}
if collector.len() == num {
let Reverse(entry) = collector
if account_balances.len() == num {
let Reverse(entry) = account_balances
.peek()
.expect("BinaryHeap::peek should succeed when len > 0");
if *entry >= (account.lamports(), *pubkey) {
return;
}
collector.pop();
account_balances.pop();
}
collector.push(Reverse((account.lamports(), *pubkey)));
account_balances.push(Reverse((account.lamports(), *pubkey)));
}
},
)?;
Expand Down Expand Up @@ -717,15 +718,17 @@ impl Accounts {
bank_id: BankId,
program_id: &Pubkey,
) -> ScanResult<Vec<(Pubkey, AccountSharedData)>> {
let mut collector = Vec::new();
self.accounts_db.scan_accounts(
ancestors,
bank_id,
|collector: &mut Vec<(Pubkey, AccountSharedData)>, some_account_tuple| {
Self::load_while_filtering(collector, some_account_tuple, |account| {
|some_account_tuple| {
Self::load_while_filtering(&mut collector, some_account_tuple, |account| {
account.owner() == program_id
})
},
)
.map(|_| collector)
}

pub fn load_by_program_with_filter<F: Fn(&AccountSharedData) -> bool>(
Expand All @@ -735,15 +738,17 @@ impl Accounts {
program_id: &Pubkey,
filter: F,
) -> ScanResult<Vec<(Pubkey, AccountSharedData)>> {
let mut collector = Vec::new();
self.accounts_db.scan_accounts(
ancestors,
bank_id,
|collector: &mut Vec<(Pubkey, AccountSharedData)>, some_account_tuple| {
Self::load_while_filtering(collector, some_account_tuple, |account| {
|some_account_tuple| {
Self::load_while_filtering(&mut collector, some_account_tuple, |account| {
account.owner() == program_id && filter(account)
})
},
)
.map(|_| collector)
}

pub fn load_by_index_key_with_filter<F: Fn(&AccountSharedData) -> bool>(
Expand All @@ -753,18 +758,20 @@ impl Accounts {
index_key: &IndexKey,
filter: F,
) -> ScanResult<Vec<(Pubkey, AccountSharedData)>> {
self.accounts_db
let mut collector = Vec::new();
self
.accounts_db
.index_scan_accounts(
ancestors,
bank_id,
*index_key,
|collector: &mut Vec<(Pubkey, AccountSharedData)>, some_account_tuple| {
Self::load_while_filtering(collector, some_account_tuple, |account| {
|some_account_tuple| {
Self::load_while_filtering(&mut collector, some_account_tuple, |account| {
filter(account)
})
},
)
.map(|result| result.0)
.map(|_| collector)
}

pub fn account_indexes_include_key(&self, key: &Pubkey) -> bool {
Expand All @@ -776,32 +783,36 @@ impl Accounts {
ancestors: &Ancestors,
bank_id: BankId,
) -> ScanResult<Vec<(Pubkey, AccountSharedData, Slot)>> {
let mut collector = Vec::new();
self.accounts_db.scan_accounts(
ancestors,
bank_id,
|collector: &mut Vec<(Pubkey, AccountSharedData, Slot)>, some_account_tuple| {
|some_account_tuple| {
if let Some((pubkey, account, slot)) = some_account_tuple
.filter(|(_, account, _)| Self::is_loadable(account.lamports()))
{
collector.push((*pubkey, account, slot))
}
},
)
.map(|_| collector)
}

pub fn load_to_collect_rent_eagerly<R: RangeBounds<Pubkey>>(
&self,
ancestors: &Ancestors,
range: R,
) -> Vec<(Pubkey, AccountSharedData)> {
let mut collector = Vec::new();
self.accounts_db.range_scan_accounts(
"load_to_collect_rent_eagerly_scan_elapsed",
ancestors,
range,
|collector: &mut Vec<(Pubkey, AccountSharedData)>, option| {
Self::load_while_filtering(collector, option, |_| true)
|option| {
Self::load_while_filtering(&mut collector, option, |_| true)
},
)
);
collector
}

/// Slow because lock is held for 1 operation instead of many.
Expand Down
119 changes: 56 additions & 63 deletions runtime/src/accounts_db.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2506,42 +2506,36 @@ impl AccountsDb {
}
}

pub fn scan_accounts<F, A>(
pub fn scan_accounts<F>(
&self,
ancestors: &Ancestors,
bank_id: BankId,
scan_func: F,
) -> ScanResult<A>
mut scan_func: F,
) -> ScanResult<()>
where
F: Fn(&mut A, Option<(&Pubkey, AccountSharedData, Slot)>),
A: Default,
F: FnMut(Option<(&Pubkey, AccountSharedData, Slot)>),
{
let mut collector = A::default();

// This can error out if the slots being scanned over are aborted
self.accounts_index
.scan_accounts(ancestors, bank_id, |pubkey, (account_info, slot)| {
let account_slot = self
.get_account_accessor(slot, pubkey, account_info.store_id, account_info.offset)
.get_loaded_account()
.map(|loaded_account| (pubkey, loaded_account.take_account(), slot));
scan_func(&mut collector, account_slot)
scan_func(account_slot)
})?;

Ok(collector)
Ok(())
}

pub fn unchecked_scan_accounts<F, A>(
pub fn unchecked_scan_accounts<F>(
&self,
metric_name: &'static str,
ancestors: &Ancestors,
scan_func: F,
) -> A
where
F: Fn(&mut A, (&Pubkey, LoadedAccount, Slot)),
A: Default,
mut scan_func: F,
) where
F: FnMut(&Pubkey, LoadedAccount, Slot),
{
let mut collector = A::default();
self.accounts_index.unchecked_scan_accounts(
metric_name,
ancestors,
Expand All @@ -2550,26 +2544,22 @@ impl AccountsDb {
.get_account_accessor(slot, pubkey, account_info.store_id, account_info.offset)
.get_loaded_account()
{
scan_func(&mut collector, (pubkey, loaded_account, slot));
scan_func(pubkey, loaded_account, slot);
}
},
);
collector
}

pub fn range_scan_accounts<F, A, R>(
pub fn range_scan_accounts<F, R>(
&self,
metric_name: &'static str,
ancestors: &Ancestors,
range: R,
scan_func: F,
) -> A
where
F: Fn(&mut A, Option<(&Pubkey, AccountSharedData, Slot)>),
A: Default,
mut scan_func: F,
) where
F: FnMut(Option<(&Pubkey, AccountSharedData, Slot)>),
R: RangeBounds<Pubkey>,
{
let mut collector = A::default();
self.accounts_index.range_scan_accounts(
metric_name,
ancestors,
Expand All @@ -2584,27 +2574,26 @@ impl AccountsDb {
// meaning no other subsystems can invalidate the account_info before making their
// changes to the index entry.
// For details, see the comment in retry_to_get_account_accessor()
let account_slot = self
if let Some(account_slot) = self
.get_account_accessor(slot, pubkey, account_info.store_id, account_info.offset)
.get_loaded_account()
.map(|loaded_account| (pubkey, loaded_account.take_account(), slot))
.unwrap();
scan_func(&mut collector, Some(account_slot))
{
scan_func(Some(account_slot))
}
},
);
collector
}

pub fn index_scan_accounts<F, A>(
pub fn index_scan_accounts<F>(
&self,
ancestors: &Ancestors,
bank_id: BankId,
index_key: IndexKey,
scan_func: F,
) -> ScanResult<(A, bool)>
mut scan_func: F,
) -> ScanResult<bool>
where
F: Fn(&mut A, Option<(&Pubkey, AccountSharedData, Slot)>),
A: Default,
F: FnMut(Option<(&Pubkey, AccountSharedData, Slot)>),
{
let key = match &index_key {
IndexKey::ProgramId(key) => key,
Expand All @@ -2614,11 +2603,10 @@ impl AccountsDb {
if !self.account_indexes.include_key(key) {
// the requested key was not indexed in the secondary index, so do a normal scan
let used_index = false;
let scan_result = self.scan_accounts(ancestors, bank_id, scan_func)?;
return Ok((scan_result, used_index));
self.scan_accounts(ancestors, bank_id, scan_func)?;
return Ok(used_index);
}

let mut collector = A::default();
self.accounts_index.index_scan_accounts(
ancestors,
bank_id,
Expand All @@ -2628,11 +2616,11 @@ impl AccountsDb {
.get_account_accessor(slot, pubkey, account_info.store_id, account_info.offset)
.get_loaded_account()
.map(|loaded_account| (pubkey, loaded_account.take_account(), slot));
scan_func(&mut collector, account_slot)
scan_func(account_slot)
},
)?;
let used_index = true;
Ok((collector, used_index))
Ok(used_index)
}

/// Scan a specific slot through all the account storage in parallel
Expand Down Expand Up @@ -6932,11 +6920,12 @@ pub mod tests {
&account1
);

let accounts: Vec<AccountSharedData> = db.unchecked_scan_accounts(
let mut accounts = Vec::new();
db.unchecked_scan_accounts(
"",
&ancestors,
|accounts: &mut Vec<AccountSharedData>, option| {
accounts.push(option.1.take_account());
|_, account, _| {
accounts.push(account.take_account());
},
);
assert_eq!(accounts, vec![account1]);
Expand Down Expand Up @@ -7855,38 +7844,40 @@ pub mod tests {
keys: [mint_key].iter().cloned().collect::<HashSet<Pubkey>>(),
});
// Secondary index can't be used - do normal scan: should still find both pubkeys
let found_accounts = accounts
let mut found_accounts = HashSet::new();
let used_index = accounts
.index_scan_accounts(
&Ancestors::default(),
bank_id,
index_key,
|collection: &mut HashSet<Pubkey>, account| {
collection.insert(*account.unwrap().0);
|account| {
found_accounts.insert(*account.unwrap().0);
},
)
.unwrap();
assert!(!found_accounts.1);
assert_eq!(found_accounts.0.len(), 2);
assert!(found_accounts.0.contains(&pubkey1));
assert!(found_accounts.0.contains(&pubkey2));
assert!(!used_index);
assert_eq!(found_accounts.len(), 2);
assert!(found_accounts.contains(&pubkey1));
assert!(found_accounts.contains(&pubkey2));

accounts.account_indexes.keys = None;

// Secondary index can now be used since it isn't marked as excluded
let found_accounts = accounts
let mut found_accounts = HashSet::new();
let used_index = accounts
.index_scan_accounts(
&Ancestors::default(),
bank_id,
index_key,
|collection: &mut HashSet<Pubkey>, account| {
collection.insert(*account.unwrap().0);
|account| {
found_accounts.insert(*account.unwrap().0);
},
)
.unwrap();
assert!(found_accounts.1);
assert_eq!(found_accounts.0.len(), 2);
assert!(found_accounts.0.contains(&pubkey1));
assert!(found_accounts.0.contains(&pubkey2));
assert!(used_index);
assert_eq!(found_accounts.len(), 2);
assert!(found_accounts.contains(&pubkey1));
assert!(found_accounts.contains(&pubkey2));

accounts.account_indexes.keys = None;
}
Expand Down Expand Up @@ -8468,21 +8459,23 @@ pub mod tests {
db.store_uncached(1, &[(&key1, &account1)]);

let ancestors = vec![(0, 0)].into_iter().collect();
let accounts: Vec<AccountSharedData> = db.unchecked_scan_accounts(
let mut accounts = Vec::new();
db.unchecked_scan_accounts(
"",
&ancestors,
|accounts: &mut Vec<AccountSharedData>, option| {
accounts.push(option.1.take_account());
|_, account, _| {
accounts.push(account.take_account());
},
);
assert_eq!(accounts, vec![account0]);

let ancestors = vec![(1, 1), (0, 0)].into_iter().collect();
let accounts: Vec<AccountSharedData> = db.unchecked_scan_accounts(
let mut accounts = Vec::new();
db.unchecked_scan_accounts(
"",
&ancestors,
|accounts: &mut Vec<AccountSharedData>, option| {
accounts.push(option.1.take_account());
|_, account, _| {
accounts.push(account.take_account());
},
);
assert_eq!(accounts.len(), 2);
Expand Down Expand Up @@ -10457,7 +10450,7 @@ pub mod tests {
db.scan_accounts(
&scan_ancestors,
bank_id,
|_collector: &mut Vec<(Pubkey, AccountSharedData)>, maybe_account| {
|maybe_account| {
ready_.store(true, Ordering::Relaxed);
if let Some((pubkey, _, _)) = maybe_account {
if *pubkey == stall_key {
Expand Down

0 comments on commit 3d40915

Please sign in to comment.