Minimal parachains part 2: Parachain statement and data routing (#173)

* dynamic inclusion threshold calculator

* collators interface

* collation helpers

* initial proposal-creation future

* create proposer when asked to propose

* remove local_availability duty

* statement table tracks includable parachain count

* beginnings of timing future

* finish proposal logic

* remove stray println

* extract shared table to separate module

* change ordering

* includability tracking

* fix doc

* initial changes to parachains module

* initialise dummy block before API calls

* give polkadot control over round proposer based on random seed

* propose only after enough candidates

* flesh out parachains module a bit more

* set_heads

* actually introduce set_heads to runtime

* update block_builder to accept parachains

* split block validity errors from real errors in evaluation

* update WASM runtimes

* polkadot-api methods for parachains additions

* delay evaluation until candidates are ready

* comments

* fix dynamic inclusion with zero initial

* test for includability tracker

* wasm validation of parachain candidates

* move primitives to primitives crate

* remove runtime-std dependency from codec

* adjust doc

* polkadot-parachain-primitives

* kill legacy polkadot-validator crate

* basic-add test chain

* test for basic_add parachain

* move to test-chains dir

* use wasm-build

* new wasm directory layout

* reorganize a bit more

* Fix for rh-minimal-parachain (#141)

* Remove extern "C"

We already encountered such behavior (bug?) in pwasm-std, I believe.

* Fix `panic_fmt` signature by adding `_col`

Wrong `panic_fmt` signature can inhibit some optimizations in LTO mode.

* Add linker flags and use wasm-gc in build script

Pass --import-memory to LLD to emit wasm binary with imported memory.

Also use wasm-gc instead of wasm-build.

* Fix effective_max.

I'm not sure why it was the way it was actually.

* Recompile wasm.

* Fix indent

* more basic_add tests

* validate parachain WASM

* produce statements on receiving statements

* tests for reactive statement production

* fix build

* add OOM lang item to runtime-io

* use dynamic_inclusion when evaluating as well

* fix update_includable_count

* remove dead code

* grumbles

* actually defer round_proposer logic

* update wasm

* address a few more grumbles

* schedule collation work as soon as BFT is started

* impl future in collator

* fix comment

* governance proposals for adding and removing parachains

* bump protocol version

* tear out polkadot-specific pieces of substrate-network

* extract out polkadot-specific stuff from substrate-network

* begin polkadot network subsystem

* grumbles

* update WASM checkins

* parse status from polkadot peer

* allow invoke of network specialization

* begin statement router implementation

* remove dependency on tokio-timer

* fix sanity check and have proposer factory create communication streams

* pull out statement routing from consensus library

* fix comments

* adjust typedefs

* extract consensus_gossip out of main network protocol handler

* port substrate-bft to new tokio

* port polkadot-consensus to new tokio

* fix typo

* start message processing task

* initial consensus network implementation

* remove known tracking from statement-table crate

* extract router into separate module

* defer statements until later

* double signature is invalid

* propagating statements

* grumbles

* request block data

* fix compilation

* embed new consensus network into service

* port demo CLI to tokio

* all test crates compile

* some tests for fetching block data

* whitespace

* adjusting some tokio stuff

* update exit-future

* remove overly noisy warning

* clean up collation work a bit

* address review grumbles

* fix lock order in protocol handler

* rebuild wasm artifacts

* tag AuthorityId::from_slice for std only

* address formatting grumbles

* rename event_loop to executor

* some more docs for polkadot-network crate

polkadot/consensus/src/service.rs

@@ -18,6 +18,10 @@
 
 /// Consensus service. A long runnung service that manages BFT agreement and parachain
 /// candidate agreement over the network.
+///
+/// This uses a handle to an underlying thread pool to dispatch heavy work
+/// such as candidate verification while performing event-driven work
+/// on a local event loop.
 
 use std::thread;
 use std::time::{Duration, Instant};
@@ -27,197 +31,37 @@ use bft::{self, BftService};
 use client::{BlockchainEvents, ChainHead};
 use ed25519;
 use futures::prelude::*;
-use futures::{future, Canceled};
 use polkadot_api::LocalPolkadotApi;
-use polkadot_primitives::{BlockId, Block, Header, Hash, AccountId};
-use polkadot_primitives::parachain::{Id as ParaId, BlockData, Extrinsic, CandidateReceipt};
-use primitives::AuthorityId;
-use runtime_support::Hashable;
-use substrate_network as net;
-use tokio_core::reactor;
+use polkadot_primitives::{Block, Header};
 use transaction_pool::TransactionPool;
 
-use super::{TableRouter, SharedTable, ProposerFactory};
+use tokio::executor::current_thread::TaskExecutor as LocalThreadHandle;
+use tokio::runtime::TaskExecutor as ThreadPoolHandle;
+use tokio::runtime::current_thread::Runtime as LocalRuntime;
+use tokio::timer::Interval;
+
+use super::{Network, Collators, ProposerFactory};
 use error;
 
 const TIMER_DELAY_MS: u64 = 5000;
 const TIMER_INTERVAL_MS: u64 = 500;
 
-struct BftSink<E> {
-	network: Arc<net::ConsensusService<Block>>,
-	parent_hash: Hash,
-	_e: ::std::marker::PhantomData<E>,
-}
-
-struct Messages {
-	network_stream: net::BftMessageStream<Block>,
-	local_id: AuthorityId,
-	authorities: Vec<AuthorityId>,
-}
-
-impl Stream for Messages {
-	type Item = bft::Communication<Block>;
-	type Error = bft::Error;
-
-	fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
-		// check the network
-		loop {
-			match self.network_stream.poll() {
-				Err(_) => return Err(bft::InputStreamConcluded.into()),
-				Ok(Async::NotReady) => return Ok(Async::NotReady),
-				Ok(Async::Ready(None)) => return Ok(Async::NotReady), // the input stream for agreements is never meant to logically end.
-				Ok(Async::Ready(Some(message))) => {
-					match process_message(message, &self.local_id, &self.authorities) {
-						Ok(Some(message)) => return Ok(Async::Ready(Some(message))),
-						Ok(None) => {} // ignored local message.
-						Err(e) => {
-							debug!("Message validation failed: {:?}", e);
-						}
-					}
-				}
-			}
-		}
-	}
-}
-
-fn process_message(msg: net::LocalizedBftMessage<Block>, local_id: &AuthorityId, authorities: &[AuthorityId]) -> Result<Option<bft::Communication<Block>>, bft::Error> {
-	Ok(Some(match msg.message {
-		net::generic_message::BftMessage::Consensus(c) => bft::generic::Communication::Consensus(match c {
-			net::generic_message::SignedConsensusMessage::Propose(proposal) => bft::generic::LocalizedMessage::Propose({
-				if &proposal.sender == local_id { return Ok(None) }
-				let proposal = bft::generic::LocalizedProposal {
-					round_number: proposal.round_number as usize,
-					proposal: proposal.proposal,
-					digest: proposal.digest,
-					sender: proposal.sender,
-					digest_signature: ed25519::LocalizedSignature {
-						signature: proposal.digest_signature,
-						signer: proposal.sender.into(),
-					},
-					full_signature: ed25519::LocalizedSignature {
-						signature: proposal.full_signature,
-						signer: proposal.sender.into(),
-					}
-				};
-				bft::check_proposal(authorities, &msg.parent_hash, &proposal)?;
-
-				trace!(target: "bft", "importing proposal message for round {} from {}", proposal.round_number, proposal.sender);
-				proposal
-			}),
-			net::generic_message::SignedConsensusMessage::Vote(vote) => bft::generic::LocalizedMessage::Vote({
-				if &vote.sender == local_id { return Ok(None) }
-				let vote = bft::generic::LocalizedVote {
-					sender: vote.sender,
-					signature: ed25519::LocalizedSignature {
-						signature: vote.signature,
-						signer: vote.sender.into(),
-					},
-					vote: match vote.vote {
-						net::generic_message::ConsensusVote::Prepare(r, h) => bft::generic::Vote::Prepare(r as usize, h),
-						net::generic_message::ConsensusVote::Commit(r, h) => bft::generic::Vote::Commit(r as usize, h),
-						net::generic_message::ConsensusVote::AdvanceRound(r) => bft::generic::Vote::AdvanceRound(r as usize),
-					}
-				};
-				bft::check_vote::<Block>(authorities, &msg.parent_hash, &vote)?;
-
-				trace!(target: "bft", "importing vote {:?} from {}", vote.vote, vote.sender);
-				vote
-			}),
-		}),
-		net::generic_message::BftMessage::Auxiliary(a) => {
-			let justification = bft::UncheckedJustification::<Hash>::from(a);
-			// TODO: get proper error
-			let justification: Result<_, bft::Error> = bft::check_prepare_justification::<Block>(authorities, msg.parent_hash, justification)
-				.map_err(|_| bft::ErrorKind::InvalidJustification.into());
-			bft::generic::Communication::Auxiliary(justification?)
-		},
-	}))
-}
-
-impl<E> Sink for BftSink<E> {
-	type SinkItem = bft::Communication<Block>;
-	// TODO: replace this with the ! type when that's stabilized
-	type SinkError = E;
-
-	fn start_send(&mut self, message: bft::Communication<Block>) -> ::futures::StartSend<bft::Communication<Block>, E> {
-		let network_message = net::generic_message::LocalizedBftMessage {
-			message: match message {
-				bft::generic::Communication::Consensus(c) => net::generic_message::BftMessage::Consensus(match c {
-					bft::generic::LocalizedMessage::Propose(proposal) => net::generic_message::SignedConsensusMessage::Propose(net::generic_message::SignedConsensusProposal {
-						round_number: proposal.round_number as u32,
-						proposal: proposal.proposal,
-						digest: proposal.digest,
-						sender: proposal.sender,
-						digest_signature: proposal.digest_signature.signature,
-						full_signature: proposal.full_signature.signature,
-					}),
-					bft::generic::LocalizedMessage::Vote(vote) => net::generic_message::SignedConsensusMessage::Vote(net::generic_message::SignedConsensusVote {
-						sender: vote.sender,
-						signature: vote.signature.signature,
-						vote: match vote.vote {
-							bft::generic::Vote::Prepare(r, h) => net::generic_message::ConsensusVote::Prepare(r as u32, h),
-							bft::generic::Vote::Commit(r, h) => net::generic_message::ConsensusVote::Commit(r as u32, h),
-							bft::generic::Vote::AdvanceRound(r) => net::generic_message::ConsensusVote::AdvanceRound(r as u32),
-						}
-					}),
-				}),
-				bft::generic::Communication::Auxiliary(justification) => net::generic_message::BftMessage::Auxiliary(justification.uncheck().into()),
-			},
-			parent_hash: self.parent_hash,
-		};
-		self.network.send_bft_message(network_message);
-		Ok(::futures::AsyncSink::Ready)
-	}
-
-	fn poll_complete(&mut self) -> ::futures::Poll<(), E> {
-		Ok(Async::Ready(()))
-	}
-}
-
-struct Network(Arc<net::ConsensusService<Block>>);
-
-impl super::Network for Network {
-	type TableRouter = Router;
-	fn table_router(&self, _table: Arc<SharedTable>) -> Self::TableRouter {
-		Router {
-			network: self.0.clone()
-		}
-	}
-}
-
+// spin up an instance of BFT agreement on the current thread's executor.
+// panics if there is no current thread executor.
 fn start_bft<F, C>(
 	header: &Header,
-	handle: reactor::Handle,
-	client: &bft::Authorities<Block>,
-	network: Arc<net::ConsensusService<Block>>,
 	bft_service: &BftService<Block, F, C>,
 ) where
-	F: bft::ProposerFactory<Block> + 'static,
+	F: bft::Environment<Block> + 'static,
 	C: bft::BlockImport<Block> + bft::Authorities<Block> + 'static,
-	<F as bft::ProposerFactory<Block>>::Error: ::std::fmt::Debug,
+	F::Error: ::std::fmt::Debug,
 	<F::Proposer as bft::Proposer<Block>>::Error: ::std::fmt::Display + Into<error::Error>,
 {
-	let parent_hash = header.hash();
-	if bft_service.live_agreement().map_or(false, |h| h == parent_hash) {
-		return;
-	}
-	let authorities = match client.authorities(&BlockId::hash(parent_hash)) {
-		Ok(authorities) => authorities,
-		Err(e) => {
-			debug!("Error reading authorities: {:?}", e);
-			return;
-		}
-	};
-
-	let input = Messages {
-		network_stream: network.bft_messages(parent_hash),
-		local_id: bft_service.local_id(),
-		authorities,
-	};
-
-	let output = BftSink { network: network, parent_hash: parent_hash, _e: Default::default() };
-	match bft_service.build_upon(&header, input.map_err(Into::into), output) {
-		Ok(Some(bft)) => handle.spawn(bft),
+	let mut handle = LocalThreadHandle::current();
+	match bft_service.build_upon(&header) {
+		Ok(Some(bft)) => if let Err(e) = handle.spawn_local(Box::new(bft)) {
+			debug!(target: "bft", "Couldn't initialize BFT agreement: {:?}", e);
+		},
 		Ok(None) => {},
 		Err(e) => debug!(target: "bft", "BFT agreement error: {:?}", e),
 	}
@@ -231,54 +75,56 @@ pub struct Service {
 
 impl Service {
 	/// Create and start a new instance.
-	pub fn new<A, C>(
+	pub fn new<A, C, N>(
 		client: Arc<C>,
 		api: Arc<A>,
-		network: Arc<net::ConsensusService<Block>>,
+		network: N,
 		transaction_pool: Arc<TransactionPool<A>>,
+		thread_pool: ThreadPoolHandle,
 		parachain_empty_duration: Duration,
 		key: ed25519::Pair,
 	) -> Service
 		where
 			A: LocalPolkadotApi + Send + Sync + 'static,
 			C: BlockchainEvents<Block> + ChainHead<Block> + bft::BlockImport<Block> + bft::Authorities<Block> + Send + Sync + 'static,
+			N: Network + Collators + Send + 'static,
+			N::TableRouter: Send + 'static,
+			<N::Collation as IntoFuture>::Future: Send + 'static,
 	{
 		let (signal, exit) = ::exit_future::signal();
 		let thread = thread::spawn(move || {
-			let mut core = reactor::Core::new().expect("tokio::Core could not be created");
+			let mut runtime = LocalRuntime::new().expect("Could not create local runtime");
 			let key = Arc::new(key);
 
 			let factory = ProposerFactory {
 				client: api.clone(),
 				transaction_pool: transaction_pool.clone(),
-				network: Network(network.clone()),
-				collators: NoCollators,
+				collators: network.clone(),
+				network,
 				parachain_empty_duration,
-				handle: core.handle(),
+				handle: thread_pool,
 			};
 			let bft_service = Arc::new(BftService::new(client.clone(), key, factory));
 
 			let notifications = {
-				let handle = core.handle();
-				let network = network.clone();
 				let client = client.clone();
 				let bft_service = bft_service.clone();
 
 				client.import_notification_stream().for_each(move |notification| {
 					if notification.is_new_best {
-						start_bft(&notification.header, handle.clone(), &*client, network.clone(), &*bft_service);
+						start_bft(&notification.header, &*bft_service);
 					}
 					Ok(())
 				})
 			};
 
-			let interval = reactor::Interval::new_at(
+			let interval = Interval::new(
 				Instant::now() + Duration::from_millis(TIMER_DELAY_MS),
 				Duration::from_millis(TIMER_INTERVAL_MS),
-				&core.handle(),
-			).expect("it is always possible to create an interval with valid params");
+			);
+
 			let mut prev_best = match client.best_block_header() {
-				Ok(header) => header.blake2_256(),
+				Ok(header) => header.hash(),
 				Err(e) => {
 					warn!("Cant's start consensus service. Error reading best block header: {:?}", e);
 					return;
@@ -288,15 +134,13 @@ impl Service {
 			let timed = {
 				let c = client.clone();
 				let s = bft_service.clone();
-				let n = network.clone();
-				let handle = core.handle();
 
 				interval.map_err(|e| debug!("Timer error: {:?}", e)).for_each(move |_| {
 					if let Ok(best_block) = c.best_block_header() {
-						let hash = best_block.blake2_256();
+						let hash = best_block.hash();
 						if hash == prev_best {
 							debug!("Starting consensus round after a timeout");
-							start_bft(&best_block, handle.clone(), &*c, n.clone(), &*s);
+							start_bft(&best_block, &*s);
 						}
 						prev_best = hash;
 					}
@@ -304,9 +148,9 @@ impl Service {
 				})
 			};
 
-			core.handle().spawn(notifications);
-			core.handle().spawn(timed);
-			if let Err(e) = core.run(exit) {
+			runtime.spawn(notifications);
+			runtime.spawn(timed);
+			if let Err(e) = runtime.block_on(exit) {
 				debug!("BFT event loop error {:?}", e);
 			}
 		});
@@ -328,42 +172,3 @@ impl Drop for Service {
 		}
 	}
 }
-
-// Collators implementation which never collates anything.
-// TODO: do a real implementation.
-#[derive(Clone, Copy)]
-struct NoCollators;
-
-impl ::collation::Collators for NoCollators {
-	type Error = ();
-	type Collation = future::Empty<::collation::Collation, ()>;
-
-	fn collate(&self, _parachain: ParaId, _relay_parent: Hash) -> Self::Collation {
-		future::empty()
-	}
-
-	fn note_bad_collator(&self, _collator: AccountId) { }
-}
-
-#[derive(Clone)]
-struct Router {
-	network: Arc<net::ConsensusService<Block>>,
-}
-
-impl TableRouter for Router {
-	type Error = Canceled;
-	type FetchCandidate =  future::Empty<BlockData, Self::Error>;
-	type FetchExtrinsic = future::FutureResult<Extrinsic, Self::Error>;
-
-	fn local_candidate_data(&self, _hash: Hash, _block_data: BlockData, _extrinsic: Extrinsic) {
-		// TODO
-	}
-
-	fn fetch_block_data(&self, _candidate: &CandidateReceipt) -> Self::FetchCandidate {
-		future::empty()
-	}
-
-	fn fetch_extrinsic_data(&self, _candidate: &CandidateReceipt) -> Self::FetchExtrinsic {
-		future::ok(Extrinsic)
-	}
-}