From 43135484ca7f3831ea9e7c63e22e330a931c4644 Mon Sep 17 00:00:00 2001 From: Age Manning Date: Mon, 13 May 2019 17:50:11 +1000 Subject: [PATCH] Update to lastest libp2p --- beacon_node/client/src/client_config.rs | 2 +- beacon_node/eth2-libp2p/src/behaviour.rs | 42 +-- beacon_node/eth2-libp2p/src/core-behaviour.rs | 279 ------------------ beacon_node/eth2-libp2p/src/service.rs | 9 +- 4 files changed, 31 insertions(+), 301 deletions(-) delete mode 100644 beacon_node/eth2-libp2p/src/core-behaviour.rs diff --git a/beacon_node/client/src/client_config.rs b/beacon_node/client/src/client_config.rs index 4d0e286b0..74115c547 100644 --- a/beacon_node/client/src/client_config.rs +++ b/beacon_node/client/src/client_config.rs @@ -1,6 +1,5 @@ use clap::ArgMatches; use eth2_libp2p::multiaddr::Protocol; -use eth2_libp2p::multiaddr::ToMultiaddr; use eth2_libp2p::Multiaddr; use fork_choice::ForkChoiceAlgorithm; use http_server::HttpServerConfig; @@ -51,6 +50,7 @@ impl Default for ClientConfig { } impl ClientConfig { +<<<<<<< HEAD /// Returns the path to which the client may initialize an on-disk database. pub fn db_path(&self) -> Option { self.data_dir() diff --git a/beacon_node/eth2-libp2p/src/behaviour.rs b/beacon_node/eth2-libp2p/src/behaviour.rs index 7ddbd95b7..58f603276 100644 --- a/beacon_node/eth2-libp2p/src/behaviour.rs +++ b/beacon_node/eth2-libp2p/src/behaviour.rs @@ -3,7 +3,6 @@ use crate::rpc::{RPCEvent, RPCMessage, Rpc}; use crate::NetworkConfig; use crate::{Topic, TopicHash}; use futures::prelude::*; -use libp2p::Multiaddr; use libp2p::{ core::{ swarm::{NetworkBehaviourAction, NetworkBehaviourEventProcess}, @@ -12,23 +11,24 @@ use libp2p::{ gossipsub::{Gossipsub, GossipsubEvent}, identify::{protocol::IdentifyInfo, Identify, IdentifyEvent}, kad::KademliaOut, - ping::{Ping, PingEvent}, + ping::{Ping, PingConfig, PingEvent}, tokio_io::{AsyncRead, AsyncWrite}, NetworkBehaviour, PeerId, }; use slog::{debug, o, trace, warn}; use ssz::{ssz_encode, Decode, DecodeError, Encode}; +use std::collections::HashMap; +use std::num::NonZeroU32; use std::time::{Duration, Instant}; use tokio_timer::Delay; -use std::collections::HashMap; use types::{Attestation, BeaconBlock}; /// Builds the network behaviour that manages the core protocols of eth2. /// This core behaviour is managed by `Behaviour` which adds peer management to all core /// behaviours. #[derive(NetworkBehaviour)] -#[behaviour(out_event = "CoreBehaviourEvent", poll_method = "poll")] -pub struct CoreCoreBehaviourTSubstream: AsyncRead + AsyncWrite> { +#[behaviour(out_event = "BehaviourEvent", poll_method = "poll")] +pub struct Behaviour { /// The routing pub-sub mechanism for eth2. gossipsub: Gossipsub, /// The serenity RPC specified in the wire-0 protocol. @@ -41,7 +41,7 @@ pub struct CoreCoreBehaviourTSubstream: AsyncRead + AsyncWrite> { discovery: Discovery, #[behaviour(ignore)] /// The events generated by this behaviour to be consumed in the swarm poll. - events: Vec, + events: Vec, /// Logger for behaviour actions. #[behaviour(ignore)] log: slog::Logger, @@ -49,7 +49,7 @@ pub struct CoreCoreBehaviourTSubstream: AsyncRead + AsyncWrite> { // Implement the NetworkBehaviourEventProcess trait so that we can derive NetworkBehaviour for Behaviour impl NetworkBehaviourEventProcess - for CoreBehaviourTSubstream> + for Behaviour { fn inject_event(&mut self, event: GossipsubEvent) { match event { @@ -82,7 +82,7 @@ impl NetworkBehaviourEventProcess NetworkBehaviourEventProcess - for CoreBehaviourTSubstream> + for Behaviour { fn inject_event(&mut self, event: RPCMessage) { match event { @@ -97,7 +97,7 @@ impl NetworkBehaviourEventProcess NetworkBehaviourEventProcess - for CoreBehaviourTSubstream> + for Behaviour { fn inject_event(&mut self, event: IdentifyEvent) { match event { @@ -127,7 +127,7 @@ impl NetworkBehaviourEventProcess NetworkBehaviourEventProcess - for CoreBehaviourTSubstream> + for Behaviour { fn inject_event(&mut self, _event: PingEvent) { // not interested in ping responses at the moment. @@ -136,19 +136,28 @@ impl NetworkBehaviourEventProcess // implement the discovery behaviour (currently kademlia) impl NetworkBehaviourEventProcess - for CoreBehaviourTSubstream> + for Behaviour { fn inject_event(&mut self, _out: KademliaOut) { // not interested in kademlia results at the moment } } -impl CoreBehaviourTSubstream> { +impl Behaviour { pub fn new(local_public_key: PublicKey, net_conf: &NetworkConfig, log: &slog::Logger) -> Self { let local_peer_id = local_public_key.clone().into_peer_id(); - let identify_config = net_conf.identify_config.clone(); let behaviour_log = log.new(o!()); + // identify configuration + let identify_config = net_conf.identify_config.clone(); + + // ping configuration + let ping_config = PingConfig::new() + .with_timeout(Duration::from_secs(30)) + .with_interval(Duration::from_secs(20)) + .with_max_failures(NonZeroU32::new(2).expect("2 != 0")) + .with_keep_alive(false); + Behaviour { serenity_rpc: Rpc::new(log), gossipsub: Gossipsub::new(local_peer_id.clone(), net_conf.gs_config.clone()), @@ -158,7 +167,7 @@ impl CoreBehaviourTSubstream> { identify_config.user_agent, local_public_key, ), - ping: Ping::new(), + ping: Ping::new(ping_config), events: Vec::new(), log: behaviour_log, } @@ -177,7 +186,7 @@ impl CoreBehaviourTSubstream> { } /// Implements the combined behaviour for the libp2p service. -impl CoreBehaviourTSubstream> { +impl Behaviour { /* Pubsub behaviour functions */ /// Subscribes to a gossipsub topic. @@ -199,11 +208,10 @@ impl CoreBehaviourTSubstream> { pub fn send_rpc(&mut self, peer_id: PeerId, rpc_event: RPCEvent) { self.serenity_rpc.send_rpc(peer_id, rpc_event); } - } /// The types of events than can be obtained from polling the behaviour. -pub enum CoreBehaviourEvent { +pub enum BehaviourEvent { RPC(PeerId, RPCEvent), PeerDialed(PeerId), Identified(PeerId, Box), diff --git a/beacon_node/eth2-libp2p/src/core-behaviour.rs b/beacon_node/eth2-libp2p/src/core-behaviour.rs deleted file mode 100644 index e59183b4c..000000000 --- a/beacon_node/eth2-libp2p/src/core-behaviour.rs +++ /dev/null @@ -1,279 +0,0 @@ -use crate::discovery::Discovery; -use crate::rpc::{RPCEvent, RPCMessage, Rpc}; -use crate::NetworkConfig; -use crate::{Topic, TopicHash}; -use futures::prelude::*; -use libp2p::Multiaddr; -use libp2p::{ - core::{ - swarm::{NetworkBehaviourAction, NetworkBehaviourEventProcess}, - PublicKey, - }, - gossipsub::{Gossipsub, GossipsubEvent}, - identify::{protocol::IdentifyInfo, Identify, IdentifyEvent}, - kad::KademliaOut, - ping::{Ping, PingEvent}, - tokio_io::{AsyncRead, AsyncWrite}, - NetworkBehaviour, PeerId, -}; -use slog::{debug, o, trace, warn}; -use ssz::{ssz_encode, Decodable, DecodeError, Encodable, SszStream}; -use std::collections::HashMap; -use types::{Attestation, BeaconBlock}; - -/// Builds the network behaviour that manages the core protocols of eth2. -/// This core behaviour is managed by `Behaviour` which adds peer management to all core -/// behaviours. -#[derive(NetworkBehaviour)] -#[behaviour(out_event = "CoreBehaviourEvent", poll_method = "poll")] -pub struct CoreBehaviour { - /// The routing pub-sub mechanism for eth2. - gossipsub: Gossipsub, - /// The serenity RPC specified in the wire-0 protocol. - serenity_rpc: Rpc, - /// Allows discovery of IP addresses for peers on the network. - identify: Identify, - /// Keep regular connection to peers and disconnect if absent. - ping: Ping, - /// Kademlia for peer discovery. - discovery: Discovery, - #[behaviour(ignore)] - /// The events generated by this behaviour to be consumed by the global behaviour. - events: Vec, - /// Logger for behaviour actions. - #[behaviour(ignore)] - log: slog::Logger, -} - -impl CoreBehaviour { - pub fn new(local_public_key: PublicKey, net_conf: &NetworkConfig, log: &slog::Logger) -> Self { - let local_peer_id = local_public_key.clone().into_peer_id(); - let identify_config = net_conf.identify_config.clone(); - let behaviour_log = log.new(o!()); - - CoreBehaviour { - serenity_rpc: Rpc::new(log), - gossipsub: Gossipsub::new(local_peer_id.clone(), net_conf.gs_config.clone()), - discovery: Discovery::new(local_peer_id, log), - identify: Identify::new( - identify_config.version, - identify_config.user_agent, - local_public_key, - ), - ping: Ping::new(), - events: Vec::new(), - log: behaviour_log, - } - } - - /// Consumes the events list when polled. - fn poll( - &mut self, - ) -> Async> { - if !self.events.is_empty() { - return Async::Ready(NetworkBehaviourAction::GenerateEvent(self.events.remove(0))); - } - - Async::NotReady - } -} - -// Implement the NetworkBehaviourEventProcess trait so that we can derive NetworkBehaviour for CoreBehaviour -impl NetworkBehaviourEventProcess - for CoreBehaviour -{ - fn inject_event(&mut self, event: GossipsubEvent) { - match event { - GossipsubEvent::Message(gs_msg) => { - trace!(self.log, "Received GossipEvent"; "msg" => format!("{:?}", gs_msg)); - - let pubsub_message = match PubsubMessage::ssz_decode(&gs_msg.data, 0) { - //TODO: Punish peer on error - Err(e) => { - warn!( - self.log, - "Received undecodable message from Peer {:?} error", gs_msg.source; - "error" => format!("{:?}", e) - ); - return; - } - Ok((msg, _index)) => msg, - }; - - self.events.push(BehaviourEvent::GossipMessage { - source: gs_msg.source, - topics: gs_msg.topics, - message: pubsub_message, - }); - } - GossipsubEvent::Subscribed { - peer_id: _, - topic: _, - } - | GossipsubEvent::Unsubscribed { - peer_id: _, - topic: _, - } => {} - } - } -} - -impl NetworkBehaviourEventProcess - for CoreBehaviour -{ - fn inject_event(&mut self, event: RPCMessage) { - match event { - RPCMessage::PeerDialed(peer_id) => { - self.events.push(BehaviourEvent::PeerDialed(peer_id)) - } - RPCMessage::RPC(peer_id, rpc_event) => { - self.events.push(BehaviourEvent::RPC(peer_id, rpc_event)) - } - } - } -} - -impl NetworkBehaviourEventProcess - for CoreBehaviour -{ - fn inject_event(&mut self, event: IdentifyEvent) { - match event { - IdentifyEvent::Identified { - peer_id, mut info, .. - } => { - if info.listen_addrs.len() > 20 { - debug!( - self.log, - "More than 20 peers have been identified, truncating" - ); - info.listen_addrs.truncate(20); - } - trace!(self.log, "Found addresses"; "Peer Id" => format!("{:?}", peer_id), "Addresses" => format!("{:?}", info.listen_addrs)); - // inject the found addresses into our discovery behaviour - for address in &info.listen_addrs { - self.discovery - .add_connected_address(&peer_id, address.clone()); - } - self.events.push(BehaviourEvent::Identified(peer_id, info)); - } - IdentifyEvent::Error { .. } => {} - IdentifyEvent::SendBack { .. } => {} - } - } -} - -impl NetworkBehaviourEventProcess - for CoreBehaviour -{ - fn inject_event(&mut self, _event: PingEvent) { - // not interested in ping responses at the moment. - } -} - -// implement the discovery behaviour (currently kademlia) -impl NetworkBehaviourEventProcess - for CoreBehaviour -{ - fn inject_event(&mut self, _out: KademliaOut) { - // not interested in kademlia results at the moment - } -} - -/// Implements the combined behaviour for the libp2p service. -impl CoreBehaviour { - /* Pubsub behaviour functions */ - - /// Subscribes to a gossipsub topic. - pub fn subscribe(&mut self, topic: Topic) -> bool { - self.gossipsub.subscribe(topic) - } - - /// Publishes a message on the pubsub (gossipsub) behaviour. - pub fn publish(&mut self, topics: Vec, message: PubsubMessage) { - let message_bytes = ssz_encode(&message); - for topic in topics { - self.gossipsub.publish(topic, message_bytes.clone()); - } - } - - /* Eth2 RPC behaviour functions */ - - /// Sends an RPC Request/Response via the RPC protocol. - pub fn send_rpc(&mut self, peer_id: PeerId, rpc_event: RPCEvent) { - self.serenity_rpc.send_rpc(peer_id, rpc_event); - } -} - -/// The types of events than can be obtained from polling the behaviour. -pub enum CoreBehaviourEvent { - RPC(PeerId, RPCEvent), - PeerDialed(PeerId), - Identified(PeerId, IdentifyInfo), - // TODO: This is a stub at the moment - GossipMessage { - source: PeerId, - topics: Vec, - message: PubsubMessage, - }, -} - -/// Messages that are passed to and from the pubsub (Gossipsub) behaviour. -#[derive(Debug, Clone, PartialEq)] -pub enum PubsubMessage { - /// Gossipsub message providing notification of a new block. - Block(BeaconBlock), - /// Gossipsub message providing notification of a new attestation. - Attestation(Attestation), -} - -//TODO: Correctly encode/decode enums. Prefixing with integer for now. -impl Encodable for PubsubMessage { - fn ssz_append(&self, s: &mut SszStream) { - match self { - PubsubMessage::Block(block_gossip) => { - 0u32.ssz_append(s); - block_gossip.ssz_append(s); - } - PubsubMessage::Attestation(attestation_gossip) => { - 1u32.ssz_append(s); - attestation_gossip.ssz_append(s); - } - } - } -} - -impl Decodable for PubsubMessage { - fn ssz_decode(bytes: &[u8], index: usize) -> Result<(Self, usize), DecodeError> { - let (id, index) = u32::ssz_decode(bytes, index)?; - match id { - 0 => { - let (block, index) = BeaconBlock::ssz_decode(bytes, index)?; - Ok((PubsubMessage::Block(block), index)) - } - 1 => { - let (attestation, index) = Attestation::ssz_decode(bytes, index)?; - Ok((PubsubMessage::Attestation(attestation), index)) - } - _ => Err(DecodeError::Invalid), - } - } -} - -#[cfg(test)] -mod test { - use super::*; - use types::*; - - #[test] - fn ssz_encoding() { - let original = PubsubMessage::Block(BeaconBlock::empty(&ChainSpec::foundation())); - - let encoded = ssz_encode(&original); - - println!("{:?}", encoded); - - let (decoded, _i) = PubsubMessage::ssz_decode(&encoded, 0).unwrap(); - - assert_eq!(original, decoded); - } -} diff --git a/beacon_node/eth2-libp2p/src/service.rs b/beacon_node/eth2-libp2p/src/service.rs index 9cbceda8d..68ca72620 100644 --- a/beacon_node/eth2-libp2p/src/service.rs +++ b/beacon_node/eth2-libp2p/src/service.rs @@ -63,11 +63,12 @@ impl Service { .map_err(|e| format!("Invalid listen multiaddr: {}", e))? { match Swarm::listen_on(&mut swarm, address.clone()) { - Ok(mut listen_addr) => { - listen_addr.append(Protocol::P2p(local_peer_id.clone().into())); - info!(log, "Listening on: {}", listen_addr); + Ok(_) => { + let mut log_address = address.clone(); + log_address.push(Protocol::P2p(local_peer_id.clone().into())); + info!(log, "Listening on: {}", log_address); } - Err(err) => warn!(log, "Cannot listen on: {} : {:?}", address, err), + Err(err) => warn!(log, "Cannot listen on: {} because: {:?}", address, err), }; } // connect to boot nodes - these are currently stored as multiaddrs