introduce backpressure between network and protocol

core/network-libp2p/src/service_task.rs

@@ -59,6 +59,7 @@ pub enum NetworkMsg {
 	RemoveReservedPeer(PeerId),
 	AcceptUnreservedPeers,
 	DenyUnreservedPeers,
+	SkipNetworkPoll,
 }
 
 pub struct NetworkPort {
@@ -285,7 +286,10 @@ pub enum ServiceEvent {
 	DisconnectNode {
 		/// Index of the node.
 		node_index: NodeIndex,
-	}
+	},
+
+	/// Skipped a network poll, while still handling incoming protocol messages.
+	SkippedNetworkPoll,
 }
 
 /// Network service. Must be polled regularly in order for the networking to work.
@@ -1039,6 +1043,9 @@ impl Service {
 		};
 		for msg in messages {
 			match msg {
+				// Injecting a ServiceEvent results in not polling the network in the current poll.
+				// If protocol is too busy, ensure at least one event is injected.
+				NetworkMsg::SkipNetworkPoll => self.injected_events.push(ServiceEvent::SkippedNetworkPoll),
 				NetworkMsg::PeerIds(node_idxs, sender) => {
 					let reply = node_idxs.into_iter().map(|idx| {
 						(idx, self.peer_id_of_node(idx).map(|p| p.clone()))

core/network/src/service.rs

@@ -37,6 +37,9 @@ use tokio::runtime::Runtime;
 
 pub use network_libp2p::NetworkMsg;
 
+// Number of messages in the protocol queue after which we start throttling network requests.
+const PROTOCOL_HIGH_WATERMARK: usize = 10;
+
 /// Type that represents fetch completion future.
 pub type FetchFuture = oneshot::Receiver<Vec<u8>>;
 
@@ -352,6 +355,9 @@ fn run_thread<B: BlockT + 'static>(
 	// The network service produces events about what happens on the network. Let's process them.
 	stream::poll_fn(move || network_service.lock().poll()).for_each(move |event| {
 		match event {
+			NetworkServiceEvent::SkippedNetworkPoll => {
+				// Network skipped a network poll, while still remaining responsive to messages from protocol.
+			}
 			NetworkServiceEvent::DisconnectNode { node_index } => {
 				let _ = protocol_sender.send(ProtocolMsg::PeerDisconnected(node_index));
 			}
@@ -394,6 +400,11 @@ fn run_thread<B: BlockT + 'static>(
 				};
 			}
 		};
+		// If protocol is too busy, tell network to skip one polling of the network,
+		// while remaining responsive to incoming messages from protocol
+		if protocol_sender.len() > PROTOCOL_HIGH_WATERMARK {
+			let _ = network_sender.send(NetworkMsg::SkipNetworkPoll);
+		}
 		Ok(())
 	})
 }