pub fn spawn_block_delay_queue<T: BeaconChainTypes>(
    ready_blocks_tx: Sender<QueuedBlock<T>>,
    executor: &TaskExecutor,
    slot_clock: T::SlotClock,
    log: Logger,
) -> Sender<QueuedBlock<T>> {
    let (early_blocks_tx, early_blocks_rx): (_, Receiver<QueuedBlock<_>>) =
        mpsc::channel(MAX_DELAYED_BLOCK_QUEUE_LEN);

    let queue_future = async move {
        let mut queued_block_roots = HashSet::new();

        let mut inbound_events = InboundEvents {
            early_blocks_rx,
            delay_queue: DelayQueue::new(),
        };

        loop {
            match inbound_events.next().await {
                // Some block has been indicated as "early" and should be processed when the
                // appropriate slot arrives.
                Some(InboundEvent::EarlyBlock(early_block)) => {
                    let block_slot = early_block.block.block.slot();
                    let block_root = early_block.block.block_root;

                    // Don't add the same block to the queue twice. This prevents DoS attacks.
                    if queued_block_roots.contains(&block_root) {
                        continue;
                    }

                    if let Some(duration_till_slot) = slot_clock.duration_to_slot(block_slot) {
                        // Check to ensure this won't over-fill the queue.
                        if queued_block_roots.len() >= MAXIMUM_QUEUED_BLOCKS {
                            error!(
                                log,
                                "Early blocks queue is full";
                                "queue_size" => MAXIMUM_QUEUED_BLOCKS,
                                "msg" => "check system clock"
                            );
                            // Drop the block.
                            continue;
                        }

                        queued_block_roots.insert(block_root);
                        // Queue the block until the start of the appropriate slot, plus
                        // `ADDITIONAL_DELAY`.
                        inbound_events
                            .delay_queue
                            .insert(early_block, duration_till_slot + ADDITIONAL_DELAY);
                    } else {
                        // If there is no duration till the next slot, check to see if the slot
                        // has already arrived. If it has already arrived, send it out for
                        // immediate processing.
                        //
                        // If we can't read the slot or the slot hasn't arrived, simply drop the
                        // block.
                        //
                        // This logic is slightly awkward since `SlotClock::duration_to_slot`
                        // doesn't distinguish between a slot that has already arrived and an
                        // error reading the slot clock.
                        if let Some(now) = slot_clock.now() {
                            if block_slot <= now && ready_blocks_tx.try_send(early_block).is_err() {
                                error!(
                                    log,
                                    "Failed to send block";
                                );
                            }
                        }
                    }
                }
                // A block that was queued for later processing is now ready to be processed.
                Some(InboundEvent::ReadyBlock(ready_block)) => {
                    let block_root = ready_block.block.block_root;

                    if !queued_block_roots.remove(&block_root) {
                        // Log an error to alert that we've made a bad assumption about how this
                        // program works, but still process the block anyway.
                        error!(
                            log,
                            "Unknown block in delay queue";
                            "block_root" => ?block_root
                        );
                    }

                    if ready_blocks_tx.try_send(ready_block).is_err() {
                        error!(
                            log,
                            "Failed to pop queued block";
                        );
                    }
                }
                Some(InboundEvent::DelayQueueError(e)) => crit!(
                    log,
                    "Failed to poll block delay queue";
                    "e" => ?e
                ),
                None => {
                    debug!(
                        log,
                        "Block delay queue stopped";
                        "msg" => "shutting down"
                    );
                    break;
                }
            }
        }
    };

    executor.spawn(queue_future, TASK_NAME);

    early_blocks_tx
}
/// Spawn a queue which will accept blocks via the returned `Sender`, potentially queue them until
/// their slot arrives, then send them back out via `ready_blocks_tx`.
Spawn a queue which will accept blocks via the returned `Sender`, potentially queue them until
their slot arrives, then send them back out via `ready_blocks_tx`.
Spawn a queue which will accept blocks via the returned ` Sender ` potentially queue them until their slot arrives then send them back out via ` ready_blocks_tx ` .
pub fn spawn_block_delay_queue<T: BeaconChainTypes>(
    ready_blocks_tx: Sender<QueuedBlock<T>>,
    executor: &TaskExecutor,
    slot_clock: T::SlotClock,
    log: Logger,
) -> Sender<QueuedBlock<T>> {
    let (early_blocks_tx, early_blocks_rx): (_, Receiver<QueuedBlock<_>>) =
        mpsc::channel(MAX_DELAYED_BLOCK_QUEUE_LEN);
    let queue_future = async move {
        let mut queued_block_roots = HashSet::new();
        let mut inbound_events = InboundEvents {
            early_blocks_rx,
            delay_queue: DelayQueue::new(),
        };
        loop {
            match inbound_events.next().await {
                // Some block has been indicated as "early" and should be processed when the
                // appropriate slot arrives.
                Some(InboundEvent::EarlyBlock(early_block)) => {
                    let block_slot = early_block.block.block.slot();
                    let block_root = early_block.block.block_root;
                    // Don't add the same block to the queue twice. This prevents DoS attacks.
                    if queued_block_roots.contains(&block_root) {
                        continue;
                    }
                    if let Some(duration_till_slot) = slot_clock.duration_to_slot(block_slot) {
                        // Check to ensure this won't over-fill the queue.
                        if queued_block_roots.len() >= MAXIMUM_QUEUED_BLOCKS {
                            error!(
                                log,
                                "Early blocks queue is full";
                                "queue_size" => MAXIMUM_QUEUED_BLOCKS,
                                "msg" => "check system clock"
                            );
                            // Drop the block.
                            continue;
                        }
                        queued_block_roots.insert(block_root);
                        // Queue the block until the start of the appropriate slot, plus
                        // `ADDITIONAL_DELAY`.
                        inbound_events
                            .delay_queue
                            .insert(early_block, duration_till_slot + ADDITIONAL_DELAY);
                    } else {
                        // If there is no duration till the next slot, check to see if the slot
                        // has already arrived. If it has already arrived, send it out for
                        // immediate processing.
                        //
                        // If we can't read the slot or the slot hasn't arrived, simply drop the
                        // block.
                        //
                        // This logic is slightly awkward since `SlotClock::duration_to_slot`
                        // doesn't distinguish between a slot that has already arrived and an
                        // error reading the slot clock.
                        if let Some(now) = slot_clock.now() {
                            if block_slot <= now && ready_blocks_tx.try_send(early_block).is_err() {
                                error!(
                                    log,
                                    "Failed to send block";
                                );
                            }
                        }
                    }
                }
                // A block that was queued for later processing is now ready to be processed.
                Some(InboundEvent::ReadyBlock(ready_block)) => {
                    let block_root = ready_block.block.block_root;
                    if !queued_block_roots.remove(&block_root) {
                        // Log an error to alert that we've made a bad assumption about how this
                        // program works, but still process the block anyway.
                        error!(
                            log,
                            "Unknown block in delay queue";
                            "block_root" => ?block_root
                        );
                    }
                    if ready_blocks_tx.try_send(ready_block).is_err() {
                        error!(
                            log,
                            "Failed to pop queued block";
                        );
                    }
                }
                Some(InboundEvent::DelayQueueError(e)) => crit!(
                    log,
                    "Failed to poll block delay queue";
                    "e" => ?e
                ),
                None => {
                    debug!(
                        log,
                        "Block delay queue stopped";
                        "msg" => "shutting down"
                    );
                    break;
                }
            }
        }
    };
    executor.spawn(queue_future, TASK_NAME);
    early_blocks_tx
}
pub fn spawn_block_delay_queue < T : BeaconChainTypes > ( ready_blocks_tx : Sender < QueuedBlock < T > > , executor : & TaskExecutor , slot_clock : T :: SlotClock , log : Logger , ) -> Sender < QueuedBlock < T > > { let ( early_blocks_tx , early_blocks_rx ) : ( _ , Receiver < QueuedBlock < _ > > ) = mpsc :: channel ( MAX_DELAYED_BLOCK_QUEUE_LEN ) ; let queue_future = async move { let mut queued_block_roots = HashSet :: new ( ) ; let mut inbound_events = InboundEvents { early_blocks_rx , delay_queue : DelayQueue :: new ( ) , } ; loop { match inbound_events . next ( ) . await { // Some block has been indicated as "early" and should be processed when the // appropriate slot arrives. Some ( InboundEvent :: EarlyBlock ( early_block ) ) => { let block_slot = early_block . block . block . slot ( ) ; let block_root = early_block . block . block_root ; // Don't add the same block to the queue twice. This prevents DoS attacks. if queued_block_roots . contains ( & block_root ) { continue ; } if let Some ( duration_till_slot ) = slot_clock . duration_to_slot ( block_slot ) { // Check to ensure this won't over-fill the queue. if queued_block_roots . len ( ) >= MAXIMUM_QUEUED_BLOCKS { error ! ( log " " " " MAXIMUM_QUEUED_BLOCKS " " " " ) ; // Drop the block. continue ; } queued_block_roots . insert ( block_root ) ; // Queue the block until the start of the appropriate slot, plus // `ADDITIONAL_DELAY`. inbound_events . delay_queue . insert ( early_block , duration_till_slot + ADDITIONAL_DELAY ) ; } else { // If there is no duration till the next slot, check to see if the slot // has already arrived. If it has already arrived, send it out for // immediate processing. // // If we can't read the slot or the slot hasn't arrived, simply drop the // block. // // This logic is slightly awkward since `SlotClock::duration_to_slot` // doesn't distinguish between a slot that has already arrived and an // error reading the slot clock. if let Some ( now ) = slot_clock . now ( ) { if block_slot <= now && ready_blocks_tx . try_send ( early_block ) . is_err ( ) { error ! ( log " " ) ; } } } } // A block that was queued for later processing is now ready to be processed. Some ( InboundEvent :: ReadyBlock ( ready_block ) ) => { let block_root = ready_block . block . block_root ; if ! queued_block_roots . remove ( & block_root ) { // Log an error to alert that we've made a bad assumption about how this // program works, but still process the block anyway. error ! ( log " " " " block_root ) ; } if ready_blocks_tx . try_send ( ready_block ) . is_err ( ) { error ! ( log " " ) ; } } Some ( InboundEvent :: DelayQueueError ( e ) ) => crit ! ( log " " " " e ) , None => { debug ! ( log " " " " " " ) ; break ; } } } } ; executor . spawn ( queue_future , TASK_NAME ) ; early_blocks_tx }
Spawn a queue which will accept blocks via the returned `Sender`, potentially queue them until
their slot arrives, then send them back out via `ready_blocks_tx`.
Spawn a queue which will accept blocks via the returned ` Sender ` potentially queue them until their slot arrives then send them back out via ` ready_blocks_tx ` .
ready_blocks_tx
Sender<QueuedBlock<T>>
executor
&TaskExecutor
slot_clock
T::SlotClock
log
Logger