Propagations & local transactions tracking

ethcore/Cargo.toml

@@ -27,6 +27,7 @@ time = "0.1"
 rand = "0.3"
 byteorder = "0.5"
 transient-hashmap = "0.1"
+linked-hash-map = "0.3.0"
 evmjit = { path = "../evmjit", optional = true }
 clippy = { version = "0.0.96", optional = true}
 ethash = { path = "../ethash" }

ethcore/src/client/client.rs

@@ -262,7 +262,7 @@ impl Client {
 		}
 	}
 
-	/// Register an action to be done if a mode change happens. 
+	/// Register an action to be done if a mode change happens.
 	pub fn on_mode_change<F>(&self, f: F) where F: 'static + FnMut(&Mode) + Send {
 		*self.on_mode_change.lock() = Some(Box::new(f));
 	}
@@ -891,12 +891,9 @@ impl BlockChainClient for Client {
 			let mut mode = self.mode.lock();
 			*mode = new_mode.clone().into();
 			trace!(target: "mode", "Mode now {:?}", &*mode);
-			match *self.on_mode_change.lock() {
-				Some(ref mut f) => {
-					trace!(target: "mode", "Making callback...");
-					f(&*mode)
-				},
-				_ => {} 
+			if let Some(ref mut f) = *self.on_mode_change.lock() {
+				trace!(target: "mode", "Making callback...");
+				f(&*mode)
 			}
 		}
 		match new_mode {

ethcore/src/lib.rs

@@ -102,6 +102,7 @@ extern crate rlp;
 extern crate ethcore_bloom_journal as bloom_journal;
 extern crate byteorder;
 extern crate transient_hashmap;
+extern crate linked_hash_map;
 
 #[macro_use]
 extern crate log;

ethcore/src/miner/local_transactions.rs

@@ -0,0 +1,196 @@
+// Copyright 2015, 2016 Ethcore (UK) Ltd.
+// This file is part of Parity.
+
+// Parity is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// Parity is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+
+// You should have received a copy of the GNU General Public License
+// along with Parity.  If not, see <http://www.gnu.org/licenses/>.
+
+//! Local Transactions List.
+
+use linked_hash_map::LinkedHashMap;
+use transaction::SignedTransaction;
+use error::TransactionError;
+use util::{U256, H256};
+
+/// Status of local transaction.
+/// Can indicate that the transaction is currently part of the queue (`Pending/Future`)
+/// or gives a reason why the transaction was removed.
+#[derive(Debug, PartialEq, Clone)]
+pub enum Status {
+	/// The transaction is currently in the transaction queue.
+	Pending,
+	/// The transaction is in future part of the queue.
+	Future,
+	/// Transaction is already mined.
+	Mined(SignedTransaction),
+	/// Transaction is dropped because of limit
+	Dropped(SignedTransaction),
+	/// Replaced because of higher gas price of another transaction.
+	Replaced(SignedTransaction, U256, H256),
+	/// Transaction was never accepted to the queue.
+	Rejected(SignedTransaction, TransactionError),
+	/// Transaction is invalid.
+	Invalid(SignedTransaction),
+}
+
+impl Status {
+	fn is_current(&self) -> bool {
+		*self == Status::Pending || *self == Status::Future
+	}
+}
+
+/// Keeps track of local transactions that are in the queue or were mined/dropped recently.
+#[derive(Debug)]
+pub struct LocalTransactionsList {
+	max_old: usize,
+	transactions: LinkedHashMap<H256, Status>,
+}
+
+impl Default for LocalTransactionsList {
+	fn default() -> Self {
+		Self::new(10)
+	}
+}
+
+impl LocalTransactionsList {
+	pub fn new(max_old: usize) -> Self {
+		LocalTransactionsList {
+			max_old: max_old,
+			transactions: Default::default(),
+		}
+	}
+
+	pub fn mark_pending(&mut self, hash: H256) {
+		self.clear_old();
+		self.transactions.insert(hash, Status::Pending);
+	}
+
+	pub fn mark_future(&mut self, hash: H256) {
+		self.transactions.insert(hash, Status::Future);
+		self.clear_old();
+	}
+
+	pub fn mark_rejected(&mut self, tx: SignedTransaction, err: TransactionError) {
+		self.transactions.insert(tx.hash(), Status::Rejected(tx, err));
+		self.clear_old();
+	}
+
+	pub fn mark_replaced(&mut self, tx: SignedTransaction, gas_price: U256, hash: H256) {
+		self.transactions.insert(tx.hash(), Status::Replaced(tx, gas_price, hash));
+		self.clear_old();
+	}
+
+	pub fn mark_invalid(&mut self, tx: SignedTransaction) {
+		self.transactions.insert(tx.hash(), Status::Invalid(tx));
+		self.clear_old();
+	}
+
+	pub fn mark_dropped(&mut self, tx: SignedTransaction) {
+		self.transactions.insert(tx.hash(), Status::Dropped(tx));
+		self.clear_old();
+	}
+
+	pub fn mark_mined(&mut self, tx: SignedTransaction) {
+		self.transactions.insert(tx.hash(), Status::Mined(tx));
+		self.clear_old();
+	}
+
+	pub fn contains(&self, hash: &H256) -> bool {
+		self.transactions.contains_key(hash)
+	}
+
+	pub fn all_transactions(&self) -> &LinkedHashMap<H256, Status> {
+		&self.transactions
+	}
+
+	fn clear_old(&mut self) {
+		let number_of_old = self.transactions
+			.values()
+			.filter(|status| !status.is_current())
+			.count();
+
+		if self.max_old >= number_of_old {
+			return;
+		}
+
+		let to_remove = self.transactions
+			.iter()
+			.filter(|&(_, status)| !status.is_current())
+			.map(|(hash, _)| *hash)
+			.take(number_of_old - self.max_old)
+			.collect::<Vec<_>>();
+
+		for hash in to_remove {
+			self.transactions.remove(&hash);
+		}
+	}
+}
+
+#[cfg(test)]
+mod tests {
+	use util::U256;
+	use ethkey::{Random, Generator};
+	use transaction::{Action, Transaction, SignedTransaction};
+	use super::{LocalTransactionsList, Status};
+
+	#[test]
+	fn should_add_transaction_as_pending() {
+		// given
+		let mut list = LocalTransactionsList::default();
+
+		// when
+		list.mark_pending(10.into());
+		list.mark_future(20.into());
+
+		// then
+		assert!(list.contains(&10.into()), "Should contain the transaction.");
+		assert!(list.contains(&20.into()), "Should contain the transaction.");
+		let statuses = list.all_transactions().values().cloned().collect::<Vec<Status>>();
+		assert_eq!(statuses, vec![Status::Pending, Status::Future]);
+	}
+
+	#[test]
+	fn should_clear_old_transactions() {
+		// given
+		let mut list = LocalTransactionsList::new(1);
+		let tx1 = new_tx(10.into());
+		let tx1_hash = tx1.hash();
+		let tx2 = new_tx(50.into());
+		let tx2_hash = tx2.hash();
+
+		list.mark_pending(10.into());
+		list.mark_invalid(tx1);
+		list.mark_dropped(tx2);
+		assert!(list.contains(&tx2_hash));
+		assert!(!list.contains(&tx1_hash));
+		assert!(list.contains(&10.into()));
+
+		// when
+		list.mark_future(15.into());
+
+		// then
+		assert!(list.contains(&10.into()));
+		assert!(list.contains(&15.into()));
+	}
+
+	fn new_tx(nonce: U256) -> SignedTransaction {
+		let keypair = Random.generate().unwrap();
+		Transaction {
+			action: Action::Create,
+			value: U256::from(100),
+			data: Default::default(),
+			gas: U256::from(10),
+			gas_price: U256::from(1245),
+			nonce: nonce
+		}.sign(keypair.secret(), None)
+	}
+}

ethcore/src/miner/miner.rs

@@ -24,6 +24,7 @@ use views::{BlockView, HeaderView};
 use header::Header;
 use state::{State, CleanupMode};
 use client::{MiningBlockChainClient, Executive, Executed, EnvInfo, TransactOptions, BlockID, CallAnalytics};
+use client::TransactionImportResult;
 use executive::contract_address;
 use block::{ClosedBlock, SealedBlock, IsBlock, Block};
 use error::*;
@@ -34,8 +35,8 @@ use engines::Engine;
 use miner::{MinerService, MinerStatus, TransactionQueue, PrioritizationStrategy, AccountDetails, TransactionOrigin};
 use miner::banning_queue::{BanningTransactionQueue, Threshold};
 use miner::work_notify::WorkPoster;
-use client::TransactionImportResult;
 use miner::price_info::PriceInfo;
+use miner::local_transactions::{Status as LocalTransactionStatus};
 use header::BlockNumber;
 
 /// Different possible definitions for pending transaction set.
@@ -563,14 +564,18 @@ impl Miner {
 		prepare_new
 	}
 
-	fn add_transactions_to_queue(&self, chain: &MiningBlockChainClient, transactions: Vec<SignedTransaction>, origin: TransactionOrigin, transaction_queue: &mut BanningTransactionQueue) ->
+	fn add_transactions_to_queue(&self, chain: &MiningBlockChainClient, transactions: Vec<SignedTransaction>, default_origin: TransactionOrigin, transaction_queue: &mut BanningTransactionQueue) ->
 		Vec<Result<TransactionImportResult, Error>> {
 
 		let fetch_account = |a: &Address| AccountDetails {
 			nonce: chain.latest_nonce(a),
 			balance: chain.latest_balance(a),
 		};
 
+		let accounts = self.accounts.as_ref()
+			.and_then(|provider| provider.accounts().ok())
+			.map(|accounts| accounts.into_iter().collect::<HashSet<_>>());
+
 		let schedule = chain.latest_schedule();
 		let gas_required = |tx: &SignedTransaction| tx.gas_required(&schedule).into();
 		let best_block_header: Header = ::rlp::decode(&chain.best_block_header());
@@ -583,12 +588,21 @@ impl Miner {
 					}
 				}
 			)
-			.map(|tx| match origin {
-				TransactionOrigin::Local | TransactionOrigin::RetractedBlock => {
-					transaction_queue.add(tx, origin, &fetch_account, &gas_required)
-				},
-				TransactionOrigin::External => {
-					transaction_queue.add_with_banlist(tx, &fetch_account, &gas_required)
+			.map(|tx| {
+				let origin = accounts.as_ref().and_then(|accounts| {
+					tx.sender().ok().and_then(|sender| match accounts.contains(&sender) {
+						true => Some(TransactionOrigin::Local),
+						false => None,
+					})
+				}).unwrap_or(default_origin);
+
+				match origin {
+					TransactionOrigin::Local | TransactionOrigin::RetractedBlock => {
+						transaction_queue.add(tx, origin, &fetch_account, &gas_required)
+					},
+					TransactionOrigin::External => {
+						transaction_queue.add_with_banlist(tx, &fetch_account, &gas_required)
+					}
 				}
 			})
 			.collect()
@@ -863,6 +877,14 @@ impl MinerService for Miner {
 		queue.top_transactions()
 	}
 
+	fn local_transactions(&self) -> BTreeMap<H256, LocalTransactionStatus> {
+		let queue = self.transaction_queue.lock();
+		queue.local_transactions()
+			.iter()
+			.map(|(hash, status)| (*hash, status.clone()))
+			.collect()
+	}
+
 	fn pending_transactions(&self, best_block: BlockNumber) -> Vec<SignedTransaction> {
 		let queue = self.transaction_queue.lock();
 		match self.options.pending_set {

ethcore/src/miner/mod.rs

@@ -41,16 +41,18 @@
 //! }
 //! ```
 
-mod miner;
+mod banning_queue;
 mod external;
+mod local_transactions;
+mod miner;
+mod price_info;
 mod transaction_queue;
-mod banning_queue;
 mod work_notify;
-mod price_info;
 
-pub use self::transaction_queue::{TransactionQueue, PrioritizationStrategy, AccountDetails, TransactionOrigin};
-pub use self::miner::{Miner, MinerOptions, Banning, PendingSet, GasPricer, GasPriceCalibratorOptions, GasLimit};
 pub use self::external::{ExternalMiner, ExternalMinerService};
+pub use self::miner::{Miner, MinerOptions, Banning, PendingSet, GasPricer, GasPriceCalibratorOptions, GasLimit};
+pub use self::transaction_queue::{TransactionQueue, PrioritizationStrategy, AccountDetails, TransactionOrigin};
+pub use self::local_transactions::{Status as LocalTransactionStatus};
 pub use client::TransactionImportResult;
 
 use std::collections::BTreeMap;
@@ -145,6 +147,9 @@ pub trait MinerService : Send + Sync {
 	/// Get a list of all pending transactions.
 	fn pending_transactions(&self, best_block: BlockNumber) -> Vec<SignedTransaction>;
 
+	/// Get a list of local transactions with statuses.
+	fn local_transactions(&self) -> BTreeMap<H256, LocalTransactionStatus>;
+
 	/// Get a list of all pending receipts.
 	fn pending_receipts(&self, best_block: BlockNumber) -> BTreeMap<H256, Receipt>;