gaspersic/solana
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Invoke cost tracker from its bank (backport #20627) (#20800)

Browse Source 
* - make cost_tracker a member of bank, remove shared instance from TPU; (#20627)

- decouple cost_model from cost_tracker; allowing one cost_model
  instance being shared within a validator;
- update cost_model api to calculate_cost(&self...)->transaction_cost

(cherry picked from commit 7496b5784b)

# Conflicts:
#	core/src/banking_stage.rs
#	ledger-tool/src/main.rs
#	runtime/src/bank.rs
#	runtime/src/cost_model.rs
#	runtime/src/cost_tracker.rs

* manual fix merge conflicts

Co-authored-by: Tao Zhu <82401714+taozhu-chicago@users.noreply.github.com>
Co-authored-by: Tao Zhu <tao@solana.com>
This commit is contained in:
mergify[bot]
2021-10-20 00:22:38 +00:00
committed by GitHub
parent de32ab4d57
commit 28eb6ff796
8 changed files with 216 additions and 354 deletions
Download Patch File Download Diff File Expand all files Collapse all files

289
core/src/banking_stage.rs
View File

@@ -24,6 +24,7 @@ use solana_runtime::{
TransactionExecutionResult,
},
bank_utils,
cost_model::CostModel,
cost_tracker::CostTracker,
cost_tracker_stats::CostTrackerStats,
hashed_transaction::HashedTransaction,
@@ -55,7 +56,7 @@ use std::{
net::{SocketAddr, UdpSocket},
ops::DerefMut,
sync::atomic::{AtomicU64, AtomicUsize, Ordering},
sync::{Arc, Mutex, RwLock},
sync::{Arc, Mutex, RwLock, RwLockReadGuard},
thread::{self, Builder, JoinHandle},
time::Duration,
time::Instant,
@@ -96,7 +97,6 @@ pub struct BankingStageStats {
current_buffered_packet_batches_count: AtomicUsize,
rebuffered_packets_count: AtomicUsize,
consumed_buffered_packets_count: AtomicUsize,
reset_cost_tracker_count: AtomicUsize,
cost_tracker_check_count: AtomicUsize,
cost_forced_retry_transactions_count: AtomicUsize,
@@ -175,11 +175,6 @@ impl BankingStageStats {
.swap(0, Ordering::Relaxed) as i64,
i64
),
(
"reset_cost_tracker_count",
self.reset_cost_tracker_count.swap(0, Ordering::Relaxed) as i64,
i64
),
(
"cost_tracker_check_count",
self.cost_tracker_check_count.swap(0, Ordering::Relaxed) as i64,
@@ -288,7 +283,7 @@ impl BankingStage {
verified_vote_receiver: CrossbeamReceiver<Vec<Packets>>,
transaction_status_sender: Option<TransactionStatusSender>,
gossip_vote_sender: ReplayVoteSender,
cost_tracker: Arc<RwLock<CostTracker>>,
cost_model: Arc<RwLock<CostModel>>,
) -> Self {
Self::new_num_threads(
cluster_info,
@@ -299,7 +294,7 @@ impl BankingStage {
Self::num_threads(),
transaction_status_sender,
gossip_vote_sender,
cost_tracker,
cost_model,
)
}
@@ -312,7 +307,7 @@ impl BankingStage {
num_threads: u32,
transaction_status_sender: Option<TransactionStatusSender>,
gossip_vote_sender: ReplayVoteSender,
cost_tracker: Arc<RwLock<CostTracker>>,
cost_model: Arc<RwLock<CostModel>>,
) -> Self {
let batch_limit = TOTAL_BUFFERED_PACKETS / ((num_threads - 1) as usize * PACKETS_PER_BATCH);
// Single thread to generate entries from many banks.
@@ -348,7 +343,7 @@ impl BankingStage {
let gossip_vote_sender = gossip_vote_sender.clone();
let duplicates = duplicates.clone();
let data_budget = data_budget.clone();
let cost_tracker = cost_tracker.clone();
let cost_model = cost_model.clone();
Builder::new()
.name("solana-banking-stage-tx".to_string())
.spawn(move || {
@@ -365,7 +360,7 @@ impl BankingStage {
gossip_vote_sender,
&duplicates,
&data_budget,
&cost_tracker,
cost_model,
);
})
.unwrap()
@@ -423,24 +418,6 @@ impl BankingStage {
has_more_unprocessed_transactions
}
fn reset_cost_tracker_if_new_bank(
cost_tracker: &Arc<RwLock<CostTracker>>,
bank: Arc<Bank>,
banking_stage_stats: &BankingStageStats,
cost_tracker_stats: &mut CostTrackerStats,
) {
if cost_tracker
.write()
.unwrap()
.reset_if_new_bank(bank.slot(), cost_tracker_stats)
{
// only increase counter when bank changed
banking_stage_stats
.reset_cost_tracker_count
.fetch_add(1, Ordering::Relaxed);
}
}
#[allow(clippy::too_many_arguments)]
pub fn consume_buffered_packets(
my_pubkey: &Pubkey,
@@ -452,7 +429,7 @@ impl BankingStage {
test_fn: Option<impl Fn()>,
banking_stage_stats: &BankingStageStats,
recorder: &TransactionRecorder,
cost_tracker: &Arc<RwLock<CostTracker>>,
cost_model: &Arc<RwLock<CostModel>>,
cost_tracker_stats: &mut CostTrackerStats,
) {
let mut rebuffered_packets_len = 0;
@@ -471,8 +448,8 @@ impl BankingStage {
original_unprocessed_indexes,
my_pubkey,
*next_leader,
cost_tracker,
banking_stage_stats,
cost_model,
cost_tracker_stats,
);
Self::update_buffered_packets_with_new_unprocessed(
@@ -482,12 +459,6 @@ impl BankingStage {
} else {
let bank_start = poh_recorder.lock().unwrap().bank_start();
if let Some((bank, bank_creation_time)) = bank_start {
Self::reset_cost_tracker_if_new_bank(
cost_tracker,
bank.clone(),
banking_stage_stats,
cost_tracker_stats,
);
let (processed, verified_txs_len, new_unprocessed_indexes) =
Self::process_packets_transactions(
&bank,
@@ -498,7 +469,7 @@ impl BankingStage {
transaction_status_sender.clone(),
gossip_vote_sender,
banking_stage_stats,
cost_tracker,
cost_model,
cost_tracker_stats,
);
if processed < verified_txs_len
@@ -603,7 +574,7 @@ impl BankingStage {
banking_stage_stats: &BankingStageStats,
recorder: &TransactionRecorder,
data_budget: &DataBudget,
cost_tracker: &Arc<RwLock<CostTracker>>,
cost_model: &Arc<RwLock<CostModel>>,
cost_tracker_stats: &mut CostTrackerStats,
) -> BufferedPacketsDecision {
let bank_start;
@@ -615,14 +586,6 @@ impl BankingStage {
) = {
let poh = poh_recorder.lock().unwrap();
bank_start = poh.bank_start();
if let Some((ref bank, _)) = bank_start {
Self::reset_cost_tracker_if_new_bank(
cost_tracker,
bank.clone(),
banking_stage_stats,
cost_tracker_stats,
);
};
(
poh.leader_after_n_slots(FORWARD_TRANSACTIONS_TO_LEADER_AT_SLOT_OFFSET),
PohRecorder::get_bank_still_processing_txs(&bank_start),
@@ -653,7 +616,7 @@ impl BankingStage {
None::<Box<dyn Fn()>>,
banking_stage_stats,
recorder,
cost_tracker,
cost_model,
cost_tracker_stats,
);
}
@@ -734,7 +697,7 @@ impl BankingStage {
gossip_vote_sender: ReplayVoteSender,
duplicates: &Arc<Mutex<(LruCache<u64, ()>, PacketHasher)>>,
data_budget: &DataBudget,
cost_tracker: &Arc<RwLock<CostTracker>>,
cost_model: Arc<RwLock<CostModel>>,
) {
let recorder = poh_recorder.lock().unwrap().recorder();
let socket = UdpSocket::bind("0.0.0.0:0").unwrap();
@@ -755,7 +718,7 @@ impl BankingStage {
&banking_stage_stats,
&recorder,
data_budget,
cost_tracker,
&cost_model,
&mut cost_tracker_stats,
);
if matches!(decision, BufferedPacketsDecision::Hold)
@@ -791,7 +754,7 @@ impl BankingStage {
&banking_stage_stats,
duplicates,
&recorder,
cost_tracker,
&cost_model,
&mut cost_tracker_stats,
) {
Ok(()) | Err(RecvTimeoutError::Timeout) => (),
@@ -896,7 +859,6 @@ impl BankingStage {
};
let mut execute_timings = ExecuteTimings::default();
let (
mut loaded_accounts,
results,
@@ -1131,10 +1093,11 @@ impl BankingStage {
msgs: &Packets,
transaction_indexes: &[usize],
feature_set: &Arc<feature_set::FeatureSet>,
votes_only: bool,
cost_tracker: &Arc<RwLock<CostTracker>>,
read_cost_tracker: &RwLockReadGuard<CostTracker>,
banking_stage_stats: &BankingStageStats,
demote_program_write_locks: bool,
votes_only: bool,
cost_model: &Arc<RwLock<CostModel>>,
cost_tracker_stats: &mut CostTrackerStats,
) -> (Vec<HashedTransaction<'static>>, Vec<usize>, Vec<usize>) {
let mut retryable_transaction_packet_indexes: Vec<usize> = vec![];
@@ -1160,7 +1123,6 @@ impl BankingStage {
let mut cost_tracker_check_time = Measure::start("cost_tracker_check_time");
let filtered_transactions_with_packet_indexes: Vec<_> = {
let cost_tracker_readonly = cost_tracker.read().unwrap();
verified_transactions_with_packet_indexes
.into_iter()
.filter_map(|(tx, tx_index)| {
@@ -1170,10 +1132,13 @@ impl BankingStage {
// excluding vote TX from cost_model, for now
if !is_vote
&& cost_tracker_readonly
&& read_cost_tracker
.would_transaction_fit(
&tx,
demote_program_write_locks,
&cost_model
.read()
.unwrap()
.calculate_cost(&tx, demote_program_write_locks),
cost_tracker_stats,
)
.is_err()
@@ -1262,7 +1227,7 @@ impl BankingStage {
transaction_status_sender: Option<TransactionStatusSender>,
gossip_vote_sender: &ReplayVoteSender,
banking_stage_stats: &BankingStageStats,
cost_tracker: &Arc<RwLock<CostTracker>>,
cost_model: &Arc<RwLock<CostModel>>,
cost_tracker_stats: &mut CostTrackerStats,
) -> (usize, usize, Vec<usize>) {
let mut packet_conversion_time = Measure::start("packet_conversion");
@@ -1271,10 +1236,11 @@ impl BankingStage {
msgs,
&packet_indexes,
&bank.feature_set,
bank.vote_only_bank(),
cost_tracker,
&bank.read_cost_tracker().unwrap(),
banking_stage_stats,
bank.demote_program_write_locks(),
bank.vote_only_bank(),
cost_model,
cost_tracker_stats,
);
packet_conversion_time.stop();
@@ -1312,9 +1278,12 @@ impl BankingStage {
let mut cost_tracking_time = Measure::start("cost_tracking_time");
transactions.iter().enumerate().for_each(|(index, tx)| {
if unprocessed_tx_indexes.iter().all(|&i| i != index) {
cost_tracker.write().unwrap().add_transaction_cost(
bank.write_cost_tracker().unwrap().add_transaction_cost(
tx.transaction(),
bank.demote_program_write_locks(),
&cost_model
.read()
.unwrap()
.calculate_cost(tx.transaction(), bank.demote_program_write_locks()),
cost_tracker_stats,
);
}
@@ -1361,8 +1330,8 @@ impl BankingStage {
transaction_indexes: &[usize],
my_pubkey: &Pubkey,
next_leader: Option<Pubkey>,
cost_tracker: &Arc<RwLock<CostTracker>>,
banking_stage_stats: &BankingStageStats,
cost_model: &Arc<RwLock<CostModel>>,
cost_tracker_stats: &mut CostTrackerStats,
) -> Vec<usize> {
// Check if we are the next leader. If so, let's not filter the packets
@@ -1381,10 +1350,11 @@ impl BankingStage {
msgs,
&transaction_indexes,
&bank.feature_set,
bank.vote_only_bank(),
cost_tracker,
&bank.read_cost_tracker().unwrap(),
banking_stage_stats,
bank.demote_program_write_locks(),
bank.vote_only_bank(),
cost_model,
cost_tracker_stats,
);
unprocessed_packet_conversion_time.stop();
@@ -1446,7 +1416,7 @@ impl BankingStage {
banking_stage_stats: &BankingStageStats,
duplicates: &Arc<Mutex<(LruCache<u64, ()>, PacketHasher)>>,
recorder: &TransactionRecorder,
cost_tracker: &Arc<RwLock<CostTracker>>,
cost_model: &Arc<RwLock<CostModel>>,
cost_tracker_stats: &mut CostTrackerStats,
) -> Result<(), RecvTimeoutError> {
let mut recv_time = Measure::start("process_packets_recv");
@@ -1488,12 +1458,6 @@ impl BankingStage {
continue;
}
let (bank, bank_creation_time) = bank_start.unwrap();
Self::reset_cost_tracker_if_new_bank(
cost_tracker,
bank.clone(),
banking_stage_stats,
cost_tracker_stats,
);
let (processed, verified_txs_len, unprocessed_indexes) =
Self::process_packets_transactions(
@@ -1505,7 +1469,7 @@ impl BankingStage {
transaction_status_sender.clone(),
gossip_vote_sender,
banking_stage_stats,
cost_tracker,
cost_model,
cost_tracker_stats,
);
@@ -1538,8 +1502,8 @@ impl BankingStage {
&packet_indexes,
my_pubkey,
next_leader,
cost_tracker,
banking_stage_stats,
cost_model,
cost_tracker_stats,
);
Self::push_unprocessed(
@@ -1768,9 +1732,7 @@ mod tests {
gossip_verified_vote_receiver,
None,
vote_forward_sender,
Arc::new(RwLock::new(CostTracker::new(Arc::new(RwLock::new(
CostModel::default(),
))))),
Arc::new(RwLock::new(CostModel::default())),
);
drop(verified_sender);
drop(gossip_verified_vote_sender);
@@ -1819,9 +1781,7 @@ mod tests {
verified_gossip_vote_receiver,
None,
vote_forward_sender,
Arc::new(RwLock::new(CostTracker::new(Arc::new(RwLock::new(
CostModel::default(),
))))),
Arc::new(RwLock::new(CostModel::default())),
);
trace!("sending bank");
drop(verified_sender);
@@ -1894,9 +1854,7 @@ mod tests {
gossip_verified_vote_receiver,
None,
gossip_vote_sender,
Arc::new(RwLock::new(CostTracker::new(Arc::new(RwLock::new(
CostModel::default(),
))))),
Arc::new(RwLock::new(CostModel::default())),
);
// fund another account so we can send 2 good transactions in a single batch.
@@ -2047,9 +2005,7 @@ mod tests {
3,
None,
gossip_vote_sender,
Arc::new(RwLock::new(CostTracker::new(Arc::new(RwLock::new(
CostModel::default(),
))))),
Arc::new(RwLock::new(CostModel::default())),
);
// wait for banking_stage to eat the packets
@@ -2871,9 +2827,7 @@ mod tests {
None::<Box<dyn Fn()>>,
&BankingStageStats::default(),
&recorder,
&Arc::new(RwLock::new(CostTracker::new(Arc::new(RwLock::new(
CostModel::default(),
))))),
&Arc::new(RwLock::new(CostModel::default())),
&mut CostTrackerStats::default(),
);
assert_eq!(buffered_packets[0].1.len(), num_conflicting_transactions);
@@ -2891,9 +2845,7 @@ mod tests {
None::<Box<dyn Fn()>>,
&BankingStageStats::default(),
&recorder,
&Arc::new(RwLock::new(CostTracker::new(Arc::new(RwLock::new(
CostModel::default(),
))))),
&Arc::new(RwLock::new(CostModel::default())),
&mut CostTrackerStats::default(),
);
if num_expected_unprocessed == 0 {
@@ -2960,9 +2912,7 @@ mod tests {
test_fn,
&BankingStageStats::default(),
&recorder,
&Arc::new(RwLock::new(CostTracker::new(Arc::new(RwLock::new(
CostModel::default(),
))))),
&Arc::new(RwLock::new(CostModel::default())),
&mut CostTrackerStats::default(),
);
@@ -3204,7 +3154,6 @@ mod tests {
&keypair,
None,
);
let features = Arc::new(feature_set::FeatureSet::default());
// packets with no votes
{
@@ -3213,35 +3162,34 @@ mod tests {
make_test_packets(vec![transfer_tx.clone(), transfer_tx.clone()], vote_indexes);
let mut votes_only = false;
let (txs, tx_packet_index, _) = BankingStage::transactions_from_packets(
&packets,
&packet_indexes,
&features,
votes_only,
&Arc::new(RwLock::new(CostTracker::new(Arc::new(RwLock::new(
CostModel::default(),
))))),
&BankingStageStats::default(),
false,
&mut CostTrackerStats::default(),
);
let (txs, tx_packet_index, _retryable_packet_indexes) =
BankingStage::transactions_from_packets(
&packets,
&packet_indexes,
&Arc::new(feature_set::FeatureSet::default()),
&RwLock::new(CostTracker::default()).read().unwrap(),
&BankingStageStats::default(),
false,
votes_only,
&Arc::new(RwLock::new(CostModel::default())),
&mut CostTrackerStats::default(),
);
assert_eq!(2, txs.len());
assert_eq!(vec![0, 1], tx_packet_index);
votes_only = true;
let (txs, tx_packet_index, _) = BankingStage::transactions_from_packets(
&packets,
&packet_indexes,
&features,
votes_only,
&Arc::new(RwLock::new(CostTracker::new(Arc::new(RwLock::new(
CostModel::default(),
))))),
&BankingStageStats::default(),
false,
&mut CostTrackerStats::default(),
);
let (txs, tx_packet_index, _retryable_packet_indexes) =
BankingStage::transactions_from_packets(
&packets,
&packet_indexes,
&Arc::new(feature_set::FeatureSet::default()),
&RwLock::new(CostTracker::default()).read().unwrap(),
&BankingStageStats::default(),
false,
votes_only,
&Arc::new(RwLock::new(CostModel::default())),
&mut CostTrackerStats::default(),
);
assert_eq!(0, txs.len());
assert_eq!(0, tx_packet_index.len());
}
@@ -3255,34 +3203,34 @@ mod tests {
);
let mut votes_only = false;
let (txs, tx_packet_index, _) = BankingStage::transactions_from_packets(
&packets,
&packet_indexes,
&features,
votes_only,
&Arc::new(RwLock::new(CostTracker::new(Arc::new(RwLock::new(
CostModel::default(),
))))),
&BankingStageStats::default(),
false,
&mut CostTrackerStats::default(),
);
let (txs, tx_packet_index, _retryable_packet_indexes) =
BankingStage::transactions_from_packets(
&packets,
&packet_indexes,
&Arc::new(feature_set::FeatureSet::default()),
&RwLock::new(CostTracker::default()).read().unwrap(),
&BankingStageStats::default(),
false,
votes_only,
&Arc::new(RwLock::new(CostModel::default())),
&mut CostTrackerStats::default(),
);
assert_eq!(3, txs.len());
assert_eq!(vec![0, 1, 2], tx_packet_index);
votes_only = true;
let (txs, tx_packet_index, _) = BankingStage::transactions_from_packets(
&packets,
&packet_indexes,
&features,
votes_only,
&Arc::new(RwLock::new(CostTracker::new(Arc::new(RwLock::new(
CostModel::default(),
))))),
&BankingStageStats::default(),
false,
&mut CostTrackerStats::default(),
);
let (txs, tx_packet_index, _retryable_packet_indexes) =
BankingStage::transactions_from_packets(
&packets,
&packet_indexes,
&Arc::new(feature_set::FeatureSet::default()),
&RwLock::new(CostTracker::default()).read().unwrap(),
&BankingStageStats::default(),
false,
votes_only,
&Arc::new(RwLock::new(CostModel::default())),
&mut CostTrackerStats::default(),
);
assert_eq!(2, txs.len());
assert_eq!(vec![0, 2], tx_packet_index);
}
@@ -3294,37 +3242,36 @@ mod tests {
vec![vote_tx.clone(), vote_tx.clone(), vote_tx],
vote_indexes,
);
let features = Arc::new(feature_set::FeatureSet::default());
let mut votes_only = false;
let (txs, tx_packet_index, _) = BankingStage::transactions_from_packets(
&packets,
&packet_indexes,
&features,
votes_only,
&Arc::new(RwLock::new(CostTracker::new(Arc::new(RwLock::new(
CostModel::default(),
))))),
&BankingStageStats::default(),
false,
&mut CostTrackerStats::default(),
);
let (txs, tx_packet_index, _retryable_packet_indexes) =
BankingStage::transactions_from_packets(
&packets,
&packet_indexes,
&Arc::new(feature_set::FeatureSet::default()),
&RwLock::new(CostTracker::default()).read().unwrap(),
&BankingStageStats::default(),
false,
votes_only,
&Arc::new(RwLock::new(CostModel::default())),
&mut CostTrackerStats::default(),
);
assert_eq!(3, txs.len());
assert_eq!(vec![0, 1, 2], tx_packet_index);
votes_only = true;
let (txs, tx_packet_index, _) = BankingStage::transactions_from_packets(
&packets,
&packet_indexes,
&features,
votes_only,
&Arc::new(RwLock::new(CostTracker::new(Arc::new(RwLock::new(
CostModel::default(),
))))),
&BankingStageStats::default(),
false,
&mut CostTrackerStats::default(),
);
let (txs, tx_packet_index, _retryable_packet_indexes) =
BankingStage::transactions_from_packets(
&packets,
&packet_indexes,
&Arc::new(feature_set::FeatureSet::default()),
&RwLock::new(CostTracker::default()).read().unwrap(),
&BankingStageStats::default(),
false,
votes_only,
&Arc::new(RwLock::new(CostModel::default())),
&mut CostTrackerStats::default(),
);
assert_eq!(3, txs.len());
assert_eq!(vec![0, 1, 2], tx_packet_index);
}

4
core/src/tpu.rs
View File

@@ -23,7 +23,6 @@ use solana_rpc::{
use solana_runtime::{
bank_forks::BankForks,
cost_model::CostModel,
cost_tracker::CostTracker,
vote_sender_types::{ReplayVoteReceiver, ReplayVoteSender},
};
use std::{
@@ -123,7 +122,6 @@ impl Tpu {
cluster_confirmed_slot_sender,
);
let cost_tracker = Arc::new(RwLock::new(CostTracker::new(cost_model.clone())));
let banking_stage = BankingStage::new(
cluster_info,
poh_recorder,
@@ -132,7 +130,7 @@ impl Tpu {
verified_gossip_vote_packets_receiver,
transaction_status_sender,
replay_vote_sender,
cost_tracker,
cost_model.clone(),
);
let broadcast_stage = broadcast_type.new_broadcast_stage(