Increase parallel downloads to 5 (#4045)

* Increase parallel downloads to 5

* CLI param

substrate/core/network/src/config.rs

@@ -81,7 +81,7 @@ pub struct Params<B: BlockT, S, H: ExHashT> {
 	pub specialization: S,
 
 	/// Type to check incoming block announcements.
-	pub block_announce_validator: Box<dyn BlockAnnounceValidator<B> + Send>
+	pub block_announce_validator: Box<dyn BlockAnnounceValidator<B> + Send>,
 }
 
 bitflags! {
@@ -261,6 +261,8 @@ pub struct NetworkConfiguration {
 	pub node_name: String,
 	/// Configuration for the transport layer.
 	pub transport: TransportConfig,
+	/// Maximum number of peers to ask the same blocks in parallel.
+	pub max_parallel_downloads: u32,
 }
 
 impl Default for NetworkConfiguration {
@@ -282,6 +284,7 @@ impl Default for NetworkConfiguration {
 				enable_mdns: false,
 				wasm_external_transport: None,
 			},
+			max_parallel_downloads: 5,
 		}
 	}
 }

substrate/core/network/src/protocol.rs

@@ -362,12 +362,15 @@ struct ContextData<B: BlockT, H: ExHashT> {
 pub struct ProtocolConfig {
 	/// Assigned roles.
 	pub roles: Roles,
+	/// Maximum number of peers to ask the same blocks in parallel.
+	pub max_parallel_downloads: u32,
 }
 
 impl Default for ProtocolConfig {
 	fn default() -> ProtocolConfig {
 		ProtocolConfig {
 			roles: Roles::FULL,
+			max_parallel_downloads: 5,
 		}
 	}
 }
@@ -393,6 +396,7 @@ impl<B: BlockT, S: NetworkSpecialization<B>, H: ExHashT> Protocol<B, S, H> {
 			&info,
 			finality_proof_request_builder,
 			block_announce_validator,
+			config.max_parallel_downloads,
 		);
 		let (peerset, peerset_handle) = peerset::Peerset::from_config(peerset_config);
 		let versions = &((MIN_VERSION as u8)..=(CURRENT_VERSION as u8)).collect::<Vec<u8>>();

substrate/core/network/src/protocol/sync.rs

@@ -127,7 +127,9 @@ pub struct ChainSync<B: BlockT> {
 	/// A flag that caches idle state with no pending requests.
 	is_idle: bool,
 	/// A type to check incoming block announcements.
-	block_announce_validator: Box<dyn BlockAnnounceValidator<B> + Send>
+	block_announce_validator: Box<dyn BlockAnnounceValidator<B> + Send>,
+	/// Maximum number of peers to ask the same blocks in parallel.
+	max_parallel_downloads: u32,
 }
 
 /// All the data we have about a Peer that we are trying to sync with
@@ -282,7 +284,8 @@ impl<B: BlockT> ChainSync<B> {
 		client: Arc<dyn crate::chain::Client<B>>,
 		info: &ClientInfo<B>,
 		request_builder: Option<BoxFinalityProofRequestBuilder<B>>,
-		block_announce_validator: Box<dyn BlockAnnounceValidator<B> + Send>
+		block_announce_validator: Box<dyn BlockAnnounceValidator<B> + Send>,
+		max_parallel_downloads: u32,
 	) -> Self {
 		let mut required_block_attributes = BlockAttributes::HEADER | BlockAttributes::JUSTIFICATION;
 
@@ -306,6 +309,7 @@ impl<B: BlockT> ChainSync<B> {
 			fork_targets: Default::default(),
 			is_idle: false,
 			block_announce_validator,
+			max_parallel_downloads,
 		}
 	}
 
@@ -571,6 +575,7 @@ impl<B: BlockT> ChainSync<B> {
 		let best_queued = self.best_queued_number;
 		let client = &self.client;
 		let queue = &self.queue_blocks;
+		let max_parallel = if major_sync { 1 } else { self.max_parallel_downloads };
 		let iter = self.peers.iter_mut().filter_map(move |(id, peer)| {
 			if !peer.state.is_available() {
 				trace!(target: "sync", "Peer {} is busy", id);
@@ -592,13 +597,19 @@ impl<B: BlockT> ChainSync<B> {
 				peer.state = PeerSyncState::DownloadingStale(hash);
 				have_requests = true;
 				Some((id.clone(), req))
-			} else if let Some((range, req)) = peer_block_request(id, peer, blocks, attrs, major_sync) {
+			} else if let Some((range, req)) = peer_block_request(id, peer, blocks, attrs, max_parallel) {
 				peer.state = PeerSyncState::DownloadingNew(range.start);
-				trace!(target: "sync", "New block request for {}", id);
+				trace!(
+					target: "sync",
+					"New block request for {}, (best:{}, common:{}) {:?}",
+					id,
+					peer.best_number,
+					peer.common_number,
+					req,
+				);
 				have_requests = true;
 				Some((id.clone(), req))
 			} else {
-				trace!(target: "sync", "No new block request for {}", id);
 				None
 			}
 		});
@@ -1006,7 +1017,7 @@ impl<B: BlockT> ChainSync<B> {
 	{
 		let header = &announce.header;
 		let number = *header.number();
-		debug!(target: "sync", "Received block announcement with number {:?}", number);
+		debug!(target: "sync", "Received block announcement {:?} with number {:?} from {}", hash, number, who);
 		if number.is_zero() {
 			warn!(target: "sync", "Ignored genesis block (#0) announcement from {}: {}", who, hash);
 			return OnBlockAnnounce::Nothing
@@ -1226,15 +1237,14 @@ fn peer_block_request<B: BlockT>(
 	peer: &PeerSync<B>,
 	blocks: &mut BlockCollection<B>,
 	attrs: &message::BlockAttributes,
-	major_sync: bool,
+	max_parallel_downloads: u32,
 ) -> Option<(Range<NumberFor<B>>, BlockRequest<B>)> {
-	let max_parallel = if major_sync { 1 } else { 3 };
 	if let Some(range) = blocks.needed_blocks(
 		id.clone(),
 		MAX_BLOCKS_TO_REQUEST,
 		peer.best_number,
 		peer.common_number,
-		max_parallel,
+		max_parallel_downloads,
 	) {
 		let request = message::generic::BlockRequest {
 			id: 0,

substrate/core/network/src/protocol/sync/blocks.rs

@@ -105,6 +105,10 @@ impl<B: BlockT> BlockCollection<B> {
 		max_parallel: u32,
 	) -> Option<Range<NumberFor<B>>>
 	{
+		if peer_best <= common {
+			// Bail out early
+			return None;
+		}
 		// First block number that we need to download
 		let first_different = common + <NumberFor<B>>::one();
 		let count = (count as u32).into();

substrate/core/network/src/service.rs

@@ -194,7 +194,10 @@ impl<B: BlockT + 'static, S: NetworkSpecialization<B>, H: ExHashT> NetworkWorker
 		let num_connected = Arc::new(AtomicUsize::new(0));
 		let is_major_syncing = Arc::new(AtomicBool::new(false));
 		let (protocol, peerset_handle) = Protocol::new(
-			protocol::ProtocolConfig { roles: params.roles },
+			protocol::ProtocolConfig {
+				roles: params.roles,
+				max_parallel_downloads: params.network_config.max_parallel_downloads,
+			},
 			params.chain,
 			params.on_demand.as_ref().map(|od| od.checker().clone())
 				.unwrap_or(Arc::new(AlwaysBadChecker)),