Skip to content
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

optimize mapping of produced block #382

Merged
merged 2 commits into from
Apr 3, 2024
Merged

optimize mapping of produced block #382

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
4909e14
optimize mapping of produced block
grooviegermanikus Apr 2, 2024
dc86430
remove the owned pattern
grooviegermanikus Apr 3, 2024
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
33 changes: 25 additions & 8 deletions cluster-endpoints/src/grpc_multiplex.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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!(
Expand All @@ -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 {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

minor: avoid one level of indentation with

let Some(slot) = extracted-slot else {
    continue 'recv_loop;
}

// 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)
Expand Down Expand Up @@ -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),
Expand Down
Loading