fn spawn_worker(&mut self, work: Work<T>, toolbox: Toolbox<T>) {
        let idle_tx = toolbox.idle_tx;
        let delayed_block_tx = toolbox.delayed_block_tx;

        // Wrap the `idle_tx` in a struct that will fire the idle message whenever it is dropped.
        //
        // This helps ensure that the worker is always freed in the case of an early exit or panic.
        // As such, this instantiation should happen as early in the function as possible.
        let send_idle_on_drop = SendOnDrop {
            tx: idle_tx,
            log: self.log.clone(),
        };

        let work_id = work.str_id();
        let worker_timer =
            metrics::start_timer_vec(&metrics::BEACON_PROCESSOR_WORKER_TIME, &[work_id]);
        metrics::inc_counter(&metrics::BEACON_PROCESSOR_WORKERS_SPAWNED_TOTAL);
        metrics::inc_counter_vec(
            &metrics::BEACON_PROCESSOR_WORK_EVENTS_STARTED_COUNT,
            &[work.str_id()],
        );

        let worker_id = self.current_workers;
        self.current_workers = self.current_workers.saturating_add(1);

        let chain = if let Some(chain) = self.beacon_chain.upgrade() {
            chain
        } else {
            debug!(
                self.log,
                "Beacon chain dropped, shutting down";
            );
            return;
        };

        let log = self.log.clone();
        let executor = self.executor.clone();

        let worker = Worker {
            chain,
            network_tx: self.network_tx.clone(),
            sync_tx: self.sync_tx.clone(),
            log: self.log.clone(),
        };

        trace!(
            self.log,
            "Spawning beacon processor worker";
            "work" => work_id,
            "worker" => worker_id,
        );

        executor.spawn_blocking(
            move || {
                let _worker_timer = worker_timer;

                match work {
                    /*
                     * Unaggregated attestation verification.
                     */
                    Work::GossipAttestation {
                        message_id,
                        peer_id,
                        attestation,
                        subnet_id,
                        should_import,
                        seen_timestamp,
                    } => worker.process_gossip_attestation(
                        message_id,
                        peer_id,
                        *attestation,
                        subnet_id,
                        should_import,
                        seen_timestamp,
                    ),
                    /*
                     * Aggregated attestation verification.
                     */
                    Work::GossipAggregate {
                        message_id,
                        peer_id,
                        aggregate,
                        seen_timestamp,
                    } => worker.process_gossip_aggregate(
                        message_id,
                        peer_id,
                        *aggregate,
                        seen_timestamp,
                    ),
                    /*
                     * Verification for beacon blocks received on gossip.
                     */
                    Work::GossipBlock {
                        message_id,
                        peer_id,
                        block,
                        seen_timestamp,
                    } => worker.process_gossip_block(
                        message_id,
                        peer_id,
                        *block,
                        delayed_block_tx,
                        seen_timestamp,
                    ),
                    /*
                     * Import for blocks that we received earlier than their intended slot.
                     */
                    Work::DelayedImportBlock {
                        peer_id,
                        block,
                        seen_timestamp,
                    } => worker.process_gossip_verified_block(peer_id, *block, seen_timestamp),
                    /*
                     * Voluntary exits received on gossip.
                     */
                    Work::GossipVoluntaryExit {
                        message_id,
                        peer_id,
                        voluntary_exit,
                    } => worker.process_gossip_voluntary_exit(message_id, peer_id, *voluntary_exit),
                    /*
                     * Proposer slashings received on gossip.
                     */
                    Work::GossipProposerSlashing {
                        message_id,
                        peer_id,
                        proposer_slashing,
                    } => worker.process_gossip_proposer_slashing(
                        message_id,
                        peer_id,
                        *proposer_slashing,
                    ),
                    /*
                     * Attester slashings received on gossip.
                     */
                    Work::GossipAttesterSlashing {
                        message_id,
                        peer_id,
                        attester_slashing,
                    } => worker.process_gossip_attester_slashing(
                        message_id,
                        peer_id,
                        *attester_slashing,
                    ),
                    /*
                     * Verification for beacon blocks received during syncing via RPC.
                     */
                    Work::RpcBlock { block, result_tx } => {
                        worker.process_rpc_block(*block, result_tx)
                    }
                    /*
                     * Verification for a chain segment (multiple blocks).
                     */
                    Work::ChainSegment { process_id, blocks } => {
                        worker.process_chain_segment(process_id, blocks)
                    }
                    /*
                     * Processing of Status Messages.
                     */
                    Work::Status { peer_id, message } => worker.process_status(peer_id, message),
                    /*
                     * Processing of range syncing requests from other peers.
                     */
                    Work::BlocksByRangeRequest {
                        peer_id,
                        request_id,
                        request,
                    } => worker.handle_blocks_by_range_request(peer_id, request_id, request),
                    /*
                     * Processing of blocks by roots requests from other peers.
                     */
                    Work::BlocksByRootsRequest {
                        peer_id,
                        request_id,
                        request,
                    } => worker.handle_blocks_by_root_request(peer_id, request_id, request),
                };

                trace!(
                    log,
                    "Beacon processor worker done";
                    "work" => work_id,
                    "worker" => worker_id,
                );

                // This explicit `drop` is used to remind the programmer that this variable must
                // not be dropped until the worker is complete. Dropping it early will cause the
                // worker to be marked as "free" and cause an over-spawning of workers.
                drop(send_idle_on_drop);
            },
            WORKER_TASK_NAME,
        );
    }
/// Spawns a blocking worker thread to process some `Work`.
///
/// Sends an message on `idle_tx` when the work is complete and the task is stopping.
Spawns a blocking worker thread to process some `Work`.
Sends an message on `idle_tx` when the work is complete and the task is stopping.
Spawns a blocking worker thread to process some ` Work ` . Sends an message on ` idle_tx ` when the work is complete and the task is stopping .
fn spawn_worker(&mut self, work: Work<T>, toolbox: Toolbox<T>) {
        let idle_tx = toolbox.idle_tx;
        let delayed_block_tx = toolbox.delayed_block_tx;
        // Wrap the `idle_tx` in a struct that will fire the idle message whenever it is dropped.
        //
        // This helps ensure that the worker is always freed in the case of an early exit or panic.
        // As such, this instantiation should happen as early in the function as possible.
        let send_idle_on_drop = SendOnDrop {
            tx: idle_tx,
            log: self.log.clone(),
        };
        let work_id = work.str_id();
        let worker_timer =
            metrics::start_timer_vec(&metrics::BEACON_PROCESSOR_WORKER_TIME, &[work_id]);
        metrics::inc_counter(&metrics::BEACON_PROCESSOR_WORKERS_SPAWNED_TOTAL);
        metrics::inc_counter_vec(
            &metrics::BEACON_PROCESSOR_WORK_EVENTS_STARTED_COUNT,
            &[work.str_id()],
        );
        let worker_id = self.current_workers;
        self.current_workers = self.current_workers.saturating_add(1);
        let chain = if let Some(chain) = self.beacon_chain.upgrade() {
            chain
        } else {
            debug!(
                self.log,
                "Beacon chain dropped, shutting down";
            );
            return;
        };
        let log = self.log.clone();
        let executor = self.executor.clone();
        let worker = Worker {
            chain,
            network_tx: self.network_tx.clone(),
            sync_tx: self.sync_tx.clone(),
            log: self.log.clone(),
        };
        trace!(
            self.log,
            "Spawning beacon processor worker";
            "work" => work_id,
            "worker" => worker_id,
        );
        executor.spawn_blocking(
            move || {
                let _worker_timer = worker_timer;
                match work {
                    /*
                     * Unaggregated attestation verification.
                     */
                    Work::GossipAttestation {
                        message_id,
                        peer_id,
                        attestation,
                        subnet_id,
                        should_import,
                        seen_timestamp,
                    } => worker.process_gossip_attestation(
                        message_id,
                        peer_id,
                        *attestation,
                        subnet_id,
                        should_import,
                        seen_timestamp,
                    ),
                    /*
                     * Aggregated attestation verification.
                     */
                    Work::GossipAggregate {
                        message_id,
                        peer_id,
                        aggregate,
                        seen_timestamp,
                    } => worker.process_gossip_aggregate(
                        message_id,
                        peer_id,
                        *aggregate,
                        seen_timestamp,
                    ),
                    /*
                     * Verification for beacon blocks received on gossip.
                     */
                    Work::GossipBlock {
                        message_id,
                        peer_id,
                        block,
                        seen_timestamp,
                    } => worker.process_gossip_block(
                        message_id,
                        peer_id,
                        *block,
                        delayed_block_tx,
                        seen_timestamp,
                    ),
                    /*
                     * Import for blocks that we received earlier than their intended slot.
                     */
                    Work::DelayedImportBlock {
                        peer_id,
                        block,
                        seen_timestamp,
                    } => worker.process_gossip_verified_block(peer_id, *block, seen_timestamp),
                    /*
                     * Voluntary exits received on gossip.
                     */
                    Work::GossipVoluntaryExit {
                        message_id,
                        peer_id,
                        voluntary_exit,
                    } => worker.process_gossip_voluntary_exit(message_id, peer_id, *voluntary_exit),
                    /*
                     * Proposer slashings received on gossip.
                     */
                    Work::GossipProposerSlashing {
                        message_id,
                        peer_id,
                        proposer_slashing,
                    } => worker.process_gossip_proposer_slashing(
                        message_id,
                        peer_id,
                        *proposer_slashing,
                    ),
                    /*
                     * Attester slashings received on gossip.
                     */
                    Work::GossipAttesterSlashing {
                        message_id,
                        peer_id,
                        attester_slashing,
                    } => worker.process_gossip_attester_slashing(
                        message_id,
                        peer_id,
                        *attester_slashing,
                    ),
                    /*
                     * Verification for beacon blocks received during syncing via RPC.
                     */
                    Work::RpcBlock { block, result_tx } => {
                        worker.process_rpc_block(*block, result_tx)
                    }
                    /*
                     * Verification for a chain segment (multiple blocks).
                     */
                    Work::ChainSegment { process_id, blocks } => {
                        worker.process_chain_segment(process_id, blocks)
                    }
                    /*
                     * Processing of Status Messages.
                     */
                    Work::Status { peer_id, message } => worker.process_status(peer_id, message),
                    /*
                     * Processing of range syncing requests from other peers.
                     */
                    Work::BlocksByRangeRequest {
                        peer_id,
                        request_id,
                        request,
                    } => worker.handle_blocks_by_range_request(peer_id, request_id, request),
                    /*
                     * Processing of blocks by roots requests from other peers.
                     */
                    Work::BlocksByRootsRequest {
                        peer_id,
                        request_id,
                        request,
                    } => worker.handle_blocks_by_root_request(peer_id, request_id, request),
                };
                trace!(
                    log,
                    "Beacon processor worker done";
                    "work" => work_id,
                    "worker" => worker_id,
                );
                // This explicit `drop` is used to remind the programmer that this variable must
                // not be dropped until the worker is complete. Dropping it early will cause the
                // worker to be marked as "free" and cause an over-spawning of workers.
                drop(send_idle_on_drop);
            },
            WORKER_TASK_NAME,
        );
    }
fn spawn_worker ( & mut self , work : Work < T > , toolbox : Toolbox < T > ) { let idle_tx = toolbox . idle_tx ; let delayed_block_tx = toolbox . delayed_block_tx ; // Wrap the `idle_tx` in a struct that will fire the idle message whenever it is dropped. // // This helps ensure that the worker is always freed in the case of an early exit or panic. // As such, this instantiation should happen as early in the function as possible. let send_idle_on_drop = SendOnDrop { tx : idle_tx , log : self . log . clone ( ) , } ; let work_id = work . str_id ( ) ; let worker_timer = metrics :: start_timer_vec ( & metrics :: BEACON_PROCESSOR_WORKER_TIME , & [ work_id ] ) ; metrics :: inc_counter ( & metrics :: BEACON_PROCESSOR_WORKERS_SPAWNED_TOTAL ) ; metrics :: inc_counter_vec ( & metrics :: BEACON_PROCESSOR_WORK_EVENTS_STARTED_COUNT , & [ work . str_id ( ) ] , ) ; let worker_id = self . current_workers ; self . current_workers = self . current_workers . saturating_add ( 1 ) ; let chain = if let Some ( chain ) = self . beacon_chain . upgrade ( ) { chain } else { debug ! ( self log " " ) ; return ; } ; let log = self . log . clone ( ) ; let executor = self . executor . clone ( ) ; let worker = Worker { chain , network_tx : self . network_tx . clone ( ) , sync_tx : self . sync_tx . clone ( ) , log : self . log . clone ( ) , } ; trace ! ( self log " " " " work_id " " worker_id ) ; executor . spawn_blocking ( move | | { let _worker_timer = worker_timer ; match work { /*
                     * Unaggregated attestation verification.
                     */ Work :: GossipAttestation { message_id , peer_id , attestation , subnet_id , should_import , seen_timestamp , } => worker . process_gossip_attestation ( message_id , peer_id , * attestation , subnet_id , should_import , seen_timestamp , ) , /*
                     * Aggregated attestation verification.
                     */ Work :: GossipAggregate { message_id , peer_id , aggregate , seen_timestamp , } => worker . process_gossip_aggregate ( message_id , peer_id , * aggregate , seen_timestamp , ) , /*
                     * Verification for beacon blocks received on gossip.
                     */ Work :: GossipBlock { message_id , peer_id , block , seen_timestamp , } => worker . process_gossip_block ( message_id , peer_id , * block , delayed_block_tx , seen_timestamp , ) , /*
                     * Import for blocks that we received earlier than their intended slot.
                     */ Work :: DelayedImportBlock { peer_id , block , seen_timestamp , } => worker . process_gossip_verified_block ( peer_id , * block , seen_timestamp ) , /*
                     * Voluntary exits received on gossip.
                     */ Work :: GossipVoluntaryExit { message_id , peer_id , voluntary_exit , } => worker . process_gossip_voluntary_exit ( message_id , peer_id , * voluntary_exit ) , /*
                     * Proposer slashings received on gossip.
                     */ Work :: GossipProposerSlashing { message_id , peer_id , proposer_slashing , } => worker . process_gossip_proposer_slashing ( message_id , peer_id , * proposer_slashing , ) , /*
                     * Attester slashings received on gossip.
                     */ Work :: GossipAttesterSlashing { message_id , peer_id , attester_slashing , } => worker . process_gossip_attester_slashing ( message_id , peer_id , * attester_slashing , ) , /*
                     * Verification for beacon blocks received during syncing via RPC.
                     */ Work :: RpcBlock { block , result_tx } => { worker . process_rpc_block ( * block , result_tx ) } /*
                     * Verification for a chain segment (multiple blocks).
                     */ Work :: ChainSegment { process_id , blocks } => { worker . process_chain_segment ( process_id , blocks ) } /*
                     * Processing of Status Messages.
                     */ Work :: Status { peer_id , message } => worker . process_status ( peer_id , message ) , /*
                     * Processing of range syncing requests from other peers.
                     */ Work :: BlocksByRangeRequest { peer_id , request_id , request , } => worker . handle_blocks_by_range_request ( peer_id , request_id , request ) , /*
                     * Processing of blocks by roots requests from other peers.
                     */ Work :: BlocksByRootsRequest { peer_id , request_id , request , } => worker . handle_blocks_by_root_request ( peer_id , request_id , request ) , } ; trace ! ( log " " " " work_id " " worker_id ) ; // This explicit `drop` is used to remind the programmer that this variable must // not be dropped until the worker is complete. Dropping it early will cause the // worker to be marked as "free" and cause an over-spawning of workers. drop ( send_idle_on_drop ) ; } , WORKER_TASK_NAME , ) ; }
Spawns a blocking worker thread to process some `Work`.
Spawns a blocking worker thread to process some ` Work ` .
&mut self
work
Work<T>
toolbox
Toolbox<T>