Skip to content

Commit

Permalink
More convient way to configure
Browse files Browse the repository at this point in the history
  • Loading branch information
@apfitzge
apfitzge committed Dec 11, 2024
1 parent ed07130 commit 80c0975
Show file tree
Hide file tree
Showing 3 changed files with 22 additions and 30 deletions.
15 changes: 5 additions & 10 deletions core/src/banking_stage.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -28,11 +28,9 @@ use {
tracer_packet_stats::TracerPacketStats,
validator::BlockProductionMethod,
},
consumer::TARGET_NUM_TRANSACTIONS_PER_BATCH,
crossbeam_channel::{unbounded, Receiver, RecvTimeoutError, Sender},
histogram::Histogram,
solana_client::connection_cache::ConnectionCache,
solana_cost_model::block_cost_limits::MAX_BLOCK_UNITS,
solana_gossip::{cluster_info::ClusterInfo, contact_info::ContactInfo},
solana_ledger::blockstore_processor::TransactionStatusSender,
solana_measure::measure_us,
Expand Down Expand Up @@ -624,14 +622,11 @@ impl BankingStage {
bank_forks.clone(),
forwarder.is_some(),
);
let scheduler_config = PrioGraphSchedulerConfig {
max_cu_per_thread: MAX_BLOCK_UNITS / num_threads as u64,
max_transactions_per_scheduling_pass: 100_000,
look_ahead_window_size: 2048,
target_transactions_per_batch: TARGET_NUM_TRANSACTIONS_PER_BATCH,
};
let scheduler =
PrioGraphScheduler::new(work_senders, finished_work_receiver, scheduler_config);
let scheduler = PrioGraphScheduler::new(
work_senders,
finished_work_receiver,
PrioGraphSchedulerConfig::default(),
);
let scheduler_controller = SchedulerController::new(
decision_maker.clone(),
receive_and_buffer,
Expand Down
27 changes: 16 additions & 11 deletions core/src/banking_stage/transaction_scheduler/prio_graph_scheduler.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@ use {
transaction_state::SanitizedTransactionTTL,
},
crate::banking_stage::{
consumer::TARGET_NUM_TRANSACTIONS_PER_BATCH,
read_write_account_set::ReadWriteAccountSet,
scheduler_messages::{
ConsumeWork, FinishedConsumeWork, MaxAge, TransactionBatchId, TransactionId,
Expand All @@ -18,6 +19,7 @@ use {
crossbeam_channel::{Receiver, Sender, TryRecvError},
itertools::izip,
prio_graph::{AccessKind, GraphNode, PrioGraph},
solana_cost_model::block_cost_limits::MAX_BLOCK_UNITS,
solana_measure::measure_us,
solana_runtime_transaction::transaction_with_meta::TransactionWithMeta,
solana_sdk::{pubkey::Pubkey, saturating_add_assign},
Expand All @@ -40,12 +42,23 @@ type SchedulerPrioGraph = PrioGraph<
>;

pub(crate) struct PrioGraphSchedulerConfig {
pub max_cu_per_thread: u64,
pub max_scheduled_cus: u64,
pub max_transactions_per_scheduling_pass: usize,
pub look_ahead_window_size: usize,
pub target_transactions_per_batch: usize,
}

impl Default for PrioGraphSchedulerConfig {
fn default() -> Self {
Self {
max_scheduled_cus: MAX_BLOCK_UNITS,
max_transactions_per_scheduling_pass: 100_000,
look_ahead_window_size: 2048,
target_transactions_per_batch: TARGET_NUM_TRANSACTIONS_PER_BATCH,
}
}
}

pub(crate) struct PrioGraphScheduler<Tx> {
in_flight_tracker: InFlightTracker,
account_locks: ThreadAwareAccountLocks,
Expand Down Expand Up @@ -95,7 +108,7 @@ impl<Tx: TransactionWithMeta> PrioGraphScheduler<Tx> {
pre_lock_filter: impl Fn(&Tx) -> bool,
) -> Result<SchedulingSummary, SchedulerError> {
let num_threads = self.consume_work_senders.len();
let max_cu_per_thread = self.config.max_cu_per_thread;
let max_cu_per_thread = self.config.max_scheduled_cus / num_threads as u64;

let mut schedulable_threads = ThreadSet::any(num_threads);
for thread_id in 0..num_threads {
Expand Down Expand Up @@ -614,13 +627,11 @@ mod tests {
use {
super::*,
crate::banking_stage::{
consumer::TARGET_NUM_TRANSACTIONS_PER_BATCH,
immutable_deserialized_packet::ImmutableDeserializedPacket,
transaction_scheduler::transaction_state_container::TransactionStateContainer,
},
crossbeam_channel::{unbounded, Receiver},
itertools::Itertools,
solana_cost_model::block_cost_limits::MAX_BLOCK_UNITS,
solana_runtime_transaction::runtime_transaction::RuntimeTransaction,
solana_sdk::{
compute_budget::ComputeBudgetInstruction,
Expand All @@ -647,16 +658,10 @@ mod tests {
let (consume_work_senders, consume_work_receivers) =
(0..num_threads).map(|_| unbounded()).unzip();
let (finished_consume_work_sender, finished_consume_work_receiver) = unbounded();
let scheduler_config = PrioGraphSchedulerConfig {
max_cu_per_thread: MAX_BLOCK_UNITS / num_threads as u64,
max_transactions_per_scheduling_pass: 100_000,
look_ahead_window_size: 2048,
target_transactions_per_batch: TARGET_NUM_TRANSACTIONS_PER_BATCH,
};
let scheduler = PrioGraphScheduler::new(
consume_work_senders,
finished_consume_work_receiver,
scheduler_config,
PrioGraphSchedulerConfig::default(),
);
(
scheduler,
Expand Down
10 changes: 1 addition & 9 deletions core/src/banking_stage/transaction_scheduler/scheduler_controller.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -451,7 +451,6 @@ mod tests {
},
crossbeam_channel::{unbounded, Receiver, Sender},
itertools::Itertools,
solana_cost_model::block_cost_limits::MAX_BLOCK_UNITS,
solana_gossip::cluster_info::ClusterInfo,
solana_ledger::{
blockstore::Blockstore, genesis_utils::GenesisConfigInfo,
Expand Down Expand Up @@ -554,18 +553,11 @@ mod tests {
false,
);

let scheduler_config = PrioGraphSchedulerConfig {
max_cu_per_thread: MAX_BLOCK_UNITS / num_threads as u64,
max_transactions_per_scheduling_pass: 100_000,
look_ahead_window_size: 2048,
target_transactions_per_batch: TARGET_NUM_TRANSACTIONS_PER_BATCH,
};
let scheduler = PrioGraphScheduler::new(
consume_work_senders,
finished_consume_work_receiver,
scheduler_config,
PrioGraphSchedulerConfig::default(),
);

let scheduler_controller = SchedulerController::new(
decision_maker,
receive_and_buffer,
Expand Down

0 comments on commit 80c0975

Please sign in to comment.