Upgrade to latest libp2p (#2605)

This is a pre-cursor to the next libp2p upgrade. 

It is currently being used for staging a number of PR upgrades which are contingent on the latest libp2p.
This commit is contained in:
Age Manning 2021-10-29 01:59:29 +00:00
parent 2c4413454a
commit 1790010260
6 changed files with 393 additions and 742 deletions

1028
Cargo.lock generated

File diff suppressed because it is too large Load Diff

View File

@ -39,15 +39,9 @@ regex = "1.3.9"
strum = { version = "0.21.0", features = ["derive"] }
superstruct = "0.2.0"
# TODO: remove the rev-reference and go back to release versions once there is a release that
# includes the following PR:
#
# https://github.com/libp2p/rust-libp2p/pull/2175
[dependencies.libp2p]
#version = "0.39.1"
#default-features = false
git = "https://github.com/libp2p/rust-libp2p"
rev = "ce23cbe76a0382b6fcb0e49f1b2612df86f6465d"
version = "0.40.0-rc.3"
default-features = false
features = ["websocket", "identify", "mplex", "yamux", "noise", "gossipsub", "dns-tokio", "tcp-tokio"]
[dev-dependencies]

View File

@ -4,7 +4,8 @@ use crate::behaviour::gossipsub_scoring_parameters::{
use crate::config::gossipsub_config;
use crate::discovery::{subnet_predicate, Discovery, DiscoveryEvent, TARGET_SUBNET_PEERS};
use crate::peer_manager::{
peerdb::score::ReportSource, ConnectionDirection, PeerManager, PeerManagerEvent,
peerdb::score::PeerAction, peerdb::score::ReportSource, ConnectionDirection, PeerManager,
PeerManagerEvent,
};
use crate::rpc::*;
use crate::service::METADATA_FILENAME;
@ -26,7 +27,7 @@ use libp2p::{
},
identify::{Identify, IdentifyConfig, IdentifyEvent},
swarm::{
AddressScore, DialPeerCondition, NetworkBehaviourAction as NBAction,
AddressScore, DialPeerCondition, NetworkBehaviour, NetworkBehaviourAction as NBAction,
NetworkBehaviourEventProcess, PollParameters,
},
NetworkBehaviour, PeerId,
@ -121,7 +122,11 @@ enum InternalBehaviourMessage {
/// This core behaviour is managed by `Behaviour` which adds peer management to all core
/// behaviours.
#[derive(NetworkBehaviour)]
#[behaviour(out_event = "BehaviourEvent<TSpec>", poll_method = "poll")]
#[behaviour(
out_event = "BehaviourEvent<TSpec>",
poll_method = "poll",
event_process = true
)]
pub struct Behaviour<TSpec: EthSpec> {
/* Sub-Behaviours */
/// The routing pub-sub mechanism for eth2.
@ -192,9 +197,11 @@ impl<TSpec: EthSpec> Behaviour<TSpec> {
"".into(),
local_key.public(), // Still send legitimate public key
)
.with_cache_size(0)
} else {
IdentifyConfig::new("eth2/1.0.0".into(), local_key.public())
.with_agent_version(lighthouse_version::version_with_platform())
.with_cache_size(0)
};
// Build and start the discovery sub-behaviour
@ -848,6 +855,15 @@ impl<TSpec: EthSpec> NetworkBehaviourEventProcess<GossipsubEvent> for Behaviour<
.remove_subscription(&peer_id, &subnet_id);
}
}
GossipsubEvent::GossipsubNotSupported { peer_id } => {
debug!(self.log, "Peer does not support gossipsub"; "peer_id" => %peer_id);
self.peer_manager.report_peer(
&peer_id,
PeerAction::LowToleranceError,
ReportSource::Gossipsub,
Some(GoodbyeReason::Unknown),
);
}
}
}
}
@ -1031,11 +1047,13 @@ impl<TSpec: EthSpec> NetworkBehaviourEventProcess<IdentifyEvent> for Behaviour<T
impl<TSpec: EthSpec> Behaviour<TSpec> {
/// Consumes the events list and drives the Lighthouse global NetworkBehaviour.
fn poll<THandlerIn>(
fn poll(
&mut self,
cx: &mut Context,
_: &mut impl PollParameters,
) -> Poll<NBAction<THandlerIn, BehaviourEvent<TSpec>>> {
) -> Poll<
NBAction<BehaviourEvent<TSpec>, <Behaviour<TSpec> as NetworkBehaviour>::ProtocolsHandler>,
> {
if let Some(waker) = &self.waker {
if waker.will_wake(cx.waker()) {
self.waker = Some(cx.waker().clone());
@ -1048,9 +1066,11 @@ impl<TSpec: EthSpec> Behaviour<TSpec> {
if let Some(event) = self.internal_events.pop_front() {
match event {
InternalBehaviourMessage::DialPeer(peer_id) => {
let handler = self.new_handler();
return Poll::Ready(NBAction::DialPeer {
peer_id,
condition: DialPeerCondition::Disconnected,
handler,
});
}
InternalBehaviourMessage::SocketUpdated(address) => {

View File

@ -23,8 +23,8 @@ use futures::stream::FuturesUnordered;
pub use libp2p::{
core::{connection::ConnectionId, ConnectedPoint, Multiaddr, PeerId},
swarm::{
protocols_handler::ProtocolsHandler, NetworkBehaviour, NetworkBehaviourAction as NBAction,
NotifyHandler, PollParameters, SubstreamProtocol,
protocols_handler::ProtocolsHandler, DialError, NetworkBehaviour,
NetworkBehaviourAction as NBAction, NotifyHandler, PollParameters, SubstreamProtocol,
},
};
use lru::LruCache;
@ -933,9 +933,10 @@ impl<TSpec: EthSpec> NetworkBehaviour for Discovery<TSpec> {
fn inject_disconnected(&mut self, _peer_id: &PeerId) {}
fn inject_connection_established(
&mut self,
_: &PeerId,
_: &ConnectionId,
_connected_point: &ConnectedPoint,
_peer_id: &PeerId,
_connection_id: &ConnectionId,
_endpoint: &ConnectedPoint,
_failed_addresses: Option<&Vec<Multiaddr>>,
) {
}
fn inject_connection_closed(
@ -943,6 +944,7 @@ impl<TSpec: EthSpec> NetworkBehaviour for Discovery<TSpec> {
_: &PeerId,
_: &ConnectionId,
_connected_point: &ConnectedPoint,
_handler: Self::ProtocolsHandler,
) {
}
fn inject_event(
@ -953,10 +955,17 @@ impl<TSpec: EthSpec> NetworkBehaviour for Discovery<TSpec> {
) {
}
fn inject_dial_failure(&mut self, peer_id: &PeerId) {
// set peer as disconnected in discovery DHT
debug!(self.log, "Marking peer disconnected in DHT"; "peer_id" => %peer_id);
self.disconnect_peer(peer_id);
fn inject_dial_failure(
&mut self,
peer_id: Option<PeerId>,
_handler: Self::ProtocolsHandler,
_error: &DialError,
) {
if let Some(peer_id) = peer_id {
// set peer as disconnected in discovery DHT
debug!(self.log, "Marking peer disconnected in DHT"; "peer_id" => %peer_id);
self.disconnect_peer(&peer_id);
}
}
// Main execution loop to drive the behaviour
@ -964,7 +973,7 @@ impl<TSpec: EthSpec> NetworkBehaviour for Discovery<TSpec> {
&mut self,
cx: &mut Context,
_: &mut impl PollParameters,
) -> Poll<NBAction<<Self::ProtocolsHandler as ProtocolsHandler>::InEvent, Self::OutEvent>> {
) -> Poll<NBAction<Self::OutEvent, Self::ProtocolsHandler>> {
if !self.started {
return Poll::Pending;
}

View File

@ -101,7 +101,7 @@ pub struct RPC<TSpec: EthSpec> {
/// Rate limiter
limiter: RateLimiter,
/// Queue of events to be processed.
events: Vec<NetworkBehaviourAction<RPCSend<TSpec>, RPCMessage<TSpec>>>,
events: Vec<NetworkBehaviourAction<RPCMessage<TSpec>, RPCHandler<TSpec>>>,
fork_context: Arc<ForkContext>,
/// Slog logger for RPC behaviour.
log: slog::Logger,
@ -218,8 +218,9 @@ where
fn inject_connection_established(
&mut self,
_peer_id: &PeerId,
_: &ConnectionId,
_connected_point: &ConnectedPoint,
_connection_id: &ConnectionId,
_endpoint: &ConnectedPoint,
_failed_addresses: Option<&Vec<Multiaddr>>,
) {
}
@ -228,6 +229,7 @@ where
_peer_id: &PeerId,
_: &ConnectionId,
_connected_point: &ConnectedPoint,
_handler: Self::ProtocolsHandler,
) {
}
@ -297,12 +299,7 @@ where
&mut self,
cx: &mut Context,
_: &mut impl PollParameters,
) -> Poll<
NetworkBehaviourAction<
<Self::ProtocolsHandler as ProtocolsHandler>::InEvent,
Self::OutEvent,
>,
> {
) -> Poll<NetworkBehaviourAction<Self::OutEvent, Self::ProtocolsHandler>> {
// let the rate limiter prune
let _ = self.limiter.poll_unpin(cx);
if !self.events.is_empty() {

View File

@ -320,6 +320,7 @@ impl<TSpec: EthSpec> Service<TSpec> {
peer_id,
endpoint,
num_established,
concurrent_dial_errors: _,
} => {
// Inform the peer manager.
// We require the ENR to inject into the peer db, if it exists.
@ -364,20 +365,14 @@ impl<TSpec: EthSpec> Service<TSpec> {
SwarmEvent::BannedPeer { peer_id, .. } => {
debug!(self.log, "Banned peer connection rejected"; "peer_id" => %peer_id);
}
SwarmEvent::UnreachableAddr {
peer_id,
address,
error,
attempts_remaining,
} => {
debug!(self.log, "Failed to dial address"; "peer_id" => %peer_id, "address" => %address, "error" => %error, "attempts_remaining" => attempts_remaining);
self.swarm
.behaviour_mut()
.peer_manager_mut()
.inject_dial_failure(&peer_id);
}
SwarmEvent::UnknownPeerUnreachableAddr { address, error } => {
debug!(self.log, "Peer not known at dialed address"; "address" => %address, "error" => %error);
SwarmEvent::OutgoingConnectionError { peer_id, error } => {
debug!(self.log, "Failed to dial address"; "peer_id" => ?peer_id, "error" => %error);
if let Some(peer_id) = peer_id {
self.swarm
.behaviour_mut()
.peer_manager_mut()
.inject_dial_failure(&peer_id);
}
}
SwarmEvent::ExpiredListenAddr { address, .. } => {
debug!(self.log, "Listen address expired"; "address" => %address)