Tendermint Engine

ethcore/res/tendermint.json

@@ -0,0 +1,40 @@
+{
+	"name": "TestBFT",
+	"engine": {
+		"Tendermint": {
+			"params": {
+				"gasLimitBoundDivisor": "0x0400",
+				"authorities" : [
+					"0x82a978b3f5962a5b0957d9ee9eef472ee55b42f1",
+					"0x7d577a597b2742b498cb5cf0c26cdcd726d39e6e"
+				]
+			}
+		}
+	},
+	"params": {
+		"accountStartNonce": "0x0",
+		"maximumExtraDataSize": "0x20",
+		"minGasLimit": "0x1388",
+		"networkID" : "0x2323"
+	},
+	"genesis": {
+		"seal": {
+			"generic": {
+				"rlp": "f88980b8410000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000f843b8410000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000"
+			}
+		},
+		"difficulty": "0x20000",
+		"author": "0x0000000000000000000000000000000000000000",
+		"timestamp": "0x00",
+		"parentHash": "0x0000000000000000000000000000000000000000000000000000000000000000",
+		"extraData": "0x",
+		"gasLimit": "0x2fefd8"
+	},
+	"accounts": {
+		"0000000000000000000000000000000000000001": { "balance": "1", "builtin": { "name": "ecrecover", "pricing": { "linear": { "base": 3000, "word": 0 } } } },
+		"0000000000000000000000000000000000000002": { "balance": "1", "builtin": { "name": "sha256", "pricing": { "linear": { "base": 60, "word": 12 } } } },
+		"0000000000000000000000000000000000000003": { "balance": "1", "builtin": { "name": "ripemd160", "pricing": { "linear": { "base": 600, "word": 120 } } } },
+		"0000000000000000000000000000000000000004": { "balance": "1", "builtin": { "name": "identity", "pricing": { "linear": { "base": 15, "word": 3 } } } },
+		"9cce34f7ab185c7aba1b7c8140d620b4bda941d6": { "balance": "1606938044258990275541962092341162602522202993782792835301376" }
+	}
+}

ethcore/src/client/chain_notify.rs

@@ -15,7 +15,7 @@
 // along with Parity.  If not, see <http://www.gnu.org/licenses/>.
 
 use ipc::IpcConfig;
-use util::H256;
+use util::{H256, Bytes};
 
 /// Represents what has to be handled by actor listening to chain events
 #[ipc]
@@ -27,6 +27,8 @@ pub trait ChainNotify : Send + Sync {
 		_enacted: Vec<H256>,
 		_retracted: Vec<H256>,
 		_sealed: Vec<H256>,
+		// Block bytes.
+		_proposed: Vec<Bytes>,
 		_duration: u64) {
 		// does nothing by default
 	}
@@ -41,6 +43,9 @@ pub trait ChainNotify : Send + Sync {
 		// does nothing by default
 	}
 
+	/// fires when chain broadcasts a message
+	fn broadcast(&self, _data: Vec<u8>) {}
+
 	/// fires when new transactions are received from a peer
 	fn transactions_received(&self,
 		_hashes: Vec<H256>,

ethcore/src/client/client.rs

@@ -24,8 +24,8 @@ use time::precise_time_ns;
 // util
 use util::{Bytes, PerfTimer, Itertools, Mutex, RwLock, Hashable};
 use util::{journaldb, TrieFactory, Trie};
-use util::trie::TrieSpec;
 use util::{U256, H256, Address, H2048, Uint, FixedHash};
+use util::trie::TrieSpec;
 use util::kvdb::*;
 
 // other
@@ -396,9 +396,10 @@ impl Client {
 	/// This is triggered by a message coming from a block queue when the block is ready for insertion
 	pub fn import_verified_blocks(&self) -> usize {
 		let max_blocks_to_import = 4;
-		let (imported_blocks, import_results, invalid_blocks, imported, duration, is_empty) = {
+		let (imported_blocks, import_results, invalid_blocks, imported, proposed_blocks, duration, is_empty) = {
 			let mut imported_blocks = Vec::with_capacity(max_blocks_to_import);
 			let mut invalid_blocks = HashSet::new();
+			let mut proposed_blocks = Vec::with_capacity(max_blocks_to_import);
 			let mut import_results = Vec::with_capacity(max_blocks_to_import);
 
 			let _import_lock = self.import_lock.lock();
@@ -417,12 +418,17 @@ impl Client {
 					continue;
 				}
 				if let Ok(closed_block) = self.check_and_close_block(&block) {
-					imported_blocks.push(header.hash());
+					if self.engine.is_proposal(&block.header) {
+						self.block_queue.mark_as_good(&[header.hash()]);
+						proposed_blocks.push(block.bytes);
+					} else {
+						imported_blocks.push(header.hash());
 
-					let route = self.commit_block(closed_block, &header.hash(), &block.bytes);
-					import_results.push(route);
+						let route = self.commit_block(closed_block, &header.hash(), &block.bytes);
+						import_results.push(route);
 
-					self.report.write().accrue_block(&block);
+						self.report.write().accrue_block(&block);
+					}
 				} else {
 					invalid_blocks.insert(header.hash());
 				}
@@ -436,7 +442,7 @@ impl Client {
 			}
 			let is_empty = self.block_queue.mark_as_good(&imported_blocks);
 			let duration_ns = precise_time_ns() - start;
-			(imported_blocks, import_results, invalid_blocks, imported, duration_ns, is_empty)
+			(imported_blocks, import_results, invalid_blocks, imported, proposed_blocks, duration_ns, is_empty)
 		};
 
 		{
@@ -454,6 +460,7 @@ impl Client {
 						enacted.clone(),
 						retracted.clone(),
 						Vec::new(),
+						proposed_blocks.clone(),
 						duration,
 					);
 				});
@@ -577,9 +584,10 @@ impl Client {
 		self.miner.clone()
 	}
 
-	/// Used by PoA to try sealing on period change.
-	pub fn update_sealing(&self) {
-		self.miner.update_sealing(self)
+
+	/// Replace io channel. Useful for testing.
+	pub fn set_io_channel(&self, io_channel: IoChannel<ClientIoMessage>) {
+		*self.io_channel.lock() = io_channel;
 	}
 
 	/// Attempt to get a copy of a specific block's final state.
@@ -1290,6 +1298,18 @@ impl BlockChainClient for Client {
 		self.miner.pending_transactions(self.chain.read().best_block_number())
 	}
 
+	fn queue_consensus_message(&self, message: Bytes) {
+		let channel = self.io_channel.lock().clone();
+		if let Err(e) = channel.send(ClientIoMessage::NewMessage(message)) {
+			debug!("Ignoring the message, error queueing: {}", e);
+		}
+	}
+
+	fn broadcast_consensus_message(&self, message: Bytes) {
+		self.notify(|notify| notify.broadcast(message.clone()));
+	}
+
+
 	fn signing_network_id(&self) -> Option<u64> {
 		self.engine.signing_network_id(&self.latest_env_info())
 	}
@@ -1314,7 +1334,6 @@ impl BlockChainClient for Client {
 }
 
 impl MiningBlockChainClient for Client {
-
 	fn latest_schedule(&self) -> Schedule {
 		self.engine.schedule(&self.latest_env_info())
 	}
@@ -1357,6 +1376,30 @@ impl MiningBlockChainClient for Client {
 		&self.factories.vm
 	}
 
+	fn update_sealing(&self) {
+		self.miner.update_sealing(self)
+	}
+
+	fn submit_seal(&self, block_hash: H256, seal: Vec<Bytes>) {
+		if self.miner.submit_seal(self, block_hash, seal).is_err() {
+			warn!(target: "poa", "Wrong internal seal submission!")
+		}
+	}
+
+	fn broadcast_proposal_block(&self, block: SealedBlock) {
+		self.notify(|notify| {
+			notify.new_blocks(
+				vec![],
+				vec![],
+				vec![],
+				vec![],
+				vec![],
+				vec![block.rlp_bytes()],
+				0,
+			);
+		});
+	}
+
 	fn import_sealed_block(&self, block: SealedBlock) -> ImportResult {
 		let h = block.header().hash();
 		let start = precise_time_ns();
@@ -1381,6 +1424,7 @@ impl MiningBlockChainClient for Client {
 				enacted.clone(),
 				retracted.clone(),
 				vec![h.clone()],
+				vec![],
 				precise_time_ns() - start,
 			);
 		});
@@ -1416,6 +1460,12 @@ impl ::client::ProvingBlockChainClient for Client {
 	}
 }
 
+impl Drop for Client {
+	fn drop(&mut self) {
+		self.engine.stop();
+	}
+}
+
 #[cfg(test)]
 mod tests {
 

ethcore/src/client/test_client.rs

@@ -360,6 +360,18 @@ impl MiningBlockChainClient for TestBlockChainClient {
 	fn import_sealed_block(&self, _block: SealedBlock) -> ImportResult {
 		Ok(H256::default())
 	}
+
+	fn broadcast_proposal_block(&self, _block: SealedBlock) {}
+
+	fn update_sealing(&self) {
+		self.miner.update_sealing(self)
+	}
+
+	fn submit_seal(&self, block_hash: H256, seal: Vec<Bytes>) {
+		if self.miner.submit_seal(self, block_hash, seal).is_err() {
+			warn!(target: "poa", "Wrong internal seal submission!")
+		}
+	}
 }
 
 impl BlockChainClient for TestBlockChainClient {
@@ -663,6 +675,12 @@ impl BlockChainClient for TestBlockChainClient {
 		self.miner.import_external_transactions(self, txs);
 	}
 
+	fn queue_consensus_message(&self, message: Bytes) {
+		self.spec.engine.handle_message(&message).unwrap();
+	}
+
+	fn broadcast_consensus_message(&self, _message: Bytes) {}
+
 	fn pending_transactions(&self) -> Vec<SignedTransaction> {
 		self.miner.pending_transactions(self.chain_info().best_block_number)
 	}

ethcore/src/client/traits.rs

@@ -202,6 +202,12 @@ pub trait BlockChainClient : Sync + Send {
 	/// Queue transactions for importing.
 	fn queue_transactions(&self, transactions: Vec<Bytes>, peer_id: usize);
 
+	/// Queue conensus engine message.
+	fn queue_consensus_message(&self, message: Bytes);
+
+	/// Used by PoA to communicate with peers.
+	fn broadcast_consensus_message(&self, message: Bytes);
+
 	/// list all transactions
 	fn pending_transactions(&self) -> Vec<SignedTransaction>;
 
@@ -273,6 +279,15 @@ pub trait MiningBlockChainClient: BlockChainClient {
 	/// Returns EvmFactory.
 	fn vm_factory(&self) -> &EvmFactory;
 
+	/// Used by PoA to try sealing on period change.
+	fn update_sealing(&self);
+
+	/// Used by PoA to submit gathered signatures.
+	fn submit_seal(&self, block_hash: H256, seal: Vec<Bytes>);
+
+	/// Broadcast a block proposal.
+	fn broadcast_proposal_block(&self, block: SealedBlock);
+
 	/// Import sealed block. Skips all verifications.
 	fn import_sealed_block(&self, block: SealedBlock) -> ImportResult;
 

ethcore/src/engines/authority_round.rs

@@ -25,7 +25,7 @@ use rlp::{UntrustedRlp, Rlp, View, encode};
 use account_provider::AccountProvider;
 use block::*;
 use spec::CommonParams;
-use engines::Engine;
+use engines::{Engine, Seal, EngineError};
 use header::Header;
 use error::{Error, BlockError};
 use blockchain::extras::BlockDetails;
@@ -225,8 +225,8 @@ impl Engine for AuthorityRound {
 	///
 	/// This operation is synchronous and may (quite reasonably) not be available, in which `false` will
 	/// be returned.
-	fn generate_seal(&self, block: &ExecutedBlock) -> Option<Vec<Bytes>> {
-		if self.proposed.load(AtomicOrdering::SeqCst) { return None; }
+	fn generate_seal(&self, block: &ExecutedBlock) -> Seal {
+		if self.proposed.load(AtomicOrdering::SeqCst) { return Seal::None; }
 		let header = block.header();
 		let step = self.step();
 		if self.is_step_proposer(step, header.author()) {
@@ -235,7 +235,8 @@ impl Engine for AuthorityRound {
 				if let Ok(signature) = ap.sign(*header.author(), self.password.read().clone(), header.bare_hash()) {
 					trace!(target: "poa", "generate_seal: Issuing a block for step {}.", step);
 					self.proposed.store(true, AtomicOrdering::SeqCst);
-					return Some(vec![encode(&step).to_vec(), encode(&(&*signature as &[u8])).to_vec()]);
+					let rlps = vec![encode(&step).to_vec(), encode(&(&*signature as &[u8])).to_vec()];
+					return Seal::Regular(rlps);
 				} else {
 					warn!(target: "poa", "generate_seal: FAIL: Accounts secret key unavailable.");
 				}
@@ -245,7 +246,7 @@ impl Engine for AuthorityRound {
 		} else {
 			trace!(target: "poa", "generate_seal: Not a proposer for step {}.", step);
 		}
-		None
+		Seal::None
 	}
 
 	/// Check the number of seal fields.
@@ -288,7 +289,7 @@ impl Engine for AuthorityRound {
 		// Check if parent is from a previous step.
 		if step == try!(header_step(parent)) {
 			trace!(target: "poa", "Multiple blocks proposed for step {}.", step);
-			try!(Err(BlockError::DoubleVote(header.author().clone())));
+			try!(Err(EngineError::DoubleVote(header.author().clone())));
 		}
 
 		let gas_limit_divisor = self.our_params.gas_limit_bound_divisor;
@@ -347,6 +348,7 @@ mod tests {
 	use tests::helpers::*;
 	use account_provider::AccountProvider;
 	use spec::Spec;
+	use engines::Seal;
 
 	#[test]
 	fn has_valid_metadata() {
@@ -416,17 +418,17 @@ mod tests {
 		let b2 = b2.close_and_lock();
 
 		engine.set_signer(addr1, "1".into());
-		if let Some(seal) = engine.generate_seal(b1.block()) {
+		if let Seal::Regular(seal) = engine.generate_seal(b1.block()) {
 			assert!(b1.clone().try_seal(engine, seal).is_ok());
 			// Second proposal is forbidden.
-			assert!(engine.generate_seal(b1.block()).is_none());
+			assert!(engine.generate_seal(b1.block()) == Seal::None);
 		}
 
 		engine.set_signer(addr2, "2".into());
-		if let Some(seal) = engine.generate_seal(b2.block()) {
+		if let Seal::Regular(seal) = engine.generate_seal(b2.block()) {
 			assert!(b2.clone().try_seal(engine, seal).is_ok());
 			// Second proposal is forbidden.
-			assert!(engine.generate_seal(b2.block()).is_none());
+			assert!(engine.generate_seal(b2.block()) == Seal::None);
 		}
 	}