Skip to content

Commit

Permalink
Add limit to looping in banking-stage (#35342)
Browse files Browse the repository at this point in the history
  • Loading branch information
pgarg66 authored and gregcusack committed Mar 1, 2024
1 parent 992b444 commit 6182226
Show file tree
Hide file tree
Showing 4 changed files with 35 additions and 9 deletions.
3 changes: 2 additions & 1 deletion core/src/banking_stage/consumer.rs
Original file line number Diff line number Diff line change
Expand Up @@ -598,7 +598,8 @@ impl Consumer {
transaction_status_sender_enabled,
&mut execute_and_commit_timings.execute_timings,
None, // account_overrides
self.log_messages_bytes_limit
self.log_messages_bytes_limit,
true,
));
execute_and_commit_timings.load_execute_us = load_execute_us;

Expand Down
9 changes: 6 additions & 3 deletions program-runtime/src/loaded_programs.rs
Original file line number Diff line number Diff line change
Expand Up @@ -195,7 +195,7 @@ impl Stats {
("reloads", reloads, i64),
("insertions", insertions, i64),
("lost_insertions", lost_insertions, i64),
("replacements", replacements, i64),
("replace_entry", replacements, i64),
("one_hit_wonders", one_hit_wonders, i64),
("prunes_orphan", prunes_orphan, i64),
("prunes_environment", prunes_environment, i64),
Expand Down Expand Up @@ -618,6 +618,7 @@ pub struct LoadedProgramsForTxBatch {
entries: HashMap<Pubkey, Arc<LoadedProgram>>,
slot: Slot,
pub environments: ProgramRuntimeEnvironments,
pub hit_max_limit: bool,
}

impl LoadedProgramsForTxBatch {
Expand All @@ -626,6 +627,7 @@ impl LoadedProgramsForTxBatch {
entries: HashMap::new(),
slot,
environments,
hit_max_limit: false,
}
}

Expand Down Expand Up @@ -964,7 +966,7 @@ impl<FG: ForkGraph> LoadedPrograms<FG> {
slot: Slot,
key: Pubkey,
loaded_program: Arc<LoadedProgram>,
) {
) -> bool {
let second_level = self.entries.entry(key).or_default();
debug_assert_eq!(
second_level.cooperative_loading_lock,
Expand All @@ -985,8 +987,9 @@ impl<FG: ForkGraph> LoadedPrograms<FG> {
{
self.stats.lost_insertions.fetch_add(1, Ordering::Relaxed);
}
self.assign_program(key, loaded_program);
let was_occupied = self.assign_program(key, loaded_program);
self.loading_task_waiter.notify();
was_occupied
}

pub fn merge(&mut self, tx_batch_cache: &LoadedProgramsForTxBatch) {
Expand Down
6 changes: 5 additions & 1 deletion runtime/src/bank.rs
Original file line number Diff line number Diff line change
Expand Up @@ -4299,6 +4299,7 @@ impl Bank {
&mut timings,
Some(&account_overrides),
None,
true,
);

let post_simulation_accounts = loaded_transactions
Expand Down Expand Up @@ -4537,7 +4538,7 @@ impl Bank {
balances
}

#[allow(clippy::type_complexity)]
#[allow(clippy::too_many_arguments, clippy::type_complexity)]
pub fn load_and_execute_transactions(
&self,
batch: &TransactionBatch,
Expand All @@ -4548,6 +4549,7 @@ impl Bank {
timings: &mut ExecuteTimings,
account_overrides: Option<&AccountOverrides>,
log_messages_bytes_limit: Option<usize>,
limit_to_load_programs: bool,
) -> LoadAndExecuteTransactionsOutput {
let sanitized_txs = batch.sanitized_transactions();
debug!("processing transactions: {}", sanitized_txs.len());
Expand Down Expand Up @@ -4614,6 +4616,7 @@ impl Bank {
account_overrides,
self.builtin_programs.iter(),
log_messages_bytes_limit,
limit_to_load_programs,
);

let mut signature_count = 0;
Expand Down Expand Up @@ -5663,6 +5666,7 @@ impl Bank {
timings,
None,
log_messages_bytes_limit,
false,
);

let (last_blockhash, lamports_per_signature) =
Expand Down
26 changes: 22 additions & 4 deletions svm/src/transaction_processor.rs
Original file line number Diff line number Diff line change
Expand Up @@ -190,6 +190,7 @@ impl<FG: ForkGraph> TransactionBatchProcessor<FG> {
account_overrides: Option<&AccountOverrides>,
builtin_programs: impl Iterator<Item = &'a Pubkey>,
log_messages_bytes_limit: Option<usize>,
limit_to_load_programs: bool,
) -> LoadAndExecuteSanitizedTransactionsOutput {
let mut program_accounts_map = Self::filter_executable_program_accounts(
callbacks,
Expand All @@ -202,9 +203,18 @@ impl<FG: ForkGraph> TransactionBatchProcessor<FG> {
program_accounts_map.insert(*builtin_program, (&native_loader, 0));
}

let programs_loaded_for_tx_batch = Rc::new(RefCell::new(
self.replenish_program_cache(callbacks, &program_accounts_map),
));
let programs_loaded_for_tx_batch = Rc::new(RefCell::new(self.replenish_program_cache(
callbacks,
&program_accounts_map,
limit_to_load_programs,
)));

if programs_loaded_for_tx_batch.borrow().hit_max_limit {
return LoadAndExecuteSanitizedTransactionsOutput {
loaded_transactions: vec![],
execution_results: vec![],
};
}

let mut load_time = Measure::start("accounts_load");
let mut loaded_transactions = load_accounts(
Expand Down Expand Up @@ -356,6 +366,7 @@ impl<FG: ForkGraph> TransactionBatchProcessor<FG> {
&self,
callback: &CB,
program_accounts_map: &HashMap<Pubkey, (&Pubkey, u64)>,
limit_to_load_programs: bool,
) -> LoadedProgramsForTxBatch {
let mut missing_programs: Vec<(Pubkey, (LoadedProgramMatchCriteria, u64))> =
if self.check_program_modification_slot {
Expand Down Expand Up @@ -401,7 +412,14 @@ impl<FG: ForkGraph> TransactionBatchProcessor<FG> {
}
// Submit our last completed loading task.
if let Some((key, program)) = program_to_store.take() {
loaded_programs_cache.finish_cooperative_loading_task(self.slot, key, program);
if loaded_programs_cache
.finish_cooperative_loading_task(self.slot, key, program)
&& limit_to_load_programs
{
let mut ret = LoadedProgramsForTxBatch::default();
ret.hit_max_limit = true;
return ret;
}
}
// Figure out which program needs to be loaded next.
let program_to_load = loaded_programs_cache.extract(
Expand Down

0 comments on commit 6182226

Please sign in to comment.