Skip to content

Commit

Permalink
prioritization fee cache: remove lru crate (#30)
Browse files Browse the repository at this point in the history
  • Loading branch information
@fanatid
fanatid authored Mar 27, 2024
1 parent cfd5b71 commit ba9c25c
Show file tree
Hide file tree
Showing 5 changed files with 56 additions and 65 deletions.
1 change: 0 additions & 1 deletion Cargo.lock
View file

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

1 change: 0 additions & 1 deletion programs/sbf/Cargo.lock
View file

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

1 change: 0 additions & 1 deletion runtime/Cargo.toml
View file
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,6 @@ itertools = { workspace = true }
lazy_static = { workspace = true }
libc = { workspace = true }
log = { workspace = true }
lru = { workspace = true }
lz4 = { workspace = true }
memmap2 = { workspace = true }
mockall = { workspace = true }
Expand Down
19 changes: 4 additions & 15 deletions runtime/src/prioritization_fee.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -165,11 +165,7 @@ impl Default for PrioritizationFee {

impl PrioritizationFee {
/// Update self for minimum transaction fee in the block and minimum fee for each writable account.
pub fn update(
&mut self,
transaction_fee: u64,
writable_accounts: Vec<Pubkey>,
) -> Result<(), PrioritizationFeeError> {
pub fn update(&mut self, transaction_fee: u64, writable_accounts: Vec<Pubkey>) {
let (_, update_time) = measure!(
{
if !self.is_finalized {
Expand Down Expand Up @@ -199,7 +195,6 @@ impl PrioritizationFee {

self.metrics
.accumulate_total_update_elapsed_us(update_time.as_us());
Ok(())
}

/// Accounts that have minimum fees lesser or equal to the minimum fee in the block are redundant, they are
Expand Down Expand Up @@ -283,9 +278,7 @@ mod tests {
// -----------------------------------------------------------------------
// [5, a, b ] --> [5, 5, 5, nil ]
{
assert!(prioritization_fee
.update(5, vec![write_account_a, write_account_b])
.is_ok());
prioritization_fee.update(5, vec![write_account_a, write_account_b]);
assert_eq!(5, prioritization_fee.get_min_transaction_fee().unwrap());
assert_eq!(
5,
Expand All @@ -309,9 +302,7 @@ mod tests {
// -----------------------------------------------------------------------
// [9, b, c ] --> [5, 5, 5, 9 ]
{
assert!(prioritization_fee
.update(9, vec![write_account_b, write_account_c])
.is_ok());
prioritization_fee.update(9, vec![write_account_b, write_account_c]);
assert_eq!(5, prioritization_fee.get_min_transaction_fee().unwrap());
assert_eq!(
5,
Expand All @@ -338,9 +329,7 @@ mod tests {
// -----------------------------------------------------------------------
// [2, a, c ] --> [2, 2, 5, 2 ]
{
assert!(prioritization_fee
.update(2, vec![write_account_a, write_account_c])
.is_ok());
prioritization_fee.update(2, vec![write_account_a, write_account_c]);
assert_eq!(2, prioritization_fee.get_min_transaction_fee().unwrap());
assert_eq!(
2,
Expand Down
99 changes: 52 additions & 47 deletions runtime/src/prioritization_fee_cache.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,6 @@ use {
crate::{bank::Bank, compute_budget_details::GetComputeBudgetDetails, prioritization_fee::*},
crossbeam_channel::{unbounded, Receiver, Sender},
log::*,
lru::LruCache,
solana_measure::measure,
solana_sdk::{
clock::{BankId, Slot},
Expand Down Expand Up @@ -122,6 +121,7 @@ impl PrioritizationFeeCacheMetrics {
}
}

#[derive(Debug)]
enum CacheServiceUpdate {
TransactionUpdate {
slot: Slot,
Expand All @@ -141,7 +141,7 @@ enum CacheServiceUpdate {
/// and collecting stats and reporting metrics.
#[derive(Debug)]
pub struct PrioritizationFeeCache {
cache: Arc<RwLock<LruCache<Slot, PrioritizationFee>>>,
cache: Arc<RwLock<BTreeMap<Slot, PrioritizationFee>>>,
service_thread: Option<JoinHandle<()>>,
sender: Sender<CacheServiceUpdate>,
metrics: Arc<PrioritizationFeeCacheMetrics>,
Expand All @@ -166,17 +166,17 @@ impl Drop for PrioritizationFeeCache {

impl PrioritizationFeeCache {
pub fn new(capacity: u64) -> Self {
let metrics = Arc::new(PrioritizationFeeCacheMetrics::default());
let cache = Arc::new(RwLock::new(BTreeMap::new()));
let (sender, receiver) = unbounded();
let cache = Arc::new(RwLock::new(LruCache::new(capacity as usize)));
let metrics = Arc::new(PrioritizationFeeCacheMetrics::default());

let cache_clone = cache.clone();
let metrics_clone = metrics.clone();
let service_thread = Some(
Builder::new()
.name("solPrFeeCachSvc".to_string())
.spawn(move || {
Self::service_loop(cache_clone, receiver, metrics_clone);
.spawn({
let cache = cache.clone();
let metrics = metrics.clone();
move || Self::service_loop(cache, capacity as usize, receiver, metrics)
})
.unwrap(),
);
Expand Down Expand Up @@ -261,8 +261,7 @@ impl PrioritizationFeeCache {
});
}

/// Internal function is invoked by worker thread to update slot's minimum prioritization fee,
/// Cache lock contends here.
/// Internal function is invoked by worker thread to update slot's minimum prioritization fee.
fn update_cache(
unfinalized: &mut UnfinalizedPrioritizationFees,
slot: Slot,
Expand All @@ -273,7 +272,7 @@ impl PrioritizationFeeCache {
) {
let (_, entry_update_time) = measure!(
{
let _ = unfinalized
unfinalized
.entry(slot)
.or_default()
.entry(bank_id)
Expand All @@ -288,7 +287,8 @@ impl PrioritizationFeeCache {

fn finalize_slot(
unfinalized: &mut UnfinalizedPrioritizationFees,
cache: &RwLock<LruCache<Slot, PrioritizationFee>>,
cache: &RwLock<BTreeMap<Slot, PrioritizationFee>>,
cache_max_size: usize,
slot: Slot,
bank_id: BankId,
metrics: &PrioritizationFeeCacheMetrics,
Expand Down Expand Up @@ -340,7 +340,10 @@ impl PrioritizationFeeCache {
let (_, cache_lock_time) = measure!(
{
let mut cache = cache.write().unwrap();
cache.put(slot, slot_prioritization_fee);
while cache.len() >= cache_max_size {
cache.pop_first();
}
cache.insert(slot, slot_prioritization_fee);
},
"cache_lock_time"
);
Expand All @@ -349,7 +352,8 @@ impl PrioritizationFeeCache {
}

fn service_loop(
cache: Arc<RwLock<LruCache<Slot, PrioritizationFee>>>,
cache: Arc<RwLock<BTreeMap<Slot, PrioritizationFee>>>,
cache_max_size: usize,
receiver: Receiver<CacheServiceUpdate>,
metrics: Arc<PrioritizationFeeCacheMetrics>,
) {
Expand All @@ -373,7 +377,14 @@ impl PrioritizationFeeCache {
&metrics,
),
CacheServiceUpdate::BankFinalized { slot, bank_id } => {
Self::finalize_slot(&mut unfinalized, &cache, slot, bank_id, &metrics);
Self::finalize_slot(
&mut unfinalized,
&cache,
cache_max_size,
slot,
bank_id,
&metrics,
);
metrics.report(slot);
}
CacheServiceUpdate::Exit => {
Expand All @@ -385,20 +396,14 @@ impl PrioritizationFeeCache {

/// Returns number of blocks that have finalized minimum fees collection
pub fn available_block_count(&self) -> usize {
self.cache
.read()
.unwrap()
.iter()
.filter(|(_slot, slot_prioritization_fee)| slot_prioritization_fee.is_finalized())
.count()
self.cache.read().unwrap().len()
}

pub fn get_prioritization_fees(&self, account_keys: &[Pubkey]) -> Vec<(Slot, u64)> {
self.cache
.read()
.unwrap()
.iter()
.filter(|(_slot, slot_prioritization_fee)| slot_prioritization_fee.is_finalized())
.map(|(slot, slot_prioritization_fee)| {
let mut fee = slot_prioritization_fee
.get_min_transaction_fee()
Expand Down Expand Up @@ -471,7 +476,7 @@ mod tests {
.load(Ordering::Relaxed)
!= expected_update_count
{
std::thread::sleep(std::time::Duration::from_millis(100));
std::thread::sleep(std::time::Duration::from_millis(10));
}
}

Expand All @@ -486,7 +491,7 @@ mod tests {

// wait till finalization is done
loop {
let mut cache = prioritization_fee_cache.cache.write().unwrap();
let cache = prioritization_fee_cache.cache.read().unwrap();
if let Some(slot_cache) = cache.get(&slot) {
if slot_cache.is_finalized() {
return;
Expand Down Expand Up @@ -528,14 +533,14 @@ mod tests {

// assert empty cache
{
let mut lock = prioritization_fee_cache.cache.write().unwrap();
let lock = prioritization_fee_cache.cache.read().unwrap();
assert!(lock.get(&slot).is_none());
}

// assert after prune, account a and c should be removed from cache to save space
{
sync_finalize_priority_fee_for_test(&prioritization_fee_cache, slot, bank.bank_id());
let mut lock = prioritization_fee_cache.cache.write().unwrap();
let lock = prioritization_fee_cache.cache.read().unwrap();
let fee = lock.get(&slot).unwrap();
assert_eq!(2, fee.get_min_transaction_fee().unwrap());
assert!(fee.get_writable_account_fee(&write_account_a).is_none());
Expand Down Expand Up @@ -568,15 +573,15 @@ mod tests {
sync_finalize_priority_fee_for_test(&prioritization_fee_cache, 1, bank1.bank_id());

// add slot 2 entry to cache, but not finalize it
let bank2 = Arc::new(Bank::new_from_parent(bank.clone(), &collector, 3));
let bank2 = Arc::new(Bank::new_from_parent(bank.clone(), &collector, 2));
let txs = vec![build_sanitized_transaction_for_test(
1,
&Pubkey::new_unique(),
&Pubkey::new_unique(),
)];
sync_update(&prioritization_fee_cache, bank2.clone(), txs.iter());

let bank3 = Arc::new(Bank::new_from_parent(bank.clone(), &collector, 2));
let bank3 = Arc::new(Bank::new_from_parent(bank.clone(), &collector, 3));
sync_update(
&prioritization_fee_cache,
bank3.clone(),
Expand All @@ -587,7 +592,7 @@ mod tests {
)]
.iter(),
);
sync_finalize_priority_fee_for_test(&prioritization_fee_cache, 2, bank3.bank_id());
sync_finalize_priority_fee_for_test(&prioritization_fee_cache, 3, bank3.bank_id());

// assert available block count should be 2 finalized blocks
assert_eq!(2, prioritization_fee_cache.available_block_count());
Expand Down Expand Up @@ -738,28 +743,28 @@ mod tests {
// after block is completed
sync_finalize_priority_fee_for_test(&prioritization_fee_cache, 2, bank2.bank_id());
assert_eq!(
vec![(2, 3), (1, 1)],
vec![(1, 1), (2, 3)],
prioritization_fee_cache.get_prioritization_fees(&[]),
);
assert_eq!(
vec![(2, 3), (1, 2)],
vec![(1, 2), (2, 3)],
prioritization_fee_cache.get_prioritization_fees(&[write_account_a]),
);
assert_eq!(
vec![(2, 4), (1, 2)],
vec![(1, 2), (2, 4)],
prioritization_fee_cache.get_prioritization_fees(&[write_account_b]),
);
assert_eq!(
vec![(2, 4), (1, 1)],
vec![(1, 1), (2, 4)],
prioritization_fee_cache.get_prioritization_fees(&[write_account_c]),
);
assert_eq!(
vec![(2, 4), (1, 2)],
vec![(1, 2), (2, 4)],
prioritization_fee_cache
.get_prioritization_fees(&[write_account_a, write_account_b]),
);
assert_eq!(
vec![(2, 4), (1, 2)],
vec![(1, 2), (2, 4)],
prioritization_fee_cache.get_prioritization_fees(&[
write_account_a,
write_account_b,
Expand All @@ -781,28 +786,28 @@ mod tests {
sync_update(&prioritization_fee_cache, bank3.clone(), txs.iter());
// before block is marked as completed
assert_eq!(
vec![(2, 3), (1, 1)],
vec![(1, 1), (2, 3)],
prioritization_fee_cache.get_prioritization_fees(&[]),
);
assert_eq!(
vec![(2, 3), (1, 2)],
vec![(1, 2), (2, 3)],
prioritization_fee_cache.get_prioritization_fees(&[write_account_a]),
);
assert_eq!(
vec![(2, 4), (1, 2)],
vec![(1, 2), (2, 4)],
prioritization_fee_cache.get_prioritization_fees(&[write_account_b]),
);
assert_eq!(
vec![(2, 4), (1, 1)],
vec![(1, 1), (2, 4)],
prioritization_fee_cache.get_prioritization_fees(&[write_account_c]),
);
assert_eq!(
vec![(2, 4), (1, 2)],
vec![(1, 2), (2, 4)],
prioritization_fee_cache
.get_prioritization_fees(&[write_account_a, write_account_b]),
);
assert_eq!(
vec![(2, 4), (1, 2)],
vec![(1, 2), (2, 4)],
prioritization_fee_cache.get_prioritization_fees(&[
write_account_a,
write_account_b,
Expand All @@ -812,28 +817,28 @@ mod tests {
// after block is completed
sync_finalize_priority_fee_for_test(&prioritization_fee_cache, 3, bank3.bank_id());
assert_eq!(
vec![(3, 5), (2, 3), (1, 1)],
vec![(1, 1), (2, 3), (3, 5)],
prioritization_fee_cache.get_prioritization_fees(&[]),
);
assert_eq!(
vec![(3, 6), (2, 3), (1, 2)],
vec![(1, 2), (2, 3), (3, 6)],
prioritization_fee_cache.get_prioritization_fees(&[write_account_a]),
);
assert_eq!(
vec![(3, 5), (2, 4), (1, 2)],
vec![(1, 2), (2, 4), (3, 5)],
prioritization_fee_cache.get_prioritization_fees(&[write_account_b]),
);
assert_eq!(
vec![(3, 6), (2, 4), (1, 1)],
vec![(1, 1), (2, 4), (3, 6)],
prioritization_fee_cache.get_prioritization_fees(&[write_account_c]),
);
assert_eq!(
vec![(3, 6), (2, 4), (1, 2)],
vec![(1, 2), (2, 4), (3, 6)],
prioritization_fee_cache
.get_prioritization_fees(&[write_account_a, write_account_b]),
);
assert_eq!(
vec![(3, 6), (2, 4), (1, 2)],
vec![(1, 2), (2, 4), (3, 6)],
prioritization_fee_cache.get_prioritization_fees(&[
write_account_a,
write_account_b,
Expand Down

0 comments on commit ba9c25c

Please sign in to comment.