Move requests-responses and polling from ChainSync to `SyncingEngin…

…e` (#1650)

Move request-response handling from `ChainSync` to `SyncingEngine` as
part of [Sync
2.0](#534) refactoring
aimed at making `ChainSync` a pure state machine.

Resolves #502.

---------

Co-authored-by: Aaro Altonen <[email protected]>
Co-authored-by: Bastian Köcher <[email protected]>

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

@@ -36,7 +36,7 @@ use sp_runtime::{
 };
 use warp::WarpSyncProgress;
 
-use std::{any::Any, fmt, fmt::Formatter, pin::Pin, sync::Arc, task::Poll};
+use std::{any::Any, fmt, fmt::Formatter, pin::Pin, sync::Arc};
 
 /// The sync status of a peer we are trying to sync with
 #[derive(Debug)]
@@ -204,6 +204,23 @@ pub enum PeerRequest<B: BlockT> {
 	WarpProof,
 }
 
+#[derive(Debug)]
+pub enum PeerRequestType {
+	Block,
+	State,
+	WarpProof,
+}
+
+impl<B: BlockT> PeerRequest<B> {
+	pub fn get_type(&self) -> PeerRequestType {
+		match self {
+			PeerRequest::Block(_) => PeerRequestType::Block,
+			PeerRequest::State => PeerRequestType::State,
+			PeerRequest::WarpProof => PeerRequestType::WarpProof,
+		}
+	}
+}
+
 /// Wrapper for implementation-specific state request.
 ///
 /// NOTE: Implementation must be able to encode and decode it for network purposes.
@@ -289,9 +306,6 @@ pub trait ChainSync<Block: BlockT>: Send {
 	/// Returns the current number of peers stored within this state machine.
 	fn num_peers(&self) -> usize;
 
-	/// Returns the number of peers we're connected to and that are being queried.
-	fn num_active_peers(&self) -> usize;
-
 	/// Handle a new connected peer.
 	///
 	/// Call this method whenever we connect to a new peer.
@@ -369,10 +383,4 @@ pub trait ChainSync<Block: BlockT>: Send {
 
 	/// Return some key metrics.
 	fn metrics(&self) -> Metrics;
-
-	/// Advance the state of `ChainSync`
-	fn poll(&mut self, cx: &mut std::task::Context) -> Poll<()>;
-
-	/// Send block request to peer
-	fn send_block_request(&mut self, who: PeerId, request: BlockRequest<Block>);
 }

substrate/client/network/sync/Cargo.toml

@@ -29,6 +29,7 @@ prost = "0.11"
 schnellru = "0.2.1"
 smallvec = "1.11.0"
 thiserror = "1.0"
+tokio-stream = "0.1.14"
 fork-tree = { path = "../../../utils/fork-tree" }
 prometheus-endpoint = { package = "substrate-prometheus-endpoint", path = "../../../utils/prometheus" }
 sc-client-api = { path = "../../api" }