Use nonce reservation per address

rpc/src/v1/helpers/dispatch.rs

@@ -19,6 +19,7 @@
 use std::fmt::Debug;
 use std::ops::Deref;
 use std::sync::Arc;
+use std::collections::HashMap;
 
 use light::cache::Cache as LightDataCache;
 use light::client::LightChainClient;
@@ -32,6 +33,7 @@ use util::Address;
 use bytes::Bytes;
 use parking_lot::{Mutex, RwLock};
 use stats::Corpus;
+use futures_cpupool::CpuPool;
 
 use ethkey::Signature;
 use ethsync::LightSync;
@@ -87,16 +89,20 @@ pub trait Dispatcher: Send + Sync + Clone {
 pub struct FullDispatcher<C, M> {
 	client: Arc<C>,
 	miner: Arc<M>,
-	nonces: Arc<Mutex<nonce::Reservations>>,
+	nonces: Arc<Mutex<HashMap<Address, nonce::Reservations>>>,
+	pool: CpuPool,
 }
 
 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>, nonces: Arc<Mutex<nonce::Reservations>>) -> Self {
+	pub fn new(client: Arc<C>, miner: Arc<M>, pool: CpuPool) -> Self {
+		let nonces = Arc::new(Mutex::new(HashMap::new()));
+
 		FullDispatcher {
 			client,
 			miner,
 			nonces,
+			pool,
 		}
 	}
 }
@@ -107,6 +113,7 @@ impl<C, M> Clone for FullDispatcher<C, M> {
 			client: self.client.clone(),
 			miner: self.miner.clone(),
 			nonces: self.nonces.clone(),
+			pool: self.pool.clone(),
 		}
 	}
 }
@@ -162,7 +169,10 @@ impl<C: MiningBlockChainClient, M: MinerService> Dispatcher for FullDispatcher<C
 		}
 
 		let state = self.state_nonce(&filled.from);
-		let reserved = self.nonces.lock().reserve_nonce(state);
+		let reserved = self.nonces.lock().entry(filled.from)
+			.or_insert(nonce::Reservations::with_pool(self.pool.clone()))
+			.reserve_nonce(state);
+
 		Box::new(ProspectiveSigner::new(accounts, filled, chain_id, reserved, password))
 	}
 
@@ -252,7 +262,9 @@ pub struct LightDispatcher {
 	/// Transaction queue.
 	pub transaction_queue: Arc<RwLock<LightTransactionQueue>>,
 	/// Nonce reservations
-	pub nonces: Arc<Mutex<nonce::Reservations>>,
+	pub nonces: Arc<Mutex<HashMap<Address, nonce::Reservations>>>,
+	/// Cpu pool
+	pub pool: CpuPool,
 }
 
 impl LightDispatcher {
@@ -265,8 +277,10 @@ impl LightDispatcher {
 		on_demand: Arc<OnDemand>,
 		cache: Arc<Mutex<LightDataCache>>,
 		transaction_queue: Arc<RwLock<LightTransactionQueue>>,
-		nonces: Arc<Mutex<nonce::Reservations>>,
+		pool: CpuPool,
 	) -> Self {
+		let nonces = Arc::new(Mutex::new(HashMap::new()));
+
 		LightDispatcher {
 			sync,
 			client,
@@ -274,6 +288,7 @@ impl LightDispatcher {
 			cache,
 			transaction_queue,
 			nonces,
+			pool,
 		}
 	}
 
@@ -379,10 +394,13 @@ impl Dispatcher for LightDispatcher {
 		}
 
 		let nonces = self.nonces.clone();
+		let pool = self.pool.clone();
 		Box::new(self.next_nonce(filled.from)
 			.map_err(|_| errors::no_light_peers())
 			.and_then(move |nonce| {
-				let reserved = nonces.lock().reserve_nonce(nonce);
+				let reserved = nonces.lock().entry(filled.from)
+					.or_insert(nonce::Reservations::with_pool(pool))
+					.reserve_nonce(nonce);
 				ProspectiveSigner::new(accounts, filled, chain_id, reserved, password)
 			}))
 	}