Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat(swarm): replace address scoring with explicit candidates #3954

Merged
merged 20 commits into from
May 24, 2023
Merged
Show file tree
Hide file tree
Changes from 7 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions examples/rendezvous/src/bin/rzv-register.rs
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,7 @@ use futures::StreamExt;
use libp2p::{
core::transport::upgrade::Version,
identity, noise, ping, rendezvous,
swarm::{keep_alive, AddressScore, NetworkBehaviour, SwarmBuilder, SwarmEvent},
swarm::{keep_alive, NetworkBehaviour, SwarmBuilder, SwarmEvent},
tcp, yamux, Multiaddr, PeerId, Transport,
};
use std::time::Duration;
Expand Down Expand Up @@ -55,7 +55,7 @@ async fn main() {
// In production the external address should be the publicly facing IP address of the rendezvous point.
// This address is recorded in the registration entry by the rendezvous point.
let external_address = "/ip4/127.0.0.1/tcp/0".parse::<Multiaddr>().unwrap();
swarm.add_external_address(external_address, AddressScore::Infinite);
swarm.add_external_address(external_address);

log::info!("Local peer id: {}", swarm.local_peer_id());

Expand Down
20 changes: 5 additions & 15 deletions misc/allow-block-list/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -243,8 +243,9 @@ where
FromSwarm::ExpiredListenAddr(_) => {}
FromSwarm::ListenerError(_) => {}
FromSwarm::ListenerClosed(_) => {}
FromSwarm::NewExternalAddr(_) => {}
FromSwarm::ExpiredExternalAddr(_) => {}
FromSwarm::NewExternalAddrCandidate(_) => {}
FromSwarm::ExternalAddrExpired(_) => {}
FromSwarm::ExternalAddrConfirmed(_) => {}
}
}

Expand Down Expand Up @@ -410,13 +411,7 @@ mod tests {
dialer
.dial(
DialOpts::unknown_peer_id()
.address(
listener
.external_addresses()
.map(|a| a.addr.clone())
.next()
.unwrap(),
)
.address(listener.external_addresses().next().cloned().unwrap())
.build(),
)
.unwrap();
Expand Down Expand Up @@ -470,12 +465,7 @@ mod tests {
{
dialer.dial(
DialOpts::peer_id(*listener.local_peer_id())
.addresses(
listener
.external_addresses()
.map(|a| a.addr.clone())
.collect(),
)
.addresses(listener.external_addresses().cloned().collect())
.build(),
)
}
Expand Down
5 changes: 3 additions & 2 deletions misc/connection-limits/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -337,8 +337,9 @@ impl NetworkBehaviour for Behaviour {
FromSwarm::ExpiredListenAddr(_) => {}
FromSwarm::ListenerError(_) => {}
FromSwarm::ListenerClosed(_) => {}
FromSwarm::NewExternalAddr(_) => {}
FromSwarm::ExpiredExternalAddr(_) => {}
FromSwarm::NewExternalAddrCandidate(_) => {}
FromSwarm::ExternalAddrExpired(_) => {}
FromSwarm::ExternalAddrConfirmed(_) => {}
}
}

Expand Down
26 changes: 15 additions & 11 deletions protocols/autonat/src/behaviour.rs
Original file line number Diff line number Diff line change
Expand Up @@ -36,10 +36,10 @@ use libp2p_request_response::{
};
use libp2p_swarm::{
behaviour::{
AddressChange, ConnectionClosed, ConnectionEstablished, DialFailure, ExpiredExternalAddr,
ExpiredListenAddr, FromSwarm,
AddressChange, ConnectionClosed, ConnectionEstablished, DialFailure, ExpiredListenAddr,
ExternalAddrExpired, FromSwarm,
},
ConnectionDenied, ConnectionId, ExternalAddresses, ListenAddresses, NetworkBehaviour,
ConnectionDenied, ConnectionId, ListenAddresses, NetworkBehaviour, NewExternalAddrCandidate,
PollParameters, THandler, THandlerInEvent, THandlerOutEvent, ToSwarm,
};
use std::{
Expand Down Expand Up @@ -214,7 +214,7 @@ pub struct Behaviour {
probe_id: ProbeId,

listen_addresses: ListenAddresses,
external_addresses: ExternalAddresses,
other_candidates: HashSet<Multiaddr>,
}

impl Behaviour {
Expand All @@ -240,7 +240,7 @@ impl Behaviour {
pending_actions: VecDeque::new(),
probe_id: ProbeId(0),
listen_addresses: Default::default(),
external_addresses: Default::default(),
other_candidates: Default::default(),
}
}

Expand Down Expand Up @@ -294,7 +294,7 @@ impl Behaviour {
last_probe: &mut self.last_probe,
schedule_probe: &mut self.schedule_probe,
listen_addresses: &self.listen_addresses,
external_addresses: &self.external_addresses,
other_candidates: &self.other_candidates,
}
}

Expand Down Expand Up @@ -532,7 +532,6 @@ impl NetworkBehaviour for Behaviour {

fn on_swarm_event(&mut self, event: FromSwarm<Self::ConnectionHandler>) {
self.listen_addresses.on_swarm_event(&event);
self.external_addresses.on_swarm_event(&event);

match event {
FromSwarm::ConnectionEstablished(connection_established) => {
Expand Down Expand Up @@ -561,13 +560,17 @@ impl NetworkBehaviour for Behaviour {
}));
self.as_client().on_expired_address(addr);
}
FromSwarm::ExpiredExternalAddr(ExpiredExternalAddr { addr }) => {
FromSwarm::ExternalAddrExpired(ExternalAddrExpired { addr }) => {
self.inner
.on_swarm_event(FromSwarm::ExpiredExternalAddr(ExpiredExternalAddr { addr }));
.on_swarm_event(FromSwarm::ExternalAddrExpired(ExternalAddrExpired { addr }));
self.as_client().on_expired_address(addr);
}
external_addr @ FromSwarm::NewExternalAddr(_) => {
self.inner.on_swarm_event(external_addr);
FromSwarm::NewExternalAddrCandidate(NewExternalAddrCandidate { addr }) => {
self.inner
.on_swarm_event(FromSwarm::NewExternalAddrCandidate(
NewExternalAddrCandidate { addr },
));
self.other_candidates.insert(addr.to_owned());
self.as_client().on_new_address();
}
listen_failure @ FromSwarm::ListenFailure(_) => {
Expand All @@ -580,6 +583,7 @@ impl NetworkBehaviour for Behaviour {
listener_closed @ FromSwarm::ListenerClosed(_) => {
self.inner.on_swarm_event(listener_closed)
}
confirmed @ FromSwarm::ExternalAddrConfirmed(_) => self.inner.on_swarm_event(confirmed),
}
}

Expand Down
24 changes: 5 additions & 19 deletions protocols/autonat/src/behaviour/as_client.rs
Original file line number Diff line number Diff line change
Expand Up @@ -30,9 +30,7 @@ use instant::Instant;
use libp2p_core::Multiaddr;
use libp2p_identity::PeerId;
use libp2p_request_response::{self as request_response, OutboundFailure, RequestId};
use libp2p_swarm::{
AddressScore, ConnectionId, ExternalAddresses, ListenAddresses, PollParameters, ToSwarm,
};
use libp2p_swarm::{ConnectionId, ListenAddresses, PollParameters, ToSwarm};
use rand::{seq::SliceRandom, thread_rng};
use std::{
collections::{HashMap, HashSet, VecDeque},
Expand Down Expand Up @@ -97,13 +95,13 @@ pub(crate) struct AsClient<'a> {
pub(crate) last_probe: &'a mut Option<Instant>,
pub(crate) schedule_probe: &'a mut Delay,
pub(crate) listen_addresses: &'a ListenAddresses,
pub(crate) external_addresses: &'a ExternalAddresses,
pub(crate) other_candidates: &'a HashSet<Multiaddr>,
}

impl<'a> HandleInnerEvent for AsClient<'a> {
fn handle_event(
&mut self,
params: &mut impl PollParameters,
_: &mut impl PollParameters,
event: request_response::Event<DialRequest, DialResponse>,
) -> VecDeque<Action> {
match event {
Expand Down Expand Up @@ -147,19 +145,7 @@ impl<'a> HandleInnerEvent for AsClient<'a> {
}

if let Ok(address) = response.result {
// Update observed address score if it is finite.
#[allow(deprecated)]
// TODO: Fix once we report `AddressScore` through `FromSwarm` event.
let score = params
.external_addresses()
.find_map(|r| (r.addr == address).then_some(r.score))
.unwrap_or(AddressScore::Finite(0));
if let AddressScore::Finite(finite_score) = score {
actions.push_back(ToSwarm::ReportObservedAddr {
address,
score: AddressScore::Finite(finite_score + 1),
});
}
actions.push_back(ToSwarm::ExternalAddrConfirmed(address));
}

actions
Expand Down Expand Up @@ -201,7 +187,7 @@ impl<'a> AsClient<'a> {
self.schedule_probe.reset(self.config.retry_interval);

let addresses = self
.external_addresses
.other_candidates
.iter()
.chain(self.listen_addresses.iter())
.cloned()
Expand Down
6 changes: 3 additions & 3 deletions protocols/autonat/tests/test_client.rs
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@ use libp2p_autonat::{
};
use libp2p_core::Multiaddr;
use libp2p_identity::PeerId;
use libp2p_swarm::{AddressScore, Swarm, SwarmEvent};
use libp2p_swarm::{Swarm, SwarmEvent};
use libp2p_swarm_test::SwarmExt as _;
use std::time::Duration;

Expand Down Expand Up @@ -74,7 +74,7 @@ async fn test_auto_probe() {

// Artificially add a faulty address.
let unreachable_addr: Multiaddr = "/ip4/127.0.0.1/tcp/42".parse().unwrap();
client.add_external_address(unreachable_addr.clone(), AddressScore::Infinite);
client.add_external_address(unreachable_addr.clone());

let id = match client.next_behaviour_event().await {
Event::OutboundProbe(OutboundProbeEvent::Request { probe_id, peer }) => {
Expand Down Expand Up @@ -198,7 +198,7 @@ async fn test_confidence() {
client.listen().await;
} else {
let unreachable_addr = "/ip4/127.0.0.1/tcp/42".parse().unwrap();
client.add_external_address(unreachable_addr, AddressScore::Infinite);
client.add_external_address(unreachable_addr);
}

for i in 0..MAX_CONFIDENCE + 1 {
Expand Down
12 changes: 3 additions & 9 deletions protocols/autonat/tests/test_server.rs
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@ use libp2p_autonat::{
use libp2p_core::{multiaddr::Protocol, ConnectedPoint, Endpoint, Multiaddr};
use libp2p_identity::PeerId;
use libp2p_swarm::DialError;
use libp2p_swarm::{AddressScore, Swarm, SwarmEvent};
use libp2p_swarm::{Swarm, SwarmEvent};
use libp2p_swarm_test::SwarmExt as _;
use std::{num::NonZeroU32, time::Duration};

Expand Down Expand Up @@ -127,10 +127,7 @@ async fn test_dial_back() {
async fn test_dial_error() {
let (mut server, server_id, server_addr) = new_server_swarm(None).await;
let (mut client, client_id) = new_client_swarm(server_id, server_addr).await;
client.add_external_address(
"/ip4/127.0.0.1/tcp/12345".parse().unwrap(),
AddressScore::Infinite,
);
client.add_external_address("/ip4/127.0.0.1/tcp/12345".parse().unwrap());
async_std::task::spawn(client.loop_on_next());

let request_probe_id = match server.next_behaviour_event().await {
Expand Down Expand Up @@ -267,10 +264,7 @@ async fn test_dial_multiple_addr() {

let (mut client, client_id) = new_client_swarm(server_id, server_addr.clone()).await;
client.listen().await;
client.add_external_address(
"/ip4/127.0.0.1/tcp/12345".parse().unwrap(),
AddressScore::Infinite,
);
client.add_external_address("/ip4/127.0.0.1/tcp/12345".parse().unwrap());
async_std::task::spawn(client.loop_on_next());

let dial_addresses = match server.next_behaviour_event().await {
Expand Down
5 changes: 3 additions & 2 deletions protocols/dcutr/src/behaviour_impl.rs
Original file line number Diff line number Diff line change
Expand Up @@ -441,8 +441,9 @@ impl NetworkBehaviour for Behaviour {
| FromSwarm::ExpiredListenAddr(_)
| FromSwarm::ListenerError(_)
| FromSwarm::ListenerClosed(_)
| FromSwarm::NewExternalAddr(_)
| FromSwarm::ExpiredExternalAddr(_) => {}
| FromSwarm::NewExternalAddrCandidate(_)
| FromSwarm::ExternalAddrExpired(_)
| FromSwarm::ExternalAddrConfirmed(_) => {}
}
}
}
Expand Down
5 changes: 3 additions & 2 deletions protocols/floodsub/src/layer.rs
Original file line number Diff line number Diff line change
Expand Up @@ -494,8 +494,9 @@ impl NetworkBehaviour for Floodsub {
| FromSwarm::ExpiredListenAddr(_)
| FromSwarm::ListenerError(_)
| FromSwarm::ListenerClosed(_)
| FromSwarm::NewExternalAddr(_)
| FromSwarm::ExpiredExternalAddr(_) => {}
| FromSwarm::NewExternalAddrCandidate(_)
| FromSwarm::ExternalAddrExpired(_)
| FromSwarm::ExternalAddrConfirmed(_) => {}
}
}
}
Expand Down
5 changes: 3 additions & 2 deletions protocols/gossipsub/src/behaviour.rs
Original file line number Diff line number Diff line change
Expand Up @@ -3500,8 +3500,9 @@ where
| FromSwarm::ExpiredListenAddr(_)
| FromSwarm::ListenerError(_)
| FromSwarm::ListenerClosed(_)
| FromSwarm::NewExternalAddr(_)
| FromSwarm::ExpiredExternalAddr(_) => {}
| FromSwarm::NewExternalAddrCandidate(_)
| FromSwarm::ExternalAddrExpired(_)
| FromSwarm::ExternalAddrConfirmed(_) => {}
}
}
}
Expand Down
18 changes: 8 additions & 10 deletions protocols/identify/src/behaviour.rs
Original file line number Diff line number Diff line change
Expand Up @@ -25,8 +25,8 @@ use libp2p_identity::PeerId;
use libp2p_identity::PublicKey;
use libp2p_swarm::behaviour::{ConnectionClosed, ConnectionEstablished, DialFailure, FromSwarm};
use libp2p_swarm::{
AddressScore, ConnectionDenied, DialError, ExternalAddresses, ListenAddresses,
NetworkBehaviour, NotifyHandler, PollParameters, StreamUpgradeError, THandlerInEvent, ToSwarm,
ConnectionDenied, DialError, ExternalAddresses, ListenAddresses, NetworkBehaviour,
NotifyHandler, PollParameters, StreamUpgradeError, THandlerInEvent, ToSwarm,
};
use libp2p_swarm::{ConnectionId, THandler, THandlerOutEvent};
use lru::LruCache;
Expand All @@ -43,8 +43,7 @@ use std::{
/// about them, and answers identify queries from other nodes.
///
/// All external addresses of the local node supposedly observed by remotes
/// are reported via [`ToSwarm::ReportObservedAddr`] with a
/// [score](AddressScore) of `1`.
/// are reported via [`ToSwarm::NewExternalAddrCandidate`].
pub struct Behaviour {
config: Config,
/// For each peer we're connected to, the observed address to send back to it.
Expand Down Expand Up @@ -295,10 +294,8 @@ impl NetworkBehaviour for Behaviour {
let observed = info.observed_addr.clone();
self.events
.push_back(ToSwarm::GenerateEvent(Event::Received { peer_id, info }));
self.events.push_back(ToSwarm::ReportObservedAddr {
address: observed,
score: AddressScore::Finite(1),
});
self.events
.push_back(ToSwarm::NewExternalAddrCandidate(observed));
}
handler::Event::Identification => {
self.events
Expand Down Expand Up @@ -405,8 +402,9 @@ impl NetworkBehaviour for Behaviour {
| FromSwarm::NewListener(_)
| FromSwarm::ListenerError(_)
| FromSwarm::ListenerClosed(_)
| FromSwarm::NewExternalAddr(_)
| FromSwarm::ExpiredExternalAddr(_) => {}
| FromSwarm::NewExternalAddrCandidate(_)
| FromSwarm::ExternalAddrExpired(_) => {}
FromSwarm::ExternalAddrConfirmed(_) => {}
}
}
}
Expand Down
5 changes: 3 additions & 2 deletions protocols/kad/src/behaviour.rs
Original file line number Diff line number Diff line change
Expand Up @@ -2431,13 +2431,14 @@ where
FromSwarm::DialFailure(dial_failure) => self.on_dial_failure(dial_failure),
FromSwarm::AddressChange(address_change) => self.on_address_change(address_change),
FromSwarm::ExpiredListenAddr(_)
| FromSwarm::NewExternalAddr(_)
| FromSwarm::NewExternalAddrCandidate(_)
| FromSwarm::NewListenAddr(_)
| FromSwarm::ListenFailure(_)
| FromSwarm::NewListener(_)
| FromSwarm::ListenerClosed(_)
| FromSwarm::ListenerError(_)
| FromSwarm::ExpiredExternalAddr(_) => {}
| FromSwarm::ExternalAddrExpired(_)
| FromSwarm::ExternalAddrConfirmed(_) => {}
}
}
}
Expand Down
5 changes: 3 additions & 2 deletions protocols/mdns/src/behaviour.rs
Original file line number Diff line number Diff line change
Expand Up @@ -243,8 +243,9 @@ where
| FromSwarm::ExpiredListenAddr(_)
| FromSwarm::ListenerError(_)
| FromSwarm::ListenerClosed(_)
| FromSwarm::NewExternalAddr(_)
| FromSwarm::ExpiredExternalAddr(_) => {}
| FromSwarm::NewExternalAddrCandidate(_)
| FromSwarm::ExternalAddrExpired(_)
| FromSwarm::ExternalAddrConfirmed(_) => {}
}
}

Expand Down
5 changes: 3 additions & 2 deletions protocols/perf/src/client/behaviour.rs
Original file line number Diff line number Diff line change
Expand Up @@ -128,8 +128,9 @@ impl NetworkBehaviour for Behaviour {
| FromSwarm::ExpiredListenAddr(_)
| FromSwarm::ListenerError(_)
| FromSwarm::ListenerClosed(_)
| FromSwarm::NewExternalAddr(_)
| FromSwarm::ExpiredExternalAddr(_) => {}
| FromSwarm::NewExternalAddrCandidate(_)
| FromSwarm::ExternalAddrExpired(_)
| FromSwarm::ExternalAddrConfirmed(_) => {}
}
}

Expand Down
Loading