Skip to content
This repository was archived by the owner on Nov 15, 2023. It is now read-only.
73 changes: 56 additions & 17 deletions client/network/src/service.rs
Original file line number Diff line number Diff line change
Expand Up @@ -973,23 +973,7 @@ impl<B: BlockT + 'static, H: ExHashT> NetworkService<B, H> {
/// Returns an `Err` if one of the given addresses is invalid or contains an
/// invalid peer ID (which includes the local peer ID).
pub fn set_priority_group(&self, group_id: String, peers: HashSet<Multiaddr>) -> Result<(), String> {
let peers = peers.into_iter()
.map(|mut addr| {
let peer = match addr.pop() {
Some(multiaddr::Protocol::P2p(key)) => PeerId::from_multihash(key)
.map_err(|_| "Invalid PeerId format".to_string())?,
_ => return Err("Missing PeerId from address".to_string()),
};

// Make sure the local peer ID is never added to the PSM
// or added as a "known address", even if given.
if peer == self.local_peer_id {
Err("Local peer ID in priority group.".to_string())
} else {
Ok((peer, addr))
}
})
.collect::<Result<Vec<(PeerId, Multiaddr)>, String>>()?;
let peers = self.parse_multiaddr(peers)?;

let peer_ids = peers.iter().map(|(peer_id, _addr)| peer_id.clone()).collect();

Expand All @@ -1004,6 +988,40 @@ impl<B: BlockT + 'static, H: ExHashT> NetworkService<B, H> {
Ok(())
}

/// Add peers to a peerset priority group.
///
/// Each `Multiaddr` must end with a `/p2p/` component containing the `PeerId`.
///
/// Returns an `Err` if one of the given addresses is invalid or contains an
/// invalid peer ID (which includes the local peer ID).
pub fn add_to_priority_group(&self, group_id: String, peers: HashSet<Multiaddr>) -> Result<(), String> {

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
pub fn add_to_priority_group(&self, group_id: String, peers: HashSet<Multiaddr>) -> Result<(), String> {
pub async fn add_to_priority_group(&self, group_id: String, peers: HashSet<Multiaddr>) -> Result<(), String> {

Declaring this method as async would grant us a greater implementation flexibility in the future, e.g. a bounded channel between NetworkService and NetworkWorker.

@tomaka what do you think of requiring all new methods on NetworkService to be able to yield?

@ordian Are you able to await this method within Polkadot?

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

yes, awaiting is fine, but would we need to change all three methods then (set_, add_, remove_)?

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done in b1c28b2.

@tomaka tomaka Oct 9, 2020

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

👍 As eventually it'd be nice if the queue from the network service to the worker was bounded as well.

let peers = self.parse_multiaddr(peers)?;

for (peer_id, addr) in peers.into_iter() {
self.peerset.add_to_priority_group(group_id.clone(), peer_id.clone());

let _ = self
.to_worker
.unbounded_send(ServiceToWorkerMsg::AddKnownAddress(peer_id, addr));
}

Ok(())
}

/// Remove peers from a peerset priority group.
///
/// Each `Multiaddr` must end with a `/p2p/` component containing the `PeerId`.
///
/// Returns an `Err` if one of the given addresses is invalid or contains an
/// invalid peer ID (which includes the local peer ID).
pub fn remove_from_priority_group(&self, group_id: String, peers: HashSet<Multiaddr>) -> Result<(), String> {
let peers = self.parse_multiaddr(peers)?;
for (peer_id, _) in peers.into_iter() {
self.peerset.remove_from_priority_group(group_id.clone(), peer_id);
}
Ok(())
}

/// Returns the number of peers we're connected to.
pub fn num_connected(&self) -> usize {
self.num_connected.load(Ordering::Relaxed)
Expand All @@ -1025,6 +1043,27 @@ impl<B: BlockT + 'static, H: ExHashT> NetworkService<B, H> {
.to_worker
.unbounded_send(ServiceToWorkerMsg::OwnBlockImported(hash, number));
}

// Utility function to extract `PeerId` from each `Multiaddr` for priority group updates.
Comment thread
ordian marked this conversation as resolved.
Outdated
fn parse_multiaddr(&self, peers: HashSet<Multiaddr>) -> Result<Vec<(PeerId, Multiaddr)>, String> {
Comment thread
ordian marked this conversation as resolved.
Outdated
peers.into_iter()
.map(|mut addr| {
let peer = match addr.pop() {
Some(multiaddr::Protocol::P2p(key)) => PeerId::from_multihash(key)
.map_err(|_| "Invalid PeerId format".to_string())?,
_ => return Err("Missing PeerId from address".to_string()),
};

// Make sure the local peer ID is never added to the PSM
// or added as a "known address", even if given.
if peer == self.local_peer_id {
Err("Local peer ID in priority group.".to_string())
} else {
Ok((peer, addr))
}
})
.collect::<Result<Vec<(PeerId, Multiaddr)>, String>>()
}
}

impl<B: BlockT + 'static, H: ExHashT> sp_consensus::SyncOracle
Expand Down