Flexible peer filtering (#3458)

* first pass at peers iter cleanup

* more flexible peer with diff lookups

* PeersIter with impl Iterator

* sync against outbound peers
reorder peers filters so expensive filters come later

* filter peers by capabilities during sync

* prefer outbound peers with high total difficulty

* with_difficulty now takes a fn to allow more flexible comparisons based on difficulty

* rename peers_iter() -> iter()

servers/src/common/types.rs

@@ -345,11 +345,11 @@ impl DandelionEpoch {
 	/// Choose a new outbound stem relay peer.
 	pub fn next_epoch(&mut self, peers: &Arc<p2p::Peers>) {
 		self.start_time = Some(Utc::now().timestamp());
-		self.relay_peer = peers.outgoing_connected_peers().first().cloned();
+		self.relay_peer = peers.iter().outbound().connected().choose_random();
 
 		// If stem_probability == 90 then we stem 90% of the time.
-		let mut rng = rand::thread_rng();
 		let stem_probability = self.config.stem_probability;
+		let mut rng = rand::thread_rng();
 		self.is_stem = rng.gen_range(0, 100) < stem_probability;
 
 		let addr = self.relay_peer.clone().map(|p| p.info.addr);
@@ -386,7 +386,7 @@ impl DandelionEpoch {
 		}
 
 		if update_relay {
-			self.relay_peer = peers.outgoing_connected_peers().first().cloned();
+			self.relay_peer = peers.iter().outbound().connected().choose_random();
 			info!(
 				"DandelionEpoch: relay_peer: new peer chosen: {:?}",
 				self.relay_peer.clone().map(|p| p.info.addr)

servers/src/grin/seed.rs

@@ -151,7 +151,7 @@ fn monitor_peers(
 	let mut banned_count = 0;
 	let mut defuncts = vec![];
 
-	for x in peers.all_peers().into_iter() {
+	for x in peers.all_peer_data().into_iter() {
 		match x.flags {
 			p2p::State::Banned => {
 				let interval = Utc::now().timestamp() - x.last_banned;
@@ -174,13 +174,23 @@ fn monitor_peers(
 		total_count += 1;
 	}
 
+	let peers_count = peers.iter().connected().count();
+
+	let max_diff = peers.max_peer_difficulty();
+	let most_work_count = peers
+		.iter()
+		.outbound()
+		.with_difficulty(|x| x >= max_diff)
+		.connected()
+		.count();
+
 	debug!(
 		"monitor_peers: on {}:{}, {} connected ({} most_work). \
 		 all {} = {} healthy + {} banned + {} defunct",
 		config.host,
 		config.port,
-		peers.peer_count(),
-		peers.most_work_peers().len(),
+		peers_count,
+		most_work_count,
 		total_count,
 		healthy_count,
 		banned_count,
@@ -198,10 +208,14 @@ fn monitor_peers(
 		return;
 	}
 
-	// loop over connected peers
+	// loop over connected peers that can provide peer lists
 	// ask them for their list of peers
 	let mut connected_peers: Vec<PeerAddr> = vec![];
-	for p in peers.connected_peers() {
+	for p in peers
+		.iter()
+		.with_capabilities(p2p::Capabilities::PEER_LIST)
+		.connected()
+	{
 		trace!(
 			"monitor_peers: {}:{} ask {} for more peers",
 			config.host,
@@ -331,9 +345,10 @@ fn listen_for_addrs(
 			.name("peer_connect".to_string())
 			.spawn(move || match p2p_c.connect(addr) {
 				Ok(p) => {
-					if p.send_peer_request(capab).is_ok() {
-						let _ = peers_c.update_state(addr, p2p::State::Healthy);
+					if p.info.capabilities.contains(p2p::Capabilities::PEER_LIST) {
+						let _ = p.send_peer_request(capab);
 					}
+					let _ = peers_c.update_state(addr, p2p::State::Healthy);
 				}
 				Err(_) => {
 					let _ = peers_c.update_state(addr, p2p::State::Defunct);

servers/src/grin/server.rs

@@ -16,11 +16,11 @@
 //! the peer-to-peer server, the blockchain and the transaction pool) and acts
 //! as a facade.
 
-use std::fs;
 use std::fs::File;
 use std::io::prelude::*;
 use std::path::Path;
 use std::sync::{mpsc, Arc};
+use std::{convert::TryInto, fs};
 use std::{
 	thread::{self, JoinHandle},
 	time::{self, Duration},
@@ -356,7 +356,13 @@ impl Server {
 
 	/// Number of peers
 	pub fn peer_count(&self) -> u32 {
-		self.p2p.peers.peer_count()
+		self.p2p
+			.peers
+			.iter()
+			.connected()
+			.count()
+			.try_into()
+			.unwrap()
 	}
 
 	/// Start a minimal "stratum" mining service on a separate thread
@@ -486,7 +492,8 @@ impl Server {
 		let peer_stats = self
 			.p2p
 			.peers
-			.connected_peers()
+			.iter()
+			.connected()
 			.into_iter()
 			.map(|p| PeerStats::from_peer(&p))
 			.collect();

servers/src/grin/sync/body_sync.rs

@@ -14,6 +14,7 @@
 
 use chrono::prelude::{DateTime, Utc};
 use chrono::Duration;
+use rand::prelude::*;
 use std::cmp;
 use std::sync::Arc;
 
@@ -89,7 +90,17 @@ impl BodySync {
 
 		hashes.reverse();
 
-		let peers = self.peers.more_work_peers()?;
+		let head = self.chain.head()?;
+
+		// Find connected peers with strictly greater difficulty than us.
+		let peers: Vec<_> = self
+			.peers
+			.iter()
+			.outbound()
+			.with_difficulty(|x| x > head.total_difficulty)
+			.connected()
+			.into_iter()
+			.collect();
 
 		// if we have 5 peers to sync from then ask for 50 blocks total (peer_count *
 		// 10) max will be 80 if all 8 peers are advertising more work
@@ -125,9 +136,9 @@ impl BodySync {
 			self.blocks_requested = 0;
 			self.receive_timeout = Utc::now() + Duration::seconds(6);
 
-			let mut peers_iter = peers.iter().cycle();
+			let mut rng = rand::thread_rng();
 			for hash in hashes_to_get.clone() {
-				if let Some(peer) = peers_iter.next() {
+				if let Some(peer) = peers.choose(&mut rng) {
 					if let Err(e) = peer.send_block_request(*hash, chain::Options::SYNC) {
 						debug!("Skipped request to {}: {:?}", peer.info.addr, e);
 						peer.stop();

servers/src/grin/sync/header_sync.rs

@@ -19,7 +19,7 @@ use std::sync::Arc;
 use crate::chain::{self, SyncState, SyncStatus};
 use crate::common::types::Error;
 use crate::core::core::hash::{Hash, Hashed};
-use crate::p2p::{self, types::ReasonForBan, Peer};
+use crate::p2p::{self, types::ReasonForBan, Capabilities, Peer};
 
 pub struct HeaderSync {
 	sync_state: Arc<SyncState>,
@@ -170,10 +170,17 @@ impl HeaderSync {
 
 	fn header_sync(&mut self) -> Option<Arc<Peer>> {
 		if let Ok(header_head) = self.chain.header_head() {
-			let difficulty = header_head.total_difficulty;
-
-			if let Some(peer) = self.peers.most_work_peer() {
-				if peer.info.total_difficulty() > difficulty {
+			let max_diff = self.peers.max_peer_difficulty();
+			let peer = self
+				.peers
+				.iter()
+				.outbound()
+				.with_capabilities(Capabilities::HEADER_HIST)
+				.with_difficulty(|x| x >= max_diff)
+				.connected()
+				.choose_random();
+			if let Some(peer) = peer {
+				if peer.info.total_difficulty() > header_head.total_difficulty {
 					return self.request_headers(peer);
 				}
 			}

servers/src/grin/sync/state_sync.rs

@@ -19,7 +19,7 @@ use std::sync::Arc;
 use crate::chain::{self, SyncState, SyncStatus};
 use crate::core::core::hash::Hashed;
 use crate::core::global;
-use crate::p2p::{self, Peer};
+use crate::p2p::{self, Capabilities, Peer};
 
 /// Fast sync has 3 "states":
 /// * syncing headers
@@ -158,7 +158,16 @@ impl StateSync {
 		let mut txhashset_height = header_head.height.saturating_sub(threshold);
 		txhashset_height = txhashset_height.saturating_sub(txhashset_height % archive_interval);
 
-		if let Some(peer) = self.peers.most_work_peer() {
+		let max_diff = self.peers.max_peer_difficulty();
+		let peer = self
+			.peers
+			.iter()
+			.outbound()
+			.with_capabilities(Capabilities::TXHASHSET_HIST)
+			.with_difficulty(|x| x >= max_diff)
+			.connected()
+			.choose_random();
+		if let Some(peer) = peer {
 			// ask for txhashset at state_sync_threshold
 			let mut txhashset_head = self
 				.chain

servers/src/grin/sync/syncer.rs

@@ -79,7 +79,15 @@ impl SyncRunner {
 			if self.stop_state.is_stopped() {
 				break;
 			}
-			let wp = self.peers.more_or_same_work_peers()?;
+			// Count peers with at least our difficulty.
+			let wp = self
+				.peers
+				.iter()
+				.outbound()
+				.with_difficulty(|x| x >= head.total_difficulty)
+				.connected()
+				.count();
+
 			// exit loop when:
 			// * we have more than MIN_PEERS more_or_same_work peers
 			// * we are synced already, e.g. grin was quickly restarted
@@ -221,7 +229,16 @@ impl SyncRunner {
 	fn needs_syncing(&self) -> Result<(bool, u64), chain::Error> {
 		let local_diff = self.chain.head()?.total_difficulty;
 		let mut is_syncing = self.sync_state.is_syncing();
-		let peer = self.peers.most_work_peer();
+
+		// Find a peer with greatest known difficulty.
+		let max_diff = self.peers.max_peer_difficulty();
+		let peer = self
+			.peers
+			.iter()
+			.outbound()
+			.with_difficulty(|x| x >= max_diff)
+			.connected()
+			.choose_random();
 
 		let peer_info = if let Some(p) = peer {
 			p.info.clone()