gnosis/lighthouse
1
0
Fork 0
mirror of https://github.com/sigp/lighthouse.git synced 2026-03-20 21:34:46 +00:00
Code Issues Packages Projects Releases Wiki Activity

Initial Interop Updates (#492)

Browse Source 
* Add interop chain spec and rename chain_id

* Add ability to connect to raw libp2p nodes

* Adds Identify protocol, clean up RPC protocol name handling

* Update to latest libp2p, gossipsub improvements

* Updates to latest interop branch.

- Shifts decoding of objects into message handler.
- Updates to latest interop gossipsub.
- Adds interop spec constant.

* Configuration updates allow for verbosity CLI flag and spec constants

* Update submodules to master

* Correct minimal chainspec modifications

* Duplication of validator polls are no longer fatal

* Apply PR suggestions
This commit is contained in:
Age Manning
2019-08-10 11:44:17 +10:00
committed by Paul Hauner
parent 3a1f56a42e
commit 468015f9bb
28 changed files with 590 additions and 436 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -2,47 +2,52 @@ use crate::discovery::Discovery;
use crate::rpc::{RPCEvent, RPCMessage, RPC};
use crate::{error, NetworkConfig};
use crate::{Topic, TopicHash};
use crate::{BEACON_ATTESTATION_TOPIC, BEACON_BLOCK_TOPIC};
use futures::prelude::*;
use libp2p::{
core::{
identity::Keypair,
swarm::{NetworkBehaviourAction, NetworkBehaviourEventProcess},
},
core::identity::Keypair,
discv5::Discv5Event,
gossipsub::{Gossipsub, GossipsubEvent},
identify::{Identify, IdentifyEvent},
ping::{Ping, PingConfig, PingEvent},
swarm::{NetworkBehaviourAction, NetworkBehaviourEventProcess},
tokio_io::{AsyncRead, AsyncWrite},
NetworkBehaviour, PeerId,
};
use slog::{o, trace, warn};
use ssz::{ssz_encode, Decode, DecodeError, Encode};
use slog::{debug, o, trace};
use ssz::{ssz_encode, Encode};
use std::num::NonZeroU32;
use std::time::Duration;
use types::{Attestation, BeaconBlock, EthSpec};
const MAX_IDENTIFY_ADDRESSES: usize = 20;
/// 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 = "BehaviourEvent<E>", poll_method = "poll")]
pub struct Behaviour<TSubstream: AsyncRead + AsyncWrite, E: EthSpec> {
#[behaviour(out_event = "BehaviourEvent", poll_method = "poll")]
pub struct Behaviour<TSubstream: AsyncRead + AsyncWrite> {
/// The routing pub-sub mechanism for eth2.
gossipsub: Gossipsub<TSubstream>,
/// The serenity RPC specified in the wire-0 protocol.
serenity_rpc: RPC<TSubstream, E>,
/// The Eth2 RPC specified in the wire-0 protocol.
eth2_rpc: RPC<TSubstream>,
/// Keep regular connection to peers and disconnect if absent.
// TODO: Remove Libp2p ping in favour of discv5 ping.
ping: Ping<TSubstream>,
/// Kademlia for peer discovery.
// TODO: Using id for initial interop. This will be removed by mainnet.
/// Provides IP addresses and peer information.
identify: Identify<TSubstream>,
/// Discovery behaviour.
discovery: Discovery<TSubstream>,
#[behaviour(ignore)]
/// The events generated by this behaviour to be consumed in the swarm poll.
events: Vec<BehaviourEvent<E>>,
events: Vec<BehaviourEvent>,
/// Logger for behaviour actions.
#[behaviour(ignore)]
log: slog::Logger,
}
impl<TSubstream: AsyncRead + AsyncWrite, E: EthSpec> Behaviour<TSubstream, E> {
impl<TSubstream: AsyncRead + AsyncWrite> Behaviour<TSubstream> {
pub fn new(
local_key: &Keypair,
net_conf: &NetworkConfig,
@@ -50,17 +55,25 @@ impl<TSubstream: AsyncRead + AsyncWrite, E: EthSpec> Behaviour<TSubstream, E> {
) -> error::Result<Self> {
let local_peer_id = local_key.public().clone().into_peer_id();
let behaviour_log = log.new(o!());
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);
let identify = Identify::new(
"lighthouse/libp2p".into(),
version::version(),
local_key.public(),
);
Ok(Behaviour {
serenity_rpc: RPC::new(log),
eth2_rpc: RPC::new(log),
gossipsub: Gossipsub::new(local_peer_id.clone(), net_conf.gs_config.clone()),
discovery: Discovery::new(local_key, net_conf, log)?,
ping: Ping::new(ping_config),
identify,
events: Vec::new(),
log: behaviour_log,
})
@@ -68,31 +81,20 @@ impl<TSubstream: AsyncRead + AsyncWrite, E: EthSpec> Behaviour<TSubstream, E> {
}
// Implement the NetworkBehaviourEventProcess trait so that we can derive NetworkBehaviour for Behaviour
impl<TSubstream: AsyncRead + AsyncWrite, E: EthSpec> NetworkBehaviourEventProcess<GossipsubEvent>
for Behaviour<TSubstream, E>
impl<TSubstream: AsyncRead + AsyncWrite> NetworkBehaviourEventProcess<GossipsubEvent>
for Behaviour<TSubstream>
{
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::from_ssz_bytes(&gs_msg.data) {
//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) => msg,
};
let msg = PubsubMessage::from_topics(&gs_msg.topics, gs_msg.data);
self.events.push(BehaviourEvent::GossipMessage {
source: gs_msg.source,
topics: gs_msg.topics,
message: Box::new(pubsub_message),
message: msg,
});
}
GossipsubEvent::Subscribed { .. } => {}
@@ -101,8 +103,8 @@ impl<TSubstream: AsyncRead + AsyncWrite, E: EthSpec> NetworkBehaviourEventProces
}
}
impl<TSubstream: AsyncRead + AsyncWrite, E: EthSpec> NetworkBehaviourEventProcess<RPCMessage>
for Behaviour<TSubstream, E>
impl<TSubstream: AsyncRead + AsyncWrite> NetworkBehaviourEventProcess<RPCMessage>
for Behaviour<TSubstream>
{
fn inject_event(&mut self, event: RPCMessage) {
match event {
@@ -119,19 +121,19 @@ impl<TSubstream: AsyncRead + AsyncWrite, E: EthSpec> NetworkBehaviourEventProces
}
}
impl<TSubstream: AsyncRead + AsyncWrite, E: EthSpec> NetworkBehaviourEventProcess<PingEvent>
for Behaviour<TSubstream, E>
impl<TSubstream: AsyncRead + AsyncWrite> NetworkBehaviourEventProcess<PingEvent>
for Behaviour<TSubstream>
{
fn inject_event(&mut self, _event: PingEvent) {
// not interested in ping responses at the moment.
}
}
impl<TSubstream: AsyncRead + AsyncWrite, E: EthSpec> Behaviour<TSubstream, E> {
impl<TSubstream: AsyncRead + AsyncWrite> Behaviour<TSubstream> {
/// Consumes the events list when polled.
fn poll<TBehaviourIn>(
&mut self,
) -> Async<NetworkBehaviourAction<TBehaviourIn, BehaviourEvent<E>>> {
) -> Async<NetworkBehaviourAction<TBehaviourIn, BehaviourEvent>> {
if !self.events.is_empty() {
return Async::Ready(NetworkBehaviourAction::GenerateEvent(self.events.remove(0)));
}
@@ -140,8 +142,36 @@ impl<TSubstream: AsyncRead + AsyncWrite, E: EthSpec> Behaviour<TSubstream, E> {
}
}
impl<TSubstream: AsyncRead + AsyncWrite, E: EthSpec> NetworkBehaviourEventProcess<Discv5Event>
for Behaviour<TSubstream, E>
impl<TSubstream: AsyncRead + AsyncWrite> NetworkBehaviourEventProcess<IdentifyEvent>
for Behaviour<TSubstream>
{
fn inject_event(&mut self, event: IdentifyEvent) {
match event {
IdentifyEvent::Identified {
peer_id, mut info, ..
} => {
if info.listen_addrs.len() > MAX_IDENTIFY_ADDRESSES {
debug!(
self.log,
"More than 20 addresses have been identified, truncating"
);
info.listen_addrs.truncate(MAX_IDENTIFY_ADDRESSES);
}
debug!(self.log, "Identified Peer"; "Peer" => format!("{}", peer_id),
"Protocol Version" => info.protocol_version,
"Agent Version" => info.agent_version,
"Listening Addresses" => format!("{:?}", info.listen_addrs),
"Protocols" => format!("{:?}", info.protocols)
);
}
IdentifyEvent::Error { .. } => {}
IdentifyEvent::SendBack { .. } => {}
}
}
}
impl<TSubstream: AsyncRead + AsyncWrite> NetworkBehaviourEventProcess<Discv5Event>
for Behaviour<TSubstream>
{
fn inject_event(&mut self, _event: Discv5Event) {
// discv5 has no events to inject
@@ -149,7 +179,7 @@ impl<TSubstream: AsyncRead + AsyncWrite, E: EthSpec> NetworkBehaviourEventProces
}
/// Implements the combined behaviour for the libp2p service.
impl<TSubstream: AsyncRead + AsyncWrite, E: EthSpec> Behaviour<TSubstream, E> {
impl<TSubstream: AsyncRead + AsyncWrite> Behaviour<TSubstream> {
/* Pubsub behaviour functions */
/// Subscribes to a gossipsub topic.
@@ -158,7 +188,7 @@ impl<TSubstream: AsyncRead + AsyncWrite, E: EthSpec> Behaviour<TSubstream, E> {
}
/// Publishes a message on the pubsub (gossipsub) behaviour.
pub fn publish(&mut self, topics: Vec<Topic>, message: PubsubMessage<E>) {
pub fn publish(&mut self, topics: Vec<Topic>, message: PubsubMessage) {
let message_bytes = ssz_encode(&message);
for topic in topics {
self.gossipsub.publish(topic, message_bytes.clone());
@@ -169,7 +199,7 @@ impl<TSubstream: AsyncRead + AsyncWrite, E: EthSpec> Behaviour<TSubstream, E> {
/// 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);
self.eth2_rpc.send_rpc(peer_id, rpc_event);
}
/* Discovery / Peer management functions */
@@ -179,99 +209,60 @@ impl<TSubstream: AsyncRead + AsyncWrite, E: EthSpec> Behaviour<TSubstream, E> {
}
/// The types of events than can be obtained from polling the behaviour.
pub enum BehaviourEvent<E: EthSpec> {
pub enum BehaviourEvent {
RPC(PeerId, RPCEvent),
PeerDialed(PeerId),
PeerDisconnected(PeerId),
GossipMessage {
source: PeerId,
topics: Vec<TopicHash>,
message: Box<PubsubMessage<E>>,
message: PubsubMessage,
},
}
/// Messages that are passed to and from the pubsub (Gossipsub) behaviour.
#[derive(Debug, Clone, PartialEq)]
pub enum PubsubMessage<E: EthSpec> {
pub enum PubsubMessage {
/// Gossipsub message providing notification of a new block.
Block(BeaconBlock<E>),
Block(Vec<u8>),
/// Gossipsub message providing notification of a new attestation.
Attestation(Attestation<E>),
Attestation(Vec<u8>),
/// Gossipsub message from an unknown topic.
Unknown(Vec<u8>),
}
//TODO: Correctly encode/decode enums. Prefixing with integer for now.
impl<E: EthSpec> Encode for PubsubMessage<E> {
impl PubsubMessage {
/* Note: This is assuming we are not hashing topics. If we choose to hash topics, these will
* need to be modified.
*
* Also note that a message can be associated with many topics. As soon as one of the topics is
* known we match. If none of the topics are known we return an unknown state.
*/
fn from_topics(topics: &Vec<TopicHash>, data: Vec<u8>) -> Self {
for topic in topics {
match topic.as_str() {
BEACON_BLOCK_TOPIC => return PubsubMessage::Block(data),
BEACON_ATTESTATION_TOPIC => return PubsubMessage::Attestation(data),
_ => {}
}
}
PubsubMessage::Unknown(data)
}
}
impl Encode for PubsubMessage {
fn is_ssz_fixed_len() -> bool {
false
}
fn ssz_append(&self, buf: &mut Vec<u8>) {
let offset = <u32 as Encode>::ssz_fixed_len() + <Vec<u8> as Encode>::ssz_fixed_len();
let mut encoder = ssz::SszEncoder::container(buf, offset);
match self {
PubsubMessage::Block(block_gossip) => {
encoder.append(&0_u32);
PubsubMessage::Block(inner)
| PubsubMessage::Attestation(inner)
| PubsubMessage::Unknown(inner) => {
// Encode the gossip as a Vec<u8>;
encoder.append(&block_gossip.as_ssz_bytes());
}
PubsubMessage::Attestation(attestation_gossip) => {
encoder.append(&1_u32);
// Encode the gossip as a Vec<u8>;
encoder.append(&attestation_gossip.as_ssz_bytes());
buf.append(&mut inner.as_ssz_bytes());
}
}
encoder.finalize();
}
}
impl<E: EthSpec> Decode for PubsubMessage<E> {
fn is_ssz_fixed_len() -> bool {
false
}
fn from_ssz_bytes(bytes: &[u8]) -> Result<Self, ssz::DecodeError> {
let mut builder = ssz::SszDecoderBuilder::new(&bytes);
builder.register_type::<u32>()?;
builder.register_type::<Vec<u8>>()?;
let mut decoder = builder.build()?;
let id: u32 = decoder.decode_next()?;
let body: Vec<u8> = decoder.decode_next()?;
match id {
0 => Ok(PubsubMessage::Block(BeaconBlock::from_ssz_bytes(&body)?)),
1 => Ok(PubsubMessage::Attestation(Attestation::from_ssz_bytes(
&body,
)?)),
_ => Err(DecodeError::BytesInvalid(
"Invalid PubsubMessage id".to_string(),
)),
}
}
}
#[cfg(test)]
mod test {
use super::*;
use types::*;
#[test]
fn ssz_encoding() {
let original = PubsubMessage::Block(BeaconBlock::<MainnetEthSpec>::empty(
&MainnetEthSpec::default_spec(),
));
let encoded = ssz_encode(&original);
let decoded = PubsubMessage::from_ssz_bytes(&encoded).unwrap();
assert_eq!(original, decoded);
}
}

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

@@ -1,12 +1,13 @@
use clap::ArgMatches;
use enr::Enr;
use libp2p::gossipsub::{GossipsubConfig, GossipsubConfigBuilder};
use libp2p::Multiaddr;
use serde_derive::{Deserialize, Serialize};
use std::path::PathBuf;
use std::time::Duration;
/// The beacon node topic string to subscribe to.
pub const BEACON_PUBSUB_TOPIC: &str = "beacon_block";
pub const BEACON_BLOCK_TOPIC: &str = "beacon_block";
pub const BEACON_ATTESTATION_TOPIC: &str = "beacon_attestation";
pub const SHARD_TOPIC_PREFIX: &str = "shard";
@@ -39,6 +40,9 @@ pub struct Config {
/// List of nodes to initially connect to.
pub boot_nodes: Vec<Enr>,
/// List of libp2p nodes to initially connect to.
pub libp2p_nodes: Vec<Multiaddr>,
/// Client version
pub client_version: String,
@@ -60,12 +64,13 @@ impl Default for Config {
discovery_port: 9000,
max_peers: 10,
//TODO: Set realistic values for production
// Note: This defaults topics to plain strings. Not hashes
gs_config: GossipsubConfigBuilder::new()
.max_gossip_size(4_000_000)
.inactivity_timeout(Duration::from_secs(90))
.max_transmit_size(1_000_000)
.heartbeat_interval(Duration::from_secs(20))
.build(),
boot_nodes: vec![],
libp2p_nodes: vec![],
client_version: version::version(),
topics: Vec::new(),
}
@@ -118,6 +123,21 @@ impl Config {
.collect::<Result<Vec<Enr>, _>>()?;
}
if let Some(libp2p_addresses_str) = args.value_of("libp2p-addresses") {
self.libp2p_nodes = libp2p_addresses_str
.split(',')
.map(|multiaddr| {
multiaddr
.parse()
.map_err(|_| format!("Invalid Multiaddr: {}", multiaddr))
})
.collect::<Result<Vec<Multiaddr>, _>>()?;
}
if let Some(topics_str) = args.value_of("topics") {
self.topics = topics_str.split(',').map(|s| s.into()).collect();
}
if let Some(discovery_address_str) = args.value_of("discovery-address") {
self.discovery_address = discovery_address_str
.parse()

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

@@ -4,13 +4,11 @@ use crate::{error, NetworkConfig};
/// Currently using discv5 for peer discovery.
///
use futures::prelude::*;
use libp2p::core::swarm::{
ConnectedPoint, NetworkBehaviour, NetworkBehaviourAction, PollParameters,
};
use libp2p::core::{identity::Keypair, Multiaddr, PeerId, ProtocolsHandler};
use libp2p::core::{identity::Keypair, ConnectedPoint, Multiaddr, PeerId};
use libp2p::discv5::{Discv5, Discv5Event};
use libp2p::enr::{Enr, EnrBuilder, NodeId};
use libp2p::multiaddr::Protocol;
use libp2p::swarm::{NetworkBehaviour, NetworkBehaviourAction, PollParameters, ProtocolsHandler};
use slog::{debug, info, o, warn};
use std::collections::HashSet;
use std::fs::File;
@@ -37,6 +35,9 @@ pub struct Discovery<TSubstream> {
/// The target number of connected peers on the libp2p interface.
max_peers: usize,
/// directory to save ENR to
enr_dir: String,
/// The delay between peer discovery searches.
peer_discovery_delay: Delay,
@@ -54,9 +55,6 @@ pub struct Discovery<TSubstream> {
/// Logger for the discovery behaviour.
log: slog::Logger,
/// directory to save ENR to
enr_dir: String,
}
impl<TSubstream> Discovery<TSubstream> {

4
beacon_node/eth2-libp2p/src/lib.rs
View File

@@ -11,9 +11,9 @@ mod service;
pub use behaviour::PubsubMessage;
pub use config::{
Config as NetworkConfig, BEACON_ATTESTATION_TOPIC, BEACON_PUBSUB_TOPIC, SHARD_TOPIC_PREFIX,
Config as NetworkConfig, BEACON_ATTESTATION_TOPIC, BEACON_BLOCK_TOPIC, SHARD_TOPIC_PREFIX,
};
pub use libp2p::floodsub::{Topic, TopicBuilder, TopicHash};
pub use libp2p::gossipsub::{Topic, TopicHash};
pub use libp2p::multiaddr;
pub use libp2p::Multiaddr;
pub use libp2p::{

27
beacon_node/eth2-libp2p/src/rpc/handler.rs
View File

@@ -5,23 +5,21 @@ use crate::rpc::protocol::{InboundFramed, OutboundFramed};
use core::marker::PhantomData;
use fnv::FnvHashMap;
use futures::prelude::*;
use libp2p::core::protocols_handler::{
use libp2p::core::upgrade::{InboundUpgrade, OutboundUpgrade};
use libp2p::swarm::protocols_handler::{
KeepAlive, ProtocolsHandler, ProtocolsHandlerEvent, ProtocolsHandlerUpgrErr, SubstreamProtocol,
};
use libp2p::core::upgrade::{InboundUpgrade, OutboundUpgrade};
use smallvec::SmallVec;
use std::time::{Duration, Instant};
use tokio_io::{AsyncRead, AsyncWrite};
use types::EthSpec;
/// The time (in seconds) before a substream that is awaiting a response times out.
pub const RESPONSE_TIMEOUT: u64 = 9;
/// Implementation of `ProtocolsHandler` for the RPC protocol.
pub struct RPCHandler<TSubstream, E>
pub struct RPCHandler<TSubstream>
where
TSubstream: AsyncRead + AsyncWrite,
E: EthSpec,
{
/// The upgrade for inbound substreams.
listen_protocol: SubstreamProtocol<RPCProtocol>,
@@ -56,8 +54,8 @@ where
/// After the given duration has elapsed, an inactive connection will shutdown.
inactive_timeout: Duration,
/// Phantom EthSpec.
_phantom: PhantomData<E>,
/// Marker to pin the generic stream.
_phantom: PhantomData<TSubstream>,
}
/// An outbound substream is waiting a response from the user.
@@ -90,10 +88,9 @@ where
},
}
impl<TSubstream, E> RPCHandler<TSubstream, E>
impl<TSubstream> RPCHandler<TSubstream>
where
TSubstream: AsyncRead + AsyncWrite,
E: EthSpec,
{
pub fn new(
listen_protocol: SubstreamProtocol<RPCProtocol>,
@@ -145,20 +142,18 @@ where
}
}
impl<TSubstream, E> Default for RPCHandler<TSubstream, E>
impl<TSubstream> Default for RPCHandler<TSubstream>
where
TSubstream: AsyncRead + AsyncWrite,
E: EthSpec,
{
fn default() -> Self {
RPCHandler::new(SubstreamProtocol::new(RPCProtocol), Duration::from_secs(30))
}
}
impl<TSubstream, E> ProtocolsHandler for RPCHandler<TSubstream, E>
impl<TSubstream> ProtocolsHandler for RPCHandler<TSubstream>
where
TSubstream: AsyncRead + AsyncWrite,
E: EthSpec,
{
type InEvent = RPCEvent;
type OutEvent = RPCEvent;
@@ -273,7 +268,11 @@ where
Self::Error,
> {
if let Some(err) = self.pending_error.take() {
return Err(err);
// Returning an error here will result in dropping any peer that doesn't support any of
// the RPC protocols. For our immediate purposes we permit this and simply log that an
// upgrade was not supported.
// TODO: Add a logger to the handler for trace output.
dbg!(&err);
}
// return any events that need to be reported

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

@@ -6,9 +6,9 @@
use futures::prelude::*;
use handler::RPCHandler;
use libp2p::core::protocols_handler::ProtocolsHandler;
use libp2p::core::swarm::{
ConnectedPoint, NetworkBehaviour, NetworkBehaviourAction, PollParameters,
use libp2p::core::ConnectedPoint;
use libp2p::swarm::{
protocols_handler::ProtocolsHandler, NetworkBehaviour, NetworkBehaviourAction, PollParameters,
};
use libp2p::{Multiaddr, PeerId};
pub use methods::{ErrorMessage, HelloMessage, RPCErrorResponse, RPCResponse, RequestId};
@@ -16,7 +16,6 @@ pub use protocol::{RPCError, RPCProtocol, RPCRequest};
use slog::o;
use std::marker::PhantomData;
use tokio::io::{AsyncRead, AsyncWrite};
use types::EthSpec;
pub(crate) mod codec;
mod handler;
@@ -50,16 +49,16 @@ impl RPCEvent {
/// Implements the libp2p `NetworkBehaviour` trait and therefore manages network-level
/// logic.
pub struct RPC<TSubstream, E: EthSpec> {
pub struct RPC<TSubstream> {
/// Queue of events to processed.
events: Vec<NetworkBehaviourAction<RPCEvent, RPCMessage>>,
/// Pins the generic substream.
marker: PhantomData<(TSubstream, E)>,
marker: PhantomData<(TSubstream)>,
/// Slog logger for RPC behaviour.
_log: slog::Logger,
}
impl<TSubstream, E: EthSpec> RPC<TSubstream, E> {
impl<TSubstream> RPC<TSubstream> {
pub fn new(log: &slog::Logger) -> Self {
let log = log.new(o!("Service" => "Libp2p-RPC"));
RPC {
@@ -80,12 +79,11 @@ impl<TSubstream, E: EthSpec> RPC<TSubstream, E> {
}
}
impl<TSubstream, E> NetworkBehaviour for RPC<TSubstream, E>
impl<TSubstream> NetworkBehaviour for RPC<TSubstream>
where
TSubstream: AsyncRead + AsyncWrite,
E: EthSpec,
{
type ProtocolsHandler = RPCHandler<TSubstream, E>;
type ProtocolsHandler = RPCHandler<TSubstream>;
type OutEvent = RPCMessage;
fn new_handler(&mut self) -> Self::ProtocolsHandler {

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

@@ -8,7 +8,7 @@ use futures::{
future::{self, FutureResult},
sink, stream, Sink, Stream,
};
use libp2p::core::{upgrade, InboundUpgrade, OutboundUpgrade, UpgradeInfo};
use libp2p::core::{upgrade, InboundUpgrade, OutboundUpgrade, ProtocolName, UpgradeInfo};
use std::io;
use std::time::Duration;
use tokio::codec::Framed;
@@ -28,24 +28,22 @@ const REQUEST_TIMEOUT: u64 = 3;
pub struct RPCProtocol;
impl UpgradeInfo for RPCProtocol {
type Info = RawProtocolId;
type Info = ProtocolId;
type InfoIter = Vec<Self::Info>;
fn protocol_info(&self) -> Self::InfoIter {
vec![
ProtocolId::new("hello", "1.0.0", "ssz").into(),
ProtocolId::new("goodbye", "1.0.0", "ssz").into(),
ProtocolId::new("beacon_block_roots", "1.0.0", "ssz").into(),
ProtocolId::new("beacon_block_headers", "1.0.0", "ssz").into(),
ProtocolId::new("beacon_block_bodies", "1.0.0", "ssz").into(),
ProtocolId::new("hello", "1.0.0", "ssz"),
ProtocolId::new("goodbye", "1.0.0", "ssz"),
ProtocolId::new("beacon_block_roots", "1.0.0", "ssz"),
ProtocolId::new("beacon_block_headers", "1.0.0", "ssz"),
ProtocolId::new("beacon_block_bodies", "1.0.0", "ssz"),
]
}
}
/// The raw protocol id sent over the wire.
type RawProtocolId = Vec<u8>;
/// Tracks the types in a protocol id.
#[derive(Clone)]
pub struct ProtocolId {
/// The rpc message type/name.
pub message_name: String,
@@ -55,44 +53,31 @@ pub struct ProtocolId {
/// The encoding of the RPC.
pub encoding: String,
/// The protocol id that is formed from the above fields.
protocol_id: String,
}
/// An RPC protocol ID.
impl ProtocolId {
pub fn new(message_name: &str, version: &str, encoding: &str) -> Self {
let protocol_id = format!(
"{}/{}/{}/{}",
PROTOCOL_PREFIX, message_name, version, encoding
);
ProtocolId {
message_name: message_name.into(),
version: version.into(),
encoding: encoding.into(),
protocol_id,
}
}
/// Converts a raw RPC protocol id string into an `RPCProtocolId`
pub fn from_bytes(bytes: &[u8]) -> Result<Self, RPCError> {
let protocol_string = String::from_utf8(bytes.to_vec())
.map_err(|_| RPCError::InvalidProtocol("Invalid protocol Id"))?;
let protocol_list: Vec<&str> = protocol_string.as_str().split('/').take(7).collect();
if protocol_list.len() != 7 {
return Err(RPCError::InvalidProtocol("Not enough '/'"));
}
Ok(ProtocolId {
message_name: protocol_list[4].into(),
version: protocol_list[5].into(),
encoding: protocol_list[6].into(),
})
}
}
impl Into<RawProtocolId> for ProtocolId {
fn into(self) -> RawProtocolId {
format!(
"{}/{}/{}/{}",
PROTOCOL_PREFIX, self.message_name, self.version, self.encoding
)
.as_bytes()
.to_vec()
impl ProtocolName for ProtocolId {
fn protocol_name(&self) -> &[u8] {
self.protocol_id.as_bytes()
}
}
@@ -127,16 +112,11 @@ where
fn upgrade_inbound(
self,
socket: upgrade::Negotiated<TSocket>,
protocol: RawProtocolId,
protocol: ProtocolId,
) -> Self::Future {
// TODO: Verify this
let protocol_id =
ProtocolId::from_bytes(&protocol).expect("Can decode all supported protocols");
match protocol_id.encoding.as_str() {
match protocol.encoding.as_str() {
"ssz" | _ => {
let ssz_codec =
BaseInboundCodec::new(SSZInboundCodec::new(protocol_id, MAX_RPC_SIZE));
let ssz_codec = BaseInboundCodec::new(SSZInboundCodec::new(protocol, MAX_RPC_SIZE));
let codec = InboundCodec::SSZ(ssz_codec);
Framed::new(socket, codec)
.into_future()
@@ -171,7 +151,7 @@ pub enum RPCRequest {
}
impl UpgradeInfo for RPCRequest {
type Info = RawProtocolId;
type Info = ProtocolId;
type InfoIter = Vec<Self::Info>;
// add further protocols as we support more encodings/versions
@@ -182,22 +162,25 @@ impl UpgradeInfo for RPCRequest {
/// Implements the encoding per supported protocol for RPCRequest.
impl RPCRequest {
pub fn supported_protocols(&self) -> Vec<RawProtocolId> {
pub fn supported_protocols(&self) -> Vec<ProtocolId> {
match self {
// add more protocols when versions/encodings are supported
RPCRequest::Hello(_) => vec![ProtocolId::new("hello", "1.0.0", "ssz").into()],
RPCRequest::Goodbye(_) => vec![ProtocolId::new("goodbye", "1.0.0", "ssz").into()],
RPCRequest::Hello(_) => vec![
ProtocolId::new("hello", "1.0.0", "ssz"),
ProtocolId::new("goodbye", "1.0.0", "ssz"),
],
RPCRequest::Goodbye(_) => vec![ProtocolId::new("goodbye", "1.0.0", "ssz")],
RPCRequest::BeaconBlockRoots(_) => {
vec![ProtocolId::new("beacon_block_roots", "1.0.0", "ssz").into()]
vec![ProtocolId::new("beacon_block_roots", "1.0.0", "ssz")]
}
RPCRequest::BeaconBlockHeaders(_) => {
vec![ProtocolId::new("beacon_block_headers", "1.0.0", "ssz").into()]
vec![ProtocolId::new("beacon_block_headers", "1.0.0", "ssz")]
}
RPCRequest::BeaconBlockBodies(_) => {
vec![ProtocolId::new("beacon_block_bodies", "1.0.0", "ssz").into()]
vec![ProtocolId::new("beacon_block_bodies", "1.0.0", "ssz")]
}
RPCRequest::BeaconChainState(_) => {
vec![ProtocolId::new("beacon_block_state", "1.0.0", "ssz").into()]
vec![ProtocolId::new("beacon_block_state", "1.0.0", "ssz")]
}
}
}
@@ -230,12 +213,9 @@ where
socket: upgrade::Negotiated<TSocket>,
protocol: Self::Info,
) -> Self::Future {
let protocol_id =
ProtocolId::from_bytes(&protocol).expect("Can decode all supported protocols");
match protocol_id.encoding.as_str() {
match protocol.encoding.as_str() {
"ssz" | _ => {
let ssz_codec = BaseOutboundCodec::new(SSZOutboundCodec::new(protocol_id, 4096));
let ssz_codec = BaseOutboundCodec::new(SSZOutboundCodec::new(protocol, 4096));
let codec = OutboundCodec::SSZ(ssz_codec);
Framed::new(socket, codec).send(self)
}

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

@@ -3,8 +3,8 @@ use crate::error;
use crate::multiaddr::Protocol;
use crate::rpc::RPCEvent;
use crate::NetworkConfig;
use crate::{TopicBuilder, TopicHash};
use crate::{BEACON_ATTESTATION_TOPIC, BEACON_PUBSUB_TOPIC};
use crate::{Topic, TopicHash};
use crate::{BEACON_ATTESTATION_TOPIC, BEACON_BLOCK_TOPIC};
use futures::prelude::*;
use futures::Stream;
use libp2p::core::{
@@ -21,25 +21,24 @@ use std::fs::File;
use std::io::prelude::*;
use std::io::{Error, ErrorKind};
use std::time::Duration;
use types::EthSpec;
type Libp2pStream = Boxed<(PeerId, StreamMuxerBox), Error>;
type Libp2pBehaviour<E> = Behaviour<Substream<StreamMuxerBox>, E>;
type Libp2pBehaviour = Behaviour<Substream<StreamMuxerBox>>;
const NETWORK_KEY_FILENAME: &str = "key";
/// The configuration and state of the libp2p components for the beacon node.
pub struct Service<E: EthSpec> {
pub struct Service {
/// The libp2p Swarm handler.
//TODO: Make this private
pub swarm: Swarm<Libp2pStream, Libp2pBehaviour<E>>,
pub swarm: Swarm<Libp2pStream, Libp2pBehaviour>,
/// This node's PeerId.
_local_peer_id: PeerId,
/// The libp2p logger handle.
pub log: slog::Logger,
}
impl<E: EthSpec> Service<E> {
impl Service {
pub fn new(config: NetworkConfig, log: slog::Logger) -> error::Result<Self> {
debug!(log, "Network-libp2p Service starting");
@@ -76,18 +75,35 @@ impl<E: EthSpec> Service<E> {
),
};
// attempt to connect to user-input libp2p nodes
for multiaddr in config.libp2p_nodes {
match Swarm::dial_addr(&mut swarm, multiaddr.clone()) {
Ok(()) => debug!(log, "Dialing libp2p node: {}", multiaddr),
Err(err) => debug!(
log,
"Could not connect to node: {} error: {:?}", multiaddr, err
),
};
}
// subscribe to default gossipsub topics
let mut topics = vec![];
//TODO: Handle multiple shard attestations. For now we simply use a separate topic for
//attestations
topics.push(BEACON_ATTESTATION_TOPIC.to_string());
topics.push(BEACON_PUBSUB_TOPIC.to_string());
topics.append(&mut config.topics.clone());
// attestations
topics.push(Topic::new(BEACON_ATTESTATION_TOPIC.into()));
topics.push(Topic::new(BEACON_BLOCK_TOPIC.into()));
topics.append(
&mut config
.topics
.iter()
.cloned()
.map(|s| Topic::new(s))
.collect(),
);
let mut subscribed_topics = vec![];
for topic in topics {
let t = TopicBuilder::new(topic.clone()).build();
if swarm.subscribe(t) {
if swarm.subscribe(topic.clone()) {
trace!(log, "Subscribed to topic: {:?}", topic);
subscribed_topics.push(topic);
} else {
@@ -104,8 +120,8 @@ impl<E: EthSpec> Service<E> {
}
}
impl<E: EthSpec> Stream for Service<E> {
type Item = Libp2pEvent<E>;
impl Stream for Service {
type Item = Libp2pEvent;
type Error = crate::error::Error;
fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
@@ -119,7 +135,7 @@ impl<E: EthSpec> Stream for Service<E> {
topics,
message,
} => {
trace!(self.log, "Pubsub message received: {:?}", message);
trace!(self.log, "Gossipsub message received"; "Message" => format!("{:?}", message));
return Ok(Async::Ready(Some(Libp2pEvent::PubsubMessage {
source,
topics,
@@ -179,7 +195,7 @@ fn build_transport(local_private_key: Keypair) -> Boxed<(PeerId, StreamMuxerBox)
}
/// Events that can be obtained from polling the Libp2p Service.
pub enum Libp2pEvent<E: EthSpec> {
pub enum Libp2pEvent {
/// An RPC response request has been received on the swarm.
RPC(PeerId, RPCEvent),
/// Initiated the connection to a new peer.
@@ -190,7 +206,7 @@ pub enum Libp2pEvent<E: EthSpec> {
PubsubMessage {
source: PeerId,
topics: Vec<TopicHash>,
message: Box<PubsubMessage<E>>,
message: PubsubMessage,
},
}