-
Notifications
You must be signed in to change notification settings - Fork 4.3k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
When forwarding buffered packets, sort by fee-per-cu, then bucket by …
…account limit. then forward by bucket to up to 10 block-limit CUs
- Loading branch information
1 parent
5af8e7c
commit 8a29354
Showing
4 changed files
with
212 additions
and
11 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,147 @@ | ||
| use { | ||
| crate::{ | ||
| banking_stage::{TOTAL_BUFFERED_PACKETS}, | ||
| qos_service::QosService, | ||
| unprocessed_packet_batches::*, | ||
| }, | ||
| solana_perf::{ | ||
| packet::{Packet}, | ||
| }, | ||
| solana_runtime::{ | ||
| bank::Bank, | ||
| cost_model::TransactionCost, | ||
| cost_tracker::CostTracker, | ||
| }, | ||
| solana_sdk::{ | ||
| transaction::{ | ||
| SanitizedTransaction, | ||
| }, | ||
| }, | ||
| std::{ | ||
| sync::{ | ||
| Arc, | ||
| }, | ||
| }, | ||
| }; | ||
|
|
||
| #[derive(Debug)] | ||
| pub struct PacketForwardManager<'a> { | ||
| cost_trackers: Vec<CostTracker>, | ||
| forwardable_packets: Vec<Vec<&'a Packet>>, | ||
| } | ||
|
|
||
| impl<'a> PacketForwardManager<'a> { | ||
| pub fn new( | ||
| max_forward_block_count: u64, | ||
| ) -> Self { | ||
| Self { | ||
| cost_trackers: Vec::<CostTracker>::with_capacity(max_forward_block_count as usize), | ||
| forwardable_packets: Vec::<Vec<&Packet>>::with_capacity(max_forward_block_count as usize), | ||
| } | ||
| } | ||
|
|
||
| pub fn take( | ||
| &mut self, | ||
| unprocessed_packet_batches: &'a UnprocessedPacketBatches, | ||
| working_bank: Arc<Bank>, | ||
| qos_service: &'a QosService, | ||
| ) -> Vec<&'a Packet> { | ||
| let prioritized_forwardable_packet_locators = Self::prioritize_unforwarded_packets_by_fee( | ||
| unprocessed_packet_batches, | ||
| Some(working_bank.clone()), | ||
| ); | ||
|
|
||
| // if we have a bank to work with, then sort packets by write-account buckets | ||
| let (transactions, sanitized_locators) = sanitize_transactions( | ||
| unprocessed_packet_batches, | ||
| &prioritized_forwardable_packet_locators, | ||
| &working_bank.feature_set, | ||
| working_bank.vote_only_bank(), | ||
| working_bank.as_ref(), | ||
| ); | ||
| let transactions_costs = qos_service.compute_transaction_costs(transactions.iter()); | ||
|
|
||
| transactions | ||
| .iter() | ||
| .zip(transactions_costs.iter()) | ||
| .zip(sanitized_locators.iter()) | ||
| .for_each(|((tx, cost), packet_locator)| { | ||
| if let Some(packet) = Self::locate_packet(unprocessed_packet_batches, packet_locator) { | ||
| self.sort_into_buckets(tx, cost, packet) | ||
| } | ||
| }); | ||
|
|
||
| let mut result = Vec::<&'a Packet>::new(); | ||
| self.forwardable_packets.iter().for_each(|v| result.extend(v)); | ||
| result | ||
| } | ||
|
|
||
| // prioritize unforwarded, unprocessed packets in buffered packet_batches by its fee/CU | ||
| fn prioritize_unforwarded_packets_by_fee( | ||
| unprocessed_packet_batches: &'a UnprocessedPacketBatches, | ||
| working_bank: Option<Arc<Bank>>, | ||
| ) -> Vec<PacketLocator> { | ||
| let mut locators = Vec::<PacketLocator>::with_capacity(TOTAL_BUFFERED_PACKETS); | ||
| unprocessed_packet_batches | ||
| .iter() | ||
| .filter(|deserialized_packet_batch| !deserialized_packet_batch.forwarded) | ||
| .enumerate() | ||
| .for_each(|(batch_index, deserialized_packet_batch)| { | ||
| deserialized_packet_batch | ||
| .unprocessed_packets | ||
| .keys() | ||
| .for_each(|packet_index| { | ||
| locators.push(PacketLocator { | ||
| batch_index, | ||
| packet_index: *packet_index, | ||
| }); | ||
| }) | ||
| }); | ||
|
|
||
| prioritize_by_fee( | ||
| unprocessed_packet_batches, | ||
| &locators, | ||
| working_bank, | ||
| ) | ||
| } | ||
|
|
||
| fn sort_into_buckets( | ||
| &mut self, | ||
| transaction: &SanitizedTransaction, | ||
| cost: &TransactionCost, | ||
| packet: &'a Packet, | ||
| ) { | ||
| // try to sort the `transaction` into one of outbound (virtual) blocks | ||
| self.cost_trackers | ||
| .iter_mut() | ||
| .zip(self.forwardable_packets.iter_mut()) | ||
| .for_each(|(cost_tracker, forwardable_packets)| { | ||
| match cost_tracker.try_add(transaction, cost) { | ||
| Ok(_) => { | ||
| forwardable_packets.push(packet); | ||
| return; | ||
| }, | ||
| Err(_) => {} | ||
| } | ||
| }); | ||
| } | ||
|
|
||
| fn locate_packet( | ||
| unprocessed_packet_batches: &'a UnprocessedPacketBatches, | ||
| locator: &PacketLocator | ||
| ) -> Option<&'a Packet> { | ||
| let deserialized_packet_batch = unprocessed_packet_batches.get(locator.batch_index)?; | ||
| deserialized_packet_batch.get_packet(locator.packet_index) | ||
| } | ||
| } | ||
|
|
||
| #[cfg(test)] | ||
| mod tests { | ||
| use { | ||
| super::*, | ||
| }; | ||
|
|
||
| #[test] | ||
| fn test_() { | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters