network: Use "one shot" protocol handler.

Add two new `NetworkBehaviour`s, one handling remote block requests
and another one to handle light client requests (both local and from
remote). The change is motivated by the desire to use multiple
substreams of a single connection for different protocols. To achieve
this, libp2p's `OneShotHandler` is used as a protocol handler in each
behaviour. It will open a fresh substream for the duration of the
request and close it afterwards. For block requests, we currently only
handle incoming requests from remote and tests are missing. For light
client handling we support incoming requests from remote and also
ported a substantial amount of functionality over from
`light_dispatch.rs` (including several tests). However the result lacks
in at least two aspects:

(1) We require external updates w.r.t. the best block per peer and
currently nothing updates this information.
(2) We carry a lot of peer-related state around.

Both aspects could be simplified by externalising peer selection and
just requiring a specific peer ID where the request should be sent to.
We still have to maintain some peer related state due to the way
libp2p's swarm and network behaviour work (e.g. we must make sure to
always issue `NetworkBehaviourAction::SendEvent`s to peers we are
connected to, otherwise the actions die a silent death.

Another change implemented here is the use of protocol buffers as the
encoding for network messages. Certain individual fields of messages
are still SCALE encoded. There has been some discussion about this
in another PR (paritytech#3452), so
far without resolution.

client/network/Cargo.toml

@@ -6,6 +6,9 @@ license = "GPL-3.0"
 authors = ["Parity Technologies <admin@parity.io>"]
 edition = "2018"
 
+[build-dependencies]
+prost-build = "0.6.1"
+
 [dependencies]
 bitflags = "1.2.0"
 bytes = "0.5.0"
@@ -24,7 +27,9 @@ linked-hash-map = "0.5.2"
 linked_hash_set = "0.1.3"
 log = "0.4.8"
 lru = "0.4.0"
+nohash-hasher = "0.1.3"
 parking_lot = "0.10.0"
+prost = "0.6.1"
 rand = "0.7.2"
 rustc-hex = "2.0.1"
 sc-block-builder = { version = "0.8", path = "../block-builder" }
@@ -45,16 +50,21 @@ sp-keyring = { version = "2.0.0", optional = true, path = "../../primitives/keyr
 sp-runtime = { version = "2.0.0", path = "../../primitives/runtime" }
 substrate-test-client = { version = "2.0.0", optional = true, path = "../../test-utils/client" }
 substrate-test-runtime-client = { version = "2.0.0", optional = true, path = "../../test-utils/runtime/client" }
+thiserror = "1"
 unsigned-varint = { version = "0.3.0", features = ["codec"] }
 void = "1.0.2"
 zeroize = "1.0.0"
 
 [dev-dependencies]
+async-std = "1.5"
+assert_matches = "1.3"
 env_logger = "0.7.0"
 quickcheck = "0.9.0"
 rand = "0.7.2"
 sp-keyring = { version = "2.0.0", path = "../../primitives/keyring" }
 sp-test-primitives = { version = "2.0.0", path = "../../primitives/test-primitives" }
+substrate-test-runtime = { version = "2.0.0", path = "../../test-utils/runtime" }
+substrate-test-runtime-client = { version = "2.0.0", path = "../../test-utils/runtime/client" }
 tempfile = "3.1.0"
 
 [features]

client/network/build.rs

@@ -0,0 +1,8 @@
+const PROTOS: &[&str] = &[
+	"src/protocol/schema/api.v1.proto",
+	"src/protocol/schema/light.v1.proto"
+];
+
+fn main() {
+	prost_build::compile_protos(PROTOS, &["src/protocol"]).unwrap();
+}

client/network/src/behaviour.rs

@@ -19,7 +19,7 @@ use crate::{
 	Event, protocol::event::DhtEvent
 };
 use crate::{ExHashT, specialization::NetworkSpecialization};
-use crate::protocol::{CustomMessageOutcome, Protocol};
+use crate::protocol::{self, CustomMessageOutcome, Protocol};
 use libp2p::NetworkBehaviour;
 use libp2p::core::{Multiaddr, PeerId, PublicKey};
 use libp2p::kad::record;
@@ -42,7 +42,10 @@ pub struct Behaviour<B: BlockT, S: NetworkSpecialization<B>, H: ExHashT> {
 	debug_info: debug_info::DebugInfoBehaviour<Substream<StreamMuxerBox>>,
 	/// Discovers nodes of the network.
 	discovery: DiscoveryBehaviour<Substream<StreamMuxerBox>>,
-
+	/// Block request handling.
+	block_requests: protocol::BlockRequests<Substream<StreamMuxerBox>, B>,
+	/// Light client request handling.
+	light_client_handler: protocol::LightClientHandler<Substream<StreamMuxerBox>, B>,
 	/// Queue of events to produce for the outside.
 	#[behaviour(ignore)]
 	events: Vec<BehaviourOut<B>>,
@@ -65,6 +68,8 @@ impl<B: BlockT, S: NetworkSpecialization<B>, H: ExHashT> Behaviour<B, S, H> {
 		known_addresses: Vec<(PeerId, Multiaddr)>,
 		enable_mdns: bool,
 		allow_private_ipv4: bool,
+		block_requests: protocol::BlockRequests<Substream<StreamMuxerBox>, B>,
+		light_client_handler: protocol::LightClientHandler<Substream<StreamMuxerBox>, B>
 	) -> Self {
 		Behaviour {
 			substrate,
@@ -75,7 +80,9 @@ impl<B: BlockT, S: NetworkSpecialization<B>, H: ExHashT> Behaviour<B, S, H> {
 				enable_mdns,
 				allow_private_ipv4
 			).await,
-			events: Vec::new(),
+			block_requests,
+			light_client_handler,
+			events: Vec::new()
 		}
 	}
 
@@ -117,6 +124,11 @@ impl<B: BlockT, S: NetworkSpecialization<B>, H: ExHashT> Behaviour<B, S, H> {
 	pub fn put_value(&mut self, key: record::Key, value: Vec<u8>) {
 		self.discovery.put_value(key, value);
 	}
+
+//	/// Issue a light client request.
+//	pub fn light_client_request(&mut self, r: light_client_handler::Request<B>) -> Result<(), light_client_handler::Error> {
+//		self.light_client_handler.request(r)
+//	}
 }
 
 impl<B: BlockT, S: NetworkSpecialization<B>, H: ExHashT> NetworkBehaviourEventProcess<void::Void> for

client/network/src/protocol.rs

@@ -54,15 +54,30 @@ use crate::error;
 use util::LruHashSet;
 use wasm_timer::Instant;
 
+// Include sources generated from protobuf definitions.
+pub mod api {
+	pub mod v1 {
+		include!(concat!(env!("OUT_DIR"), "/api.v1.rs"));
+		pub mod light {
+			include!(concat!(env!("OUT_DIR"), "/api.v1.light.rs"));
+		}
+	}
+}
+
 mod legacy_proto;
 mod util;
 
+pub mod block_requests;
 pub mod message;
 pub mod event;
+pub mod light_client_handler;
 pub mod light_dispatch;
 pub mod specialization;
 pub mod sync;
 
+pub use block_requests::BlockRequests;
+pub use light_client_handler::LightClientHandler;
+
 const REQUEST_TIMEOUT_SEC: u64 = 40;
 /// Interval at which we perform time based maintenance
 const TICK_TIMEOUT: time::Duration = time::Duration::from_millis(1100);