Merge pull request #1335 from ethcore/tx_perf

Transaction processing queue

ethcore/src/client/client.rs

@@ -18,6 +18,7 @@
 
 use std::marker::PhantomData;
 use std::path::PathBuf;
+use std::sync::atomic::{AtomicUsize, Ordering as AtomicOrdering};
 use util::*;
 use util::panics::*;
 use views::BlockView;
@@ -50,6 +51,8 @@ pub use types::block_status::BlockStatus;
 use evm::Factory as EvmFactory;
 use miner::{Miner, MinerService, TransactionImportResult, AccountDetails};
 
+const MAX_TX_QUEUE_SIZE: usize = 4096;
+
 impl fmt::Display for BlockChainInfo {
 	fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
 		write!(f, "#{}.{}", self.best_block_number, self.best_block_hash)
@@ -92,6 +95,8 @@ pub struct Client<V = CanonVerifier> where V: Verifier {
 	verifier: PhantomData<V>,
 	vm_factory: Arc<EvmFactory>,
 	miner: Arc<Miner>,
+	io_channel: IoChannel<NetSyncMessage>,
+	queue_transactions: AtomicUsize,
 }
 
 const HISTORY: u64 = 1200;
@@ -152,7 +157,7 @@ impl<V> Client<V> where V: Verifier {
 
 		let engine = Arc::new(spec.engine);
 
-		let block_queue = BlockQueue::new(config.queue, engine.clone(), message_channel);
+		let block_queue = BlockQueue::new(config.queue, engine.clone(), message_channel.clone());
 		let panic_handler = PanicHandler::new_in_arc();
 		panic_handler.forward_from(&block_queue);
 
@@ -168,6 +173,8 @@ impl<V> Client<V> where V: Verifier {
 			verifier: PhantomData,
 			vm_factory: Arc::new(EvmFactory::new(config.vm_type)),
 			miner: miner,
+			io_channel: message_channel,
+			queue_transactions: AtomicUsize::new(0),
 		};
 
 		Ok(Arc::new(client))
@@ -274,6 +281,7 @@ impl<V> Client<V> where V: Verifier {
 		let mut import_results = Vec::with_capacity(max_blocks_to_import);
 
 		let _import_lock = self.import_lock.lock();
+		let _timer = PerfTimer::new("import_verified_blocks");
 		let blocks = self.block_queue.drain(max_blocks_to_import);
 
 		let original_best = self.chain_info().best_block_hash;
@@ -364,6 +372,19 @@ impl<V> Client<V> where V: Verifier {
 		imported
 	}
 
+	/// Import transactions from the IO queue
+	pub fn import_queued_transactions(&self, transactions: &[Bytes]) -> usize {
+		let _timer = PerfTimer::new("import_queued_transactions");
+		self.queue_transactions.fetch_sub(transactions.len(), AtomicOrdering::SeqCst);
+		let fetch_account = |a: &Address| AccountDetails {
+			nonce: self.latest_nonce(a),
+			balance: self.latest_balance(a),
+		};
+		let tx = transactions.iter().filter_map(|bytes| UntrustedRlp::new(&bytes).as_val().ok()).collect();
+		let results = self.miner.import_transactions(tx, fetch_account);
+		results.len()
+	}
+
 	/// Attempt to get a copy of a specific block's state.
 	///
 	/// This will not fail if given BlockID::Latest.
@@ -753,6 +774,22 @@ impl<V> BlockChainClient for Client<V> where V: Verifier {
 		self.miner.import_transactions(transactions, fetch_account)
 	}
 
+	fn queue_transactions(&self, transactions: Vec<Bytes>) {
+		if self.queue_transactions.load(AtomicOrdering::Relaxed) > MAX_TX_QUEUE_SIZE {
+			debug!("Ignoring {} transactions: queue is full", transactions.len());
+		} else {
+			let len = transactions.len();
+			match self.io_channel.send(NetworkIoMessage::User(SyncMessage::NewTransactions(transactions))) {
+				Ok(_) => {
+					self.queue_transactions.fetch_add(len, AtomicOrdering::SeqCst);
+				}
+				Err(e) => {
+					debug!("Ignoring {} transactions: error queueing: {}", len, e);
+				}
+			}
+		}
+	}
+
 	fn all_transactions(&self) -> Vec<SignedTransaction> {
 		self.miner.all_transactions()
 	}

ethcore/src/client/mod.rs

@@ -193,6 +193,9 @@ pub trait BlockChainClient : Sync + Send {
 	/// import transactions from network/other 3rd party
 	fn import_transactions(&self, transactions: Vec<SignedTransaction>) -> Vec<Result<TransactionImportResult, EthError>>;
 
+	/// Queue transactions for importing.
+	fn queue_transactions(&self, transactions: Vec<Bytes>);
+
 	/// list all transactions
 	fn all_transactions(&self) -> Vec<SignedTransaction>;
 

ethcore/src/client/test_client.rs

@@ -493,6 +493,12 @@ impl BlockChainClient for TestBlockChainClient {
 		self.miner.import_transactions(transactions, &fetch_account)
 	}
 
+	fn queue_transactions(&self, transactions: Vec<Bytes>) {
+		// import right here
+		let tx = transactions.into_iter().filter_map(|bytes| UntrustedRlp::new(&bytes).as_val().ok()).collect();
+		self.import_transactions(tx);
+	}
+
 	fn all_transactions(&self) -> Vec<SignedTransaction> {
 		self.miner.all_transactions()
 	}

ethcore/src/service.rs

@@ -41,6 +41,8 @@ pub enum SyncMessage {
 	NewChainHead,
 	/// A block is ready
 	BlockVerified,
+	/// New transaction RLPs are ready to be imported
+	NewTransactions(Vec<Bytes>),
 	/// Start network command.
 	StartNetwork,
 	/// Stop network command.
@@ -136,6 +138,9 @@ impl IoHandler<NetSyncMessage> for ClientIoHandler {
 				SyncMessage::BlockVerified => {
 					self.client.import_verified_blocks(&io.channel());
 				},
+				SyncMessage::NewTransactions(ref transactions) => {
+					self.client.import_queued_transactions(&transactions);
+				},
 				_ => {}, // ignore other messages
 			}
 		}