mirror/lighthouse-pulse
1
0
Fork 0
mirror of https://gitlab.com/pulsechaincom/lighthouse-pulse.git synced 2024-12-27 22:37:25 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

Merge branch 'sync-threading' into interop

Browse Source
This commit is contained in:
Age Manning 2019-09-07 09:32:09 +10:00
commit 1eafa63c2a
No known key found for this signature in database
GPG Key ID: 05EED64B79E06A93
19 changed files with 1096 additions and 569 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
account_manager/src/main.rs
View File

@ -125,9 +125,13 @@ fn main() {
}
}
}
_ => panic!(
"The account manager must be run with a subcommand. See help for more information."
),
_ => {
crit!(
log,
"The account manager must be run with a subcommand. See help for more information."
);
return;
}
}
}

11
beacon_node/beacon_chain/src/beacon_chain.rs
View File

@ -459,6 +459,15 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
None
}
/// Returns the block canonical root of the current canonical chain at a given slot.
///
/// Returns None if a block doesn't exist at the slot.
pub fn root_at_slot(&self, target_slot: Slot) -> Option<Hash256> {
self.rev_iter_block_roots()
.find(|(_root, slot)| *slot == target_slot)
.map(|(root, _slot)| root)
}
/// Reads the slot clock (see `self.read_slot_clock()` and returns the number of slots since
/// genesis.
pub fn slots_since_genesis(&self) -> Option<SlotHeight> {
@ -1017,7 +1026,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
};
// Load the parent blocks state from the database, returning an error if it is not found.
// It is an error because if know the parent block we should also know the parent state.
// It is an error because if we know the parent block we should also know the parent state.
let parent_state_root = parent_block.state_root;
let parent_state = self
.store

4
beacon_node/client/src/notifier.rs
View File

@ -34,10 +34,10 @@ pub fn run<T: BeaconChainTypes>(client: &Client<T>, executor: TaskExecutor, exit
// Panics if libp2p is poisoned.
let connected_peer_count = libp2p.lock().swarm.connected_peers();
debug!(log, "Libp2p connected peer status"; "peer_count" => connected_peer_count);
debug!(log, "Connected peer status"; "peer_count" => connected_peer_count);
if connected_peer_count <= WARN_PEER_COUNT {
warn!(log, "Low libp2p peer count"; "peer_count" => connected_peer_count);
warn!(log, "Low peer count"; "peer_count" => connected_peer_count);
}
Ok(())

4
beacon_node/eth2-libp2p/Cargo.toml
View File

@ -7,8 +7,8 @@ edition = "2018"
[dependencies]
clap = "2.32.0"
#SigP repository
libp2p = { git = "https://github.com/SigP/rust-libp2p", rev = "61036890d574f5b46573952b20def2baafd6a6e9" }
enr = { git = "https://github.com/SigP/rust-libp2p/", rev = "61036890d574f5b46573952b20def2baafd6a6e9", features = ["serde"] }
libp2p = { git = "https://github.com/SigP/rust-libp2p", rev = "76f7475e4b7063e663ad03c7524cf091f9961968" }
enr = { git = "https://github.com/SigP/rust-libp2p/", rev = "76f7475e4b7063e663ad03c7524cf091f9961968", features = ["serde"] }
types = { path = "../../eth2/types" }
serde = "1.0"
serde_derive = "1.0"

28
beacon_node/eth2-libp2p/src/behaviour.rs
View File

@ -15,7 +15,7 @@ use libp2p::{
tokio_io::{AsyncRead, AsyncWrite},
NetworkBehaviour, PeerId,
};
use slog::{debug, o, trace};
use slog::{debug, o};
use std::num::NonZeroU32;
use std::time::Duration;
@ -90,13 +90,15 @@ impl<TSubstream: AsyncRead + AsyncWrite> NetworkBehaviourEventProcess<GossipsubE
{
fn inject_event(&mut self, event: GossipsubEvent) {
match event {
GossipsubEvent::Message(gs_msg) => {
trace!(self.log, "Received GossipEvent");
GossipsubEvent::Message(propagation_source, gs_msg) => {
let id = gs_msg.id();
let msg = PubsubMessage::from_topics(&gs_msg.topics, gs_msg.data);
// Note: We are keeping track here of the peer that sent us the message, not the
// peer that originally published the message.
self.events.push(BehaviourEvent::GossipMessage {
source: gs_msg.source,
id,
source: propagation_source,
topics: gs_msg.topics,
message: msg,
});
@ -199,6 +201,13 @@ impl<TSubstream: AsyncRead + AsyncWrite> Behaviour<TSubstream> {
}
}
/// Forwards a message that is waiting in gossipsub's mcache. Messages are only propagated
/// once validated by the beacon chain.
pub fn propagate_message(&mut self, propagation_source: &PeerId, message_id: String) {
self.gossipsub
.propagate_message(&message_id, propagation_source);
}
/* Eth2 RPC behaviour functions */
/// Sends an RPC Request/Response via the RPC protocol.
@ -214,12 +223,21 @@ impl<TSubstream: AsyncRead + AsyncWrite> Behaviour<TSubstream> {
/// The types of events than can be obtained from polling the behaviour.
pub enum BehaviourEvent {
/// A received RPC event and the peer that it was received from.
RPC(PeerId, RPCEvent),
/// We have completed an initial connection to a new peer.
PeerDialed(PeerId),
/// A peer has disconnected.
PeerDisconnected(PeerId),
/// A gossipsub message has been received.
GossipMessage {
/// The gossipsub message id. Used when propagating blocks after validation.
id: String,
/// The peer from which we received this message, not the peer that published it.
source: PeerId,
/// The topics that this message was sent on.
topics: Vec<TopicHash>,
/// The message itself.
message: PubsubMessage,
},
}

3
beacon_node/eth2-libp2p/src/config.rs
View File

@ -74,7 +74,8 @@ impl Default for Config {
// parameter.
gs_config: GossipsubConfigBuilder::new()
.max_transmit_size(1_048_576)
.heartbeat_interval(Duration::from_secs(20))
.heartbeat_interval(Duration::from_secs(20)) // TODO: Reduce for mainnet
.propagate_messages(false) // require validation before propagation
.build(),
boot_nodes: vec![],
libp2p_nodes: vec![],

3
beacon_node/eth2-libp2p/src/discovery.rs
View File

@ -114,7 +114,7 @@ impl<TSubstream> Discovery<TSubstream> {
self.find_peers();
}
/// Add an Enr to the routing table of the discovery mechanism.
/// Add an ENR to the routing table of the discovery mechanism.
pub fn add_enr(&mut self, enr: Enr) {
self.discovery.add_enr(enr);
}
@ -169,6 +169,7 @@ where
fn inject_connected(&mut self, peer_id: PeerId, _endpoint: ConnectedPoint) {
self.connected_peers.insert(peer_id);
// TODO: Drop peers if over max_peer limit
metrics::inc_counter(&metrics::PEER_CONNECT_EVENT_COUNT);
metrics::set_gauge(&metrics::PEERS_CONNECTED, self.connected_peers() as i64);

36
beacon_node/eth2-libp2p/src/rpc/codec/ssz.rs
View File

@ -152,45 +152,49 @@ impl Decoder for SSZOutboundCodec {
type Error = RPCError;
fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
match self.inner.decode(src).map_err(RPCError::from) {
Ok(Some(packet)) => match self.protocol.message_name.as_str() {
if src.is_empty() {
// the object sent could be empty. We return the empty object if this is the case
match self.protocol.message_name.as_str() {
"hello" => match self.protocol.version.as_str() {
"1" => Ok(Some(RPCResponse::Hello(HelloMessage::from_ssz_bytes(
&packet,
)?))),
"1" => Err(RPCError::Custom(
"Hello stream terminated unexpectedly".into(),
)), // cannot have an empty HELLO message. The stream has terminated unexpectedly
_ => unreachable!("Cannot negotiate an unknown version"),
},
"goodbye" => Err(RPCError::InvalidProtocol("GOODBYE doesn't have a response")),
"beacon_blocks" => match self.protocol.version.as_str() {
"1" => Ok(Some(RPCResponse::BeaconBlocks(packet.to_vec()))),
"1" => Ok(Some(RPCResponse::BeaconBlocks(Vec::new()))),
_ => unreachable!("Cannot negotiate an unknown version"),
},
"recent_beacon_blocks" => match self.protocol.version.as_str() {
"1" => Ok(Some(RPCResponse::RecentBeaconBlocks(packet.to_vec()))),
"1" => Ok(Some(RPCResponse::RecentBeaconBlocks(Vec::new()))),
_ => unreachable!("Cannot negotiate an unknown version"),
},
_ => unreachable!("Cannot negotiate an unknown protocol"),
},
Ok(None) => {
// the object sent could be a empty. We return the empty object if this is the case
match self.protocol.message_name.as_str() {
}
} else {
match self.inner.decode(src).map_err(RPCError::from) {
Ok(Some(packet)) => match self.protocol.message_name.as_str() {
"hello" => match self.protocol.version.as_str() {
"1" => Ok(None), // cannot have an empty HELLO message. The stream has terminated unexpectedly
"1" => Ok(Some(RPCResponse::Hello(HelloMessage::from_ssz_bytes(
&packet,
)?))),
_ => unreachable!("Cannot negotiate an unknown version"),
},
"goodbye" => Err(RPCError::InvalidProtocol("GOODBYE doesn't have a response")),
"beacon_blocks" => match self.protocol.version.as_str() {
"1" => Ok(Some(RPCResponse::BeaconBlocks(Vec::new()))),
"1" => Ok(Some(RPCResponse::BeaconBlocks(packet.to_vec()))),
_ => unreachable!("Cannot negotiate an unknown version"),
},
"recent_beacon_blocks" => match self.protocol.version.as_str() {
"1" => Ok(Some(RPCResponse::RecentBeaconBlocks(Vec::new()))),
"1" => Ok(Some(RPCResponse::RecentBeaconBlocks(packet.to_vec()))),
_ => unreachable!("Cannot negotiate an unknown version"),
},
_ => unreachable!("Cannot negotiate an unknown protocol"),
}
},
Ok(None) => Ok(None), // waiting for more bytes
Err(e) => Err(e),
}
Err(e) => Err(e),
}
}
}

50
beacon_node/eth2-libp2p/src/rpc/methods.rs
View File

@ -157,3 +157,53 @@ impl ErrorMessage {
String::from_utf8(self.error_message.clone()).unwrap_or_else(|_| "".into())
}
}
impl std::fmt::Display for HelloMessage {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "Hello Message: Fork Version: {:?}, Finalized Root: {}, Finalized Epoch: {}, Head Root: {}, Head Slot: {}", self.fork_version, self.finalized_root, self.finalized_epoch, self.head_root, self.head_slot)
}
}
impl std::fmt::Display for RPCResponse {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
RPCResponse::Hello(hello) => write!(f, "{}", hello),
RPCResponse::BeaconBlocks(data) => write!(f, "<BeaconBlocks>, len: {}", data.len()),
RPCResponse::RecentBeaconBlocks(data) => {
write!(f, "<RecentBeaconBlocks>, len: {}", data.len())
}
}
}
}
impl std::fmt::Display for RPCErrorResponse {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
RPCErrorResponse::Success(res) => write!(f, "{}", res),
RPCErrorResponse::InvalidRequest(err) => write!(f, "Invalid Request: {:?}", err),
RPCErrorResponse::ServerError(err) => write!(f, "Server Error: {:?}", err),
RPCErrorResponse::Unknown(err) => write!(f, "Unknown Error: {:?}", err),
}
}
}
impl std::fmt::Display for GoodbyeReason {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
GoodbyeReason::ClientShutdown => write!(f, "Client Shutdown"),
GoodbyeReason::IrrelevantNetwork => write!(f, "Irrelevant Network"),
GoodbyeReason::Fault => write!(f, "Fault"),
GoodbyeReason::Unknown => write!(f, "Unknown Reason"),
}
}
}
impl std::fmt::Display for BeaconBlocksRequest {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(
f,
"Head Block Root: {}, Start Slot: {}, Count: {}, Step: {}",
self.head_block_root, self.start_slot, self.count, self.step
)
}
}

10
beacon_node/eth2-libp2p/src/rpc/mod.rs
View File

@ -47,6 +47,16 @@ impl RPCEvent {
}
}
impl std::fmt::Display for RPCEvent {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
RPCEvent::Request(id, req) => write!(f, "RPC Request(Id: {}, {})", id, req),
RPCEvent::Response(id, res) => write!(f, "RPC Response(Id: {}, {})", id, res),
RPCEvent::Error(id, err) => write!(f, "RPC Request(Id: {}, Error: {:?})", id, err),
}
}
}
/// Implements the libp2p `NetworkBehaviour` trait and therefore manages network-level
/// logic.
pub struct RPC<TSubstream> {

11
beacon_node/eth2-libp2p/src/rpc/protocol.rs
View File

@ -288,3 +288,14 @@ impl std::error::Error for RPCError {
}
}
}
impl std::fmt::Display for RPCRequest {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
RPCRequest::Hello(hello) => write!(f, "Hello Message: {}", hello),
RPCRequest::Goodbye(reason) => write!(f, "Goodbye: {}", reason),
RPCRequest::BeaconBlocks(req) => write!(f, "Beacon Blocks: {}", req),
RPCRequest::RecentBeaconBlocks(req) => write!(f, "Recent Beacon Blocks: {:?}", req),
}
}
}

28
beacon_node/eth2-libp2p/src/service.rs
View File

@ -79,15 +79,32 @@ impl Service {
}
};
// attempt to connect to user-input libp2p nodes
for multiaddr in config.libp2p_nodes {
// helper closure for dialing peers
let mut dial_addr = |multiaddr: Multiaddr| {
match Swarm::dial_addr(&mut swarm, multiaddr.clone()) {
Ok(()) => debug!(log, "Dialing libp2p peer"; "address" => format!("{}", multiaddr)),
Err(err) => debug!(
log,
"Could not connect to peer"; "address" => format!("{}", multiaddr), "Error" => format!("{:?}", err)
"Could not connect to peer"; "address" => format!("{}", multiaddr), "error" => format!("{:?}", err)
),
};
};
// attempt to connect to user-input libp2p nodes
for multiaddr in config.libp2p_nodes {
dial_addr(multiaddr);
}
// attempt to connect to any specified boot-nodes
for bootnode_enr in config.boot_nodes {
for multiaddr in bootnode_enr.multiaddr() {
// ignore udp multiaddr if it exists
let components = multiaddr.iter().collect::<Vec<_>>();
if let Protocol::Udp(_) = components[1] {
continue;
}
dial_addr(multiaddr);
}
}
// subscribe to default gossipsub topics
@ -145,16 +162,16 @@ impl Stream for Service {
fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
loop {
match self.swarm.poll() {
//Behaviour events
Ok(Async::Ready(Some(event))) => match event {
// TODO: Stub here for debugging
BehaviourEvent::GossipMessage {
id,
source,
topics,
message,
} => {
trace!(self.log, "Gossipsub message received"; "service" => "Swarm");
return Ok(Async::Ready(Some(Libp2pEvent::PubsubMessage {
id,
source,
topics,
message,
@ -222,6 +239,7 @@ pub enum Libp2pEvent {
PeerDisconnected(PeerId),
/// Received pubsub message.
PubsubMessage {
id: String,
source: PeerId,
topics: Vec<TopicHash>,
message: PubsubMessage,

1
beacon_node/network/Cargo.toml
View File

@ -19,3 +19,4 @@ futures = "0.1.25"
error-chain = "0.12.0"
tokio = "0.1.16"
parking_lot = "0.9.0"
smallvec = "0.6.10"

105
beacon_node/network/src/message_handler.rs
View File

@ -1,6 +1,6 @@
use crate::error;
use crate::service::NetworkMessage;
use crate::sync::SimpleSync;
use crate::sync::MessageProcessor;
use beacon_chain::{BeaconChain, BeaconChainTypes};
use eth2_libp2p::{
behaviour::PubsubMessage,
@ -9,18 +9,22 @@ use eth2_libp2p::{
};
use futures::future::Future;
use futures::stream::Stream;
use slog::{debug, trace, warn};
use slog::{debug, o, trace, warn};
use ssz::{Decode, DecodeError};
use std::sync::Arc;
use tokio::sync::mpsc;
use types::{Attestation, AttesterSlashing, BeaconBlock, ProposerSlashing, VoluntaryExit};
/// Handles messages received from the network and client and organises syncing.
/// 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 MessageHandler<T: BeaconChainTypes> {
/// Currently loaded and initialised beacon chain.
_chain: Arc<BeaconChain<T>>,
/// The syncing framework.
sync: SimpleSync<T>,
/// A channel to the network service to allow for gossip propagation.
network_send: mpsc::UnboundedSender<NetworkMessage>,
/// Processes validated and decoded messages from the network. Has direct access to the
/// sync manager.
message_processor: MessageProcessor<T>,
/// The `MessageHandler` logger.
log: slog::Logger,
}
@ -34,8 +38,9 @@ pub enum HandlerMessage {
PeerDisconnected(PeerId),
/// An RPC response/request has been received.
RPC(PeerId, RPCEvent),
/// A gossip message has been received.
PubsubMessage(PeerId, PubsubMessage),
/// A gossip message has been received. The fields are: message id, the peer that sent us this
/// message and the message itself.
PubsubMessage(String, PeerId, PubsubMessage),
}
impl<T: BeaconChainTypes + 'static> MessageHandler<T> {
@ -46,17 +51,20 @@ impl<T: BeaconChainTypes + 'static> MessageHandler<T> {
executor: &tokio::runtime::TaskExecutor,
log: slog::Logger,
) -> error::Result<mpsc::UnboundedSender<HandlerMessage>> {
trace!(log, "Service starting");
let message_handler_log = log.new(o!("Service"=> "Message Handler"));
trace!(message_handler_log, "Service starting");
let (handler_send, handler_recv) = mpsc::unbounded_channel();
// Initialise sync and begin processing in thread
let sync = SimpleSync::new(beacon_chain.clone(), network_send, &log);
// Initialise a message instance, which itself spawns the syncing thread.
let message_processor =
MessageProcessor::new(executor, beacon_chain, network_send.clone(), &log);
// generate the Message handler
let mut handler = MessageHandler {
_chain: beacon_chain.clone(),
sync,
log: log.clone(),
network_send,
message_processor,
log: message_handler_log,
};
// spawn handler task and move the message handler instance into the spawned thread
@ -65,7 +73,11 @@ impl<T: BeaconChainTypes + 'static> MessageHandler<T> {
.for_each(move |msg| Ok(handler.handle_message(msg)))
.map_err(move |_| {
debug!(log, "Network message handler terminated.");
}),
}), /*
.then(move |_| {
debug!(log.clone(), "Message handler shutdown");
}),
*/
);
Ok(handler_send)
@ -76,19 +88,19 @@ impl<T: BeaconChainTypes + 'static> MessageHandler<T> {
match message {
// we have initiated a connection to a peer
HandlerMessage::PeerDialed(peer_id) => {
self.sync.on_connect(peer_id);
self.message_processor.on_connect(peer_id);
}
// A peer has disconnected
HandlerMessage::PeerDisconnected(peer_id) => {
self.sync.on_disconnect(peer_id);
self.message_processor.on_disconnect(peer_id);
}
// An RPC message request/response has been received
HandlerMessage::RPC(peer_id, rpc_event) => {
self.handle_rpc_message(peer_id, rpc_event);
}
// An RPC message request/response has been received
HandlerMessage::PubsubMessage(peer_id, gossip) => {
self.handle_gossip(peer_id, gossip);
HandlerMessage::PubsubMessage(id, peer_id, gossip) => {
self.handle_gossip(id, peer_id, gossip);
}
}
}
@ -108,7 +120,7 @@ impl<T: BeaconChainTypes + 'static> MessageHandler<T> {
fn handle_rpc_request(&mut self, peer_id: PeerId, request_id: RequestId, request: RPCRequest) {
match request {
RPCRequest::Hello(hello_message) => {
self.sync
self.message_processor
.on_hello_request(peer_id, request_id, hello_message)
}
RPCRequest::Goodbye(goodbye_reason) => {
@ -117,13 +129,13 @@ impl<T: BeaconChainTypes + 'static> MessageHandler<T> {
"peer" => format!("{:?}", peer_id),
"reason" => format!("{:?}", goodbye_reason),
);
self.sync.on_disconnect(peer_id);
self.message_processor.on_disconnect(peer_id);
}
RPCRequest::BeaconBlocks(request) => self
.sync
.message_processor
.on_beacon_blocks_request(peer_id, request_id, request),
RPCRequest::RecentBeaconBlocks(request) => self
.sync
.message_processor
.on_recent_beacon_blocks_request(peer_id, request_id, request),
}
}
@ -150,12 +162,13 @@ impl<T: BeaconChainTypes + 'static> MessageHandler<T> {
RPCErrorResponse::Success(response) => {
match response {
RPCResponse::Hello(hello_message) => {
self.sync.on_hello_response(peer_id, hello_message);
self.message_processor
.on_hello_response(peer_id, hello_message);
}
RPCResponse::BeaconBlocks(response) => {
match self.decode_beacon_blocks(&response) {
Ok(beacon_blocks) => {
self.sync.on_beacon_blocks_response(
self.message_processor.on_beacon_blocks_response(
peer_id,
request_id,
beacon_blocks,
@ -170,7 +183,7 @@ impl<T: BeaconChainTypes + 'static> MessageHandler<T> {
RPCResponse::RecentBeaconBlocks(response) => {
match self.decode_beacon_blocks(&response) {
Ok(beacon_blocks) => {
self.sync.on_recent_beacon_blocks_response(
self.message_processor.on_recent_beacon_blocks_response(
peer_id,
request_id,
beacon_blocks,
@ -194,24 +207,37 @@ impl<T: BeaconChainTypes + 'static> MessageHandler<T> {
}
/// Handle RPC messages
fn handle_gossip(&mut self, peer_id: PeerId, gossip_message: PubsubMessage) {
fn handle_gossip(&mut self, id: String, peer_id: PeerId, gossip_message: PubsubMessage) {
match gossip_message {
PubsubMessage::Block(message) => match self.decode_gossip_block(message) {
Ok(block) => {
let _should_forward_on = self.sync.on_block_gossip(peer_id, block);
let should_forward_on = self
.message_processor
.on_block_gossip(peer_id.clone(), block);
// TODO: Apply more sophisticated validation and decoding logic
if should_forward_on {
self.propagate_message(id, peer_id.clone());
}
}
Err(e) => {
debug!(self.log, "Invalid gossiped beacon block"; "peer_id" => format!("{}", peer_id), "Error" => format!("{:?}", e));
}
},
PubsubMessage::Attestation(message) => match self.decode_gossip_attestation(message) {
Ok(attestation) => self.sync.on_attestation_gossip(peer_id, attestation),
Ok(attestation) => {
// TODO: Apply more sophisticated validation and decoding logic
self.propagate_message(id, peer_id.clone());
self.message_processor
.on_attestation_gossip(peer_id, attestation);
}
Err(e) => {
debug!(self.log, "Invalid gossiped attestation"; "peer_id" => format!("{}", peer_id), "Error" => format!("{:?}", e));
}
},
PubsubMessage::VoluntaryExit(message) => match self.decode_gossip_exit(message) {
Ok(_exit) => {
// TODO: Apply more sophisticated validation and decoding logic
self.propagate_message(id, peer_id.clone());
// TODO: Handle exits
debug!(self.log, "Received a voluntary exit"; "peer_id" => format!("{}", peer_id) );
}
@ -222,6 +248,8 @@ impl<T: BeaconChainTypes + 'static> MessageHandler<T> {
PubsubMessage::ProposerSlashing(message) => {
match self.decode_gossip_proposer_slashing(message) {
Ok(_slashing) => {
// TODO: Apply more sophisticated validation and decoding logic
self.propagate_message(id, peer_id.clone());
// TODO: Handle proposer slashings
debug!(self.log, "Received a proposer slashing"; "peer_id" => format!("{}", peer_id) );
}
@ -233,6 +261,8 @@ impl<T: BeaconChainTypes + 'static> MessageHandler<T> {
PubsubMessage::AttesterSlashing(message) => {
match self.decode_gossip_attestation_slashing(message) {
Ok(_slashing) => {
// TODO: Apply more sophisticated validation and decoding logic
self.propagate_message(id, peer_id.clone());
// TODO: Handle attester slashings
debug!(self.log, "Received an attester slashing"; "peer_id" => format!("{}", peer_id) );
}
@ -248,6 +278,21 @@ impl<T: BeaconChainTypes + 'static> MessageHandler<T> {
}
}
/// Informs the network service that the message should be forwarded to other peers.
fn propagate_message(&mut self, message_id: String, propagation_source: PeerId) {
self.network_send
.try_send(NetworkMessage::Propagate {
propagation_source,
message_id,
})
.unwrap_or_else(|_| {
warn!(
self.log,
"Could not send propagation request to the network service"
)
});
}
/* Decoding of gossipsub objects from the network.
*
* The decoding is done in the message handler as it has access to to a `BeaconChain` and can

64
beacon_node/network/src/service.rs
View File

@ -34,13 +34,8 @@ impl<T: BeaconChainTypes + 'static> Service<T> {
// build the network channel
let (network_send, network_recv) = mpsc::unbounded_channel::<NetworkMessage>();
// launch message handler thread
let message_handler_log = log.new(o!("Service" => "MessageHandler"));
let message_handler_send = MessageHandler::spawn(
beacon_chain,
network_send.clone(),
executor,
message_handler_log,
)?;
let message_handler_send =
MessageHandler::spawn(beacon_chain, network_send.clone(), executor, log.clone())?;
let network_log = log.new(o!("Service" => "Network"));
// launch libp2p service
@ -159,12 +154,23 @@ fn network_service(
// poll the network channel
match network_recv.poll() {
Ok(Async::Ready(Some(message))) => match message {
NetworkMessage::Send(peer_id, outgoing_message) => match outgoing_message {
OutgoingMessage::RPC(rpc_event) => {
trace!(log, "Sending RPC Event: {:?}", rpc_event);
libp2p_service.lock().swarm.send_rpc(peer_id, rpc_event);
}
},
NetworkMessage::RPC(peer_id, rpc_event) => {
trace!(log, "{}", rpc_event);
libp2p_service.lock().swarm.send_rpc(peer_id, rpc_event);
}
NetworkMessage::Propagate {
propagation_source,
message_id,
} => {
trace!(log, "Propagating gossipsub message";
"propagation_peer" => format!("{:?}", propagation_source),
"message_id" => format!("{}", message_id),
);
libp2p_service
.lock()
.swarm
.propagate_message(&propagation_source, message_id);
}
NetworkMessage::Publish { topics, message } => {
debug!(log, "Sending pubsub message"; "topics" => format!("{:?}",topics));
libp2p_service.lock().swarm.publish(&topics, message);
@ -185,7 +191,7 @@ fn network_service(
match libp2p_service.lock().poll() {
Ok(Async::Ready(Some(event))) => match event {
Libp2pEvent::RPC(peer_id, rpc_event) => {
trace!(log, "RPC Event: RPC message received: {:?}", rpc_event);
trace!(log, "{}", rpc_event);
message_handler_send
.try_send(HandlerMessage::RPC(peer_id, rpc_event))
.map_err(|_| "Failed to send RPC to handler")?;
@ -203,13 +209,14 @@ fn network_service(
.map_err(|_| "Failed to send PeerDisconnected to handler")?;
}
Libp2pEvent::PubsubMessage {
source, message, ..
id,
source,
message,
..
} => {
//TODO: Decide if we need to propagate the topic upwards. (Potentially for
//attestations)
message_handler_send
.try_send(HandlerMessage::PubsubMessage(source, message))
.map_err(|_| " failed to send pubsub message to handler")?;
.try_send(HandlerMessage::PubsubMessage(id, source, message))
.map_err(|_| "Failed to send pubsub message to handler")?;
}
},
Ok(Async::Ready(None)) => unreachable!("Stream never ends"),
@ -225,19 +232,16 @@ fn network_service(
/// Types of messages that the network service can receive.
#[derive(Debug)]
pub enum NetworkMessage {
/// Send a message to libp2p service.
//TODO: Define typing for messages across the wire
Send(PeerId, OutgoingMessage),
/// Publish a message to pubsub mechanism.
/// Send an RPC message to the libp2p service.
RPC(PeerId, RPCEvent),
/// Publish a message to gossipsub.
Publish {
topics: Vec<Topic>,
message: PubsubMessage,
},
}
/// Type of outgoing messages that can be sent through the network service.
#[derive(Debug)]
pub enum OutgoingMessage {
/// Send an RPC request/response.
RPC(RPCEvent),
/// Propagate a received gossipsub message
Propagate {
propagation_source: PeerId,
message_id: String,
},
}

1044
beacon_node/network/src/sync/manager.rs
View File

File diff suppressed because it is too large Load Diff

2
beacon_node/network/src/sync/mod.rs
View File

@ -4,7 +4,7 @@ mod manager;
/// Stores the various syncing methods for the beacon chain.
mod simple_sync;
pub use simple_sync::SimpleSync;
pub use simple_sync::MessageProcessor;
/// Currently implemented sync methods.
pub enum SyncMethod {

231
beacon_node/network/src/sync/simple_sync.rs
View File

@ -1,23 +1,23 @@
use super::manager::{ImportManager, ImportManagerOutcome};
use crate::service::{NetworkMessage, OutgoingMessage};
use super::manager::SyncMessage;
use crate::service::NetworkMessage;
use beacon_chain::{BeaconChain, BeaconChainTypes, BlockProcessingOutcome};
use eth2_libp2p::rpc::methods::*;
use eth2_libp2p::rpc::{RPCEvent, RPCRequest, RPCResponse, RequestId};
use eth2_libp2p::PeerId;
use slog::{debug, info, o, trace, warn};
use ssz::Encode;
use std::ops::Sub;
use std::sync::Arc;
use store::Store;
use tokio::sync::mpsc;
use tokio::sync::{mpsc, oneshot};
use types::{Attestation, BeaconBlock, Epoch, EthSpec, Hash256, Slot};
//TODO: Put a maximum limit on the number of block that can be requested.
//TODO: Rate limit requests
/// If a block is more than `FUTURE_SLOT_TOLERANCE` slots ahead of our slot clock, we drop it.
/// Otherwise we queue it.
pub(crate) const FUTURE_SLOT_TOLERANCE: u64 = 1;
/// The number of slots behind our head that we still treat a peer as a fully synced peer.
const FULL_PEER_TOLERANCE: u64 = 10;
const SHOULD_FORWARD_GOSSIP_BLOCK: bool = true;
const SHOULD_NOT_FORWARD_GOSSIP_BLOCK: bool = false;
@ -49,45 +49,63 @@ impl<T: BeaconChainTypes> From<&Arc<BeaconChain<T>>> for PeerSyncInfo {
}
}
/// The current syncing state.
#[derive(PartialEq)]
pub enum SyncState {
_Idle,
_Downloading,
_Stopped,
}
/// Simple Syncing protocol.
pub struct SimpleSync<T: BeaconChainTypes> {
/// Processes validated messages from the network. It relays necessary data to the syncing thread
/// and processes blocks from the pubsub network.
pub struct MessageProcessor<T: BeaconChainTypes> {
/// A reference to the underlying beacon chain.
chain: Arc<BeaconChain<T>>,
manager: ImportManager<T>,
/// A channel to the syncing thread.
sync_send: mpsc::UnboundedSender<SyncMessage<T::EthSpec>>,
/// A oneshot channel for destroying the sync thread.
_sync_exit: oneshot::Sender<()>,
/// A nextwork context to return and handle RPC requests.
network: NetworkContext,
/// The `RPCHandler` logger.
log: slog::Logger,
}
impl<T: BeaconChainTypes> SimpleSync<T> {
/// Instantiate a `SimpleSync` instance, with no peers and an empty queue.
impl<T: BeaconChainTypes> MessageProcessor<T> {
/// Instantiate a `MessageProcessor` instance
pub fn new(
executor: &tokio::runtime::TaskExecutor,
beacon_chain: Arc<BeaconChain<T>>,
network_send: mpsc::UnboundedSender<NetworkMessage>,
log: &slog::Logger,
) -> Self {
let sync_logger = log.new(o!("Service"=> "Sync"));
let sync_network_context = NetworkContext::new(network_send.clone(), sync_logger.clone());
SimpleSync {
chain: beacon_chain.clone(),
manager: ImportManager::new(beacon_chain, log),
// spawn the sync thread
let (sync_send, _sync_exit) = super::manager::spawn(
executor,
Arc::downgrade(&beacon_chain),
sync_network_context,
sync_logger,
);
MessageProcessor {
chain: beacon_chain,
sync_send,
_sync_exit,
network: NetworkContext::new(network_send, log.clone()),
log: sync_logger,
log: log.clone(),
}
}
fn send_to_sync(&mut self, message: SyncMessage<T::EthSpec>) {
self.sync_send.try_send(message).unwrap_or_else(|_| {
warn!(
self.log,
"Could not send message to the sync service";
)
});
}
/// Handle a peer disconnect.
///
/// Removes the peer from the manager.
pub fn on_disconnect(&mut self, peer_id: PeerId) {
self.manager.peer_disconnect(&peer_id);
self.send_to_sync(SyncMessage::Disconnect(peer_id));
}
/// Handle the connection of a new peer.
@ -107,6 +125,7 @@ impl<T: BeaconChainTypes> SimpleSync<T> {
request_id: RequestId,
hello: HelloMessage,
) {
// ignore hello responses if we are shutting down
trace!(self.log, "HelloRequest"; "peer" => format!("{:?}", peer_id));
// Say hello back.
@ -149,7 +168,7 @@ impl<T: BeaconChainTypes> SimpleSync<T> {
} else if remote.finalized_epoch <= local.finalized_epoch
&& remote.finalized_root != Hash256::zero()
&& local.finalized_root != Hash256::zero()
&& (self.root_at_slot(start_slot(remote.finalized_epoch))
&& (self.chain.root_at_slot(start_slot(remote.finalized_epoch))
!= Some(remote.finalized_root))
{
// The remotes finalized epoch is less than or greater than ours, but the block root is
@ -189,18 +208,16 @@ impl<T: BeaconChainTypes> SimpleSync<T> {
.exists::<BeaconBlock<T::EthSpec>>(&remote.head_root)
.unwrap_or_else(|_| false)
{
trace!(
self.log, "Peer with known chain found";
"peer" => format!("{:?}", peer_id),
"remote_head_slot" => remote.head_slot,
"remote_latest_finalized_epoch" => remote.finalized_epoch,
);
// If the node's best-block is already known to us and they are close to our current
// head, treat them as a fully sync'd peer.
if self.chain.best_slot().sub(remote.head_slot).as_u64() < FULL_PEER_TOLERANCE {
self.manager.add_full_peer(peer_id);
self.process_sync();
} else {
debug!(
self.log,
"Out of sync peer connected";
"peer" => format!("{:?}", peer_id),
);
}
self.send_to_sync(SyncMessage::AddPeer(peer_id, remote));
} else {
// The remote node has an equal or great finalized epoch and we don't know it's head.
//
@ -212,87 +229,10 @@ impl<T: BeaconChainTypes> SimpleSync<T> {
"local_finalized_epoch" => local.finalized_epoch,
"remote_latest_finalized_epoch" => remote.finalized_epoch,
);
self.manager.add_peer(peer_id, remote);
self.process_sync();
self.send_to_sync(SyncMessage::AddPeer(peer_id, remote));
}
}
fn process_sync(&mut self) {
loop {
match self.manager.poll() {
ImportManagerOutcome::Hello(peer_id) => {
trace!(
self.log,
"RPC Request";
"method" => "HELLO",
"peer" => format!("{:?}", peer_id)
);
self.network.send_rpc_request(
None,
peer_id,
RPCRequest::Hello(hello_message(&self.chain)),
);
}
ImportManagerOutcome::RequestBlocks {
peer_id,
request_id,
request,
} => {
trace!(
self.log,
"RPC Request";
"method" => "BeaconBlocks",
"id" => request_id,
"count" => request.count,
"peer" => format!("{:?}", peer_id)
);
self.network.send_rpc_request(
Some(request_id),
peer_id.clone(),
RPCRequest::BeaconBlocks(request),
);
}
ImportManagerOutcome::RecentRequest(peer_id, req) => {
trace!(
self.log,
"RPC Request";
"method" => "RecentBeaconBlocks",
"count" => req.block_roots.len(),
"peer" => format!("{:?}", peer_id)
);
self.network.send_rpc_request(
None,
peer_id.clone(),
RPCRequest::RecentBeaconBlocks(req),
);
}
ImportManagerOutcome::DownvotePeer(peer_id) => {
trace!(
self.log,
"Peer downvoted";
"peer" => format!("{:?}", peer_id)
);
// TODO: Implement reputation
self.network
.disconnect(peer_id.clone(), GoodbyeReason::Fault);
}
ImportManagerOutcome::Idle => {
// nothing to do
return;
}
}
}
}
//TODO: Move to beacon chain
fn root_at_slot(&self, target_slot: Slot) -> Option<Hash256> {
self.chain
.rev_iter_block_roots()
.find(|(_root, slot)| *slot == target_slot)
.map(|(root, _slot)| root)
}
/// Handle a `RecentBeaconBlocks` request from the peer.
pub fn on_recent_beacon_blocks_request(
&mut self,
@ -321,7 +261,7 @@ impl<T: BeaconChainTypes> SimpleSync<T> {
debug!(
self.log,
"BlockBodiesRequest";
"RecentBeaconBlocksRequest";
"peer" => format!("{:?}", peer_id),
"requested" => request.block_roots.len(),
"returned" => blocks.len(),
@ -380,18 +320,16 @@ impl<T: BeaconChainTypes> SimpleSync<T> {
blocks.reverse();
blocks.dedup_by_key(|brs| brs.slot);
if blocks.len() as u64 != req.count {
debug!(
self.log,
"BeaconBlocksRequest response";
"peer" => format!("{:?}", peer_id),
"msg" => "Failed to return all requested hashes",
"start_slot" => req.start_slot,
"current_slot" => format!("{:?}", self.chain.slot()),
"requested" => req.count,
"returned" => blocks.len(),
);
}
debug!(
self.log,
"BeaconBlocksRequest response";
"peer" => format!("{:?}", peer_id),
"msg" => "Failed to return all requested hashes",
"start_slot" => req.start_slot,
"current_slot" => self.chain.slot().unwrap_or_else(|_| Slot::from(0_u64)).as_u64(),
"requested" => req.count,
"returned" => blocks.len(),
);
self.network.send_rpc_response(
peer_id,
@ -414,10 +352,11 @@ impl<T: BeaconChainTypes> SimpleSync<T> {
"count" => beacon_blocks.len(),
);
self.manager
.beacon_blocks_response(peer_id, request_id, beacon_blocks);
self.process_sync();
self.send_to_sync(SyncMessage::BeaconBlocksResponse {
peer_id,
request_id,
beacon_blocks,
});
}
/// Handle a `RecentBeaconBlocks` response from the peer.
@ -429,15 +368,16 @@ impl<T: BeaconChainTypes> SimpleSync<T> {
) {
debug!(
self.log,
"BeaconBlocksResponse";
"RecentBeaconBlocksResponse";
"peer" => format!("{:?}", peer_id),
"count" => beacon_blocks.len(),
);
self.manager
.recent_blocks_response(peer_id, request_id, beacon_blocks);
self.process_sync();
self.send_to_sync(SyncMessage::RecentBeaconBlocksResponse {
peer_id,
request_id,
beacon_blocks,
});
}
/// Process a gossip message declaring a new block.
@ -455,9 +395,9 @@ impl<T: BeaconChainTypes> SimpleSync<T> {
}
BlockProcessingOutcome::ParentUnknown { parent: _ } => {
// Inform the sync manager to find parents for this block
trace!(self.log, "Unknown parent gossip";
trace!(self.log, "Block with unknown parent received";
"peer_id" => format!("{:?}",peer_id));
self.manager.add_unknown_block(block.clone(), peer_id);
self.send_to_sync(SyncMessage::UnknownBlock(peer_id, block.clone()));
SHOULD_FORWARD_GOSSIP_BLOCK
}
BlockProcessingOutcome::FutureSlot {
@ -468,7 +408,7 @@ impl<T: BeaconChainTypes> SimpleSync<T> {
SHOULD_FORWARD_GOSSIP_BLOCK
}
BlockProcessingOutcome::BlockIsAlreadyKnown => SHOULD_FORWARD_GOSSIP_BLOCK,
_ => SHOULD_NOT_FORWARD_GOSSIP_BLOCK,
_ => SHOULD_NOT_FORWARD_GOSSIP_BLOCK, //TODO: Decide if we want to forward these
}
} else {
SHOULD_NOT_FORWARD_GOSSIP_BLOCK
@ -491,15 +431,10 @@ impl<T: BeaconChainTypes> SimpleSync<T> {
}
}
}
/// Generates our current state in the form of a HELLO RPC message.
pub fn generate_hello(&self) -> HelloMessage {
hello_message(&self.chain)
}
}
/// Build a `HelloMessage` representing the state of the given `beacon_chain`.
fn hello_message<T: BeaconChainTypes>(beacon_chain: &BeaconChain<T>) -> HelloMessage {
pub(crate) fn hello_message<T: BeaconChainTypes>(beacon_chain: &BeaconChain<T>) -> HelloMessage {
let state = &beacon_chain.head().beacon_state;
HelloMessage {
@ -527,7 +462,7 @@ impl NetworkContext {
pub fn disconnect(&mut self, peer_id: PeerId, reason: GoodbyeReason) {
warn!(
&self.log,
"Disconnecting peer";
"Disconnecting peer (RPC)";
"reason" => format!("{:?}", reason),
"peer_id" => format!("{:?}", peer_id),
);
@ -560,12 +495,8 @@ impl NetworkContext {
}
fn send_rpc_event(&mut self, peer_id: PeerId, rpc_event: RPCEvent) {
self.send(peer_id, OutgoingMessage::RPC(rpc_event))
}
fn send(&mut self, peer_id: PeerId, outgoing_message: OutgoingMessage) {
self.network_send
.try_send(NetworkMessage::Send(peer_id, outgoing_message))
.try_send(NetworkMessage::RPC(peer_id, rpc_event))
.unwrap_or_else(|_| {
warn!(
self.log,

20
beacon_node/src/main.rs
View File

@ -33,14 +33,14 @@ fn main() {
.arg(
Arg::with_name("logfile")
.long("logfile")
.value_name("logfile")
.value_name("FILE")
.help("File path where output will be written.")
.takes_value(true),
)
.arg(
Arg::with_name("network-dir")
.long("network-dir")
.value_name("NETWORK-DIR")
.value_name("DIR")
.help("Data directory for network keys.")
.takes_value(true)
.global(true)
@ -83,7 +83,7 @@ fn main() {
Arg::with_name("boot-nodes")
.long("boot-nodes")
.allow_hyphen_values(true)
.value_name("BOOTNODES")
.value_name("ENR-LIST")
.help("One or more comma-delimited base64-encoded ENR's to bootstrap the p2p network.")
.takes_value(true),
)
@ -128,13 +128,14 @@ fn main() {
.arg(
Arg::with_name("rpc-address")
.long("rpc-address")
.value_name("Address")
.value_name("ADDRESS")
.help("Listen address for RPC endpoint.")
.takes_value(true),
)
.arg(
Arg::with_name("rpc-port")
.long("rpc-port")
.value_name("PORT")
.help("Listen port for RPC endpoint.")
.conflicts_with("port-bump")
.takes_value(true),
@ -149,14 +150,14 @@ fn main() {
.arg(
Arg::with_name("api-address")
.long("api-address")
.value_name("APIADDRESS")
.value_name("ADDRESS")
.help("Set the listen address for the RESTful HTTP API server.")
.takes_value(true),
)
.arg(
Arg::with_name("api-port")
.long("api-port")
.value_name("APIPORT")
.value_name("PORT")
.help("Set the listen TCP port for the RESTful HTTP API server.")
.conflicts_with("port-bump")
.takes_value(true),
@ -196,13 +197,6 @@ fn main() {
.possible_values(&["info", "debug", "trace", "warn", "error", "crit"])
.default_value("trace"),
)
.arg(
Arg::with_name("verbosity")
.short("v")
.multiple(true)
.help("Sets the verbosity level")
.takes_value(true),
)
/*
* The "testnet" sub-command.
*