Reserve nonces for signing (#6834)

* Nonce future - reserve and dispatch

* Single thread nonce tests

* Track status of reserved nonces.

* Initialization of nonce reservations.

* Prospective Signer

* Fix cli tests.

ethcore/src/account_provider/mod.rs

@@ -516,8 +516,8 @@ impl AccountProvider {
 	}
 
 	/// Returns each hardware account along with name and meta.
-	pub fn is_hardware_address(&self, address: Address) -> bool {
+	pub fn is_hardware_address(&self, address: &Address) -> bool {
-		self.hardware_store.as_ref().and_then(|s| s.wallet_info(&address)).is_some()
+		self.hardware_store.as_ref().and_then(|s| s.wallet_info(address)).is_some()
 	}
 
 	/// Returns each account along with name and meta.
@@ -589,7 +589,7 @@ impl AccountProvider {
 			}
 		}
 
-		if self.unlock_keep_secret && unlock != Unlock::OneTime {
+		if self.unlock_keep_secret && unlock == Unlock::Perm {
 			// verify password and get the secret
 			let secret = self.sstore.raw_secret(&account, &password)?;
 			self.unlocked_secrets.write().insert(account.clone(), secret);
@@ -639,14 +639,22 @@ impl AccountProvider {
 	}
 
 	/// Checks if given account is unlocked
-	pub fn is_unlocked(&self, address: Address) -> bool {
+	pub fn is_unlocked(&self, address: &Address) -> bool {
 		let unlocked = self.unlocked.read();
 		let unlocked_secrets = self.unlocked_secrets.read();
-		self.sstore.account_ref(&address)
+		self.sstore.account_ref(address)
 			.map(|r| unlocked.get(&r).is_some() || unlocked_secrets.get(&r).is_some())
 			.unwrap_or(false)
 	}
 
+	/// Checks if given account is unlocked permanently
+	pub fn is_unlocked_permanently(&self, address: &Address) -> bool {
+		let unlocked = self.unlocked.read();
+		self.sstore.account_ref(address)
+			.map(|r| unlocked.get(&r).map_or(false, |account| account.unlock == Unlock::Perm))
+			.unwrap_or(false)
+	}
+
 	/// Signs the message. If password is not provided the account must be unlocked.
 	pub fn sign(&self, address: Address, password: Option<String>, message: Message) -> Result<Signature, SignError> {
 		let account = self.sstore.account_ref(&address)?;

parity/cli/mod.rs

@@ -470,7 +470,7 @@ usage! {
 			"--jsonrpc-hosts=[HOSTS]",
 			"List of allowed Host header values. This option will validate the Host header sent by the browser, it is additional security against some attack vectors. Special options: \"all\", \"none\",.",
 
-			ARG arg_jsonrpc_threads: (usize) = 0usize, or |c: &Config| otry!(c.rpc).processing_threads,
+			ARG arg_jsonrpc_threads: (usize) = 4usize, or |c: &Config| otry!(c.rpc).processing_threads,
 			"--jsonrpc-threads=[THREADS]",
 			"Turn on additional processing threads in all RPC servers. Setting this to non-zero value allows parallel cpu-heavy queries execution.",
 
@@ -1449,7 +1449,7 @@ mod tests {
 			arg_jsonrpc_apis: "web3,eth,net,parity,traces,rpc,secretstore".into(),
 			arg_jsonrpc_hosts: "none".into(),
 			arg_jsonrpc_server_threads: None,
-			arg_jsonrpc_threads: 0,
+			arg_jsonrpc_threads: 4,
 
 			// WS
 			flag_no_ws: false,

parity/rpc.rs

@@ -62,7 +62,7 @@ impl Default for HttpConfiguration {
 			cors: None,
 			hosts: Some(Vec::new()),
 			server_threads: 1,
-			processing_threads: 0,
+			processing_threads: 4,
 		}
 	}
 }

parity/rpc_apis.rs

@@ -239,10 +239,10 @@ impl FullDependencies {
 		use parity_rpc::v1::*;
 
 		macro_rules! add_signing_methods {
-			($namespace:ident, $handler:expr, $deps:expr) => {
+			($namespace:ident, $handler:expr, $deps:expr, $nonces:expr) => {
 				{
 					let deps = &$deps;
-					let dispatcher = FullDispatcher::new(deps.client.clone(), deps.miner.clone());
+					let dispatcher = FullDispatcher::new(deps.client.clone(), deps.miner.clone(), $nonces);
 					if deps.signer_service.is_enabled() {
 						$handler.extend_with($namespace::to_delegate(SigningQueueClient::new(&deps.signer_service, dispatcher, deps.remote.clone(), &deps.secret_store)))
 					} else {
@@ -252,7 +252,12 @@ impl FullDependencies {
 			}
 		}
 
-		let dispatcher = FullDispatcher::new(self.client.clone(), self.miner.clone());
+		let nonces = Arc::new(Mutex::new(dispatch::Reservations::with_pool(self.fetch.pool())));
+		let dispatcher = FullDispatcher::new(
+			self.client.clone(),
+			self.miner.clone(),
+			nonces.clone(),
+		);
 		for api in apis {
 			match *api {
 				Api::Web3 => {
@@ -281,7 +286,7 @@ impl FullDependencies {
 						let filter_client = EthFilterClient::new(self.client.clone(), self.miner.clone());
 						handler.extend_with(filter_client.to_delegate());
 
-						add_signing_methods!(EthSigning, handler, self);
+						add_signing_methods!(EthSigning, handler, self, nonces.clone());
 					}
 				},
 				Api::EthPubSub => {
@@ -318,7 +323,7 @@ impl FullDependencies {
 					).to_delegate());
 
 					if !for_generic_pubsub {
-						add_signing_methods!(ParitySigning, handler, self);
+						add_signing_methods!(ParitySigning, handler, self, nonces.clone());
 					}
 				},
 				Api::ParityPubSub => {
@@ -435,6 +440,7 @@ impl<C: LightChainClient + 'static> LightDependencies<C> {
 			self.on_demand.clone(),
 			self.cache.clone(),
 			self.transaction_queue.clone(),
+			Arc::new(Mutex::new(dispatch::Reservations::with_pool(self.fetch.pool()))),
 		);
 
 		macro_rules! add_signing_methods {

rpc/src/v1/helpers/dispatch.rs

@@ -43,9 +43,9 @@ use ethcore::account_provider::AccountProvider;
 use crypto::DEFAULT_MAC;
 
 use jsonrpc_core::{BoxFuture, Error};
-use jsonrpc_core::futures::{future, Future};
+use jsonrpc_core::futures::{future, Future, Poll, Async};
 use jsonrpc_core::futures::future::Either;
-use v1::helpers::{errors, TransactionRequest, FilledTransactionRequest, ConfirmationPayload};
+use v1::helpers::{errors, nonce, TransactionRequest, FilledTransactionRequest, ConfirmationPayload};
 use v1::types::{
 	H256 as RpcH256, H520 as RpcH520, Bytes as RpcBytes,
 	RichRawTransaction as RpcRichRawTransaction,
@@ -55,6 +55,8 @@ use v1::types::{
 	DecryptRequest as RpcDecryptRequest,
 };
 
+pub use self::nonce::Reservations;
+
 /// Has the capability to dispatch, sign, and decrypt.
 ///
 /// Requires a clone implementation, with the implication that it be cheap;
@@ -75,7 +77,8 @@ pub trait Dispatcher: Send + Sync + Clone {
 	fn enrich(&self, SignedTransaction) -> RpcRichRawTransaction;
 
 	/// "Dispatch" a local transaction.
-	fn dispatch_transaction(&self, signed_transaction: PendingTransaction) -> Result<H256, Error>;
+	fn dispatch_transaction(&self, signed_transaction: PendingTransaction)
+		-> Result<H256, Error>;
 }
 
 /// A dispatcher which uses references to a client and miner in order to sign
@@ -84,14 +87,16 @@ pub trait Dispatcher: Send + Sync + Clone {
 pub struct FullDispatcher<C, M> {
 	client: Arc<C>,
 	miner: Arc<M>,
+	nonces: Arc<Mutex<nonce::Reservations>>,
 }
 
 impl<C, M> FullDispatcher<C, M> {
 	/// Create a `FullDispatcher` from Arc references to a client and miner.
-	pub fn new(client: Arc<C>, miner: Arc<M>) -> Self {
+	pub fn new(client: Arc<C>, miner: Arc<M>, nonces: Arc<Mutex<nonce::Reservations>>) -> Self {
 		FullDispatcher {
 			client,
 			miner,
+			nonces,
 		}
 	}
 }
@@ -101,15 +106,24 @@ impl<C, M> Clone for FullDispatcher<C, M> {
 		FullDispatcher {
 			client: self.client.clone(),
 			miner: self.miner.clone(),
+			nonces: self.nonces.clone(),
 		}
 	}
 }
 
 impl<C: MiningBlockChainClient, M: MinerService> FullDispatcher<C, M> {
-	fn fill_nonce(nonce: Option<U256>, from: &Address, miner: &M, client: &C) -> U256 {
+	fn state_nonce(&self, from: &Address) -> U256 {
-		nonce
+		self.miner.last_nonce(from).map(|nonce| nonce + U256::one())
-			.or_else(|| miner.last_nonce(from).map(|nonce| nonce + U256::one()))
+			.unwrap_or_else(|| self.client.latest_nonce(from))
-			.unwrap_or_else(|| client.latest_nonce(from))
+	}
+
+	/// Imports transaction to the miner's queue.
+	pub fn dispatch_transaction(client: &C, miner: &M, signed_transaction: PendingTransaction) -> Result<H256, Error> {
+		let hash = signed_transaction.transaction.hash();
+
+		miner.import_own_transaction(client, signed_transaction)
+			.map_err(errors::transaction)
+			.map(|_| hash)
 	}
 }
 
@@ -117,20 +131,21 @@ impl<C: MiningBlockChainClient, M: MinerService> Dispatcher for FullDispatcher<C
 	fn fill_optional_fields(&self, request: TransactionRequest, default_sender: Address, force_nonce: bool)
 		-> BoxFuture<FilledTransactionRequest, Error>
 	{
-		let (client, miner) = (self.client.clone(), self.miner.clone());
 		let request = request;
 		let from = request.from.unwrap_or(default_sender);
-		let nonce = match force_nonce {
+		let nonce = if force_nonce {
-			false => request.nonce,
+			request.nonce.or_else(|| Some(self.state_nonce(&from)))
-			true => Some(Self::fill_nonce(request.nonce, &from, &miner, &client)),
+		} else {
+			request.nonce
 		};
+
 		Box::new(future::ok(FilledTransactionRequest {
-			from: from,
+			from,
 			used_default_from: request.from.is_none(),
 			to: request.to,
-			nonce: nonce,
+			nonce,
-			gas_price: request.gas_price.unwrap_or_else(|| default_gas_price(&*client, &*miner)),
+			gas_price: request.gas_price.unwrap_or_else(|| default_gas_price(&*self.client, &*self.miner)),
-			gas: request.gas.unwrap_or_else(|| miner.sensible_gas_limit()),
+			gas: request.gas.unwrap_or_else(|| self.miner.sensible_gas_limit()),
 			value: request.value.unwrap_or_else(|| 0.into()),
 			data: request.data.unwrap_or_else(Vec::new),
 			condition: request.condition,
@@ -140,30 +155,15 @@ impl<C: MiningBlockChainClient, M: MinerService> Dispatcher for FullDispatcher<C
 	fn sign(&self, accounts: Arc<AccountProvider>, filled: FilledTransactionRequest, password: SignWith)
 		-> BoxFuture<WithToken<SignedTransaction>, Error>
 	{
-		let (client, miner) = (self.client.clone(), self.miner.clone());
+		let chain_id = self.client.signing_chain_id();
-		let chain_id = client.signing_chain_id();
-		let address = filled.from;
-		Box::new(future::done({
-			let t = Transaction {
-				nonce: Self::fill_nonce(filled.nonce, &filled.from, &miner, &client),
-				action: filled.to.map_or(Action::Create, Action::Call),
-				gas: filled.gas,
-				gas_price: filled.gas_price,
-				value: filled.value,
-				data: filled.data,
-			};
 
-			if accounts.is_hardware_address(address) {
+		if let Some(nonce) = filled.nonce {
-				hardware_signature(&*accounts, address, t, chain_id).map(WithToken::No)
+			return Box::new(future::done(sign_transaction(&*accounts, filled, chain_id, nonce, password)));
-			} else {
-				let hash = t.hash(chain_id);
-				let signature = try_bf!(signature(&*accounts, address, hash, password));
-				Ok(signature.map(|sig| {
-					SignedTransaction::new(t.with_signature(sig, chain_id))
-						.expect("Transaction was signed by AccountsProvider; it never produces invalid signatures; qed")
-				}))
 		}
-		}))
+
+		let state = self.state_nonce(&filled.from);
+		let reserved = self.nonces.lock().reserve_nonce(state);
+		Box::new(ProspectiveSigner::new(accounts, filled, chain_id, reserved, password))
 	}
 
 	fn enrich(&self, signed_transaction: SignedTransaction) -> RpcRichRawTransaction {
@@ -172,11 +172,7 @@ impl<C: MiningBlockChainClient, M: MinerService> Dispatcher for FullDispatcher<C
 	}
 
 	fn dispatch_transaction(&self, signed_transaction: PendingTransaction) -> Result<H256, Error> {
-		let hash = signed_transaction.transaction.hash();
+		Self::dispatch_transaction(&*self.client, &*self.miner, signed_transaction)
-
-		self.miner.import_own_transaction(&*self.client, signed_transaction)
-			.map_err(errors::transaction)
-			.map(|_| hash)
 	}
 }
 
@@ -255,6 +251,8 @@ pub struct LightDispatcher {
 	pub cache: Arc<Mutex<LightDataCache>>,
 	/// Transaction queue.
 	pub transaction_queue: Arc<RwLock<LightTransactionQueue>>,
+	/// Nonce reservations
+	pub nonces: Arc<Mutex<nonce::Reservations>>,
 }
 
 impl LightDispatcher {
@@ -267,6 +265,7 @@ impl LightDispatcher {
 		on_demand: Arc<OnDemand>,
 		cache: Arc<Mutex<LightDataCache>>,
 		transaction_queue: Arc<RwLock<LightTransactionQueue>>,
+		nonces: Arc<Mutex<nonce::Reservations>>,
 	) -> Self {
 		LightDispatcher {
 			sync,
@@ -274,6 +273,7 @@ impl LightDispatcher {
 			on_demand,
 			cache,
 			transaction_queue,
+			nonces,
 		}
 	}
 
@@ -372,39 +372,19 @@ impl Dispatcher for LightDispatcher {
 		-> BoxFuture<WithToken<SignedTransaction>, Error>
 	{
 		let chain_id = self.client.signing_chain_id();
-		let address = filled.from;
-
-		let with_nonce = move |filled: FilledTransactionRequest, nonce| {
-			let t = Transaction {
-				nonce: nonce,
-				action: filled.to.map_or(Action::Create, Action::Call),
-				gas: filled.gas,
-				gas_price: filled.gas_price,
-				value: filled.value,
-				data: filled.data,
-			};
-
-			if accounts.is_hardware_address(address) {
-				return hardware_signature(&*accounts, address, t, chain_id).map(WithToken::No)
-			}
-
-			let hash = t.hash(chain_id);
-			let signature = signature(&*accounts, address, hash, password)?;
-
-			Ok(signature.map(|sig| {
-				SignedTransaction::new(t.with_signature(sig, chain_id))
-					.expect("Transaction was signed by AccountsProvider; it never produces invalid signatures; qed")
-			}))
-		};
 
 		// fast path for pre-filled nonce.
 		if let Some(nonce) = filled.nonce {
-			return Box::new(future::done(with_nonce(filled, nonce)))
+			return Box::new(future::done(sign_transaction(&*accounts, filled, chain_id, nonce, password)))
 		}
 
-		Box::new(self.next_nonce(address)
+		let nonces = self.nonces.clone();
+		Box::new(self.next_nonce(filled.from)
 			.map_err(|_| errors::no_light_peers())
-			.and_then(move |nonce| with_nonce(filled, nonce)))
+			.and_then(move |nonce| {
+				let reserved = nonces.lock().reserve_nonce(nonce);
+				ProspectiveSigner::new(accounts, filled, chain_id, reserved, password)
+			}))
 	}
 
 	fn enrich(&self, signed_transaction: SignedTransaction) -> RpcRichRawTransaction {
@@ -422,6 +402,147 @@ impl Dispatcher for LightDispatcher {
 	}
 }
 
+fn sign_transaction(
+	accounts: &AccountProvider,
+	filled: FilledTransactionRequest,
+	chain_id: Option<u64>,
+	nonce: U256,
+	password: SignWith,
+) -> Result<WithToken<SignedTransaction>, Error> {
+	let t = Transaction {
+		nonce: nonce,
+		action: filled.to.map_or(Action::Create, Action::Call),
+		gas: filled.gas,
+		gas_price: filled.gas_price,
+		value: filled.value,
+		data: filled.data,
+	};
+
+	if accounts.is_hardware_address(&filled.from) {
+		return hardware_signature(accounts, filled.from, t, chain_id).map(WithToken::No)
+	}
+
+	let hash = t.hash(chain_id);
+	let signature = signature(accounts, filled.from, hash, password)?;
+
+	Ok(signature.map(|sig| {
+		SignedTransaction::new(t.with_signature(sig, chain_id))
+			.expect("Transaction was signed by AccountsProvider; it never produces invalid signatures; qed")
+	}))
+}
+
+#[derive(Debug, Clone, Copy)]
+enum ProspectiveSignerState {
+	TryProspectiveSign,
+	WaitForNonce,
+	Finish,
+}
+
+struct ProspectiveSigner {
+	accounts: Arc<AccountProvider>,
+	filled: FilledTransactionRequest,
+	chain_id: Option<u64>,
+	reserved: nonce::Reserved,
+	password: SignWith,
+	state: ProspectiveSignerState,
+	prospective: Option<Result<WithToken<SignedTransaction>, Error>>,
+	ready: Option<nonce::Ready>,
+}
+
+impl ProspectiveSigner {
+	pub fn new(
+		accounts: Arc<AccountProvider>,
+		filled: FilledTransactionRequest,
+		chain_id: Option<u64>,
+		reserved: nonce::Reserved,
+		password: SignWith,
+	) -> Self {
+		// If the account is permanently unlocked we can try to sign
+		// using prospective nonce. This should speed up sending
+		// multiple subsequent transactions in multi-threaded RPC environment.
+		let is_unlocked_permanently = accounts.is_unlocked_permanently(&filled.from);
+		let has_password = password.is_password();
+
+		ProspectiveSigner {
+			accounts,
+			filled,
+			chain_id,
+			reserved,
+			password,
+			state: if is_unlocked_permanently || has_password {
+				ProspectiveSignerState::TryProspectiveSign
+			} else {
+				ProspectiveSignerState::WaitForNonce
+			},
+			prospective: None,
+			ready: None,
+		}
+	}
+
+	fn sign(&self, nonce: &U256) -> Result<WithToken<SignedTransaction>, Error> {
+		sign_transaction(
+			&*self.accounts,
+			self.filled.clone(),
+			self.chain_id,
+			*nonce,
+			self.password.clone()
+		)
+	}
+
+	fn poll_reserved(&mut self) -> Poll<nonce::Ready, Error> {
+		self.reserved.poll().map_err(|_| errors::internal("Nonce reservation failure", ""))
+	}
+}
+
+impl Future for ProspectiveSigner {
+	type Item = WithToken<SignedTransaction>;
+	type Error = Error;
+
+	fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+		use self::ProspectiveSignerState::*;
+
+		loop {
+			match self.state {
+				TryProspectiveSign => {
+					// Try to poll reserved, it might be ready.
+					match self.poll_reserved()? {
+						Async::NotReady => {
+							self.state = WaitForNonce;
+							self.prospective = Some(self.sign(self.reserved.prospective_value()));
+						},
+						Async::Ready(nonce) => {
+							self.state = Finish;
+							self.prospective = Some(self.sign(nonce.value()));
+							self.ready = Some(nonce);
+						},
+					}
+				},
+				WaitForNonce => {
+					let nonce = try_ready!(self.poll_reserved());
+					let result = match (self.prospective.take(), nonce.matches_prospective()) {
+						(Some(prospective), true) => prospective,
+						_ => self.sign(nonce.value()),
+					};
+					self.state = Finish;
+					self.prospective = Some(result);
+					self.ready = Some(nonce);
+				},
+				Finish => {
+					if let (Some(result), Some(nonce)) = (self.prospective.take(), self.ready.take()) {
+						// Mark nonce as used on successful signing
+						return result.map(move |tx| {
+							nonce.mark_used();
+							Async::Ready(tx)
+						})
+					} else {
+						panic!("Poll after ready.");
+					}
+				}
+			}
+		}
+	}
+}
+
 /// Single-use account token.
 pub type AccountToken = String;
 
@@ -436,6 +557,16 @@ pub enum SignWith {
 	Token(AccountToken),
 }
 
+impl SignWith {
+	fn is_password(&self) -> bool {
+		if let SignWith::Password(_) = *self {
+			true
+		} else {
+			false
+		}
+	}
+}
+
 /// A value, potentially accompanied by a signing token.
 #[derive(Debug)]
 pub enum WithToken<T: Debug> {
@@ -529,7 +660,7 @@ pub fn execute<D: Dispatcher + 'static>(
 				))
 		},
 		ConfirmationPayload::EthSignMessage(address, data) => {
-			if accounts.is_hardware_address(address) {
+			if accounts.is_hardware_address(&address) {
 				return Box::new(future::err(errors::unsupported("Signing via hardware wallets is not supported.", None)));
 			}
 
@@ -543,7 +674,7 @@ pub fn execute<D: Dispatcher + 'static>(
 			Box::new(future::done(res))
 		},
 		ConfirmationPayload::Decrypt(address, data) => {
-			if accounts.is_hardware_address(address) {
+			if accounts.is_hardware_address(&address) {
 				return Box::new(future::err(errors::unsupported("Decrypting via hardware wallets is not supported.", None)));
 			}
 
@@ -572,7 +703,7 @@ fn signature(accounts: &AccountProvider, address: Address, hash: H256, password:
 fn hardware_signature(accounts: &AccountProvider, address: Address, t: Transaction, chain_id: Option<u64>)
 	-> Result<SignedTransaction, Error>
 {
-	debug_assert!(accounts.is_hardware_address(address));
+	debug_assert!(accounts.is_hardware_address(&address));
 
 	let mut stream = rlp::RlpStream::new();
 	t.rlp_append_unsigned_transaction(&mut stream, chain_id);

rpc/src/v1/helpers/mod.rs

@@ -22,14 +22,15 @@ pub mod block_import;
 pub mod dapps;
 pub mod dispatch;
 pub mod fake_sign;
-pub mod light_fetch;
-pub mod oneshot;
 pub mod ipfs;
+pub mod light_fetch;
+pub mod nonce;
+pub mod oneshot;
 pub mod secretstore;
 
 mod network_settings;
-mod poll_manager;
 mod poll_filter;
+mod poll_manager;
 mod requests;
 mod signer;
 mod signing_queue;

rpc/src/v1/helpers/nonce.rs

@@ -0,0 +1,249 @@
+// Copyright 2015-2017 harity Technologies (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/>.
+
+use std::{cmp, mem};
+use std::sync::{atomic, Arc};
+use std::sync::atomic::AtomicUsize;
+
+use bigint::prelude::U256;
+use futures::{Future, future, Poll, Async};
+use futures::future::Either;
+use futures::sync::oneshot;
+use futures_cpupool::CpuPool;
+
+/// Manages currently reserved and prospective nonces.
+#[derive(Debug)]
+pub struct Reservations {
+	previous: Option<oneshot::Receiver<U256>>,
+	pool: CpuPool,
+	prospective_value: U256,
+	dropped: Arc<AtomicUsize>,
+}
+
+impl Reservations {
+	/// Create new nonces manager and spawn a single-threaded cpu pool
+	/// for progressing execution of dropped nonces.
+	pub fn new() -> Self {
+		Self::with_pool(CpuPool::new(1))
+	}
+
+	/// Create new nonces manager with given cpu pool.
+	pub fn with_pool(pool: CpuPool) -> Self {
+		Reservations {
+			previous: None,
+			pool,
+			prospective_value: Default::default(),
+			dropped: Default::default(),
+		}
+	}
+
+	/// Reserves a prospective nonce.
+	/// The caller should provide a minimal nonce that needs to be reserved (taken from state/txqueue).
+	/// If there were any previous reserved nonces the returned future will be resolved when those are finished
+	/// (confirmed that the nonce were indeed used).
+	/// The caller can use `prospective_nonce` and perform some heavy computation anticipating
+	/// that the `prospective_nonce` will be equal to the one he will get.
+	pub fn reserve_nonce(&mut self, minimal: U256) -> Reserved {
+		// Update prospective value
+		let dropped = self.dropped.swap(0, atomic::Ordering::SeqCst);
+		let prospective_value = cmp::max(minimal, self.prospective_value - dropped.into());
+		self.prospective_value = prospective_value + 1.into();
+
+		let (next, rx) = oneshot::channel();
+		let next = Some(next);
+		let pool = self.pool.clone();
+		let dropped = self.dropped.clone();
+		match mem::replace(&mut self.previous, Some(rx)) {
+			Some(previous) => Reserved {
+				previous: Either::A(previous),
+				next,
+				minimal,
+				prospective_value,
+				pool,
+				dropped,
+			},
+			None => Reserved {
+				previous: Either::B(future::ok(minimal)),
+				next,
+				minimal,
+				prospective_value,
+				pool,
+				dropped,
+			},
+		}
+	}
+}
+
+/// Represents a future nonce.
+#[derive(Debug)]
+pub struct Reserved {
+	previous: Either<
+		oneshot::Receiver<U256>,
+		future::FutureResult<U256, oneshot::Canceled>
+	>,
+	next: Option<oneshot::Sender<U256>>,
+	minimal: U256,
+	prospective_value: U256,
+	pool: CpuPool,
+	dropped: Arc<AtomicUsize>,
+}
+
+impl Reserved {
+	/// Returns a prospective value of the nonce.
+	/// NOTE: This might be different than the one we resolve to.
+	/// Make sure to check if both nonces match or use the latter one.
+	pub fn prospective_value(&self) -> &U256 {
+		&self.prospective_value
+	}
+}
+
+impl Future for Reserved {
+    type Item = Ready;
+    type Error = ();
+
+    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+		let mut value = try_ready!(self.previous.poll().map_err(|e| {
+			warn!("Unexpected nonce cancellation: {}", e);
+		}));
+
+		if value < self.minimal {
+			value = self.minimal
+		}
+		let matches_prospective = value == self.prospective_value;
+
+		Ok(Async::Ready(Ready {
+			value,
+			matches_prospective,
+			next: self.next.take(),
+			dropped: self.dropped.clone(),
+		}))
+	}
+}
+
+impl Drop for Reserved {
+	fn drop(&mut self) {
+		if let Some(next) = self.next.take() {
+			self.dropped.fetch_add(1, atomic::Ordering::SeqCst);
+			// If Reserved is dropped just pipe previous and next together.
+			let previous = mem::replace(&mut self.previous, Either::B(future::ok(U256::default())));
+			self.pool.spawn(previous.map(|nonce| {
+				next.send(nonce).expect(Ready::RECV_PROOF)
+			})).forget()
+		}
+	}
+}
+
+/// Represents a valid reserved nonce.
+/// This can be used to dispatch the transaction.
+///
+/// After this nonce is used it should be marked as such
+/// using `mark_used` method.
+#[derive(Debug)]
+pub struct Ready {
+	value: U256,
+	matches_prospective: bool,
+	next: Option<oneshot::Sender<U256>>,
+	dropped: Arc<AtomicUsize>,
+}
+
+impl Ready {
+	const RECV_PROOF: &'static str = "Receiver never dropped.";
+
+	/// Returns a value of the nonce.
+	pub fn value(&self) -> &U256 {
+		&self.value
+	}
+
+	/// Returns true if current value matches the prospective nonce.
+	pub fn matches_prospective(&self) -> bool {
+		self.matches_prospective
+	}
+
+	/// Marks this nonce as used.
+	/// Make sure to call that method after this nonce has been consumed.
+	pub fn mark_used(mut self) {
+		let next = self.next.take().expect("Nonce can be marked as used only once; qed");
+		next.send(self.value + 1.into()).expect(Self::RECV_PROOF);
+	}
+}
+
+impl Drop for Ready {
+	fn drop(&mut self) {
+		if let Some(send) = self.next.take() {
+			self.dropped.fetch_add(1, atomic::Ordering::SeqCst);
+			send.send(self.value).expect(Self::RECV_PROOF);
+		}
+	}
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+	#[test]
+	fn should_reserve_a_set_of_nonces_and_resolve_them() {
+		let mut nonces = Reservations::new();
+
+		let n1 = nonces.reserve_nonce(5.into());
+		let n2 = nonces.reserve_nonce(5.into());
+		let n3 = nonces.reserve_nonce(5.into());
+		let n4 = nonces.reserve_nonce(5.into());
+
+		// Check first nonce
+		let r = n1.wait().unwrap();
+		assert_eq!(r.value(), &U256::from(5));
+		assert!(r.matches_prospective());
+		r.mark_used();
+
+		// Drop second nonce
+		drop(n2);
+
+		// Drop third without marking as used
+		let r = n3.wait().unwrap();
+		drop(r);
+
+		// Last nonce should be resolved to 6
+		let r = n4.wait().unwrap();
+		assert_eq!(r.value(), &U256::from(6));
+		assert!(!r.matches_prospective());
+		r.mark_used();
+
+		// Next nonce should be immediately available.
+		let n5 = nonces.reserve_nonce(5.into());
+		let r = n5.wait().unwrap();
+		assert_eq!(r.value(), &U256::from(7));
+		assert!(r.matches_prospective());
+		r.mark_used();
+
+		// Should use start number if it's greater
+		let n6 = nonces.reserve_nonce(10.into());
+		let r = n6.wait().unwrap();
+		assert_eq!(r.value(), &U256::from(10));
+		assert!(r.matches_prospective());
+		r.mark_used();
+	}
+
+	#[test]
+	fn should_return_prospective_nonce() {
+		let mut nonces = Reservations::new();
+
+		let n1 = nonces.reserve_nonce(5.into());
+		let n2 = nonces.reserve_nonce(5.into());
+
+		assert_eq!(n1.prospective_value(), &U256::from(5));
+		assert_eq!(n2.prospective_value(), &U256::from(6));
+	}
+}

rpc/src/v1/helpers/requests.rs

@@ -15,8 +15,9 @@
 // along with Parity.  If not, see <http://www.gnu.org/licenses/>.
 
 use bigint::prelude::U256;
-use util::Address;
 use bytes::Bytes;
+use util::Address;
+
 use v1::types::{Origin, TransactionCondition};
 
 /// Transaction request coming from RPC

rpc/src/v1/impls/eth.rs

@@ -45,7 +45,7 @@ use jsonrpc_core::futures::future;
 use jsonrpc_macros::Trailing;
 
 use v1::helpers::{errors, limit_logs, fake_sign};
-use v1::helpers::dispatch::{Dispatcher, FullDispatcher, default_gas_price};
+use v1::helpers::dispatch::{FullDispatcher, default_gas_price};
 use v1::helpers::block_import::is_major_importing;
 use v1::helpers::accounts::unwrap_provider;
 use v1::traits::Eth;
@@ -610,8 +610,11 @@ impl<C, SN: ?Sized, S: ?Sized, M, EM> Eth for EthClient<C, SN, S, M, EM> where
 			.map_err(errors::rlp)
 			.and_then(|tx| SignedTransaction::new(tx).map_err(errors::transaction))
 			.and_then(|signed_transaction| {
-				FullDispatcher::new(self.client.clone(), self.miner.clone())
+				FullDispatcher::dispatch_transaction(
-					.dispatch_transaction(signed_transaction.into())
+					&*self.client,
+					&*self.miner,
+					signed_transaction.into(),
+				)
 			})
 			.map(Into::into)
 	}

rpc/src/v1/impls/signing.rs

@@ -125,7 +125,7 @@ impl<D: Dispatcher + 'static> SigningQueueClient<D> {
 		Box::new(dispatch::from_rpc(payload, default_account, &dispatcher)
 			.and_then(move |payload| {
 				let sender = payload.sender();
-				if accounts.is_unlocked(sender) {
+				if accounts.is_unlocked(&sender) {
 					Either::A(dispatch::execute(dispatcher, accounts, payload, dispatch::SignWith::Nothing)
 						.map(|v| v.into_value())
 						.map(DispatchResult::Value))

rpc/src/v1/tests/eth.rs

@@ -19,25 +19,27 @@ use std::env;
 use std::sync::Arc;
 use std::time::Duration;
 
-use ethcore::client::{BlockChainClient, Client, ClientConfig};
+use bigint::hash::H256;
-use ethcore::ids::BlockId;
+use bigint::prelude::U256;
-use ethcore::spec::{Genesis, Spec};
-use ethcore::block::Block;
-use ethcore::views::BlockView;
-use ethcore::ethereum;
-use ethcore::miner::{MinerOptions, Banning, GasPricer, MinerService, ExternalMiner, Miner, PendingSet, PrioritizationStrategy, GasLimit};
 use ethcore::account_provider::AccountProvider;
+use ethcore::block::Block;
+use ethcore::client::{BlockChainClient, Client, ClientConfig};
+use ethcore::ethereum;
+use ethcore::ids::BlockId;
+use ethcore::miner::{MinerOptions, Banning, GasPricer, MinerService, ExternalMiner, Miner, PendingSet, PrioritizationStrategy, GasLimit};
+use ethcore::spec::{Genesis, Spec};
+use ethcore::views::BlockView;
 use ethjson::blockchain::BlockChain;
 use ethjson::state::test::ForkSpec;
 use io::IoChannel;
-use bigint::prelude::U256;
-use bigint::hash::H256;
-use util::Address;
 use kvdb_memorydb;
+use parking_lot::Mutex;
+use util::Address;
 
 use jsonrpc_core::IoHandler;
-use v1::impls::{EthClient, SigningUnsafeClient};
 use v1::helpers::dispatch::FullDispatcher;
+use v1::helpers::nonce;
+use v1::impls::{EthClient, SigningUnsafeClient};
 use v1::metadata::Metadata;
 use v1::tests::helpers::{TestSnapshotService, TestSyncProvider, Config};
 use v1::traits::eth::Eth;
@@ -148,7 +150,9 @@ impl EthTester {
 			Default::default(),
 		);
 
-		let dispatcher = FullDispatcher::new(client.clone(), miner_service.clone());
+		let reservations = Arc::new(Mutex::new(nonce::Reservations::new()));
+
+		let dispatcher = FullDispatcher::new(client.clone(), miner_service.clone(), reservations);
 		let eth_sign = SigningUnsafeClient::new(
 			&opt_account_provider,
 			dispatcher,

rpc/src/v1/tests/mocked/eth.rs

@@ -37,6 +37,7 @@ use ethsync::SyncState;
 
 use jsonrpc_core::IoHandler;
 use v1::{Eth, EthClient, EthClientOptions, EthFilter, EthFilterClient, EthSigning, SigningUnsafeClient};
+use v1::helpers::nonce;
 use v1::helpers::dispatch::FullDispatcher;
 use v1::tests::helpers::{TestSyncProvider, Config, TestMinerService, TestSnapshotService};
 use v1::metadata::Metadata;
@@ -94,8 +95,9 @@ impl EthTester {
 		let external_miner = Arc::new(ExternalMiner::new(hashrates.clone()));
 		let eth = EthClient::new(&client, &snapshot, &sync, &opt_ap, &miner, &external_miner, options).to_delegate();
 		let filter = EthFilterClient::new(client.clone(), miner.clone()).to_delegate();
+		let reservations = Arc::new(Mutex::new(nonce::Reservations::new()));
 
-		let dispatcher = FullDispatcher::new(client.clone(), miner.clone());
+		let dispatcher = FullDispatcher::new(client.clone(), miner.clone(), reservations);
 		let sign = SigningUnsafeClient::new(&opt_ap, dispatcher).to_delegate();
 		let mut io: IoHandler<Metadata> = IoHandler::default();
 		io.extend_with(eth);

rpc/src/v1/tests/mocked/personal.rs

@@ -17,14 +17,16 @@
 use std::sync::Arc;
 use std::str::FromStr;
 
+use bigint::prelude::U256;
 use ethcore::account_provider::AccountProvider;
 use ethcore::client::TestBlockChainClient;
 use ethcore::transaction::{Action, Transaction};
 use jsonrpc_core::IoHandler;
-use bigint::prelude::U256;
+use parking_lot::Mutex;
 use util::Address;
 
 use v1::{PersonalClient, Personal, Metadata};
+use v1::helpers::nonce;
 use v1::helpers::dispatch::FullDispatcher;
 use v1::tests::helpers::TestMinerService;
 
@@ -52,8 +54,9 @@ fn setup() -> PersonalTester {
 	let opt_accounts = Some(accounts.clone());
 	let client = blockchain_client();
 	let miner = miner_service();
+	let reservations = Arc::new(Mutex::new(nonce::Reservations::new()));
 
-	let dispatcher = FullDispatcher::new(client, miner.clone());
+	let dispatcher = FullDispatcher::new(client, miner.clone(), reservations);
 	let personal = PersonalClient::new(opt_accounts, dispatcher, false);
 
 	let mut io = IoHandler::default();

rpc/src/v1/tests/mocked/signer.rs

@@ -24,6 +24,7 @@ use ethcore::account_provider::AccountProvider;
 use ethcore::client::TestBlockChainClient;
 use ethcore::transaction::{Transaction, Action, SignedTransaction};
 use parity_reactor::EventLoop;
+use parking_lot::Mutex;
 use rlp::encode;
 
 use serde_json;
@@ -32,7 +33,7 @@ use v1::{SignerClient, Signer, Origin};
 use v1::metadata::Metadata;
 use v1::tests::helpers::TestMinerService;
 use v1::types::{Bytes as RpcBytes, H520};
-use v1::helpers::{SigningQueue, SignerService, FilledTransactionRequest, ConfirmationPayload};
+use v1::helpers::{nonce, SigningQueue, SignerService, FilledTransactionRequest, ConfirmationPayload};
 use v1::helpers::dispatch::{FullDispatcher, eth_data_hash};
 
 struct SignerTester {
@@ -61,9 +62,10 @@ fn signer_tester() -> SignerTester {
 	let opt_accounts = Some(accounts.clone());
 	let client = blockchain_client();
 	let miner = miner_service();
+	let reservations = Arc::new(Mutex::new(nonce::Reservations::new()));
 	let event_loop = EventLoop::spawn();
 
-	let dispatcher = FullDispatcher::new(client, miner.clone());
+	let dispatcher = FullDispatcher::new(client, miner.clone(), reservations);
 	let mut io = IoHandler::default();
 	io.extend_with(SignerClient::new(&opt_accounts, dispatcher, &signer, event_loop.remote()).to_delegate());
 

rpc/src/v1/tests/mocked/signing.rs

@@ -25,7 +25,7 @@ use jsonrpc_core::futures::Future;
 use v1::impls::SigningQueueClient;
 use v1::metadata::Metadata;
 use v1::traits::{EthSigning, ParitySigning, Parity};
-use v1::helpers::{SignerService, SigningQueue, FullDispatcher};
+use v1::helpers::{nonce, SignerService, SigningQueue, FullDispatcher};
 use v1::types::{ConfirmationResponse, RichRawTransaction};
 use v1::tests::helpers::TestMinerService;
 use v1::tests::mocked::parity;
@@ -39,6 +39,7 @@ use ethcore::client::TestBlockChainClient;
 use ethcore::transaction::{Transaction, Action, SignedTransaction};
 use ethstore::ethkey::{Generator, Random};
 use serde_json;
+use parking_lot::Mutex;
 
 use parity_reactor::Remote;
 
@@ -57,9 +58,10 @@ impl Default for SigningTester {
 		let miner = Arc::new(TestMinerService::default());
 		let accounts = Arc::new(AccountProvider::transient_provider());
 		let opt_accounts = Some(accounts.clone());
+		let reservations = Arc::new(Mutex::new(nonce::Reservations::new()));
 		let mut io = IoHandler::default();
 
-		let dispatcher = FullDispatcher::new(client.clone(), miner.clone());
+		let dispatcher = FullDispatcher::new(client.clone(), miner.clone(), reservations);
 
 		let remote = Remote::new_thread_per_future();