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

Initial core grouping of libp2p behaviours

Browse Source
This commit is contained in:
Age Manning 2019-04-30 15:12:57 +10:00
parent cb7d5eba1c
commit b33ce5dd10
No known key found for this signature in database
GPG Key ID: 05EED64B79E06A93
5 changed files with 310 additions and 34 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -15,7 +15,7 @@ serde = "1.0"
serde_derive = "1.0"
eth2_ssz = { path = "../../eth2/utils/ssz" }
eth2_ssz_derive = { path = "../../eth2/utils/ssz_derive" }
slog = "2.4.1"
slog = { version = "^2.4.1" , features = ["max_level_trace", "release_max_level_trace"] }
version = { path = "../version" }
tokio = "0.1.16"
futures = "0.1.25"

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

@ -3,6 +3,7 @@ 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},
@ -19,16 +20,18 @@ use slog::{debug, o, trace, warn};
use ssz::{ssz_encode, Decode, DecodeError, Encode};
use std::time::{Duration, Instant};
use tokio_timer::Delay;
use std::collections::HashMap;
use types::{Attestation, BeaconBlock};
/// Builds the network behaviour for the libp2p Swarm.
/// Implements gossipsub message routing.
/// 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", poll_method = "poll")]
pub struct Behaviour<TSubstream: AsyncRead + AsyncWrite> {
#[behaviour(out_event = "CoreBehaviourEvent", poll_method = "poll")]
pub struct CoreCoreBehaviourTSubstream: AsyncRead + AsyncWrite> {
/// The routing pub-sub mechanism for eth2.
gossipsub: Gossipsub<TSubstream>,
/// The events generated by this behaviour to be consumed in the swarm poll.
/// The serenity RPC specified in the wire-0 protocol.
serenity_rpc: Rpc<TSubstream>,
/// Allows discovery of IP addresses for peers on the network.
identify: Identify<TSubstream>,
@ -36,9 +39,9 @@ pub struct Behaviour<TSubstream: AsyncRead + AsyncWrite> {
ping: Ping<TSubstream>,
/// Kademlia for peer discovery.
discovery: Discovery<TSubstream>,
/// Queue of behaviour events to be processed.
#[behaviour(ignore)]
events: Vec<BehaviourEvent>,
/// The events generated by this behaviour to be consumed in the swarm poll.
events: Vec<CoreBehaviourEvent>,
/// Logger for behaviour actions.
#[behaviour(ignore)]
log: slog::Logger,
@ -46,7 +49,7 @@ pub struct Behaviour<TSubstream: AsyncRead + AsyncWrite> {
// Implement the NetworkBehaviourEventProcess trait so that we can derive NetworkBehaviour for Behaviour
impl<TSubstream: AsyncRead + AsyncWrite> NetworkBehaviourEventProcess<GossipsubEvent>
for Behaviour<TSubstream>
for CoreBehaviourTSubstream>
{
fn inject_event(&mut self, event: GossipsubEvent) {
match event {
@ -79,7 +82,7 @@ impl<TSubstream: AsyncRead + AsyncWrite> NetworkBehaviourEventProcess<GossipsubE
}
impl<TSubstream: AsyncRead + AsyncWrite> NetworkBehaviourEventProcess<RPCMessage>
for Behaviour<TSubstream>
for CoreBehaviourTSubstream>
{
fn inject_event(&mut self, event: RPCMessage) {
match event {
@ -94,7 +97,7 @@ impl<TSubstream: AsyncRead + AsyncWrite> NetworkBehaviourEventProcess<RPCMessage
}
impl<TSubstream: AsyncRead + AsyncWrite> NetworkBehaviourEventProcess<IdentifyEvent>
for Behaviour<TSubstream>
for CoreBehaviourTSubstream>
{
fn inject_event(&mut self, event: IdentifyEvent) {
match event {
@ -124,7 +127,7 @@ impl<TSubstream: AsyncRead + AsyncWrite> NetworkBehaviourEventProcess<IdentifyEv
}
impl<TSubstream: AsyncRead + AsyncWrite> NetworkBehaviourEventProcess<PingEvent>
for Behaviour<TSubstream>
for CoreBehaviourTSubstream>
{
fn inject_event(&mut self, _event: PingEvent) {
// not interested in ping responses at the moment.
@ -133,14 +136,14 @@ impl<TSubstream: AsyncRead + AsyncWrite> NetworkBehaviourEventProcess<PingEvent>
// implement the discovery behaviour (currently kademlia)
impl<TSubstream: AsyncRead + AsyncWrite> NetworkBehaviourEventProcess<KademliaOut>
for Behaviour<TSubstream>
for CoreBehaviourTSubstream>
{
fn inject_event(&mut self, _out: KademliaOut) {
// not interested in kademlia results at the moment
}
}
impl<TSubstream: AsyncRead + AsyncWrite> Behaviour<TSubstream> {
impl<TSubstream: AsyncRead + AsyncWrite> CoreBehaviourTSubstream> {
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();
@ -174,17 +177,14 @@ impl<TSubstream: AsyncRead + AsyncWrite> Behaviour<TSubstream> {
}
/// Implements the combined behaviour for the libp2p service.
impl<TSubstream: AsyncRead + AsyncWrite> Behaviour<TSubstream> {
impl<TSubstream: AsyncRead + AsyncWrite> CoreBehaviourTSubstream> {
/* Pubsub behaviour functions */
/// Subscribes to a gossipsub topic.
pub fn subscribe(&mut self, topic: Topic) -> bool {
self.gossipsub.subscribe(topic)
}
/// 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);
}
/// Publishes a message on the pubsub (gossipsub) behaviour.
pub fn publish(&mut self, topics: Vec<Topic>, message: PubsubMessage) {
let message_bytes = ssz_encode(&message);
@ -192,10 +192,18 @@ impl<TSubstream: AsyncRead + AsyncWrite> Behaviour<TSubstream> {
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 BehaviourEvent {
pub enum CoreBehaviourEvent {
RPC(PeerId, RPCEvent),
PeerDialed(PeerId),
Identified(PeerId, Box<IdentifyInfo>),

279
beacon_node/eth2-libp2p/src/core-behaviour.rs Normal file
View File

@ -0,0 +1,279 @@
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<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>,
/// Allows discovery of IP addresses for peers on the network.
identify: Identify<TSubstream>,
/// Keep regular connection to peers and disconnect if absent.
ping: Ping<TSubstream>,
/// Kademlia for peer discovery.
discovery: Discovery<TSubstream>,
#[behaviour(ignore)]
/// The events generated by this behaviour to be consumed by the global behaviour.
events: Vec<CoreBehaviourEvent>,
/// Logger for behaviour actions.
#[behaviour(ignore)]
log: slog::Logger,
}
impl<TSubstream: AsyncRead + AsyncWrite> CoreBehaviour<TSubstream> {
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<TBehaviourIn>(
&mut self,
) -> Async<NetworkBehaviourAction<TBehaviourIn, CoreBehaviourEvent>> {
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<TSubstream: AsyncRead + AsyncWrite> NetworkBehaviourEventProcess<GossipsubEvent>
for CoreBehaviour<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::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<TSubstream: AsyncRead + AsyncWrite> NetworkBehaviourEventProcess<RPCMessage>
for CoreBehaviour<TSubstream>
{
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<TSubstream: AsyncRead + AsyncWrite> NetworkBehaviourEventProcess<IdentifyEvent>
for CoreBehaviour<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 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<TSubstream: AsyncRead + AsyncWrite> NetworkBehaviourEventProcess<PingEvent>
for CoreBehaviour<TSubstream>
{
fn inject_event(&mut self, _event: PingEvent) {
// not interested in ping responses at the moment.
}
}
// implement the discovery behaviour (currently kademlia)
impl<TSubstream: AsyncRead + AsyncWrite> NetworkBehaviourEventProcess<KademliaOut>
for CoreBehaviour<TSubstream>
{
fn inject_event(&mut self, _out: KademliaOut) {
// not interested in kademlia results at the moment
}
}
/// Implements the combined behaviour for the libp2p service.
impl<TSubstream: AsyncRead + AsyncWrite> CoreBehaviour<TSubstream> {
/* 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<Topic>, 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<TopicHash>,
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);
}
}

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

@ -9,7 +9,6 @@ use libp2p::core::swarm::{
use libp2p::core::{Multiaddr, PeerId, ProtocolsHandler};
use libp2p::kad::{Kademlia, KademliaOut};
use slog::{debug, o, warn};
use std::collections::HashMap;
use std::time::{Duration, Instant};
use tokio::io::{AsyncRead, AsyncWrite};
use tokio_timer::Delay;
@ -27,8 +26,6 @@ pub struct Discovery<TSubstream> {
discovery: Kademlia<TSubstream>,
/// The delay between peer discovery searches.
peer_discovery_delay: Delay,
/// Mapping of known addresses for peer ids.
known_peers: HashMap<PeerId, Vec<Multiaddr>>,
/// Logger for the discovery behaviour.
log: slog::Logger,
}
@ -37,10 +34,8 @@ impl<TSubstream> Discovery<TSubstream> {
pub fn new(local_peer_id: PeerId, log: &slog::Logger) -> Self {
let log = log.new(o!("Service" => "Libp2p-Discovery"));
Self {
// events: Vec::new(),
discovery: Kademlia::new(local_peer_id),
peer_discovery_delay: Delay::new(Instant::now()),
known_peers: HashMap::new(),
log,
}
}
@ -59,13 +54,6 @@ impl<TSubstream> Discovery<TSubstream> {
/// We have discovered an address for a peer, add it to known peers.
pub fn add_connected_address(&mut self, peer_id: &PeerId, address: Multiaddr) {
let known_peers = self
.known_peers
.entry(peer_id.clone())
.or_insert_with(|| vec![]);
if !known_peers.contains(&address) {
known_peers.push(address.clone());
}
// pass the address on to kademlia
self.discovery.add_connected_address(peer_id, address);
}
@ -160,6 +148,7 @@ where
},
_ => {}
};
// propagate result upwards
return Async::Ready(action);
}
Async::NotReady => (),

2
beacon_node/network/Cargo.toml
View File

@ -13,7 +13,7 @@ store = { path = "../store" }
eth2-libp2p = { path = "../eth2-libp2p" }
version = { path = "../version" }
types = { path = "../../eth2/types" }
slog = { version = "^2.2.3" , features = ["max_level_trace", "release_max_level_debug"] }
slog = { version = "^2.2.3" }
eth2_ssz = { path = "../../eth2/utils/ssz" }
tree_hash = { path = "../../eth2/utils/tree_hash" }
futures = "0.1.25"