fluencelabs/rust-libp2p
1
0
Fork 0
mirror of https://github.com/fluencelabs/rust-libp2p synced 2025-06-16 19:41:22 +00:00
Code Issues Projects Releases Wiki Activity

protocols/mdns: Make libp2p-mdns socket agnostic (#1699)

Browse Source 
Allow libp2p-mdns to use either async-std or tokio to drive required UDP
socket.

Co-authored-by: Roman Borschel <romanb@users.noreply.github.com>
This commit is contained in:
Max Inden
2020-08-18 14:51:03 +02:00
committed by GitHub
parent b4ad2d6297
commit cbdbf656c0
7 changed files with 370 additions and 302 deletions
Download Patch File Download Diff File Expand all files Collapse all files

398
protocols/mdns/src/behaviour.rs
View File

@ -18,7 +18,7 @@
// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
// DEALINGS IN THE SOFTWARE.
use crate::service::{MdnsService, MdnsPacket, build_query_response, build_service_discovery_response};
use crate::service::{MdnsPacket, build_query_response, build_service_discovery_response};
use futures::prelude::*;
use libp2p_core::{
Multiaddr,
@ -41,11 +41,14 @@ use wasm_timer::{Delay, Instant};
const MDNS_RESPONSE_TTL: std::time::Duration = Duration::from_secs(5 * 60);
macro_rules! codegen {
($feature_name:expr, $behaviour_name:ident, $maybe_busy_wrapper:ident, $service_name:ty) => {
/// A `NetworkBehaviour` for mDNS. Automatically discovers peers on the local network and adds
/// them to the topology.
pub struct Mdns {
pub struct $behaviour_name {
/// The inner service.
service: MaybeBusyMdnsService,
service: $maybe_busy_wrapper,
/// List of nodes that we have discovered, the address, and when their TTL expires.
///
@ -63,37 +66,37 @@ pub struct Mdns {
/// and a `MdnsPacket` (similar to the old Tokio socket send style). The two states are thus `Free`
/// with an `MdnsService` or `Busy` with a future returning the original `MdnsService` and an
/// `MdnsPacket`.
enum MaybeBusyMdnsService {
Free(MdnsService),
Busy(Pin<Box<dyn Future<Output = (MdnsService, MdnsPacket)> + Send>>),
enum $maybe_busy_wrapper {
Free($service_name),
Busy(Pin<Box<dyn Future<Output = ($service_name, MdnsPacket)> + Send>>),
Poisoned,
}
impl fmt::Debug for MaybeBusyMdnsService {
impl fmt::Debug for $maybe_busy_wrapper {
fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
MaybeBusyMdnsService::Free(service) => {
fmt.debug_struct("MaybeBusyMdnsService::Free")
$maybe_busy_wrapper::Free(service) => {
fmt.debug_struct("$maybe_busy_wrapper::Free")
.field("service", service)
.finish()
},
MaybeBusyMdnsService::Busy(_) => {
fmt.debug_struct("MaybeBusyMdnsService::Busy")
$maybe_busy_wrapper::Busy(_) => {
fmt.debug_struct("$maybe_busy_wrapper::Busy")
.finish()
}
MaybeBusyMdnsService::Poisoned => {
fmt.debug_struct("MaybeBusyMdnsService::Poisoned")
$maybe_busy_wrapper::Poisoned => {
fmt.debug_struct("$maybe_busy_wrapper::Poisoned")
.finish()
}
}
}
}
impl Mdns {
impl $behaviour_name {
/// Builds a new `Mdns` behaviour.
pub fn new() -> io::Result<Mdns> {
Ok(Mdns {
service: MaybeBusyMdnsService::Free(MdnsService::new()?),
pub fn new() -> io::Result<$behaviour_name> {
Ok($behaviour_name {
service: $maybe_busy_wrapper::Free(<$service_name>::new()?),
discovered_nodes: SmallVec::new(),
closest_expiration: None,
})
@ -110,6 +113,191 @@ impl Mdns {
}
}
impl NetworkBehaviour for $behaviour_name {
type ProtocolsHandler = DummyProtocolsHandler;
type OutEvent = MdnsEvent;
fn new_handler(&mut self) -> Self::ProtocolsHandler {
DummyProtocolsHandler::default()
}
fn addresses_of_peer(&mut self, peer_id: &PeerId) -> Vec<Multiaddr> {
let now = Instant::now();
self.discovered_nodes
.iter()
.filter(move |(p, _, expires)| p == peer_id && *expires > now)
.map(|(_, addr, _)| addr.clone())
.collect()
}
fn inject_connected(&mut self, _: &PeerId) {}
fn inject_disconnected(&mut self, _: &PeerId) {}
fn inject_event(
&mut self,
_: PeerId,
_: ConnectionId,
_ev: <Self::ProtocolsHandler as ProtocolsHandler>::OutEvent,
) {
void::unreachable(_ev)
}
fn poll(
&mut self,
cx: &mut Context<'_>,
params: &mut impl PollParameters,
) -> Poll<
NetworkBehaviourAction<
<Self::ProtocolsHandler as ProtocolsHandler>::InEvent,
Self::OutEvent,
>,
> {
// Remove expired peers.
if let Some(ref mut closest_expiration) = self.closest_expiration {
match Future::poll(Pin::new(closest_expiration), cx) {
Poll::Ready(Ok(())) => {
let now = Instant::now();
let mut expired = SmallVec::<[(PeerId, Multiaddr); 4]>::new();
while let Some(pos) = self.discovered_nodes.iter().position(|(_, _, exp)| *exp < now) {
let (peer_id, addr, _) = self.discovered_nodes.remove(pos);
expired.push((peer_id, addr));
}
if !expired.is_empty() {
let event = MdnsEvent::Expired(ExpiredAddrsIter {
inner: expired.into_iter(),
});
return Poll::Ready(NetworkBehaviourAction::GenerateEvent(event));
}
},
Poll::Pending => (),
Poll::Ready(Err(err)) => warn!("timer has errored: {:?}", err),
}
}
// Polling the mDNS service, and obtain the list of nodes discovered this round.
let discovered = loop {
let service = mem::replace(&mut self.service, $maybe_busy_wrapper::Poisoned);
let packet = match service {
$maybe_busy_wrapper::Free(service) => {
self.service = $maybe_busy_wrapper::Busy(Box::pin(service.next()));
continue;
},
$maybe_busy_wrapper::Busy(mut fut) => {
match fut.as_mut().poll(cx) {
Poll::Ready((service, packet)) => {
self.service = $maybe_busy_wrapper::Free(service);
packet
},
Poll::Pending => {
self.service = $maybe_busy_wrapper::Busy(fut);
return Poll::Pending;
}
}
},
$maybe_busy_wrapper::Poisoned => panic!("Mdns poisoned"),
};
match packet {
MdnsPacket::Query(query) => {
// MaybeBusyMdnsService should always be Free.
if let $maybe_busy_wrapper::Free(ref mut service) = self.service {
let resp = build_query_response(
query.query_id(),
params.local_peer_id().clone(),
params.listened_addresses().into_iter(),
MDNS_RESPONSE_TTL,
);
service.enqueue_response(resp.unwrap());
} else { debug_assert!(false); }
},
MdnsPacket::Response(response) => {
// We replace the IP address with the address we observe the
// remote as and the address they listen on.
let obs_ip = Protocol::from(response.remote_addr().ip());
let obs_port = Protocol::Udp(response.remote_addr().port());
let observed: Multiaddr = iter::once(obs_ip)
.chain(iter::once(obs_port))
.collect();
let mut discovered: SmallVec<[_; 4]> = SmallVec::new();
for peer in response.discovered_peers() {
if peer.id() == params.local_peer_id() {
continue;
}
let new_expiration = Instant::now() + peer.ttl();
let mut addrs: Vec<Multiaddr> = Vec::new();
for addr in peer.addresses() {
if let Some(new_addr) = address_translation(&addr, &observed) {
addrs.push(new_addr.clone())
}
addrs.push(addr.clone())
}
for addr in addrs {
if let Some((_, _, cur_expires)) = self.discovered_nodes.iter_mut()
.find(|(p, a, _)| p == peer.id() && *a == addr)
{
*cur_expires = cmp::max(*cur_expires, new_expiration);
} else {
self.discovered_nodes.push((peer.id().clone(), addr.clone(), new_expiration));
}
discovered.push((peer.id().clone(), addr));
}
}
break discovered;
},
MdnsPacket::ServiceDiscovery(disc) => {
// MaybeBusyMdnsService should always be Free.
if let $maybe_busy_wrapper::Free(ref mut service) = self.service {
let resp = build_service_discovery_response(
disc.query_id(),
MDNS_RESPONSE_TTL,
);
service.enqueue_response(resp);
} else { debug_assert!(false); }
},
}
};
// Getting this far implies that we discovered new nodes. As the final step, we need to
// refresh `closest_expiration`.
self.closest_expiration = self.discovered_nodes.iter()
.fold(None, |exp, &(_, _, elem_exp)| {
Some(exp.map(|exp| cmp::min(exp, elem_exp)).unwrap_or(elem_exp))
})
.map(Delay::new_at);
Poll::Ready(NetworkBehaviourAction::GenerateEvent(MdnsEvent::Discovered(DiscoveredAddrsIter {
inner: discovered.into_iter(),
})))
}
}
impl fmt::Debug for $behaviour_name {
fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
fmt.debug_struct("Mdns")
.field("service", &self.service)
.finish()
}
}
};
}
#[cfg(feature = "async-std")]
codegen!("async-std", Mdns, MaybeBusyMdnsService, crate::service::MdnsService);
#[cfg(feature = "tokio")]
codegen!("tokio", TokioMdns, MaybeBusyTokioMdnsService, crate::service::TokioMdnsService);
/// Event that can be produced by the `Mdns` behaviour.
#[derive(Debug)]
pub enum MdnsEvent {
@ -180,179 +368,3 @@ impl fmt::Debug for ExpiredAddrsIter {
.finish()
}
}
impl NetworkBehaviour for Mdns {
type ProtocolsHandler = DummyProtocolsHandler;
type OutEvent = MdnsEvent;
fn new_handler(&mut self) -> Self::ProtocolsHandler {
DummyProtocolsHandler::default()
}
fn addresses_of_peer(&mut self, peer_id: &PeerId) -> Vec<Multiaddr> {
let now = Instant::now();
self.discovered_nodes
.iter()
.filter(move |(p, _, expires)| p == peer_id && *expires > now)
.map(|(_, addr, _)| addr.clone())
.collect()
}
fn inject_connected(&mut self, _: &PeerId) {}
fn inject_disconnected(&mut self, _: &PeerId) {}
fn inject_event(
&mut self,
_: PeerId,
_: ConnectionId,
_ev: <Self::ProtocolsHandler as ProtocolsHandler>::OutEvent,
) {
void::unreachable(_ev)
}
fn poll(
&mut self,
cx: &mut Context<'_>,
params: &mut impl PollParameters,
) -> Poll<
NetworkBehaviourAction<
<Self::ProtocolsHandler as ProtocolsHandler>::InEvent,
Self::OutEvent,
>,
> {
// Remove expired peers.
if let Some(ref mut closest_expiration) = self.closest_expiration {
match Future::poll(Pin::new(closest_expiration), cx) {
Poll::Ready(Ok(())) => {
let now = Instant::now();
let mut expired = SmallVec::<[(PeerId, Multiaddr); 4]>::new();
while let Some(pos) = self.discovered_nodes.iter().position(|(_, _, exp)| *exp < now) {
let (peer_id, addr, _) = self.discovered_nodes.remove(pos);
expired.push((peer_id, addr));
}
if !expired.is_empty() {
let event = MdnsEvent::Expired(ExpiredAddrsIter {
inner: expired.into_iter(),
});
return Poll::Ready(NetworkBehaviourAction::GenerateEvent(event));
}
},
Poll::Pending => (),
Poll::Ready(Err(err)) => warn!("timer has errored: {:?}", err),
}
}
// Polling the mDNS service, and obtain the list of nodes discovered this round.
let discovered = loop {
let service = mem::replace(&mut self.service, MaybeBusyMdnsService::Poisoned);
let packet = match service {
MaybeBusyMdnsService::Free(service) => {
self.service = MaybeBusyMdnsService::Busy(Box::pin(service.next()));
continue;
},
MaybeBusyMdnsService::Busy(mut fut) => {
match fut.as_mut().poll(cx) {
Poll::Ready((service, packet)) => {
self.service = MaybeBusyMdnsService::Free(service);
packet
},
Poll::Pending => {
self.service = MaybeBusyMdnsService::Busy(fut);
return Poll::Pending;
}
}
},
MaybeBusyMdnsService::Poisoned => panic!("Mdns poisoned"),
};
match packet {
MdnsPacket::Query(query) => {
// MaybeBusyMdnsService should always be Free.
if let MaybeBusyMdnsService::Free(ref mut service) = self.service {
let resp = build_query_response(
query.query_id(),
params.local_peer_id().clone(),
params.listened_addresses().into_iter(),
MDNS_RESPONSE_TTL,
);
service.enqueue_response(resp.unwrap());
} else { debug_assert!(false); }
},
MdnsPacket::Response(response) => {
// We replace the IP address with the address we observe the
// remote as and the address they listen on.
let obs_ip = Protocol::from(response.remote_addr().ip());
let obs_port = Protocol::Udp(response.remote_addr().port());
let observed: Multiaddr = iter::once(obs_ip)
.chain(iter::once(obs_port))
.collect();
let mut discovered: SmallVec<[_; 4]> = SmallVec::new();
for peer in response.discovered_peers() {
if peer.id() == params.local_peer_id() {
continue;
}
let new_expiration = Instant::now() + peer.ttl();
let mut addrs: Vec<Multiaddr> = Vec::new();
for addr in peer.addresses() {
if let Some(new_addr) = address_translation(&addr, &observed) {
addrs.push(new_addr.clone())
}
addrs.push(addr.clone())
}
for addr in addrs {
if let Some((_, _, cur_expires)) = self.discovered_nodes.iter_mut()
.find(|(p, a, _)| p == peer.id() && *a == addr)
{
*cur_expires = cmp::max(*cur_expires, new_expiration);
} else {
self.discovered_nodes.push((peer.id().clone(), addr.clone(), new_expiration));
}
discovered.push((peer.id().clone(), addr));
}
}
break discovered;
},
MdnsPacket::ServiceDiscovery(disc) => {
// MaybeBusyMdnsService should always be Free.
if let MaybeBusyMdnsService::Free(ref mut service) = self.service {
let resp = build_service_discovery_response(
disc.query_id(),
MDNS_RESPONSE_TTL,
);
service.enqueue_response(resp);
} else { debug_assert!(false); }
},
}
};
// Getting this far implies that we discovered new nodes. As the final step, we need to
// refresh `closest_expiration`.
self.closest_expiration = self.discovered_nodes.iter()
.fold(None, |exp, &(_, _, elem_exp)| {
Some(exp.map(|exp| cmp::min(exp, elem_exp)).unwrap_or(elem_exp))
})
.map(Delay::new_at);
Poll::Ready(NetworkBehaviourAction::GenerateEvent(MdnsEvent::Discovered(DiscoveredAddrsIter {
inner: discovered.into_iter(),
})))
}
}
impl fmt::Debug for Mdns {
fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
fmt.debug_struct("Mdns")
.field("service", &self.service)
.finish()
}
}