New Transaction Queue implementation

Cargo.toml

@@ -131,5 +131,5 @@ members = [
 	"miner",
 	"transaction-pool",
 	"whisper",
-	"util/rlp_compress"
+	"util/rlp_compress",
 ]

ethcore/Cargo.toml

@@ -42,8 +42,7 @@ num = "0.1"
 num_cpus = "1.2"
 parity-machine = { path = "../machine" }
 parking_lot = "0.5"
-price-info = { path = "../price-info" }
-rayon = "0.8"
+rayon = "1.0"
 rand = "0.4"
 rlp = { path = "../util/rlp" }
 rlp_compress = { path = "../util/rlp_compress" }
@@ -62,7 +61,6 @@ stats = { path = "../util/stats" }
 time = "0.1"
 trace-time = { path = "../util/trace-time" }
 using_queue = { path = "../util/using_queue" }
-table = { path = "../util/table" }
 vm = { path = "vm" }
 wasm = { path = "wasm" }
 keccak-hash = { path = "../util/hash" }
@@ -75,11 +73,20 @@ tempdir = "0.3"
 trie-standardmap = { path = "../util/trie-standardmap" }
 
 [features]
+# Enable JIT EVM interpreter.
 jit = ["evm/jit"]
+# Display EVM debug traces.
 evm-debug = ["slow-blocks"]
+# Display EVM debug traces when running tests.
 evm-debug-tests = ["evm-debug", "evm/evm-debug-tests"]
-slow-blocks = [] # Use SLOW_TX_DURATION="50" (compile time!) to track transactions over 50ms
+# Measure time of transaction execution.
+# Whenever the transaction execution time (in millis) exceeds the value of
+# SLOW_TX_DURATION env variable (provided compile time!)
+# EVM debug traces are printed.
+slow-blocks = []
+# Run JSON consensus tests.
 json-tests = ["ethcore-transaction/json-tests"]
+# Run memory/cpu heavy tests.
 test-heavy = []
-default = []
+# Compile benches
 benches = []

ethcore/light/src/provider.rs

@@ -282,6 +282,9 @@ impl<T: ProvingBlockChainClient + ?Sized> Provider for T {
 
 	fn ready_transactions(&self) -> Vec<PendingTransaction> {
 		BlockChainClient::ready_transactions(self)
+			.into_iter()
+			.map(|tx| tx.pending().clone())
+			.collect()
 	}
 
 	fn epoch_signal(&self, req: request::CompleteSignalRequest) -> Option<request::SignalResponse> {

ethcore/light/src/transaction_queue.rs

@@ -120,6 +120,18 @@ impl AccountTransactions {
 	}
 }
 
+/// Transaction import result.
+pub enum ImportDestination {
+	/// Transaction has been imported to the current queue.
+	///
+	/// It's going to be propagated to peers.
+	Current,
+	/// Transaction has been imported to future queue.
+	///
+	/// It means it won't be propagated until the gap is filled.
+	Future,
+}
+
 type Listener = Box<Fn(&[H256]) + Send + Sync>;
 
 /// Light transaction queue. See module docs for more details.
@@ -142,7 +154,7 @@ impl fmt::Debug for TransactionQueue {
 
 impl TransactionQueue {
 	/// Import a pending transaction to be queued.
-	pub fn import(&mut self, tx: PendingTransaction) -> Result<transaction::ImportResult, transaction::Error>  {
+	pub fn import(&mut self, tx: PendingTransaction) -> Result<ImportDestination, transaction::Error>  {
 		let sender = tx.sender();
 		let hash = tx.hash();
 		let nonce = tx.nonce;
@@ -158,7 +170,7 @@ impl TransactionQueue {
 					future: BTreeMap::new(),
 				});
 
-				(transaction::ImportResult::Current, vec![hash])
+				(ImportDestination::Current, vec![hash])
 			}
 			Entry::Occupied(mut entry) => {
 				let acct_txs = entry.get_mut();
@@ -180,7 +192,7 @@ impl TransactionQueue {
 						let old = ::std::mem::replace(&mut acct_txs.current[idx], tx_info);
 						self.by_hash.remove(&old.hash);
 
-						(transaction::ImportResult::Current, vec![hash])
+						(ImportDestination::Current, vec![hash])
 					}
 					Err(idx) => {
 						let cur_len = acct_txs.current.len();
@@ -202,13 +214,13 @@ impl TransactionQueue {
 								acct_txs.future.insert(future_nonce, future);
 							}
 
-							(transaction::ImportResult::Current, vec![hash])
+							(ImportDestination::Current, vec![hash])
 						} else if idx == cur_len && acct_txs.current.last().map_or(false, |f| f.nonce + 1.into() != nonce) {
 							trace!(target: "txqueue", "Queued future transaction for {}, nonce={}", sender, nonce);
 							let future_nonce = nonce;
 							acct_txs.future.insert(future_nonce, tx_info);
 
-							(transaction::ImportResult::Future, vec![])
+							(ImportDestination::Future, vec![])
 						} else {
 							trace!(target: "txqueue", "Queued current transaction for {}, nonce={}", sender, nonce);
 
@@ -217,7 +229,7 @@ impl TransactionQueue {
 							let mut promoted = acct_txs.adjust_future();
 							promoted.insert(0, hash);
 
-							(transaction::ImportResult::Current, promoted)
+							(ImportDestination::Current, promoted)
 						}
 					}
 				}

ethcore/node_filter/src/lib.rs

@@ -139,7 +139,7 @@ mod test {
 			ClientConfig::default(),
 			&spec,
 			client_db,
-			Arc::new(Miner::with_spec(&spec)),
+			Arc::new(Miner::new_for_tests(&spec, None)),
 			IoChannel::disconnected(),
 		).unwrap();
 		let filter = NodeFilter::new(Arc::downgrade(&client) as Weak<BlockChainClient>, contract_addr);

ethcore/src/account_provider/mod.rs

@@ -66,8 +66,6 @@ pub enum SignError {
 	Hardware(HardwareError),
 	/// Low-level error from store
 	SStore(SSError),
-	/// Inappropriate chain
-	InappropriateChain,
 }
 
 impl fmt::Display for SignError {
@@ -77,7 +75,6 @@ impl fmt::Display for SignError {
 			SignError::NotFound => write!(f, "Account does not exist"),
 			SignError::Hardware(ref e) => write!(f, "{}", e),
 			SignError::SStore(ref e) => write!(f, "{}", e),
-			SignError::InappropriateChain => write!(f, "Inappropriate chain"),
 		}
 	}
 }

ethcore/src/block.rs

@@ -336,8 +336,33 @@ impl<'x> OpenBlock<'x> {
 	}
 
 	/// Push transactions onto the block.
-	pub fn push_transactions(&mut self, transactions: &[SignedTransaction]) -> Result<(), Error> {
-		push_transactions(self, transactions)
+	#[cfg(not(feature = "slow-blocks"))]
+	fn push_transactions(&mut self, transactions: Vec<SignedTransaction>) -> Result<(), Error> {
+		for t in transactions {
+			self.push_transaction(t, None)?;
+		}
+		Ok(())
+	}
+
+	/// Push transactions onto the block.
+	#[cfg(feature = "slow-blocks")]
+	fn push_transactions(&mut self, transactions: Vec<SignedTransaction>) -> Result<(), Error> {
+		use std::time;
+
+		let slow_tx = option_env!("SLOW_TX_DURATION").and_then(|v| v.parse().ok()).unwrap_or(100);
+		for t in transactions {
+			let hash = t.hash();
+			let start = time::Instant::now();
+			self.push_transaction(t, None)?;
+			let took = start.elapsed();
+			let took_ms = took.as_secs() * 1000 + took.subsec_nanos() as u64 / 1000000;
+			if took > time::Duration::from_millis(slow_tx) {
+				warn!("Heavy ({} ms) transaction in block {:?}: {:?}", took_ms, block.header().number(), hash);
+			}
+			debug!(target: "tx", "Transaction {:?} took: {} ms", hash, took_ms);
+		}
+
+		Ok(())
 	}
 
 	/// Populate self from a header.
@@ -534,10 +559,10 @@ impl IsBlock for SealedBlock {
 }
 
 /// Enact the block given by block header, transactions and uncles
-pub fn enact(
-	header: &Header,
-	transactions: &[SignedTransaction],
-	uncles: &[Header],
+fn enact(
+	header: Header,
+	transactions: Vec<SignedTransaction>,
+	uncles: Vec<Header>,
 	engine: &EthEngine,
 	tracing: bool,
 	db: StateDB,
@@ -567,48 +592,19 @@ pub fn enact(
 		is_epoch_begin,
 	)?;
 
-	b.populate_from(header);
+	b.populate_from(&header);
 	b.push_transactions(transactions)?;
 
 	for u in uncles {
-		b.push_uncle(u.clone())?;
+		b.push_uncle(u)?;
 	}
 
 	Ok(b.close_and_lock())
 }
 
-#[inline]
-#[cfg(not(feature = "slow-blocks"))]
-fn push_transactions(block: &mut OpenBlock, transactions: &[SignedTransaction]) -> Result<(), Error> {
-	for t in transactions {
-		block.push_transaction(t.clone(), None)?;
-	}
-	Ok(())
-}
-
-#[cfg(feature = "slow-blocks")]
-fn push_transactions(block: &mut OpenBlock, transactions: &[SignedTransaction]) -> Result<(), Error> {
-	use std::time;
-
-	let slow_tx = option_env!("SLOW_TX_DURATION").and_then(|v| v.parse().ok()).unwrap_or(100);
-	for t in transactions {
-		let hash = t.hash();
-		let start = time::Instant::now();
-		block.push_transaction(t.clone(), None)?;
-		let took = start.elapsed();
-		let took_ms = took.as_secs() * 1000 + took.subsec_nanos() as u64 / 1000000;
-		if took > time::Duration::from_millis(slow_tx) {
-			warn!("Heavy ({} ms) transaction in block {:?}: {:?}", took_ms, block.header().number(), hash);
-		}
-		debug!(target: "tx", "Transaction {:?} took: {} ms", hash, took_ms);
-	}
-	Ok(())
-}
-
-// TODO [ToDr] Pass `PreverifiedBlock` by move, this will avoid unecessary allocation
 /// Enact the block given by `block_bytes` using `engine` on the database `db` with given `parent` block header
 pub fn enact_verified(
-	block: &PreverifiedBlock,
+	block: PreverifiedBlock,
 	engine: &EthEngine,
 	tracing: bool,
 	db: StateDB,
@@ -620,9 +616,9 @@ pub fn enact_verified(
 	let view = BlockView::new(&block.bytes);
 
 	enact(
-		&block.header,
-		&block.transactions,
-		&view.uncles(),
+		block.header,
+		block.transactions,
+		view.uncles(),
 		engine,
 		tracing,
 		db,
@@ -690,7 +686,7 @@ mod tests {
 		)?;
 
 		b.populate_from(&header);
-		b.push_transactions(&transactions)?;
+		b.push_transactions(transactions)?;
 
 		for u in &block.uncles() {
 			b.push_uncle(u.clone())?;
@@ -783,3 +779,4 @@ mod tests {
 		assert!(orig_db.journal_db().keys().iter().filter(|k| orig_db.journal_db().get(k.0) != db.journal_db().get(k.0)).next() == None);
 	}
 }
+

ethcore/src/client/chain_notify.rs

@@ -49,7 +49,7 @@ pub trait ChainNotify : Send + Sync {
 
 	/// fires when new transactions are received from a peer
 	fn transactions_received(&self,
-		_hashes: Vec<H256>,
+		_hashes: &[H256],
 		_peer_id: usize,
 	) {
 		// does nothing by default