Share peers between syncing strategies

prdoc/pr_2814.prdoc

@@ -0,0 +1,15 @@
+# Schema: Polkadot SDK PRDoc Schema (prdoc) v1.0.0
+# See doc at https://raw.githubusercontent.com/paritytech/polkadot-sdk/master/prdoc/schema_user.json
+
+title: Share peers between syncing strategies
+
+doc:
+  - audience: Node Dev
+    description: |
+      New struct `PeerPool` is added to `SyncingStrategy` to allow running syncing strategies
+      in parellel and sharing peers. Available peers (with no active requests) are retrieved by
+      calling `PeerPool::available_peers()`, and a peer is reserved for a request by a syncing
+      strategy using `AvailablePeer::reserve()`.
+
+crates: 
+  - name: sc-network-sync

substrate/client/network/sync/Cargo.toml

@@ -28,6 +28,7 @@ futures-timer = "3.0.2"
 libp2p = "0.51.4"
 log = "0.4.17"
 mockall = "0.11.3"
+parking_lot = "0.12.1"
 prost = "0.12"
 schnellru = "0.2.1"
 smallvec = "1.11.0"

substrate/client/network/sync/src/engine.rs

@@ -214,9 +214,6 @@ pub struct SyncingEngine<B: BlockT, Client> {
 	/// Syncing strategy.
 	strategy: SyncingStrategy<B, Client>,
 
-	/// Syncing configuration for startegies.
-	syncing_config: SyncingConfig,
-
 	/// Blockchain client.
 	client: Arc<Client>,
 
@@ -441,8 +438,7 @@ where
 			.map_or(futures::future::pending().boxed().fuse(), |rx| rx.boxed().fuse());
 
 		// Initialize syncing strategy.
-		let strategy =
-			SyncingStrategy::new(syncing_config.clone(), client.clone(), warp_sync_config)?;
+		let strategy = SyncingStrategy::new(syncing_config, client.clone(), warp_sync_config)?;
 
 		let block_announce_protocol_name = block_announce_config.protocol_name().clone();
 		let (tx, service_rx) = tracing_unbounded("mpsc_chain_sync", 100_000);
@@ -471,7 +467,6 @@ where
 				roles,
 				client,
 				strategy,
-				syncing_config,
 				network_service,
 				peers: HashMap::new(),
 				block_announce_data_cache: LruMap::new(ByLength::new(cache_capacity)),
@@ -661,8 +656,15 @@ where
 					Some(event) => self.process_notification_event(event),
 					None => return,
 				},
-				warp_target_block_header = &mut self.warp_sync_target_block_header_rx_fused =>
-					self.pass_warp_sync_target_block_header(warp_target_block_header),
+				warp_target_block_header = &mut self.warp_sync_target_block_header_rx_fused => {
+					if let Err(_) = self.pass_warp_sync_target_block_header(warp_target_block_header) {
+						error!(
+							target: LOG_TARGET,
+							"Failed to set warp sync target block header, terminating `SyncingEngine`.",
+						);
+						return
+					}
+				},
 				response_event = self.pending_responses.select_next_some() =>
 					self.process_response_event(response_event),
 				validation_result = self.block_announce_validator.select_next_some() =>
@@ -675,14 +677,17 @@ where
 
 			// Process actions requested by a syncing strategy.
 			if let Err(e) = self.process_strategy_actions() {
-				error!("Terminating `SyncingEngine` due to fatal error: {e:?}");
+				error!(
+					target: LOG_TARGET,
+					"Terminating `SyncingEngine` due to fatal error: {e:?}.",
+				);
 				return
 			}
 		}
 	}
 
 	fn process_strategy_actions(&mut self) -> Result<(), ClientError> {
-		for action in self.strategy.actions() {
+		for action in self.strategy.actions()? {
 			match action {
 				SyncingAction::SendBlockRequest { peer_id, request } => {
 					// Sending block request implies dropping obsolete pending response as we are
@@ -699,7 +704,7 @@ where
 						removed,
 					)
 				},
-				SyncingAction::CancelBlockRequest { peer_id } => {
+				SyncingAction::CancelRequest { peer_id } => {
 					let removed = self.pending_responses.remove(&peer_id);
 
 					trace!(
@@ -753,20 +758,8 @@ where
 						number,
 					)
 				},
-				SyncingAction::Finished => {
-					let connected_peers = self.peers.iter().filter_map(|(peer_id, peer)| {
-						peer.info.roles.is_full().then_some((
-							*peer_id,
-							peer.info.best_hash,
-							peer.info.best_number,
-						))
-					});
-					self.strategy.switch_to_next(
-						self.syncing_config.clone(),
-						self.client.clone(),
-						connected_peers,
-					)?;
-				},
+				// Nothing to do, this is handled internally by `SyncingStrategy`.
+				SyncingAction::Finished => {},
 			}
 		}
 
@@ -948,23 +941,18 @@ where
 		}
 	}
 
-	fn pass_warp_sync_target_block_header(&mut self, header: Result<B::Header, oneshot::Canceled>) {
+	fn pass_warp_sync_target_block_header(
+		&mut self,
+		header: Result<B::Header, oneshot::Canceled>,
+	) -> Result<(), ()> {
 		match header {
-			Ok(header) =>
-				if let SyncingStrategy::WarpSyncStrategy(warp_sync) = &mut self.strategy {
-					warp_sync.set_target_block(header);
-				} else {
-					error!(
-						target: LOG_TARGET,
-						"Cannot set warp sync target block: no warp sync strategy is active."
-					);
-					debug_assert!(false);
-				},
+			Ok(header) => self.strategy.set_warp_sync_target_block_header(header),
 			Err(err) => {
 				error!(
 					target: LOG_TARGET,
 					"Failed to get target block for warp sync. Error: {err:?}",
 				);
+				Err(())
 			},
 		}
 	}

substrate/client/network/sync/src/extra_requests.rs

@@ -17,17 +17,17 @@
 // along with this program. If not, see <https://www.gnu.org/licenses/>.
 
 use crate::{
-	request_metrics::Metrics,
-	strategy::chain_sync::{PeerSync, PeerSyncState},
-	LOG_TARGET,
+	peer_pool::PeerPool, request_metrics::Metrics, strategy::chain_sync::PeerSync, LOG_TARGET,
 };
 use fork_tree::ForkTree;
 use libp2p::PeerId;
-use log::{debug, trace, warn};
+use log::{debug, error, trace, warn};
+use parking_lot::Mutex;
 use sp_blockchain::Error as ClientError;
 use sp_runtime::traits::{Block as BlockT, NumberFor, Zero};
 use std::{
 	collections::{HashMap, HashSet, VecDeque},
+	sync::Arc,
 	time::{Duration, Instant},
 };
 
@@ -288,6 +288,7 @@ impl<'a, B: BlockT> Matcher<'a, B> {
 	pub(crate) fn next(
 		&mut self,
 		peers: &HashMap<PeerId, PeerSync<B>>,
+		peer_pool: &Arc<Mutex<PeerPool>>,
 	) -> Option<(PeerId, ExtraRequest<B>)> {
 		if self.remaining == 0 {
 			return None
@@ -299,36 +300,41 @@ impl<'a, B: BlockT> Matcher<'a, B> {
 		}
 
 		while let Some(request) = self.extras.pending_requests.pop_front() {
-			for (peer, sync) in
-				peers.iter().filter(|(_, sync)| sync.state == PeerSyncState::Available)
-			{
+			for mut available_peer in peer_pool.lock().available_peers() {
+				let peer_id = available_peer.peer_id();
+				let Some(sync) = peers.get(peer_id) else {
+					error!(
+						target: LOG_TARGET,
+						"Peer {peer_id} is available, but not known to `ChainSync`",
+					);
+					debug_assert!(false);
+					continue
+				};
 				// only ask peers that have synced at least up to the block number that we're asking
 				// the extra for
 				if sync.best_number < request.1 {
 					continue
 				}
-				// don't request to any peers that already have pending requests
-				if self.extras.active_requests.contains_key(peer) {
-					continue
-				}
 				// only ask if the same request has not failed for this peer before
 				if self
 					.extras
 					.failed_requests
 					.get(&request)
-					.map(|rr| rr.iter().any(|i| &i.0 == peer))
+					.map(|rr| rr.iter().any(|i| &i.0 == peer_id))
 					.unwrap_or(false)
 				{
 					continue
 				}
-				self.extras.active_requests.insert(*peer, request);
+
+				available_peer.reserve();
+				self.extras.active_requests.insert(*peer_id, request);
 
 				trace!(target: LOG_TARGET,
 					"Sending {} request to {:?} for {:?}",
-					self.extras.request_type_name, peer, request,
+					self.extras.request_type_name, peer_id, request,
 				);
 
-				return Some((*peer, request))
+				return Some((*peer_id, request))
 			}
 
 			self.extras.pending_requests.push_back(request);
@@ -346,19 +352,26 @@ impl<'a, B: BlockT> Matcher<'a, B> {
 #[cfg(test)]
 mod tests {
 	use super::*;
-	use crate::strategy::chain_sync::PeerSync;
+	use crate::{
+		peer_pool::PeerPool,
+		strategy::chain_sync::{PeerSync, PeerSyncState},
+	};
+	use parking_lot::Mutex;
 	use quickcheck::{Arbitrary, Gen, QuickCheck};
 	use sp_blockchain::Error as ClientError;
 	use sp_test_primitives::{Block, BlockNumber, Hash};
-	use std::collections::{HashMap, HashSet};
+	use std::{
+		collections::{HashMap, HashSet},
+		sync::Arc,
+	};
 
 	#[test]
 	fn requests_are_processed_in_order() {
-		fn property(mut peers: ArbitraryPeers) {
+		fn property(mut ap: ArbitraryPeers) {
 			let mut requests = ExtraRequests::<Block>::new("test");
 
 			let num_peers_available =
-				peers.0.values().filter(|s| s.state == PeerSyncState::Available).count();
+				ap.peers.values().filter(|s| s.state == PeerSyncState::Available).count();
 
 			for i in 0..num_peers_available {
 				requests.schedule((Hash::random(), i as u64), |a, b| Ok(a[0] >= b[0]))
@@ -368,9 +381,9 @@ mod tests {
 			let mut m = requests.matcher();
 
 			for p in &pending {
-				let (peer, r) = m.next(&peers.0).unwrap();
+				let (peer, r) = m.next(&ap.peers, &ap.peer_pool).unwrap();
 				assert_eq!(p, &r);
-				peers.0.get_mut(&peer).unwrap().state =
+				ap.peers.get_mut(&peer).unwrap().state =
 					PeerSyncState::DownloadingJustification(r.0);
 			}
 		}
@@ -397,19 +410,19 @@ mod tests {
 
 	#[test]
 	fn disconnecting_implies_rescheduling() {
-		fn property(mut peers: ArbitraryPeers) -> bool {
+		fn property(mut ap: ArbitraryPeers) -> bool {
 			let mut requests = ExtraRequests::<Block>::new("test");
 
 			let num_peers_available =
-				peers.0.values().filter(|s| s.state == PeerSyncState::Available).count();
+				ap.peers.values().filter(|s| s.state == PeerSyncState::Available).count();
 
 			for i in 0..num_peers_available {
 				requests.schedule((Hash::random(), i as u64), |a, b| Ok(a[0] >= b[0]))
 			}
 
 			let mut m = requests.matcher();
-			while let Some((peer, r)) = m.next(&peers.0) {
-				peers.0.get_mut(&peer).unwrap().state =
+			while let Some((peer, r)) = m.next(&ap.peers, &ap.peer_pool) {
+				ap.peers.get_mut(&peer).unwrap().state =
 					PeerSyncState::DownloadingJustification(r.0);
 			}
 
@@ -433,19 +446,19 @@ mod tests {
 
 	#[test]
 	fn no_response_reschedules() {
-		fn property(mut peers: ArbitraryPeers) {
+		fn property(mut ap: ArbitraryPeers) {
 			let mut requests = ExtraRequests::<Block>::new("test");
 
 			let num_peers_available =
-				peers.0.values().filter(|s| s.state == PeerSyncState::Available).count();
+				ap.peers.values().filter(|s| s.state == PeerSyncState::Available).count();
 
 			for i in 0..num_peers_available {
 				requests.schedule((Hash::random(), i as u64), |a, b| Ok(a[0] >= b[0]))
 			}
 
 			let mut m = requests.matcher();
-			while let Some((peer, r)) = m.next(&peers.0) {
-				peers.0.get_mut(&peer).unwrap().state =
+			while let Some((peer, r)) = m.next(&ap.peers, &ap.peer_pool) {
+				ap.peers.get_mut(&peer).unwrap().state =
 					PeerSyncState::DownloadingJustification(r.0);
 			}
 
@@ -570,16 +583,22 @@ mod tests {
 	}
 
 	#[derive(Debug, Clone)]
-	struct ArbitraryPeers(HashMap<PeerId, PeerSync<Block>>);
+	struct ArbitraryPeers {
+		peers: HashMap<PeerId, PeerSync<Block>>,
+		peer_pool: Arc<Mutex<PeerPool>>,
+	}
 
 	impl Arbitrary for ArbitraryPeers {
 		fn arbitrary(g: &mut Gen) -> Self {
 			let mut peers = HashMap::with_capacity(g.size());
+			let peer_pool = Arc::new(Mutex::new(PeerPool::default()));
 			for _ in 0..g.size() {
 				let ps = ArbitraryPeerSync::arbitrary(g).0;
-				peers.insert(ps.peer_id, ps);
+				let peer_id = ps.peer_id;
+				peers.insert(peer_id, ps);
+				peer_pool.lock().add_peer(peer_id);
 			}
-			ArbitraryPeers(peers)
+			ArbitraryPeers { peers, peer_pool }
 		}
 	}
 }

substrate/client/network/sync/src/lib.rs

@@ -25,6 +25,7 @@ pub use types::{SyncEvent, SyncEventStream, SyncState, SyncStatus, SyncStatusPro
 mod block_announce_validator;
 mod extra_requests;
 mod futures_stream;
+mod peer_pool;
 mod pending_responses;
 mod request_metrics;
 mod schema;