cerc-io/lighthouse
16
0
Fork 0
Code Issues 1 Pull Requests Packages Projects Releases Wiki Activity

Adds Identify protocol, clean up RPC protocol name handling

Browse Source
This commit is contained in:
Age Manning 2019-07-26 14:43:42 +10:00
parent 15c4062761
commit 04ce9ec95e
No known key found for this signature in database
GPG Key ID: 05EED64B79E06A93
2 changed files with 107 additions and 86 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -10,39 +10,44 @@ use libp2p::{
},
discv5::Discv5Event,
gossipsub::{Gossipsub, GossipsubEvent},
identify::{Identify, IdentifyEvent},
ping::{Ping, PingConfig, PingEvent},
tokio_io::{AsyncRead, AsyncWrite},
NetworkBehaviour, PeerId,
};
use slog::{o, trace, warn};
use slog::{debug, o, trace, warn};
use ssz::{ssz_encode, Decode, DecodeError, Encode};
use std::num::NonZeroU32;
use std::time::Duration;
use types::{Attestation, BeaconBlock, EthSpec};
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 = "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,8 +81,8 @@ 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 {
@ -101,8 +114,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 +132,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 +153,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() > 20 {
debug!(
self.log,
"More than 20 addresses have been identified, truncating"
);
info.listen_addrs.truncate(20);
}
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 +190,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 +199,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 +210,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,28 +220,28 @@ 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: Box<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(BeaconBlock),
/// Gossipsub message providing notification of a new attestation.
Attestation(Attestation<E>),
Attestation(Attestation),
}
//TODO: Correctly encode/decode enums. Prefixing with integer for now.
impl<E: EthSpec> Encode for PubsubMessage<E> {
impl Encode for PubsubMessage {
fn is_ssz_fixed_len() -> bool {
false
}
@ -229,7 +270,7 @@ impl<E: EthSpec> Encode for PubsubMessage<E> {
}
}
impl<E: EthSpec> Decode for PubsubMessage<E> {
impl Decode for PubsubMessage {
fn is_ssz_fixed_len() -> bool {
false
}

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)
}