Revert "Asyncify network functions"

This reverts commit f20ae65.

network/src/lib.rs

@@ -28,6 +28,7 @@ pub mod gossip;
 use codec::{Decode, Encode};
 use futures::channel::{oneshot, mpsc};
 use futures::prelude::*;
+use futures::future::Either;
 use polkadot_primitives::{Block, Hash, Header};
 use polkadot_primitives::parachain::{
 	Id as ParaId, CollatorId, CandidateReceipt, Collation, PoVBlock,
@@ -825,25 +826,34 @@ impl PolkadotProtocol {
 	/// This should be called by a collator intending to get the locally-collated
 	/// block into the hands of validators.
 	/// It also places the outgoing message and block data in the local availability store.
-	pub async fn add_local_collation(
+	pub fn add_local_collation(
 		&mut self,
 		ctx: &mut dyn Context<Block>,
 		relay_parent: Hash,
 		targets: HashSet<ValidatorId>,
 		collation: Collation,
 		outgoing_targeted: OutgoingMessages,
-	) {
+	) -> impl Future<Output = ()> {
 		debug!(target: "p_net", "Importing local collation on relay parent {:?} and parachain {:?}",
 			relay_parent, collation.info.parachain_index);
 
-		if let Some(ref availability_store) = self.availability_store {
-			let collation_cloned = collation.clone();
-			let _ = availability_store.make_available(av_store::Data {
-				relay_parent,
-				parachain_id: collation_cloned.info.parachain_index,
-				block_data: collation_cloned.pov.block_data.clone(),
-				outgoing_queues: Some(outgoing_targeted.clone().into()),
-			}).await;
+		let res = match self.availability_store {
+			Some(ref availability_store) => {
+				let availability_store_cloned = availability_store.clone();
+				let collation_cloned = collation.clone();
+				Either::Left((async move {
+						let _ = availability_store_cloned.make_available(av_store::Data {
+							relay_parent,
+							parachain_id: collation_cloned.info.parachain_index,
+							block_data: collation_cloned.pov.block_data.clone(),
+							outgoing_queues: Some(outgoing_targeted.clone().into()),
+						}).await;
+					}
+				)
+					.boxed()
+				)
+			}
+			None => Either::Right(futures::future::ready(())),
 		};
 
 		for (primary, cloned_collation) in self.local_collations.add_collation(relay_parent, targets, collation.clone()) {
@@ -860,6 +870,8 @@ impl PolkadotProtocol {
 					warn!(target: "polkadot_network", "Encountered tracked but disconnected validator {:?}", primary),
 			}
 		}
+
+		res
 	}
 
 	/// Give the network protocol a handle to an availability store, used for

network/src/router.rs

@@ -175,7 +175,7 @@ impl<P: ProvideRuntimeApi + Send + Sync + 'static, E, N, T> Router<P, E, N, T> w
 				if let Some(work) = producer.map(|p| self.create_work(c_hash, p)) {
 					trace!(target: "validation", "driving statement work to completion");
 
-					let work = select(work.boxed(), self.fetcher.exit().clone())
+					let work = select(work, self.fetcher.exit().clone())
 						.map(drop);
 					let _ = self.fetcher.executor().spawn(work);
 				}
@@ -193,35 +193,35 @@ impl<P: ProvideRuntimeApi + Send + Sync + 'static, E, N, T> Router<P, E, N, T> w
 		let knowledge = self.fetcher.knowledge().clone();
 		let attestation_topic = self.attestation_topic;
 		let parent_hash = self.parent_hash();
-		let api = self.fetcher.api().clone();
-
-		async move {
-			match producer.prime(api).validate().await {
-				Ok(validated) => {
-					// store the data before broadcasting statements, so other peers can fetch.
-					knowledge.lock().note_candidate(
-						candidate_hash,
-						Some(validated.0.pov_block().clone()),
-						validated.0.outgoing_messages().cloned(),
-					);
-
-					// propagate the statement.
-					// consider something more targeted than gossip in the future.
-					let statement = GossipStatement::new(
-						parent_hash,
-						match table.import_validated(validated.0) {
-							None => return,
-							Some(s) => s,
-						}
-					);
-
-					network.gossip_message(attestation_topic, statement.into());
-				}
-				Err(err) => {
-					debug!(target: "p_net", "Failed to produce statements: {:?}", err);
+
+		producer.prime(self.fetcher.api().clone())
+			.validate()
+			.boxed()
+			.map_ok(move |validated| {
+				// store the data before broadcasting statements, so other peers can fetch.
+				knowledge.lock().note_candidate(
+					candidate_hash,
+					Some(validated.0.pov_block().clone()),
+					validated.0.outgoing_messages().cloned(),
+				);
+
+				// propagate the statement.
+				// consider something more targeted than gossip in the future.
+				let statement = GossipStatement::new(
+					parent_hash,
+					match table.import_validated(validated.0) {
+						None => return,
+						Some(s) => s,
+					}
+				);
+
+				network.gossip_message(attestation_topic, statement.into());
+			})
+			.map(|res| {
+				if let Err(e) = res {
+					debug!(target: "p_net", "Failed to produce statements: {:?}", e);
 				}
-			}
-		}
+			})
 	}
 }
 

network/src/tests/validation.rs

@@ -150,7 +150,7 @@ impl NetworkService for TestNetwork {
 	fn gossip_messages_for(&self, topic: Hash) -> GossipMessageStream {
 		let (tx, rx) = mpsc::unbounded();
 		let _  = self.gossip.send_listener.unbounded_send((topic, tx));
-		GossipMessageStream::new(rx.boxed())
+		GossipMessageStream::new(Box::new(rx))
 	}
 
 	fn gossip_message(&self, topic: Hash, message: GossipMessage) {
@@ -419,8 +419,8 @@ impl av_store::ProvideGossipMessages for DummyGossipMessages {
 	fn gossip_messages_for(
 		&self,
 		_topic: Hash
-	) -> Pin<Box<dyn futures::Stream<Item = (Hash, Hash, ErasureChunk)> + Send>> {
-		stream::empty().boxed()
+	) -> Box<dyn futures::Stream<Item = (Hash, Hash, ErasureChunk)> + Send + Unpin> {
+		Box::new(stream::empty())
 	}
 
 	fn gossip_erasure_chunk(

network/src/validation.rs

@@ -158,13 +158,14 @@ impl<P, E, N, T> ValidationNetwork<P, E, N, T> where
 
 impl<P, E, N, T> ValidationNetwork<P, E, N, T> where N: NetworkService {
 	/// Convert the given `CollatorId` to a `PeerId`.
-	pub async fn collator_id_to_peer_id(&self, collator_id: CollatorId) -> Option<PeerId> {
+	pub fn collator_id_to_peer_id(&self, collator_id: CollatorId) ->
+		impl Future<Output=Option<PeerId>> + Send
+	{
 		let (send, recv) = oneshot::channel();
 		self.network.with_spec(move |spec, _| {
 			let _ = send.send(spec.collator_id_to_peer_id(&collator_id).cloned());
 		});
-
-		recv.map(|res| res.unwrap_or(None)).await
+		recv.map(|res| res.unwrap_or(None))
 	}
 
 	/// Create a `Stream` of checked statements for the given `relay_parent`.