fluencelabs/rust-libp2p
1
0
Fork 0
mirror of https://github.com/fluencelabs/rust-libp2p synced 2025-06-24 23:31:33 +00:00
Code Issues Projects Releases Wiki Activity

feat(swarm): replace address scoring with explicit candidates

Browse Source 
Previously, a `NetworkBehaviour` could report an `AddressScore` for an external address. This score was a `u32` and addresses would be ranked amongst those.

In reality, an address is either confirmed to be publicly reachable (via a protocol such as AutoNAT) or merely represents a candidate that might be an external address. In a way, addresses are guilty (private) until proven innocent (publicly reachable).

When a `NetworkBehaviour` reports an address candidate, we perform address translation on it to potentially correct for ephemeral ports of TCP. These candidates are then injected back into the `NetworkBehaviour`. Protocols such as AutoNAT can use these addresses as a source for probing their NAT status. Once confirmed, they can emit a `ToSwarm::ExternalAddrConfirmed` event which again will be passed to all `NetworkBehaviour`s.

This simplified approach will allow us implement Kademlia's client-mode (https://github.com/libp2p/rust-libp2p/issues/2032) without additional configuration options: As soon as an address is reported as publicly reachable, we can activate server-mode for that connection.

Related: https://github.com/libp2p/rust-libp2p/pull/3877.
Related: https://github.com/libp2p/rust-libp2p/issues/3953.
Related: https://github.com/libp2p/rust-libp2p/issues/2032.
Related: https://github.com/libp2p/go-libp2p/issues/2229.

Co-authored-by: Max Inden <mail@max-inden.de>

Pull-Request: #3954.
This commit is contained in:
Thomas Eizinger
2023-05-24 09:52:16 +02:00
committed by GitHub
parent a424310d60
commit 5e8f2e82e4
32 changed files with 273 additions and 822 deletions
Download Patch File Download Diff File Expand all files Collapse all files

91
swarm/src/behaviour.rs
View File

@ -29,8 +29,8 @@ pub use listen_addresses::ListenAddresses;
use crate::connection::ConnectionId;
use crate::dial_opts::DialOpts;
use crate::{
AddressRecord, AddressScore, ConnectionDenied, ConnectionHandler, DialError, ListenError,
THandler, THandlerInEvent, THandlerOutEvent,
ConnectionDenied, ConnectionHandler, DialError, ListenError, THandler, THandlerInEvent,
THandlerOutEvent,
};
use libp2p_core::{transport::ListenerId, ConnectedPoint, Endpoint, Multiaddr};
use libp2p_identity::PeerId;
@ -221,8 +221,6 @@ pub trait PollParameters {
type SupportedProtocolsIter: ExactSizeIterator<Item = Vec<u8>>;
/// Iterator returned by [`listened_addresses`](PollParameters::listened_addresses).
type ListenedAddressesIter: ExactSizeIterator<Item = Multiaddr>;
/// Iterator returned by [`external_addresses`](PollParameters::external_addresses).
type ExternalAddressesIter: ExactSizeIterator<Item = AddressRecord>;
/// Returns the list of protocol the behaviour supports when a remote negotiates a protocol on
/// an inbound substream.
@ -242,13 +240,6 @@ pub trait PollParameters {
)]
fn listened_addresses(&self) -> Self::ListenedAddressesIter;
/// Returns the list of the addresses nodes can use to reach us.
#[deprecated(
since = "0.42.0",
note = "Use `libp2p_swarm::ExternalAddresses` instead."
)]
fn external_addresses(&self) -> Self::ExternalAddressesIter;
/// Returns the peer id of the local node.
#[deprecated(
since = "0.42.0",
@ -299,21 +290,28 @@ pub enum ToSwarm<TOutEvent, TInEvent> {
event: TInEvent,
},
/// Informs the `Swarm` about an address observed by a remote for
/// the local node by which the local node is supposedly publicly
/// reachable.
/// Reports a new candidate for an external address to the [`Swarm`](crate::Swarm).
///
/// It is advisable to issue `ReportObservedAddr` actions at a fixed frequency
/// per node. This way address information will be more accurate over time
/// and individual outliers carry less weight.
ReportObservedAddr {
/// The observed address of the local node.
address: Multiaddr,
/// The score to associate with this observation, i.e.
/// an indicator for the trusworthiness of this address
/// relative to other observed addresses.
score: AddressScore,
},
/// This address will be shared with all [`NetworkBehaviour`]s via [`FromSwarm::NewExternalAddrCandidate`].
///
/// This address could come from a variety of sources:
/// - A protocol such as identify obtained it from a remote.
/// - The user provided it based on configuration.
/// - We made an educated guess based on one of our listen addresses.
/// - We established a new relay connection.
NewExternalAddrCandidate(Multiaddr),
/// Indicates to the [`Swarm`](crate::Swarm) that the provided address is confirmed to be externally reachable.
///
/// This is intended to be issued in response to a [`FromSwarm::NewExternalAddrCandidate`] if we are indeed externally reachable on this address.
/// This address will be shared with all [`NetworkBehaviour`]s via [`FromSwarm::ExternalAddrConfirmed`].
ExternalAddrConfirmed(Multiaddr),
/// Indicates to the [`Swarm`](crate::Swarm) that we are no longer externally reachable under the provided address.
///
/// This expires an address that was earlier confirmed via [`ToSwarm::ExternalAddrConfirmed`].
/// This address will be shared with all [`NetworkBehaviour`]s via [`FromSwarm::ExternalAddrExpired`].
ExternalAddrExpired(Multiaddr),
/// Instructs the `Swarm` to initiate a graceful close of one or all connections
/// with the given peer.
@ -351,9 +349,6 @@ impl<TOutEvent, TInEventOld> ToSwarm<TOutEvent, TInEventOld> {
handler,
event: f(event),
},
ToSwarm::ReportObservedAddr { address, score } => {
ToSwarm::ReportObservedAddr { address, score }
}
ToSwarm::CloseConnection {
peer_id,
connection,
@ -361,6 +356,9 @@ impl<TOutEvent, TInEventOld> ToSwarm<TOutEvent, TInEventOld> {
peer_id,
connection,
},
ToSwarm::NewExternalAddrCandidate(addr) => ToSwarm::NewExternalAddrCandidate(addr),
ToSwarm::ExternalAddrConfirmed(addr) => ToSwarm::ExternalAddrConfirmed(addr),
ToSwarm::ExternalAddrExpired(addr) => ToSwarm::ExternalAddrExpired(addr),
}
}
}
@ -380,9 +378,9 @@ impl<TOutEvent, THandlerIn> ToSwarm<TOutEvent, THandlerIn> {
handler,
event,
},
ToSwarm::ReportObservedAddr { address, score } => {
ToSwarm::ReportObservedAddr { address, score }
}
ToSwarm::NewExternalAddrCandidate(addr) => ToSwarm::NewExternalAddrCandidate(addr),
ToSwarm::ExternalAddrConfirmed(addr) => ToSwarm::ExternalAddrConfirmed(addr),
ToSwarm::ExternalAddrExpired(addr) => ToSwarm::ExternalAddrExpired(addr),
ToSwarm::CloseConnection {
peer_id,
connection,
@ -449,10 +447,12 @@ pub enum FromSwarm<'a, Handler> {
ListenerError(ListenerError<'a>),
/// Informs the behaviour that a listener closed.
ListenerClosed(ListenerClosed<'a>),
/// Informs the behaviour that we have discovered a new external address for us.
NewExternalAddr(NewExternalAddr<'a>),
/// Informs the behaviour that an external address was removed.
ExpiredExternalAddr(ExpiredExternalAddr<'a>),
/// Informs the behaviour that we have discovered a new candidate for an external address for us.
NewExternalAddrCandidate(NewExternalAddrCandidate<'a>),
/// Informs the behaviour that an external address of the local node was removed.
ExternalAddrConfirmed(ExternalAddrConfirmed<'a>),
/// Informs the behaviour that an external address of the local node expired, i.e. is no-longer confirmed.
ExternalAddrExpired(ExternalAddrExpired<'a>),
}
/// [`FromSwarm`] variant that informs the behaviour about a newly established connection to a peer.
@ -549,15 +549,21 @@ pub struct ListenerClosed<'a> {
}
/// [`FromSwarm`] variant that informs the behaviour
/// that we have discovered a new external address for us.
/// that we have discovered a new candidate for an external address for us.
#[derive(Clone, Copy)]
pub struct NewExternalAddr<'a> {
pub struct NewExternalAddrCandidate<'a> {
pub addr: &'a Multiaddr,
}
/// [`FromSwarm`] variant that informs the behaviour that an external address was removed.
#[derive(Clone, Copy)]
pub struct ExpiredExternalAddr<'a> {
pub struct ExternalAddrConfirmed<'a> {
pub addr: &'a Multiaddr,
}
/// [`FromSwarm`] variant that informs the behaviour that an external address was removed.
#[derive(Clone, Copy)]
pub struct ExternalAddrExpired<'a> {
pub addr: &'a Multiaddr,
}
@ -657,12 +663,9 @@ impl<'a, Handler> FromSwarm<'a, Handler> {
listener_id,
reason,
})),
FromSwarm::NewExternalAddr(NewExternalAddr { addr }) => {
Some(FromSwarm::NewExternalAddr(NewExternalAddr { addr }))
}
FromSwarm::ExpiredExternalAddr(ExpiredExternalAddr { addr }) => {
Some(FromSwarm::ExpiredExternalAddr(ExpiredExternalAddr { addr }))
}
FromSwarm::NewExternalAddrCandidate(e) => Some(FromSwarm::NewExternalAddrCandidate(e)),
FromSwarm::ExternalAddrExpired(e) => Some(FromSwarm::ExternalAddrExpired(e)),
FromSwarm::ExternalAddrConfirmed(e) => Some(FromSwarm::ExternalAddrConfirmed(e)),
}
}
}

10
swarm/src/behaviour/external_addresses.rs
View File

@ -1,4 +1,4 @@
use crate::behaviour::{ExpiredExternalAddr, FromSwarm, NewExternalAddr};
use crate::behaviour::{ExternalAddrConfirmed, ExternalAddrExpired, FromSwarm};
use libp2p_core::Multiaddr;
use std::collections::HashSet;
@ -34,12 +34,12 @@ impl ExternalAddresses {
/// Returns whether the event changed our set of external addresses.
pub fn on_swarm_event<THandler>(&mut self, event: &FromSwarm<THandler>) -> bool {
match event {
FromSwarm::NewExternalAddr(NewExternalAddr { addr, .. }) => {
FromSwarm::ExternalAddrConfirmed(ExternalAddrConfirmed { addr }) => {
if self.addresses.len() < self.limit {
return self.addresses.insert((*addr).clone());
}
}
FromSwarm::ExpiredExternalAddr(ExpiredExternalAddr { addr, .. }) => {
FromSwarm::ExternalAddrExpired(ExternalAddrExpired { addr, .. }) => {
return self.addresses.remove(addr)
}
_ => {}
@ -80,11 +80,11 @@ mod tests {
}
fn new_external_addr() -> FromSwarm<'static, dummy::ConnectionHandler> {
FromSwarm::NewExternalAddr(NewExternalAddr { addr: &MEMORY_ADDR })
FromSwarm::ExternalAddrConfirmed(ExternalAddrConfirmed { addr: &MEMORY_ADDR })
}
fn expired_external_addr() -> FromSwarm<'static, dummy::ConnectionHandler> {
FromSwarm::ExpiredExternalAddr(ExpiredExternalAddr { addr: &MEMORY_ADDR })
FromSwarm::ExternalAddrExpired(ExternalAddrExpired { addr: &MEMORY_ADDR })
}
static MEMORY_ADDR: Lazy<Multiaddr> =

5
swarm/src/dummy.rs
View File

@ -70,8 +70,9 @@ impl NetworkBehaviour for Behaviour {
| FromSwarm::ExpiredListenAddr(_)
| FromSwarm::ListenerError(_)
| FromSwarm::ListenerClosed(_)
| FromSwarm::NewExternalAddr(_)
| FromSwarm::ExpiredExternalAddr(_) => {}
| FromSwarm::NewExternalAddrCandidate(_)
| FromSwarm::ExternalAddrExpired(_)
| FromSwarm::ExternalAddrConfirmed(_) => {}
}
}
}

5
swarm/src/keep_alive.rs
View File

@ -73,8 +73,9 @@ impl NetworkBehaviour for Behaviour {
| FromSwarm::ExpiredListenAddr(_)
| FromSwarm::ListenerError(_)
| FromSwarm::ListenerClosed(_)
| FromSwarm::NewExternalAddr(_)
| FromSwarm::ExpiredExternalAddr(_) => {}
| FromSwarm::NewExternalAddrCandidate(_)
| FromSwarm::ExternalAddrExpired(_)
| FromSwarm::ExternalAddrConfirmed(_) => {}
}
}
}

133
swarm/src/lib.rs
View File

@ -57,7 +57,6 @@
mod connection;
mod executor;
mod registry;
mod stream;
mod stream_protocol;
#[cfg(test)]
@ -77,13 +76,14 @@ pub mod derive_prelude {
pub use crate::behaviour::ConnectionClosed;
pub use crate::behaviour::ConnectionEstablished;
pub use crate::behaviour::DialFailure;
pub use crate::behaviour::ExpiredExternalAddr;
pub use crate::behaviour::ExpiredListenAddr;
pub use crate::behaviour::ExternalAddrConfirmed;
pub use crate::behaviour::ExternalAddrExpired;
pub use crate::behaviour::FromSwarm;
pub use crate::behaviour::ListenFailure;
pub use crate::behaviour::ListenerClosed;
pub use crate::behaviour::ListenerError;
pub use crate::behaviour::NewExternalAddr;
pub use crate::behaviour::NewExternalAddrCandidate;
pub use crate::behaviour::NewListenAddr;
pub use crate::behaviour::NewListener;
pub use crate::connection::ConnectionId;
@ -107,10 +107,10 @@ pub mod derive_prelude {
}
pub use behaviour::{
AddressChange, CloseConnection, ConnectionClosed, DialFailure, ExpiredExternalAddr,
ExpiredListenAddr, ExternalAddresses, FromSwarm, ListenAddresses, ListenFailure,
ListenerClosed, ListenerError, NetworkBehaviour, NewExternalAddr, NewListenAddr, NotifyHandler,
PollParameters, ToSwarm,
AddressChange, CloseConnection, ConnectionClosed, DialFailure, ExpiredListenAddr,
ExternalAddrExpired, ExternalAddresses, FromSwarm, ListenAddresses, ListenFailure,
ListenerClosed, ListenerError, NetworkBehaviour, NewExternalAddrCandidate, NewListenAddr,
NotifyHandler, PollParameters, ToSwarm,
};
pub use connection::pool::ConnectionCounters;
pub use connection::{ConnectionError, ConnectionId, SupportedProtocols};
@ -121,10 +121,10 @@ pub use handler::{
};
#[cfg(feature = "macros")]
pub use libp2p_swarm_derive::NetworkBehaviour;
pub use registry::{AddAddressResult, AddressRecord, AddressScore};
pub use stream::Stream;
pub use stream_protocol::{InvalidProtocol, StreamProtocol};
use crate::behaviour::ExternalAddrConfirmed;
use crate::handler::UpgradeInfoSend;
use connection::pool::{EstablishedConnection, Pool, PoolConfig, PoolEvent};
use connection::IncomingInfo;
@ -142,7 +142,6 @@ use libp2p_core::{
Endpoint, Multiaddr, Transport,
};
use libp2p_identity::PeerId;
use registry::{AddressIntoIter, Addresses};
use smallvec::SmallVec;
use std::collections::{HashMap, HashSet};
use std::num::{NonZeroU32, NonZeroU8, NonZeroUsize};
@ -340,13 +339,11 @@ where
/// List of protocols that the behaviour says it supports.
supported_protocols: SmallVec<[Vec<u8>; 16]>,
confirmed_external_addr: HashSet<Multiaddr>,
/// Multiaddresses that our listeners are listening on,
listened_addrs: HashMap<ListenerId, SmallVec<[Multiaddr; 1]>>,
/// List of multiaddresses we're listening on, after account for external IP addresses and
/// similar mechanisms.
external_addrs: Addresses,
/// Pending event to be delivered to connection handlers
/// (or dropped if the peer disconnected) before the `behaviour`
/// can be polled again.
@ -654,60 +651,30 @@ where
&self.local_peer_id
}
/// Returns an iterator for [`AddressRecord`]s of external addresses
/// of the local node, in decreasing order of their current
/// [score](AddressScore).
pub fn external_addresses(&self) -> impl Iterator<Item = &AddressRecord> {
self.external_addrs.iter()
/// TODO
pub fn external_addresses(&self) -> impl Iterator<Item = &Multiaddr> {
self.confirmed_external_addr.iter()
}
/// Adds an external address record for the local node.
/// Add a **confirmed** external address for the local node.
///
/// An external address is an address of the local node known to
/// be (likely) reachable for other nodes, possibly taking into
/// account NAT. The external addresses of the local node may be
/// shared with other nodes by the `NetworkBehaviour`.
///
/// The associated score determines both the position of the address
/// in the list of external addresses (which can determine the
/// order in which addresses are used to connect to) as well as
/// how long the address is retained in the list, depending on
/// how frequently it is reported by the `NetworkBehaviour` via
/// [`ToSwarm::ReportObservedAddr`] or explicitly
/// through this method.
pub fn add_external_address(&mut self, a: Multiaddr, s: AddressScore) -> AddAddressResult {
let result = self.external_addrs.add(a.clone(), s);
let expired = match &result {
AddAddressResult::Inserted { expired } => {
self.behaviour
.on_swarm_event(FromSwarm::NewExternalAddr(NewExternalAddr { addr: &a }));
expired
}
AddAddressResult::Updated { expired } => expired,
};
for a in expired {
self.behaviour
.on_swarm_event(FromSwarm::ExpiredExternalAddr(ExpiredExternalAddr {
addr: &a.addr,
}));
}
result
/// This function should only be called with addresses that are guaranteed to be reachable.
/// The address is broadcast to all [`NetworkBehaviour`]s via [`FromSwarm::ExternalAddrConfirmed`].
pub fn add_external_address(&mut self, a: Multiaddr) {
self.behaviour
.on_swarm_event(FromSwarm::ExternalAddrConfirmed(ExternalAddrConfirmed {
addr: &a,
}));
self.confirmed_external_addr.insert(a);
}
/// Removes an external address of the local node, regardless of
/// its current score. See [`Swarm::add_external_address`]
/// for details.
/// Remove an external address for the local node.
///
/// Returns `true` if the address existed and was removed, `false`
/// otherwise.
pub fn remove_external_address(&mut self, addr: &Multiaddr) -> bool {
if self.external_addrs.remove(addr) {
self.behaviour
.on_swarm_event(FromSwarm::ExpiredExternalAddr(ExpiredExternalAddr { addr }));
true
} else {
false
}
/// The address is broadcast to all [`NetworkBehaviour`]s via [`FromSwarm::ExternalAddrExpired`].
pub fn remove_external_address(&mut self, addr: &Multiaddr) {
self.behaviour
.on_swarm_event(FromSwarm::ExternalAddrExpired(ExternalAddrExpired { addr }));
self.confirmed_external_addr.remove(addr);
}
/// Disconnects a peer by its peer ID, closing all connections to said peer.
@ -1162,25 +1129,21 @@ where
self.pending_event = Some((peer_id, handler, event));
}
ToSwarm::ReportObservedAddr { address, score } => {
// Maps the given `observed_addr`, representing an address of the local
// node observed by a remote peer, onto the locally known listen addresses
// to yield one or more addresses of the local node that may be publicly
// reachable.
//
// I.e. self method incorporates the view of other peers into the listen
// addresses seen by the local node to account for possible IP and port
// mappings performed by intermediate network devices in an effort to
// obtain addresses for the local peer that are also reachable for peers
// other than the peer who reported the `observed_addr`.
//
// The translation is transport-specific. See [`Transport::address_translation`].
ToSwarm::NewExternalAddrCandidate(addr) => {
self.behaviour
.on_swarm_event(FromSwarm::NewExternalAddrCandidate(
NewExternalAddrCandidate { addr: &addr },
));
// Generate more candidates based on address translation.
// For TCP without port-reuse, the observed address contains an ephemeral port which needs to be replaced by the port of a listen address.
let translated_addresses = {
let mut addrs: Vec<_> = self
.listened_addrs
.values()
.flatten()
.filter_map(|server| self.transport.address_translation(server, &address))
.filter_map(|server| self.transport.address_translation(server, &addr))
.collect();
// remove duplicates
@ -1189,9 +1152,18 @@ where
addrs
};
for addr in translated_addresses {
self.add_external_address(addr, score);
self.behaviour
.on_swarm_event(FromSwarm::NewExternalAddrCandidate(
NewExternalAddrCandidate { addr: &addr },
));
}
}
ToSwarm::ExternalAddrConfirmed(addr) => {
self.add_external_address(addr);
}
ToSwarm::ExternalAddrExpired(addr) => {
self.remove_external_address(&addr);
}
ToSwarm::CloseConnection {
peer_id,
connection,
@ -1263,7 +1235,6 @@ where
local_peer_id: &this.local_peer_id,
supported_protocols: &this.supported_protocols,
listened_addrs: this.listened_addrs.values().flatten().collect(),
external_addrs: &this.external_addrs,
};
this.behaviour.poll(cx, &mut parameters)
};
@ -1431,13 +1402,11 @@ pub struct SwarmPollParameters<'a> {
local_peer_id: &'a PeerId,
supported_protocols: &'a [Vec<u8>],
listened_addrs: Vec<&'a Multiaddr>,
external_addrs: &'a Addresses,
}
impl<'a> PollParameters for SwarmPollParameters<'a> {
type SupportedProtocolsIter = std::iter::Cloned<std::slice::Iter<'a, std::vec::Vec<u8>>>;
type ListenedAddressesIter = std::iter::Cloned<std::vec::IntoIter<&'a Multiaddr>>;
type ExternalAddressesIter = AddressIntoIter;
fn supported_protocols(&self) -> Self::SupportedProtocolsIter {
self.supported_protocols.iter().cloned()
@ -1447,10 +1416,6 @@ impl<'a> PollParameters for SwarmPollParameters<'a> {
self.listened_addrs.clone().into_iter().cloned()
}
fn external_addresses(&self) -> Self::ExternalAddressesIter {
self.external_addrs.clone().into_iter()
}
fn local_peer_id(&self) -> &PeerId {
self.local_peer_id
}
@ -1639,8 +1604,8 @@ where
pool: Pool::new(self.local_peer_id, self.pool_config),
behaviour: self.behaviour,
supported_protocols: Default::default(),
confirmed_external_addr: Default::default(),
listened_addrs: HashMap::new(),
external_addrs: Addresses::default(),
pending_event: None,
}
}

504
swarm/src/registry.rs
View File

@ -1,504 +0,0 @@
// Copyright 2019 Parity Technologies (UK) Ltd.
//
// Permission is hereby granted, free of charge, to any person obtaining a
// copy of this software and associated documentation files (the "Software"),
// to deal in the Software without restriction, including without limitation
// the rights to use, copy, modify, merge, publish, distribute, sublicense,
// and/or sell copies of the Software, and to permit persons to whom the
// Software is furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in
// all copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
// DEALINGS IN THE SOFTWARE.
use libp2p_core::Multiaddr;
use smallvec::SmallVec;
use std::ops::{Add, Sub};
use std::{cmp::Ordering, collections::VecDeque, num::NonZeroUsize};
/// A ranked collection of [`Multiaddr`] values.
///
/// Every address has an associated [score](`AddressScore`) and iterating
/// over the addresses will return them in order from highest to lowest score.
///
/// In addition to the currently held addresses and their score, the collection
/// keeps track of a limited history of the most-recently added addresses.
/// This history determines how address scores are reduced over time as old
/// scores expire in the context of new addresses being added:
///
/// * An address's score is increased by a given amount whenever it is
/// [(re-)added](Addresses::add) to the collection.
/// * An address's score is decreased by the same amount used when it
/// was added when the least-recently seen addition is (as per the
/// limited history) for this address in the context of [`Addresses::add`].
/// * If an address's score reaches 0 in the context of [`Addresses::add`],
/// it is removed from the collection.
///
#[derive(Debug, Clone)]
pub(crate) struct Addresses {
/// The ranked sequence of addresses, from highest to lowest score.
///
/// By design, the number of finitely scored addresses stored here is
/// never larger (but may be smaller) than the number of historic `reports`
/// at any time.
registry: SmallVec<[AddressRecord; 8]>,
/// The configured limit of the `reports` history of added addresses,
/// and thus also of the size of the `registry` w.r.t. finitely scored
/// addresses.
limit: NonZeroUsize,
/// The limited history of added addresses. If the queue reaches the `limit`,
/// the first record, i.e. the least-recently added, is removed in the
/// context of [`Addresses::add`] and the corresponding record in the
/// `registry` has its score reduced accordingly.
reports: VecDeque<Report>,
}
/// An record in a prioritised list of addresses.
#[derive(Clone, Debug, PartialEq, Eq)]
#[non_exhaustive]
pub struct AddressRecord {
pub addr: Multiaddr,
pub score: AddressScore,
}
/// A report tracked for a finitely scored address.
#[derive(Debug, Clone)]
struct Report {
addr: Multiaddr,
score: u32,
}
impl AddressRecord {
fn new(addr: Multiaddr, score: AddressScore) -> Self {
AddressRecord { addr, score }
}
}
/// The "score" of an address w.r.t. an ordered collection of addresses.
///
/// A score is a measure of the trusworthyness of a particular
/// observation of an address. The same address may be repeatedly
/// reported with the same or differing scores.
#[derive(PartialEq, Eq, Debug, Clone, Copy, Hash)]
pub enum AddressScore {
/// The score is "infinite", i.e. an address with this score is never
/// purged from the associated address records and remains sorted at
/// the beginning (possibly with other `Infinite`ly scored addresses).
Infinite,
/// The score is finite, i.e. an address with this score has
/// its score increased and decreased as per the frequency of
/// reports (i.e. additions) of the same address relative to
/// the reports of other addresses.
Finite(u32),
}
impl AddressScore {
fn is_zero(&self) -> bool {
&AddressScore::Finite(0) == self
}
}
impl PartialOrd for AddressScore {
fn partial_cmp(&self, other: &AddressScore) -> Option<Ordering> {
Some(self.cmp(other))
}
}
impl Ord for AddressScore {
fn cmp(&self, other: &AddressScore) -> Ordering {
// Semantics of cardinal numbers with a single infinite cardinal.
match (self, other) {
(AddressScore::Infinite, AddressScore::Infinite) => Ordering::Equal,
(AddressScore::Infinite, AddressScore::Finite(_)) => Ordering::Greater,
(AddressScore::Finite(_), AddressScore::Infinite) => Ordering::Less,
(AddressScore::Finite(a), AddressScore::Finite(b)) => a.cmp(b),
}
}
}
impl Add for AddressScore {
type Output = AddressScore;
fn add(self, rhs: AddressScore) -> Self::Output {
// Semantics of cardinal numbers with a single infinite cardinal.
match (self, rhs) {
(AddressScore::Infinite, AddressScore::Infinite) => AddressScore::Infinite,
(AddressScore::Infinite, AddressScore::Finite(_)) => AddressScore::Infinite,
(AddressScore::Finite(_), AddressScore::Infinite) => AddressScore::Infinite,
(AddressScore::Finite(a), AddressScore::Finite(b)) => {
AddressScore::Finite(a.saturating_add(b))
}
}
}
}
impl Sub<u32> for AddressScore {
type Output = AddressScore;
fn sub(self, rhs: u32) -> Self::Output {
// Semantics of cardinal numbers with a single infinite cardinal.
match self {
AddressScore::Infinite => AddressScore::Infinite,
AddressScore::Finite(score) => AddressScore::Finite(score.saturating_sub(rhs)),
}
}
}
impl Default for Addresses {
fn default() -> Self {
Addresses::new(NonZeroUsize::new(200).expect("200 > 0"))
}
}
/// The result of adding an address to an ordered list of
/// addresses with associated scores.
pub enum AddAddressResult {
Inserted {
expired: SmallVec<[AddressRecord; 8]>,
},
Updated {
expired: SmallVec<[AddressRecord; 8]>,
},
}
impl Addresses {
/// Create a new ranked address collection with the given size limit
/// for [finitely scored](AddressScore::Finite) addresses.
pub(crate) fn new(limit: NonZeroUsize) -> Self {
Addresses {
registry: SmallVec::new(),
limit,
reports: VecDeque::with_capacity(limit.get()),
}
}
/// Add a [`Multiaddr`] to the collection.
///
/// If the given address already exists in the collection,
/// the given score is added to the current score of the address.
///
/// If the collection has already observed the configured
/// number of address additions, the least-recently added address
/// as per this limited history has its score reduced by the amount
/// used in this prior report, with removal from the collection
/// occurring when the score drops to 0.
pub(crate) fn add(&mut self, addr: Multiaddr, score: AddressScore) -> AddAddressResult {
// If enough reports (i.e. address additions) occurred, reduce
// the score of the least-recently added address.
if self.reports.len() == self.limit.get() {
let old_report = self.reports.pop_front().expect("len = limit > 0");
// If the address is still in the collection, decrease its score.
if let Some(record) = self.registry.iter_mut().find(|r| r.addr == old_report.addr) {
record.score = record.score - old_report.score;
isort(&mut self.registry);
}
}
// Remove addresses that have a score of 0.
let mut expired = SmallVec::new();
while self
.registry
.last()
.map(|e| e.score.is_zero())
.unwrap_or(false)
{
if let Some(addr) = self.registry.pop() {
expired.push(addr);
}
}
// If the address score is finite, remember this report.
if let AddressScore::Finite(score) = score {
self.reports.push_back(Report {
addr: addr.clone(),
score,
});
}
// If the address is already in the collection, increase its score.
for r in &mut self.registry {
if r.addr == addr {
r.score = r.score + score;
isort(&mut self.registry);
return AddAddressResult::Updated { expired };
}
}
// It is a new record.
self.registry.push(AddressRecord::new(addr, score));
AddAddressResult::Inserted { expired }
}
/// Explicitly remove an address from the collection.
///
/// Returns `true` if the address existed in the collection
/// and was thus removed, false otherwise.
pub(crate) fn remove(&mut self, addr: &Multiaddr) -> bool {
if let Some(pos) = self.registry.iter().position(|r| &r.addr == addr) {
self.registry.remove(pos);
true
} else {
false
}
}
/// Return an iterator over all [`Multiaddr`] values.
///
/// The iteration is ordered by descending score.
pub(crate) fn iter(&self) -> AddressIter<'_> {
AddressIter {
items: &self.registry,
offset: 0,
}
}
/// Return an iterator over all [`Multiaddr`] values.
///
/// The iteration is ordered by descending score.
pub(crate) fn into_iter(self) -> AddressIntoIter {
AddressIntoIter {
items: self.registry,
}
}
}
/// An iterator over [`Multiaddr`] values.
#[derive(Clone)]
pub(crate) struct AddressIter<'a> {
items: &'a [AddressRecord],
offset: usize,
}
impl<'a> Iterator for AddressIter<'a> {
type Item = &'a AddressRecord;
fn next(&mut self) -> Option<Self::Item> {
if self.offset == self.items.len() {
return None;
}
let item = &self.items[self.offset];
self.offset += 1;
Some(item)
}
fn size_hint(&self) -> (usize, Option<usize>) {
let n = self.items.len() - self.offset;
(n, Some(n))
}
}
impl<'a> ExactSizeIterator for AddressIter<'a> {}
/// An iterator over [`Multiaddr`] values.
#[derive(Clone)]
pub struct AddressIntoIter {
items: SmallVec<[AddressRecord; 8]>,
}
impl Iterator for AddressIntoIter {
type Item = AddressRecord;
fn next(&mut self) -> Option<Self::Item> {
if !self.items.is_empty() {
Some(self.items.remove(0))
} else {
None
}
}
fn size_hint(&self) -> (usize, Option<usize>) {
let n = self.items.len();
(n, Some(n))
}
}
impl ExactSizeIterator for AddressIntoIter {}
// Reverse insertion sort.
fn isort(xs: &mut [AddressRecord]) {
for i in 1..xs.len() {
for j in (1..=i).rev() {
if xs[j].score <= xs[j - 1].score {
break;
}
xs.swap(j, j - 1)
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use libp2p_core::multiaddr::{Multiaddr, Protocol};
use quickcheck::*;
use std::num::{NonZeroU8, NonZeroUsize};
impl Arbitrary for AddressScore {
fn arbitrary(g: &mut Gen) -> AddressScore {
if g.gen_range(0..10u8) == 0 {
// ~10% "Infinitely" scored addresses
AddressScore::Infinite
} else {
AddressScore::Finite(Arbitrary::arbitrary(g))
}
}
}
impl Arbitrary for AddressRecord {
fn arbitrary(g: &mut Gen) -> Self {
let addr = Protocol::Tcp(g.gen_range(0..256)).into();
let score = AddressScore::arbitrary(g);
AddressRecord::new(addr, score)
}
}
#[test]
fn isort_sorts() {
fn property(xs: Vec<AddressScore>) {
let mut xs = xs
.into_iter()
.map(|score| AddressRecord::new(Multiaddr::empty(), score))
.collect::<Vec<_>>();
isort(&mut xs);
for i in 1..xs.len() {
assert!(xs[i - 1].score >= xs[i].score)
}
}
quickcheck(property as fn(_));
}
#[test]
fn score_retention() {
fn prop(first: AddressRecord, other: AddressRecord) -> TestResult {
if first.addr == other.addr || first.score.is_zero() {
return TestResult::discard();
}
let mut addresses = Addresses::default();
// Add the first address.
addresses.add(first.addr.clone(), first.score);
assert!(addresses.iter().any(|a| a.addr == first.addr));
// Add another address so often that the initial report of
// the first address may be purged and, since it was the
// only report, the address removed.
for _ in 0..addresses.limit.get() + 1 {
addresses.add(other.addr.clone(), other.score);
}
let exists = addresses.iter().any(|a| a.addr == first.addr);
match (first.score, other.score) {
// Only finite scores push out other finite scores.
(AddressScore::Finite(_), AddressScore::Finite(_)) => assert!(!exists),
_ => assert!(exists),
}
TestResult::passed()
}
quickcheck(prop as fn(_, _) -> _);
}
#[test]
fn score_retention_finite_0() {
let first = {
let addr = Protocol::Tcp(42).into();
let score = AddressScore::Finite(0);
AddressRecord::new(addr, score)
};
let other = {
let addr = Protocol::Udp(42).into();
let score = AddressScore::Finite(42);
AddressRecord::new(addr, score)
};
let mut addresses = Addresses::default();
// Add the first address.
addresses.add(first.addr.clone(), first.score);
assert!(addresses.iter().any(|a| a.addr == first.addr));
// Add another address so the first will address be purged,
// because its score is finite(0)
addresses.add(other.addr.clone(), other.score);
assert!(addresses.iter().any(|a| a.addr == other.addr));
assert!(!addresses.iter().any(|a| a.addr == first.addr));
}
#[test]
fn finitely_scored_address_limit() {
fn prop(reports: Vec<AddressRecord>, limit: NonZeroU8) {
let mut addresses = Addresses::new(limit.into());
// Add all reports.
for r in reports {
addresses.add(r.addr, r.score);
}
// Count the finitely scored addresses.
let num_finite = addresses
.iter()
.filter(|r| {
matches!(
r,
AddressRecord {
score: AddressScore::Finite(_),
..
}
)
})
.count();
// Check against the limit.
assert!(num_finite <= limit.get() as usize);
}
quickcheck(prop as fn(_, _));
}
#[test]
fn record_score_sum() {
fn prop(records: Vec<AddressRecord>) -> bool {
// Make sure the address collection can hold all reports.
let n = std::cmp::max(records.len(), 1);
let mut addresses = Addresses::new(NonZeroUsize::new(n).unwrap());
// Add all address reports to the collection.
for r in records.iter() {
addresses.add(r.addr.clone(), r.score);
}
// Check that each address in the registry has the expected score.
for r in &addresses.registry {
let expected_score = records.iter().fold(None::<AddressScore>, |sum, rec| {
if rec.addr == r.addr {
sum.map_or(Some(rec.score), |s| Some(s + rec.score))
} else {
sum
}
});
if Some(r.score) != expected_score {
return false;
}
}
true
}
quickcheck(prop as fn(_) -> _)
}
}

19
swarm/src/test.rs
View File

@ -19,8 +19,8 @@
// DEALINGS IN THE SOFTWARE.
use crate::behaviour::{
ConnectionClosed, ConnectionEstablished, DialFailure, ExpiredExternalAddr, ExpiredListenAddr,
FromSwarm, ListenerClosed, ListenerError, NewExternalAddr, NewListenAddr, NewListener,
ConnectionClosed, ConnectionEstablished, DialFailure, ExpiredListenAddr, ExternalAddrExpired,
FromSwarm, ListenerClosed, ListenerError, NewExternalAddrCandidate, NewListenAddr, NewListener,
};
use crate::{
ConnectionDenied, ConnectionHandler, ConnectionId, NetworkBehaviour, PollParameters, THandler,
@ -130,8 +130,9 @@ where
| FromSwarm::ExpiredListenAddr(_)
| FromSwarm::ListenerError(_)
| FromSwarm::ListenerClosed(_)
| FromSwarm::NewExternalAddr(_)
| FromSwarm::ExpiredExternalAddr(_) => {}
| FromSwarm::NewExternalAddrCandidate(_)
| FromSwarm::ExternalAddrExpired(_)
| FromSwarm::ExternalAddrConfirmed(_) => {}
}
}
@ -500,15 +501,17 @@ where
addr,
}));
}
FromSwarm::NewExternalAddr(NewExternalAddr { addr }) => {
FromSwarm::NewExternalAddrCandidate(NewExternalAddrCandidate { addr }) => {
self.on_new_external_addr.push(addr.clone());
self.inner
.on_swarm_event(FromSwarm::NewExternalAddr(NewExternalAddr { addr }));
.on_swarm_event(FromSwarm::NewExternalAddrCandidate(
NewExternalAddrCandidate { addr },
));
}
FromSwarm::ExpiredExternalAddr(ExpiredExternalAddr { addr }) => {
FromSwarm::ExternalAddrExpired(ExternalAddrExpired { addr }) => {
self.on_expired_external_addr.push(addr.clone());
self.inner
.on_swarm_event(FromSwarm::ExpiredExternalAddr(ExpiredExternalAddr { addr }));
.on_swarm_event(FromSwarm::ExternalAddrExpired(ExternalAddrExpired { addr }));
}
FromSwarm::ListenerError(ListenerError { listener_id, err }) => {
self.on_listener_error.push(listener_id);