From 1ec30416738e10e110e0731e9800e0c69dade5df Mon Sep 17 00:00:00 2001 From: Daniel Ramirez Chiquillo Date: Wed, 15 Mar 2023 01:27:47 +0000 Subject: [PATCH] Remove Router/Processor Code (#4002) ## Issue Addressed #3938 ## Proposed Changes - `network::Processor` is deleted and all it's logic is moved to `network::Router`. - The `network::Router` module is moved to a single file. - The following functions are deleted: `on_disconnect` `send_status` `on_status_response` `on_blocks_by_root_request` `on_lightclient_bootstrap` `on_blocks_by_range_request` `on_block_gossip` `on_unaggregated_attestation_gossip` `on_aggregated_attestation_gossip` `on_voluntary_exit_gossip` `on_proposer_slashing_gossip` `on_attester_slashing_gossip` `on_sync_committee_signature_gossip` `on_sync_committee_contribution_gossip` `on_light_client_finality_update_gossip` `on_light_client_optimistic_update_gossip`. This deletions are possible because the updated `Router` allows the underlying methods to be called directly. --- beacon_node/network/src/router.rs | 535 ++++++++++++++++++++ beacon_node/network/src/router/mod.rs | 321 ------------ beacon_node/network/src/router/processor.rs | 472 ----------------- beacon_node/network/src/service.rs | 2 +- 4 files changed, 536 insertions(+), 794 deletions(-) create mode 100644 beacon_node/network/src/router.rs delete mode 100644 beacon_node/network/src/router/mod.rs delete mode 100644 beacon_node/network/src/router/processor.rs diff --git a/beacon_node/network/src/router.rs b/beacon_node/network/src/router.rs new file mode 100644 index 000000000..7f75a27fe --- /dev/null +++ b/beacon_node/network/src/router.rs @@ -0,0 +1,535 @@ +//! This module handles incoming network messages. +//! +//! It routes the messages to appropriate services. +//! It handles requests at the application layer in its associated processor and directs +//! syncing-related responses to the Sync manager. +#![allow(clippy::unit_arg)] + +use crate::beacon_processor::{ + BeaconProcessor, WorkEvent as BeaconWorkEvent, MAX_WORK_EVENT_QUEUE_LEN, +}; +use crate::error; +use crate::service::{NetworkMessage, RequestId}; +use crate::status::status_message; +use crate::sync::manager::RequestId as SyncId; +use crate::sync::SyncMessage; +use beacon_chain::{BeaconChain, BeaconChainTypes}; +use futures::prelude::*; +use lighthouse_network::rpc::*; +use lighthouse_network::{ + MessageId, NetworkGlobals, PeerId, PeerRequestId, PubsubMessage, Request, Response, +}; +use slog::{debug, o, trace}; +use slog::{error, warn}; +use std::cmp; +use std::sync::Arc; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; +use tokio::sync::mpsc; +use tokio_stream::wrappers::UnboundedReceiverStream; +use types::{EthSpec, SignedBeaconBlock}; + +/// Handles messages from the network and routes them to the appropriate service to be handled. +pub struct Router { + /// Access to the peer db and network information. + network_globals: Arc>, + /// A reference to the underlying beacon chain. + chain: Arc>, + /// A channel to the syncing thread. + sync_send: mpsc::UnboundedSender>, + /// A network context to return and handle RPC requests. + network: HandlerNetworkContext, + /// A multi-threaded, non-blocking processor for applying messages to the beacon chain. + beacon_processor_send: mpsc::Sender>, + /// The `Router` logger. + log: slog::Logger, +} + +/// Types of messages the router can receive. +#[derive(Debug)] +pub enum RouterMessage { + /// Peer has disconnected. + PeerDisconnected(PeerId), + /// An RPC request has been received. + RPCRequestReceived { + peer_id: PeerId, + id: PeerRequestId, + request: Request, + }, + /// An RPC response has been received. + RPCResponseReceived { + peer_id: PeerId, + request_id: RequestId, + response: Response, + }, + /// An RPC request failed + RPCFailed { + peer_id: PeerId, + request_id: RequestId, + }, + /// A gossip message has been received. The fields are: message id, the peer that sent us this + /// message, the message itself and a bool which indicates if the message should be processed + /// by the beacon chain after successful verification. + PubsubMessage(MessageId, PeerId, PubsubMessage, bool), + /// The peer manager has requested we re-status a peer. + StatusPeer(PeerId), +} + +impl Router { + /// Initializes and runs the Router. + pub fn spawn( + beacon_chain: Arc>, + network_globals: Arc>, + network_send: mpsc::UnboundedSender>, + executor: task_executor::TaskExecutor, + log: slog::Logger, + ) -> error::Result>> { + let message_handler_log = log.new(o!("service"=> "router")); + trace!(message_handler_log, "Service starting"); + + let (handler_send, handler_recv) = mpsc::unbounded_channel(); + + let (beacon_processor_send, beacon_processor_receive) = + mpsc::channel(MAX_WORK_EVENT_QUEUE_LEN); + + let sync_logger = log.new(o!("service"=> "sync")); + + // spawn the sync thread + let sync_send = crate::sync::manager::spawn( + executor.clone(), + beacon_chain.clone(), + network_globals.clone(), + network_send.clone(), + beacon_processor_send.clone(), + sync_logger, + ); + + BeaconProcessor { + beacon_chain: Arc::downgrade(&beacon_chain), + network_tx: network_send.clone(), + sync_tx: sync_send.clone(), + network_globals: network_globals.clone(), + executor: executor.clone(), + max_workers: cmp::max(1, num_cpus::get()), + current_workers: 0, + importing_blocks: Default::default(), + log: log.clone(), + } + .spawn_manager(beacon_processor_receive, None); + + // generate the Message handler + let mut handler = Router { + network_globals, + chain: beacon_chain, + sync_send, + network: HandlerNetworkContext::new(network_send, log.clone()), + beacon_processor_send, + log: message_handler_log, + }; + + // spawn handler task and move the message handler instance into the spawned thread + executor.spawn( + async move { + debug!(log, "Network message router started"); + UnboundedReceiverStream::new(handler_recv) + .for_each(move |msg| future::ready(handler.handle_message(msg))) + .await; + }, + "router", + ); + + Ok(handler_send) + } + + /// Handle all messages incoming from the network service. + fn handle_message(&mut self, message: RouterMessage) { + match message { + // we have initiated a connection to a peer or the peer manager has requested a + // re-status + RouterMessage::StatusPeer(peer_id) => { + self.send_status(peer_id); + } + // A peer has disconnected + RouterMessage::PeerDisconnected(peer_id) => { + self.send_to_sync(SyncMessage::Disconnect(peer_id)); + } + RouterMessage::RPCRequestReceived { + peer_id, + id, + request, + } => { + self.handle_rpc_request(peer_id, id, request); + } + RouterMessage::RPCResponseReceived { + peer_id, + request_id, + response, + } => { + self.handle_rpc_response(peer_id, request_id, response); + } + RouterMessage::RPCFailed { + peer_id, + request_id, + } => { + self.on_rpc_error(peer_id, request_id); + } + RouterMessage::PubsubMessage(id, peer_id, gossip, should_process) => { + self.handle_gossip(id, peer_id, gossip, should_process); + } + } + } + + /* RPC - Related functionality */ + + /// A new RPC request has been received from the network. + fn handle_rpc_request(&mut self, peer_id: PeerId, request_id: PeerRequestId, request: Request) { + if !self.network_globals.peers.read().is_connected(&peer_id) { + debug!(self.log, "Dropping request of disconnected peer"; "peer_id" => %peer_id, "request" => ?request); + return; + } + match request { + Request::Status(status_message) => { + self.on_status_request(peer_id, request_id, status_message) + } + Request::BlocksByRange(request) => self.send_beacon_processor_work( + BeaconWorkEvent::blocks_by_range_request(peer_id, request_id, request), + ), + Request::BlocksByRoot(request) => self.send_beacon_processor_work( + BeaconWorkEvent::blocks_by_roots_request(peer_id, request_id, request), + ), + Request::LightClientBootstrap(request) => self.send_beacon_processor_work( + BeaconWorkEvent::lightclient_bootstrap_request(peer_id, request_id, request), + ), + } + } + + /// An RPC response has been received from the network. + fn handle_rpc_response( + &mut self, + peer_id: PeerId, + request_id: RequestId, + response: Response, + ) { + match response { + Response::Status(status_message) => { + debug!(self.log, "Received Status Response"; "peer_id" => %peer_id, &status_message); + self.send_beacon_processor_work(BeaconWorkEvent::status_message( + peer_id, + status_message, + )) + } + Response::BlocksByRange(beacon_block) => { + self.on_blocks_by_range_response(peer_id, request_id, beacon_block); + } + Response::BlocksByRoot(beacon_block) => { + self.on_blocks_by_root_response(peer_id, request_id, beacon_block); + } + Response::LightClientBootstrap(_) => unreachable!(), + } + } + + /// Handle RPC messages. + /// Note: `should_process` is currently only useful for the `Attestation` variant. + /// if `should_process` is `false`, we only propagate the message on successful verification, + /// else, we propagate **and** import into the beacon chain. + fn handle_gossip( + &mut self, + message_id: MessageId, + peer_id: PeerId, + gossip_message: PubsubMessage, + should_process: bool, + ) { + match gossip_message { + PubsubMessage::AggregateAndProofAttestation(aggregate_and_proof) => self + .send_beacon_processor_work(BeaconWorkEvent::aggregated_attestation( + message_id, + peer_id, + *aggregate_and_proof, + timestamp_now(), + )), + PubsubMessage::Attestation(subnet_attestation) => { + self.send_beacon_processor_work(BeaconWorkEvent::unaggregated_attestation( + message_id, + peer_id, + subnet_attestation.1, + subnet_attestation.0, + should_process, + timestamp_now(), + )) + } + PubsubMessage::BeaconBlock(block) => { + self.send_beacon_processor_work(BeaconWorkEvent::gossip_beacon_block( + message_id, + peer_id, + self.network_globals.client(&peer_id), + block, + timestamp_now(), + )) + } + PubsubMessage::VoluntaryExit(exit) => { + debug!(self.log, "Received a voluntary exit"; "peer_id" => %peer_id); + self.send_beacon_processor_work(BeaconWorkEvent::gossip_voluntary_exit( + message_id, peer_id, exit, + )) + } + PubsubMessage::ProposerSlashing(proposer_slashing) => { + debug!( + self.log, + "Received a proposer slashing"; + "peer_id" => %peer_id + ); + self.send_beacon_processor_work(BeaconWorkEvent::gossip_proposer_slashing( + message_id, + peer_id, + proposer_slashing, + )) + } + PubsubMessage::AttesterSlashing(attester_slashing) => { + debug!( + self.log, + "Received a attester slashing"; + "peer_id" => %peer_id + ); + self.send_beacon_processor_work(BeaconWorkEvent::gossip_attester_slashing( + message_id, + peer_id, + attester_slashing, + )) + } + PubsubMessage::SignedContributionAndProof(contribution_and_proof) => { + trace!( + self.log, + "Received sync committee aggregate"; + "peer_id" => %peer_id + ); + self.send_beacon_processor_work(BeaconWorkEvent::gossip_sync_contribution( + message_id, + peer_id, + *contribution_and_proof, + timestamp_now(), + )) + } + PubsubMessage::SyncCommitteeMessage(sync_committtee_msg) => { + trace!( + self.log, + "Received sync committee signature"; + "peer_id" => %peer_id + ); + self.send_beacon_processor_work(BeaconWorkEvent::gossip_sync_signature( + message_id, + peer_id, + sync_committtee_msg.1, + sync_committtee_msg.0, + timestamp_now(), + )) + } + PubsubMessage::LightClientFinalityUpdate(light_client_finality_update) => { + trace!( + self.log, + "Received light client finality update"; + "peer_id" => %peer_id + ); + self.send_beacon_processor_work( + BeaconWorkEvent::gossip_light_client_finality_update( + message_id, + peer_id, + light_client_finality_update, + timestamp_now(), + ), + ) + } + PubsubMessage::LightClientOptimisticUpdate(light_client_optimistic_update) => { + trace!( + self.log, + "Received light client optimistic update"; + "peer_id" => %peer_id + ); + self.send_beacon_processor_work( + BeaconWorkEvent::gossip_light_client_optimistic_update( + message_id, + peer_id, + light_client_optimistic_update, + timestamp_now(), + ), + ) + } + PubsubMessage::BlsToExecutionChange(bls_to_execution_change) => self + .send_beacon_processor_work(BeaconWorkEvent::gossip_bls_to_execution_change( + message_id, + peer_id, + bls_to_execution_change, + )), + } + } + + fn send_status(&mut self, peer_id: PeerId) { + let status_message = status_message(&self.chain); + debug!(self.log, "Sending Status Request"; "peer" => %peer_id, &status_message); + self.network + .send_processor_request(peer_id, Request::Status(status_message)); + } + + fn send_to_sync(&mut self, message: SyncMessage) { + self.sync_send.send(message).unwrap_or_else(|e| { + warn!( + self.log, + "Could not send message to the sync service"; + "error" => %e, + ) + }); + } + + /// An error occurred during an RPC request. The state is maintained by the sync manager, so + /// this function notifies the sync manager of the error. + pub fn on_rpc_error(&mut self, peer_id: PeerId, request_id: RequestId) { + // Check if the failed RPC belongs to sync + if let RequestId::Sync(request_id) = request_id { + self.send_to_sync(SyncMessage::RpcError { + peer_id, + request_id, + }); + } + } + + /// Handle a `Status` request. + /// + /// Processes the `Status` from the remote peer and sends back our `Status`. + pub fn on_status_request( + &mut self, + peer_id: PeerId, + request_id: PeerRequestId, + status: StatusMessage, + ) { + debug!(self.log, "Received Status Request"; "peer_id" => %peer_id, &status); + + // Say status back. + self.network.send_response( + peer_id, + Response::Status(status_message(&self.chain)), + request_id, + ); + + self.send_beacon_processor_work(BeaconWorkEvent::status_message(peer_id, status)) + } + + /// Handle a `BlocksByRange` response from the peer. + /// A `beacon_block` behaves as a stream which is terminated on a `None` response. + pub fn on_blocks_by_range_response( + &mut self, + peer_id: PeerId, + request_id: RequestId, + beacon_block: Option>>, + ) { + let request_id = match request_id { + RequestId::Sync(sync_id) => match sync_id { + SyncId::SingleBlock { .. } | SyncId::ParentLookup { .. } => { + unreachable!("Block lookups do not request BBRange requests") + } + id @ (SyncId::BackFillSync { .. } | SyncId::RangeSync { .. }) => id, + }, + RequestId::Router => unreachable!("All BBRange requests belong to sync"), + }; + + trace!( + self.log, + "Received BlocksByRange Response"; + "peer" => %peer_id, + ); + + self.send_to_sync(SyncMessage::RpcBlock { + peer_id, + request_id, + beacon_block, + seen_timestamp: timestamp_now(), + }); + } + + /// Handle a `BlocksByRoot` response from the peer. + pub fn on_blocks_by_root_response( + &mut self, + peer_id: PeerId, + request_id: RequestId, + beacon_block: Option>>, + ) { + let request_id = match request_id { + RequestId::Sync(sync_id) => match sync_id { + id @ (SyncId::SingleBlock { .. } | SyncId::ParentLookup { .. }) => id, + SyncId::BackFillSync { .. } | SyncId::RangeSync { .. } => { + unreachable!("Batch syncing do not request BBRoot requests") + } + }, + RequestId::Router => unreachable!("All BBRoot requests belong to sync"), + }; + + trace!( + self.log, + "Received BlocksByRoot Response"; + "peer" => %peer_id, + ); + self.send_to_sync(SyncMessage::RpcBlock { + peer_id, + request_id, + beacon_block, + seen_timestamp: timestamp_now(), + }); + } + + fn send_beacon_processor_work(&mut self, work: BeaconWorkEvent) { + self.beacon_processor_send + .try_send(work) + .unwrap_or_else(|e| { + let work_type = match &e { + mpsc::error::TrySendError::Closed(work) + | mpsc::error::TrySendError::Full(work) => work.work_type(), + }; + error!(&self.log, "Unable to send message to the beacon processor"; + "error" => %e, "type" => work_type) + }) + } +} + +/// Wraps a Network Channel to employ various RPC related network functionality for the +/// processor. +#[derive(Clone)] +pub struct HandlerNetworkContext { + /// The network channel to relay messages to the Network service. + network_send: mpsc::UnboundedSender>, + /// Logger for the `NetworkContext`. + log: slog::Logger, +} + +impl HandlerNetworkContext { + pub fn new(network_send: mpsc::UnboundedSender>, log: slog::Logger) -> Self { + Self { network_send, log } + } + + /// Sends a message to the network task. + fn inform_network(&mut self, msg: NetworkMessage) { + self.network_send.send(msg).unwrap_or_else( + |e| warn!(self.log, "Could not send message to the network service"; "error" => %e), + ) + } + + /// Sends a request to the network task. + pub fn send_processor_request(&mut self, peer_id: PeerId, request: Request) { + self.inform_network(NetworkMessage::SendRequest { + peer_id, + request_id: RequestId::Router, + request, + }) + } + + /// Sends a response to the network task. + pub fn send_response(&mut self, peer_id: PeerId, response: Response, id: PeerRequestId) { + self.inform_network(NetworkMessage::SendResponse { + peer_id, + id, + response, + }) + } +} + +fn timestamp_now() -> Duration { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_else(|_| Duration::from_secs(0)) +} diff --git a/beacon_node/network/src/router/mod.rs b/beacon_node/network/src/router/mod.rs deleted file mode 100644 index 231f30f3e..000000000 --- a/beacon_node/network/src/router/mod.rs +++ /dev/null @@ -1,321 +0,0 @@ -//! This module handles incoming network messages. -//! -//! It routes the messages to appropriate services. -//! It handles requests at the application layer in its associated processor and directs -//! syncing-related responses to the Sync manager. -#![allow(clippy::unit_arg)] - -mod processor; - -use crate::error; -use crate::service::{NetworkMessage, RequestId}; -use beacon_chain::{BeaconChain, BeaconChainTypes}; -use futures::prelude::*; -use lighthouse_network::{ - MessageId, NetworkGlobals, PeerId, PeerRequestId, PubsubMessage, Request, Response, -}; -use processor::Processor; -use slog::{debug, o, trace}; -use std::sync::Arc; -use tokio::sync::mpsc; -use tokio_stream::wrappers::UnboundedReceiverStream; -use types::EthSpec; - -/// Handles messages received from the network and client and organises syncing. This -/// functionality of this struct is to validate an decode messages from the network before -/// passing them to the internal message processor. The message processor spawns a syncing thread -/// which manages which blocks need to be requested and processed. -pub struct Router { - /// Access to the peer db. - network_globals: Arc>, - /// Processes validated and decoded messages from the network. Has direct access to the - /// sync manager. - processor: Processor, - /// The `Router` logger. - log: slog::Logger, -} - -/// Types of messages the handler can receive. -#[derive(Debug)] -pub enum RouterMessage { - /// We have initiated a connection to a new peer. - PeerDialed(PeerId), - /// Peer has disconnected, - PeerDisconnected(PeerId), - /// An RPC request has been received. - RPCRequestReceived { - peer_id: PeerId, - id: PeerRequestId, - request: Request, - }, - /// An RPC response has been received. - RPCResponseReceived { - peer_id: PeerId, - request_id: RequestId, - response: Response, - }, - /// An RPC request failed - RPCFailed { - peer_id: PeerId, - request_id: RequestId, - }, - /// A gossip message has been received. The fields are: message id, the peer that sent us this - /// message, the message itself and a bool which indicates if the message should be processed - /// by the beacon chain after successful verification. - PubsubMessage(MessageId, PeerId, PubsubMessage, bool), - /// The peer manager has requested we re-status a peer. - StatusPeer(PeerId), -} - -impl Router { - /// Initializes and runs the Router. - pub fn spawn( - beacon_chain: Arc>, - network_globals: Arc>, - network_send: mpsc::UnboundedSender>, - executor: task_executor::TaskExecutor, - log: slog::Logger, - ) -> error::Result>> { - let message_handler_log = log.new(o!("service"=> "router")); - trace!(message_handler_log, "Service starting"); - - let (handler_send, handler_recv) = mpsc::unbounded_channel(); - - // Initialise a message instance, which itself spawns the syncing thread. - let processor = Processor::new( - executor.clone(), - beacon_chain, - network_globals.clone(), - network_send, - &log, - ); - - // generate the Message handler - let mut handler = Router { - network_globals, - processor, - log: message_handler_log, - }; - - // spawn handler task and move the message handler instance into the spawned thread - executor.spawn( - async move { - debug!(log, "Network message router started"); - UnboundedReceiverStream::new(handler_recv) - .for_each(move |msg| future::ready(handler.handle_message(msg))) - .await; - }, - "router", - ); - - Ok(handler_send) - } - - /// Handle all messages incoming from the network service. - fn handle_message(&mut self, message: RouterMessage) { - match message { - // we have initiated a connection to a peer or the peer manager has requested a - // re-status - RouterMessage::PeerDialed(peer_id) | RouterMessage::StatusPeer(peer_id) => { - self.processor.send_status(peer_id); - } - // A peer has disconnected - RouterMessage::PeerDisconnected(peer_id) => { - self.processor.on_disconnect(peer_id); - } - RouterMessage::RPCRequestReceived { - peer_id, - id, - request, - } => { - self.handle_rpc_request(peer_id, id, request); - } - RouterMessage::RPCResponseReceived { - peer_id, - request_id, - response, - } => { - self.handle_rpc_response(peer_id, request_id, response); - } - RouterMessage::RPCFailed { - peer_id, - request_id, - } => { - self.processor.on_rpc_error(peer_id, request_id); - } - RouterMessage::PubsubMessage(id, peer_id, gossip, should_process) => { - self.handle_gossip(id, peer_id, gossip, should_process); - } - } - } - - /* RPC - Related functionality */ - - /// A new RPC request has been received from the network. - fn handle_rpc_request(&mut self, peer_id: PeerId, id: PeerRequestId, request: Request) { - if !self.network_globals.peers.read().is_connected(&peer_id) { - debug!(self.log, "Dropping request of disconnected peer"; "peer_id" => %peer_id, "request" => ?request); - return; - } - match request { - Request::Status(status_message) => { - self.processor - .on_status_request(peer_id, id, status_message) - } - Request::BlocksByRange(request) => self - .processor - .on_blocks_by_range_request(peer_id, id, request), - Request::BlocksByRoot(request) => self - .processor - .on_blocks_by_root_request(peer_id, id, request), - Request::LightClientBootstrap(request) => self - .processor - .on_lightclient_bootstrap(peer_id, id, request), - } - } - - /// An RPC response has been received from the network. - // we match on id and ignore responses past the timeout. - fn handle_rpc_response( - &mut self, - peer_id: PeerId, - request_id: RequestId, - response: Response, - ) { - // an error could have occurred. - match response { - Response::Status(status_message) => { - self.processor.on_status_response(peer_id, status_message); - } - Response::BlocksByRange(beacon_block) => { - self.processor - .on_blocks_by_range_response(peer_id, request_id, beacon_block); - } - Response::BlocksByRoot(beacon_block) => { - self.processor - .on_blocks_by_root_response(peer_id, request_id, beacon_block); - } - Response::LightClientBootstrap(_) => unreachable!(), - } - } - - /// Handle RPC messages. - /// Note: `should_process` is currently only useful for the `Attestation` variant. - /// if `should_process` is `false`, we only propagate the message on successful verification, - /// else, we propagate **and** import into the beacon chain. - fn handle_gossip( - &mut self, - id: MessageId, - peer_id: PeerId, - gossip_message: PubsubMessage, - should_process: bool, - ) { - match gossip_message { - // Attestations should never reach the router. - PubsubMessage::AggregateAndProofAttestation(aggregate_and_proof) => { - self.processor - .on_aggregated_attestation_gossip(id, peer_id, *aggregate_and_proof); - } - PubsubMessage::Attestation(subnet_attestation) => { - self.processor.on_unaggregated_attestation_gossip( - id, - peer_id, - subnet_attestation.1.clone(), - subnet_attestation.0, - should_process, - ); - } - PubsubMessage::BeaconBlock(block) => { - self.processor.on_block_gossip( - id, - peer_id, - self.network_globals.client(&peer_id), - block, - ); - } - PubsubMessage::VoluntaryExit(exit) => { - debug!(self.log, "Received a voluntary exit"; "peer_id" => %peer_id); - self.processor.on_voluntary_exit_gossip(id, peer_id, exit); - } - PubsubMessage::ProposerSlashing(proposer_slashing) => { - debug!( - self.log, - "Received a proposer slashing"; - "peer_id" => %peer_id - ); - self.processor - .on_proposer_slashing_gossip(id, peer_id, proposer_slashing); - } - PubsubMessage::AttesterSlashing(attester_slashing) => { - debug!( - self.log, - "Received a attester slashing"; - "peer_id" => %peer_id - ); - self.processor - .on_attester_slashing_gossip(id, peer_id, attester_slashing); - } - PubsubMessage::SignedContributionAndProof(contribution_and_proof) => { - trace!( - self.log, - "Received sync committee aggregate"; - "peer_id" => %peer_id - ); - self.processor.on_sync_committee_contribution_gossip( - id, - peer_id, - *contribution_and_proof, - ); - } - PubsubMessage::SyncCommitteeMessage(sync_committtee_msg) => { - trace!( - self.log, - "Received sync committee signature"; - "peer_id" => %peer_id - ); - self.processor.on_sync_committee_signature_gossip( - id, - peer_id, - sync_committtee_msg.1, - sync_committtee_msg.0, - ); - } - PubsubMessage::BlsToExecutionChange(bls_to_execution_change) => { - trace!( - self.log, - "Received BLS to execution change"; - "peer_id" => %peer_id - ); - self.processor.on_bls_to_execution_change_gossip( - id, - peer_id, - bls_to_execution_change, - ); - } - PubsubMessage::LightClientFinalityUpdate(light_client_finality_update) => { - trace!( - self.log, - "Received light client finality update"; - "peer_id" => %peer_id - ); - self.processor.on_light_client_finality_update_gossip( - id, - peer_id, - light_client_finality_update, - ); - } - PubsubMessage::LightClientOptimisticUpdate(light_client_optimistic_update) => { - trace!( - self.log, - "Received light client optimistic update"; - "peer_id" => %peer_id - ); - self.processor.on_light_client_optimistic_update_gossip( - id, - peer_id, - light_client_optimistic_update, - ); - } - } - } -} diff --git a/beacon_node/network/src/router/processor.rs b/beacon_node/network/src/router/processor.rs deleted file mode 100644 index d3ba024e4..000000000 --- a/beacon_node/network/src/router/processor.rs +++ /dev/null @@ -1,472 +0,0 @@ -use crate::beacon_processor::{ - BeaconProcessor, WorkEvent as BeaconWorkEvent, MAX_WORK_EVENT_QUEUE_LEN, -}; -use crate::service::{NetworkMessage, RequestId}; -use crate::status::status_message; -use crate::sync::manager::RequestId as SyncId; -use crate::sync::SyncMessage; -use beacon_chain::{BeaconChain, BeaconChainTypes}; -use lighthouse_network::rpc::*; -use lighthouse_network::{ - Client, MessageId, NetworkGlobals, PeerId, PeerRequestId, Request, Response, -}; -use slog::{debug, error, o, trace, warn}; -use std::cmp; -use std::sync::Arc; -use std::time::{Duration, SystemTime, UNIX_EPOCH}; -use store::SyncCommitteeMessage; -use tokio::sync::mpsc; -use types::{ - Attestation, AttesterSlashing, EthSpec, LightClientFinalityUpdate, LightClientOptimisticUpdate, - ProposerSlashing, SignedAggregateAndProof, SignedBeaconBlock, SignedBlsToExecutionChange, - SignedContributionAndProof, SignedVoluntaryExit, SubnetId, SyncSubnetId, -}; - -/// Processes validated messages from the network. It relays necessary data to the syncing thread -/// and processes blocks from the pubsub network. -pub struct Processor { - /// A reference to the underlying beacon chain. - chain: Arc>, - /// A channel to the syncing thread. - sync_send: mpsc::UnboundedSender>, - /// A network context to return and handle RPC requests. - network: HandlerNetworkContext, - /// A multi-threaded, non-blocking processor for applying messages to the beacon chain. - beacon_processor_send: mpsc::Sender>, - /// The `RPCHandler` logger. - log: slog::Logger, -} - -impl Processor { - /// Instantiate a `Processor` instance - pub fn new( - executor: task_executor::TaskExecutor, - beacon_chain: Arc>, - network_globals: Arc>, - network_send: mpsc::UnboundedSender>, - log: &slog::Logger, - ) -> Self { - let sync_logger = log.new(o!("service"=> "sync")); - let (beacon_processor_send, beacon_processor_receive) = - mpsc::channel(MAX_WORK_EVENT_QUEUE_LEN); - - // spawn the sync thread - let sync_send = crate::sync::manager::spawn( - executor.clone(), - beacon_chain.clone(), - network_globals.clone(), - network_send.clone(), - beacon_processor_send.clone(), - sync_logger, - ); - - BeaconProcessor { - beacon_chain: Arc::downgrade(&beacon_chain), - network_tx: network_send.clone(), - sync_tx: sync_send.clone(), - network_globals, - executor, - max_workers: cmp::max(1, num_cpus::get()), - current_workers: 0, - importing_blocks: Default::default(), - log: log.clone(), - } - .spawn_manager(beacon_processor_receive, None); - - Processor { - chain: beacon_chain, - sync_send, - network: HandlerNetworkContext::new(network_send, log.clone()), - beacon_processor_send, - log: log.new(o!("service" => "router")), - } - } - - fn send_to_sync(&mut self, message: SyncMessage) { - self.sync_send.send(message).unwrap_or_else(|e| { - warn!( - self.log, - "Could not send message to the sync service"; - "error" => %e, - ) - }); - } - - /// Handle a peer disconnect. - /// - /// Removes the peer from the manager. - pub fn on_disconnect(&mut self, peer_id: PeerId) { - self.send_to_sync(SyncMessage::Disconnect(peer_id)); - } - - /// An error occurred during an RPC request. The state is maintained by the sync manager, so - /// this function notifies the sync manager of the error. - pub fn on_rpc_error(&mut self, peer_id: PeerId, request_id: RequestId) { - // Check if the failed RPC belongs to sync - if let RequestId::Sync(request_id) = request_id { - self.send_to_sync(SyncMessage::RpcError { - peer_id, - request_id, - }); - } - } - - /// Sends a `Status` message to the peer. - /// - /// Called when we first connect to a peer, or when the PeerManager determines we need to - /// re-status. - pub fn send_status(&mut self, peer_id: PeerId) { - let status_message = status_message(&self.chain); - debug!(self.log, "Sending Status Request"; "peer" => %peer_id, &status_message); - self.network - .send_processor_request(peer_id, Request::Status(status_message)); - } - - /// Handle a `Status` request. - /// - /// Processes the `Status` from the remote peer and sends back our `Status`. - pub fn on_status_request( - &mut self, - peer_id: PeerId, - request_id: PeerRequestId, - status: StatusMessage, - ) { - debug!(self.log, "Received Status Request"; "peer_id" => %peer_id, &status); - - // Say status back. - self.network.send_response( - peer_id, - Response::Status(status_message(&self.chain)), - request_id, - ); - - self.send_beacon_processor_work(BeaconWorkEvent::status_message(peer_id, status)) - } - - /// Process a `Status` response from a peer. - pub fn on_status_response(&mut self, peer_id: PeerId, status: StatusMessage) { - debug!(self.log, "Received Status Response"; "peer_id" => %peer_id, &status); - self.send_beacon_processor_work(BeaconWorkEvent::status_message(peer_id, status)) - } - - /// Handle a `BlocksByRoot` request from the peer. - pub fn on_blocks_by_root_request( - &mut self, - peer_id: PeerId, - request_id: PeerRequestId, - request: BlocksByRootRequest, - ) { - self.send_beacon_processor_work(BeaconWorkEvent::blocks_by_roots_request( - peer_id, request_id, request, - )) - } - - /// Handle a `LightClientBootstrap` request from the peer. - pub fn on_lightclient_bootstrap( - &mut self, - peer_id: PeerId, - request_id: PeerRequestId, - request: LightClientBootstrapRequest, - ) { - self.send_beacon_processor_work(BeaconWorkEvent::lightclient_bootstrap_request( - peer_id, request_id, request, - )) - } - - /// Handle a `BlocksByRange` request from the peer. - pub fn on_blocks_by_range_request( - &mut self, - peer_id: PeerId, - request_id: PeerRequestId, - req: BlocksByRangeRequest, - ) { - self.send_beacon_processor_work(BeaconWorkEvent::blocks_by_range_request( - peer_id, request_id, req, - )) - } - - /// Handle a `BlocksByRange` response from the peer. - /// A `beacon_block` behaves as a stream which is terminated on a `None` response. - pub fn on_blocks_by_range_response( - &mut self, - peer_id: PeerId, - request_id: RequestId, - beacon_block: Option>>, - ) { - let request_id = match request_id { - RequestId::Sync(sync_id) => match sync_id { - SyncId::SingleBlock { .. } | SyncId::ParentLookup { .. } => { - unreachable!("Block lookups do not request BBRange requests") - } - id @ (SyncId::BackFillSync { .. } | SyncId::RangeSync { .. }) => id, - }, - RequestId::Router => unreachable!("All BBRange requests belong to sync"), - }; - - trace!( - self.log, - "Received BlocksByRange Response"; - "peer" => %peer_id, - ); - - self.send_to_sync(SyncMessage::RpcBlock { - peer_id, - request_id, - beacon_block, - seen_timestamp: timestamp_now(), - }); - } - - /// Handle a `BlocksByRoot` response from the peer. - pub fn on_blocks_by_root_response( - &mut self, - peer_id: PeerId, - request_id: RequestId, - beacon_block: Option>>, - ) { - let request_id = match request_id { - RequestId::Sync(sync_id) => match sync_id { - id @ (SyncId::SingleBlock { .. } | SyncId::ParentLookup { .. }) => id, - SyncId::BackFillSync { .. } | SyncId::RangeSync { .. } => { - unreachable!("Batch syncing do not request BBRoot requests") - } - }, - RequestId::Router => unreachable!("All BBRoot requests belong to sync"), - }; - - trace!( - self.log, - "Received BlocksByRoot Response"; - "peer" => %peer_id, - ); - self.send_to_sync(SyncMessage::RpcBlock { - peer_id, - request_id, - beacon_block, - seen_timestamp: timestamp_now(), - }); - } - - /// Process a gossip message declaring a new block. - /// - /// Attempts to apply to block to the beacon chain. May queue the block for later processing. - /// - /// Returns a `bool` which, if `true`, indicates we should forward the block to our peers. - pub fn on_block_gossip( - &mut self, - message_id: MessageId, - peer_id: PeerId, - peer_client: Client, - block: Arc>, - ) { - self.send_beacon_processor_work(BeaconWorkEvent::gossip_beacon_block( - message_id, - peer_id, - peer_client, - block, - timestamp_now(), - )) - } - - pub fn on_unaggregated_attestation_gossip( - &mut self, - message_id: MessageId, - peer_id: PeerId, - unaggregated_attestation: Attestation, - subnet_id: SubnetId, - should_process: bool, - ) { - self.send_beacon_processor_work(BeaconWorkEvent::unaggregated_attestation( - message_id, - peer_id, - unaggregated_attestation, - subnet_id, - should_process, - timestamp_now(), - )) - } - - pub fn on_aggregated_attestation_gossip( - &mut self, - message_id: MessageId, - peer_id: PeerId, - aggregate: SignedAggregateAndProof, - ) { - self.send_beacon_processor_work(BeaconWorkEvent::aggregated_attestation( - message_id, - peer_id, - aggregate, - timestamp_now(), - )) - } - - pub fn on_voluntary_exit_gossip( - &mut self, - message_id: MessageId, - peer_id: PeerId, - voluntary_exit: Box, - ) { - self.send_beacon_processor_work(BeaconWorkEvent::gossip_voluntary_exit( - message_id, - peer_id, - voluntary_exit, - )) - } - - pub fn on_proposer_slashing_gossip( - &mut self, - message_id: MessageId, - peer_id: PeerId, - proposer_slashing: Box, - ) { - self.send_beacon_processor_work(BeaconWorkEvent::gossip_proposer_slashing( - message_id, - peer_id, - proposer_slashing, - )) - } - - pub fn on_attester_slashing_gossip( - &mut self, - message_id: MessageId, - peer_id: PeerId, - attester_slashing: Box>, - ) { - self.send_beacon_processor_work(BeaconWorkEvent::gossip_attester_slashing( - message_id, - peer_id, - attester_slashing, - )) - } - - pub fn on_sync_committee_signature_gossip( - &mut self, - message_id: MessageId, - peer_id: PeerId, - sync_signature: SyncCommitteeMessage, - subnet_id: SyncSubnetId, - ) { - self.send_beacon_processor_work(BeaconWorkEvent::gossip_sync_signature( - message_id, - peer_id, - sync_signature, - subnet_id, - timestamp_now(), - )) - } - - pub fn on_sync_committee_contribution_gossip( - &mut self, - message_id: MessageId, - peer_id: PeerId, - sync_contribution: SignedContributionAndProof, - ) { - self.send_beacon_processor_work(BeaconWorkEvent::gossip_sync_contribution( - message_id, - peer_id, - sync_contribution, - timestamp_now(), - )) - } - - pub fn on_bls_to_execution_change_gossip( - &mut self, - message_id: MessageId, - peer_id: PeerId, - bls_to_execution_change: Box, - ) { - self.send_beacon_processor_work(BeaconWorkEvent::gossip_bls_to_execution_change( - message_id, - peer_id, - bls_to_execution_change, - )) - } - - pub fn on_light_client_finality_update_gossip( - &mut self, - message_id: MessageId, - peer_id: PeerId, - light_client_finality_update: Box>, - ) { - self.send_beacon_processor_work(BeaconWorkEvent::gossip_light_client_finality_update( - message_id, - peer_id, - light_client_finality_update, - timestamp_now(), - )) - } - - pub fn on_light_client_optimistic_update_gossip( - &mut self, - message_id: MessageId, - peer_id: PeerId, - light_client_optimistic_update: Box>, - ) { - self.send_beacon_processor_work(BeaconWorkEvent::gossip_light_client_optimistic_update( - message_id, - peer_id, - light_client_optimistic_update, - timestamp_now(), - )) - } - - fn send_beacon_processor_work(&mut self, work: BeaconWorkEvent) { - self.beacon_processor_send - .try_send(work) - .unwrap_or_else(|e| { - let work_type = match &e { - mpsc::error::TrySendError::Closed(work) - | mpsc::error::TrySendError::Full(work) => work.work_type(), - }; - error!(&self.log, "Unable to send message to the beacon processor"; - "error" => %e, "type" => work_type) - }) - } -} - -/// Wraps a Network Channel to employ various RPC related network functionality for the -/// processor. -#[derive(Clone)] -pub struct HandlerNetworkContext { - /// The network channel to relay messages to the Network service. - network_send: mpsc::UnboundedSender>, - /// Logger for the `NetworkContext`. - log: slog::Logger, -} - -impl HandlerNetworkContext { - pub fn new(network_send: mpsc::UnboundedSender>, log: slog::Logger) -> Self { - Self { network_send, log } - } - - /// Sends a message to the network task. - fn inform_network(&mut self, msg: NetworkMessage) { - self.network_send.send(msg).unwrap_or_else( - |e| warn!(self.log, "Could not send message to the network service"; "error" => %e), - ) - } - - /// Sends a request to the network task. - pub fn send_processor_request(&mut self, peer_id: PeerId, request: Request) { - self.inform_network(NetworkMessage::SendRequest { - peer_id, - request_id: RequestId::Router, - request, - }) - } - - /// Sends a response to the network task. - pub fn send_response(&mut self, peer_id: PeerId, response: Response, id: PeerRequestId) { - self.inform_network(NetworkMessage::SendResponse { - peer_id, - id, - response, - }) - } -} - -fn timestamp_now() -> Duration { - SystemTime::now() - .duration_since(UNIX_EPOCH) - .unwrap_or_else(|_| Duration::from_secs(0)) -} diff --git a/beacon_node/network/src/service.rs b/beacon_node/network/src/service.rs index ab3d15aee..3e86d2099 100644 --- a/beacon_node/network/src/service.rs +++ b/beacon_node/network/src/service.rs @@ -472,7 +472,7 @@ impl NetworkService { ) { match ev { NetworkEvent::PeerConnectedOutgoing(peer_id) => { - self.send_to_router(RouterMessage::PeerDialed(peer_id)); + self.send_to_router(RouterMessage::StatusPeer(peer_id)); } NetworkEvent::PeerConnectedIncoming(_) | NetworkEvent::PeerBanned(_)