blockworks-foundation
/
lite-rpc
mirror of https://github.com/blockworks-foundation/lite-rpc.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

optimize mapping of produced block (#382) 

optimize mapping of produced block
This commit is contained in:
Groovie | Mango 2024-04-03 10:40:53 +02:00 committed by GitHub
parent 06373c0844
commit 69d7dbb123
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 25 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

33
cluster-endpoints/src/grpc_multiplex.rs
View File

@ -49,7 +49,7 @@ fn create_grpc_multiplex_processed_block_task(
let jh_merging_streams = tokio::task::spawn(async move {
let mut slots_processed = BTreeSet::<u64>::new();
let mut last_tick = Instant::now();
loop {
'recv_loop: loop {
// recv loop
if last_tick.elapsed() > Duration::from_millis(800) {
warn!(
@ -70,16 +70,23 @@ fn create_grpc_multiplex_processed_block_task(
};
match blocks_rx_result {
Some(Message::GeyserSubscribeUpdate(subscribe_update)) => {
let mapfilter =
map_block_from_yellowstone_update(*subscribe_update, COMMITMENT_CONFIG);
if let Some((slot, produced_block)) = mapfilter {
assert_eq!(COMMITMENT_CONFIG, produced_block.commitment_config);
// note: avoid mapping of full block as long as possible
let extracted_slot = extract_slot_from_yellowstone_update(&subscribe_update);
if let Some(slot) = extracted_slot {
// check if the slot is in the map, if not check if the container is half full and the slot in question is older than the lowest value
// it means that the slot is too old to process
if !slots_processed.contains(&slot)
&& (slots_processed.len() < MAX_SIZE / 2
|| slot > slots_processed.first().cloned().unwrap_or_default())
if slots_processed.contains(&slot) {
continue 'recv_loop;
}
if slots_processed.len() >= MAX_SIZE / 2
&& slot <= slots_processed.first().cloned().unwrap_or_default()
{
continue 'recv_loop;
}
let mapfilter =
map_block_from_yellowstone_update(*subscribe_update, COMMITMENT_CONFIG);
if let Some((_slot, produced_block)) = mapfilter {
let send_started_at = Instant::now();
let send_result = block_sender
.send(produced_block)
@ -552,6 +559,16 @@ pub fn create_grpc_multiplex_processed_slots_subscription(
(multiplexed_messages_rx, jh_multiplex_task)
}
fn extract_slot_from_yellowstone_update(update: &SubscribeUpdate) -> Option<Slot> {
match &update.update_oneof {
// list is not exhaustive
Some(UpdateOneof::Slot(update_message)) => Some(update_message.slot),
Some(UpdateOneof::BlockMeta(update_message)) => Some(update_message.slot),
Some(UpdateOneof::Block(update_message)) => Some(update_message.slot),
_ => None,
}
}
fn map_slot_from_yellowstone_update(update: SubscribeUpdate) -> Option<Slot> {
match update.update_oneof {
Some(UpdateOneof::Slot(update_slot_message)) => Some(update_slot_message.slot),