allow asynchronous collation (#290)

* allow asynchronous collation

* remove unnecessary leading colons

Co-Authored-By: Bastian Köcher <bkchr@users.noreply.github.com>

* remove unneeded bound

* Fixes compilation

polkadot/collator/src/lib.rs

@@ -98,12 +98,14 @@ impl<R: fmt::Display> fmt::Display for Error<R> {
 /// This can be implemented through an externally attached service or a stub.
 /// This is expected to be a lightweight, shared type like an Arc.
 pub trait ParachainContext: Clone {
+	type ProduceCandidate: IntoFuture<Item=(BlockData, HeadData, Extrinsic), Error=InvalidHead>;
+
 	/// Produce a candidate, given the latest ingress queue information and the last parachain head.
 	fn produce_candidate<I: IntoIterator<Item=(ParaId, Message)>>(
 		&self,
 		last_head: HeadData,
 		ingress: I,
-	) -> Result<(BlockData, HeadData, Extrinsic), InvalidHead>;
+	) -> Self::ProduceCandidate;
 }
 
 /// Relay chain context needed to collate.
@@ -134,38 +136,44 @@ pub fn collate<'a, R, P>(
 		R::Error: 'a,
 		R::FutureEgress: 'a,
 		P: ParachainContext + 'a,
+		<P::ProduceCandidate as IntoFuture>::Future: Send,
 {
 	let ingress = relay_context.unrouted_egress(local_id).into_future().map_err(Error::Polkadot);
-	ingress.and_then(move |ingress| {
+	ingress
-		let (block_data, head_data, mut extrinsic) = para_context.produce_candidate(
+		.and_then(move |ingress| {
-			last_head,
+			para_context.produce_candidate(
-			ingress.0.iter().flat_map(|&(id, ref msgs)| msgs.iter().cloned().map(move |msg| (id, msg)))
+				last_head,
-		).map_err(Error::Collator)?;
+				ingress.0.iter().flat_map(|&(id, ref msgs)| msgs.iter().cloned().map(move |msg| (id, msg)))
-
+			)
-		let block_data_hash = block_data.hash();
+				.into_future()
-		let signature = key.sign(block_data_hash.as_ref()).into();
+				.map(move |x| (ingress, x))
-		let egress_queue_roots =
+				.map_err(Error::Collator)
-			::polkadot_validation::egress_roots(&mut extrinsic.outgoing_messages);
+		})
-
+		.and_then(move |(ingress, (block_data, head_data, mut extrinsic))| {
-		let receipt = parachain::CandidateReceipt {
+			let block_data_hash = block_data.hash();
-			parachain_index: local_id,
+			let signature = key.sign(block_data_hash.as_ref()).into();
-			collator: key.public(),
+			let egress_queue_roots =
-			signature,
+				polkadot_validation::egress_roots(&mut extrinsic.outgoing_messages);
-			head_data,
+
-			egress_queue_roots,
+			let receipt = parachain::CandidateReceipt {
-			fees: 0,
+				parachain_index: local_id,
-			block_data_hash,
+				collator: key.public(),
-			upward_messages: Vec::new(),
+				signature,
-		};
+				head_data,
-
+				egress_queue_roots,
-		Ok(parachain::Collation {
+				fees: 0,
-			receipt,
+				block_data_hash,
-			pov: PoVBlock {
+				upward_messages: Vec::new(),
-				block_data,
+			};
-				ingress,
+
-			},
+			Ok(parachain::Collation {
+				receipt,
+				pov: PoVBlock {
+					block_data,
+					ingress,
+				},
+			})
 		})
-	})
 }
 
 /// Polkadot-api context.
@@ -216,7 +224,8 @@ impl<P, E> IntoExit for CollationNode<P, E> where
 
 impl<P, E> Worker for CollationNode<P, E> where
 	P: ParachainContext + Send + 'static,
-	E: Future<Item=(),Error=()> + Clone + Send + Sync + 'static
+	E: Future<Item=(),Error=()> + Clone + Send + Sync + 'static,
+	<P::ProduceCandidate as IntoFuture>::Future: Send + 'static,
 {
 	type Work = Box<Future<Item=(),Error=()> + Send>;
 
@@ -376,6 +385,7 @@ pub fn run_collator<P, E, I, ArgT>(
 	version: VersionInfo,
 ) -> polkadot_cli::error::Result<()> where
 	P: ParachainContext + Send + 'static,
+	<P::ProduceCandidate as IntoFuture>::Future: Send + 'static,
 	E: IntoFuture<Item=(),Error=()>,
 	E::Future: Send + Clone + Sync + 'static,
 	I: IntoIterator<Item=ArgT>,
@@ -413,6 +423,8 @@ mod tests {
 	struct DummyParachainContext;
 
 	impl ParachainContext for DummyParachainContext {
+		type ProduceCandidate = Result<(BlockData, HeadData, Extrinsic), InvalidHead>;
+
 		fn produce_candidate<I: IntoIterator<Item=(ParaId, Message)>>(
 			&self,
 			_last_head: HeadData,

polkadot/test-parachains/adder/collator/src/main.rs

@@ -45,6 +45,8 @@ struct AdderContext {
 
 /// The parachain context.
 impl ParachainContext for AdderContext {
+	type ProduceCandidate = Result<(BlockData, HeadData, Extrinsic), InvalidHead>;
+
 	fn produce_candidate<I: IntoIterator<Item=(ParaId, Message)>>(
 		&self,
 		last_head: HeadData,