Skip to content

prioritization fee cache: remove lru crate #30

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
CriesofCarrots merged 1 commit into from
Mar 27, 2024
Merged

prioritization fee cache: remove lru crate #30

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 0 additions & 1 deletion Cargo.lock
View file
Open in desktop

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
Open in desktop

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
Open in desktop
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
Open in desktop
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>) {
Comment thread
fanatid marked this conversation as resolved.
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
Open in desktop
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