gnosis/lighthouse
1
0
Fork 0
mirror of https://github.com/sigp/lighthouse.git synced 2026-05-30 04:37:13 +00:00
Code Issues Packages Projects Releases Wiki Activity

keyed by hash256

Browse Source
This commit is contained in:
Eitan Seri-Levi
2026-05-23 17:50:28 +03:00
parent 7f84ac18c7
commit 3febb7275c
1 changed files with 49 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files

114
beacon_node/beacon_processor/src/scheduler/work_reprocessing_queue.rs
View File

@@ -253,7 +253,7 @@ enum InboundEvent {
/// A column reconstruction that was queued is ready for processing. /// A column reconstruction that was queued is ready for processing.
ReadyColumnReconstruction(QueuedColumnReconstruction), ReadyColumnReconstruction(QueuedColumnReconstruction),
/// A gossip data column that timed out waiting for its block. /// A gossip data column that timed out waiting for its block.
ReadyDataColumn(usize), ReadyDataColumn(Hash256),
/// A message sent to the `ReprocessQueue` /// A message sent to the `ReprocessQueue`
Msg(ReprocessQueueMessage), Msg(ReprocessQueueMessage),
} }
@@ -278,7 +278,8 @@ struct ReprocessQueue<S> {
lc_updates_delay_queue: DelayQueue<QueuedLightClientUpdateId>, lc_updates_delay_queue: DelayQueue<QueuedLightClientUpdateId>,
/// Queue to manage scheduled column reconstructions. /// Queue to manage scheduled column reconstructions.
column_reconstructions_delay_queue: DelayQueue<QueuedColumnReconstruction>, column_reconstructions_delay_queue: DelayQueue<QueuedColumnReconstruction>,
data_columns_delay_queue: DelayQueue<usize>, /// Queue to manage gossip data column timeouts.
data_columns_delay_queue: DelayQueue<Hash256>,
/* Queued items */ /* Queued items */
/// Queued blocks. /// Queued blocks.
@@ -299,15 +300,12 @@ struct ReprocessQueue<S> {
queued_column_reconstructions: HashMap<Hash256, Option<DelayKey>>, queued_column_reconstructions: HashMap<Hash256, Option<DelayKey>>,
/// Queued backfill batches /// Queued backfill batches
queued_backfill_batches: Vec<QueuedBackfillBatch>, queued_backfill_batches: Vec<QueuedBackfillBatch>,
/// Queued gossip data columns awaiting their block. /// Queued gossip data columns awaiting their block, keyed by block root.
queued_gossip_data_columns: FnvHashMap<usize, (QueuedGossipDataColumn, DelayKey)>, queued_gossip_data_columns: HashMap<Hash256, (Vec<QueuedGossipDataColumn>, DelayKey)>,
/// Data columns per block root.
awaiting_data_columns_per_root: HashMap<Hash256, Vec<usize>>,
/* Aux */ /* Aux */
/// Next attestation id, used for both aggregated and unaggregated attestations /// Next attestation id, used for both aggregated and unaggregated attestations
next_attestation: usize, next_attestation: usize,
next_data_column: usize,
next_lc_update: usize, next_lc_update: usize,
early_block_debounce: TimeLatch, early_block_debounce: TimeLatch,
envelope_delay_debounce: TimeLatch, envelope_delay_debounce: TimeLatch,
@@ -408,8 +406,8 @@ impl<S: SlotClock> Stream for ReprocessQueue<S> {
} }
match self.data_columns_delay_queue.poll_expired(cx) { match self.data_columns_delay_queue.poll_expired(cx) {
Poll::Ready(Some(col_id)) => { Poll::Ready(Some(block_root)) => {
return Poll::Ready(Some(InboundEvent::ReadyDataColumn(col_id.into_inner()))); return Poll::Ready(Some(InboundEvent::ReadyDataColumn(block_root.into_inner())));
} }
Poll::Ready(None) | Poll::Pending => (), Poll::Ready(None) | Poll::Pending => (),
} }
@@ -492,10 +490,8 @@ impl<S: SlotClock> ReprocessQueue<S> {
awaiting_lc_updates_per_parent_root: HashMap::new(), awaiting_lc_updates_per_parent_root: HashMap::new(),
queued_backfill_batches: Vec::new(), queued_backfill_batches: Vec::new(),
queued_column_reconstructions: HashMap::new(), queued_column_reconstructions: HashMap::new(),
queued_gossip_data_columns: FnvHashMap::default(), queued_gossip_data_columns: HashMap::new(),
awaiting_data_columns_per_root: HashMap::new(),
next_attestation: 0, next_attestation: 0,
next_data_column: 0,
next_lc_update: 0, next_lc_update: 0,
early_block_debounce: TimeLatch::default(), early_block_debounce: TimeLatch::default(),
envelope_delay_debounce: TimeLatch::default(), envelope_delay_debounce: TimeLatch::default(),
@@ -720,27 +716,25 @@ impl<S: SlotClock> ReprocessQueue<S> {
self.next_attestation += 1; self.next_attestation += 1;
} }
InboundEvent::Msg(UnknownBlockDataColumn(queued_data_column)) => { InboundEvent::Msg(UnknownBlockDataColumn(queued_data_column)) => {
if self.queued_gossip_data_columns.len() >= MAXIMUM_QUEUED_DATA_COLUMNS { let block_root = queued_data_column.beacon_block_root;
return;
if let Some((columns, _delay_key)) =
self.queued_gossip_data_columns.get_mut(&block_root)
{
// Append to existing entry; the timer for this root is already running.
columns.push(queued_data_column);
} else {
if self.queued_gossip_data_columns.len() >= MAXIMUM_QUEUED_DATA_COLUMNS {
return;
}
let delay_key = self
.data_columns_delay_queue
.insert(block_root, QUEUED_ATTESTATION_DELAY);
self.queued_gossip_data_columns
.insert(block_root, (vec![queued_data_column], delay_key));
} }
let col_id = self.next_data_column;
let delay_key = self
.data_columns_delay_queue
.insert(col_id, QUEUED_ATTESTATION_DELAY);
// Register this column for the corresponding block root.
self.awaiting_data_columns_per_root
.entry(queued_data_column.beacon_block_root)
.or_default()
.push(col_id);
// Store the column and its info.
self.queued_gossip_data_columns
.insert(col_id, (queued_data_column, delay_key));
self.next_data_column += 1;
} }
InboundEvent::Msg(UnknownLightClientOptimisticUpdate( InboundEvent::Msg(UnknownLightClientOptimisticUpdate(
queued_light_client_optimistic_update, queued_light_client_optimistic_update,
@@ -856,19 +850,17 @@ impl<S: SlotClock> ReprocessQueue<S> {
} }
// Unqueue the data columns we have for this root, if any. // Unqueue the data columns we have for this root, if any.
if let Some(queued_ids) = self.awaiting_data_columns_per_root.remove(&block_root) { if let Some((data_columns, delay_key)) =
for col_id in queued_ids { self.queued_gossip_data_columns.remove(&block_root)
if let Some((data_column, delay_key)) = {
self.queued_gossip_data_columns.remove(&col_id) self.data_columns_delay_queue.remove(&delay_key);
for data_column in data_columns {
if self
.ready_work_tx
.try_send(ReadyWork::DataColumn(data_column))
.is_err()
{ {
self.data_columns_delay_queue.remove(&delay_key); error!(?block_root, "Failed to send data column for reprocessing");
if self
.ready_work_tx
.try_send(ReadyWork::DataColumn(data_column))
.is_err()
{
error!(?block_root, "Failed to send data column for reprocessing");
}
} }
} }
} }
@@ -1125,29 +1117,21 @@ impl<S: SlotClock> ReprocessQueue<S> {
); );
} }
} }
InboundEvent::ReadyDataColumn(col_id) => { InboundEvent::ReadyDataColumn(block_root) => {
if let Some((data_column, _)) = self.queued_gossip_data_columns.remove(&col_id) { if let Some((data_columns, _)) = self.queued_gossip_data_columns.remove(&block_root)
// Clean up the per-root index. {
let root = data_column.beacon_block_root; for data_column in data_columns {
if let Entry::Occupied(mut entry) = if self
self.awaiting_data_columns_per_root.entry(root) .ready_work_tx
{ .try_send(ReadyWork::DataColumn(data_column))
let ids = entry.get_mut(); .is_err()
ids.retain(|&id| id != col_id); {
if ids.is_empty() { error!(
entry.remove_entry(); hint = "system may be overloaded",
"Ignored expired gossip data column"
);
} }
} }
if self
.ready_work_tx
.try_send(ReadyWork::DataColumn(data_column))
.is_err()
{
error!(
hint = "system may be overloaded",
"Ignored expired gossip data column"
);
}
} }
} }
} }