Merge pull request #4594 from ethcore/lightrpc

Remainder of RPC APIs implemented for the light client

ethcore/Cargo.toml

@@ -43,6 +43,7 @@ rlp = { path = "../util/rlp" }
 ethcore-stratum = { path = "../stratum" }
 ethcore-bloom-journal = { path = "../util/bloom" }
 hardware-wallet = { path = "../hw" }
+stats = { path = "../util/stats" }
 
 [dependencies.hyper]
 git = "https://github.com/ethcore/hyper"

ethcore/light/Cargo.toml

@@ -23,6 +23,7 @@ smallvec = "0.3.1"
 futures = "0.1"
 rand = "0.3"
 itertools = "0.5"
+stats = { path = "../../util/stats" }
 
 [features]
 default = []

ethcore/light/src/cache.rs

@@ -0,0 +1,175 @@
+// Copyright 2015-2017 Parity 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/>.
+
+//! Cache for data fetched from the network.
+//!
+//! Stores ancient block headers, bodies, receipts, and total difficulties.
+//! Furthermore, stores a "gas price corpus" of relative recency, which is a sorted
+//! vector of all gas prices from a recent range of blocks.
+
+use ethcore::encoded;
+use ethcore::header::BlockNumber;
+use ethcore::receipt::Receipt;
+
+use stats::Corpus;
+use time::{SteadyTime, Duration};
+use util::{U256, H256};
+use util::cache::MemoryLruCache;
+
+/// Configuration for how much data to cache.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct CacheSizes {
+	/// Maximum size, in bytes, of cached headers.
+	pub headers: usize,
+	/// Maximum size, in bytes, of cached canonical hashes.
+	pub canon_hashes: usize,
+	/// Maximum size, in bytes, of cached block bodies.
+	pub bodies: usize,
+	/// Maximum size, in bytes, of cached block receipts.
+	pub receipts: usize,
+	/// Maximum size, in bytes, of cached chain score for the block.
+	pub chain_score: usize,
+}
+
+impl Default for CacheSizes {
+	fn default() -> Self {
+		const MB: usize = 1024 * 1024;
+		CacheSizes {
+			headers: 10 * MB,
+			canon_hashes: 3 * MB,
+			bodies: 20 * MB,
+			receipts: 10 * MB,
+			chain_score: 7 * MB,
+		}
+	}
+}
+
+/// The light client data cache.
+///
+/// Note that almost all getter methods take `&mut self` due to the necessity to update
+/// the underlying LRU-caches on read.
+pub struct Cache {
+	headers: MemoryLruCache<H256, encoded::Header>,
+	canon_hashes: MemoryLruCache<BlockNumber, H256>,
+	bodies: MemoryLruCache<H256, encoded::Body>,
+	receipts: MemoryLruCache<H256, Vec<Receipt>>,
+	chain_score: MemoryLruCache<H256, U256>,
+	corpus: Option<(Corpus<U256>, SteadyTime)>,
+	corpus_expiration: Duration,
+}
+
+impl Cache {
+	/// Create a new data cache with the given sizes and gas price corpus expiration time.
+	pub fn new(sizes: CacheSizes, corpus_expiration: Duration) -> Self {
+		Cache {
+			headers: MemoryLruCache::new(sizes.headers),
+			canon_hashes: MemoryLruCache::new(sizes.canon_hashes),
+			bodies: MemoryLruCache::new(sizes.bodies),
+			receipts: MemoryLruCache::new(sizes.receipts),
+			chain_score: MemoryLruCache::new(sizes.chain_score),
+			corpus: None,
+			corpus_expiration: corpus_expiration,
+		}
+	}
+
+	/// Query header by hash.
+	pub fn block_header(&mut self, hash: &H256) -> Option<encoded::Header> {
+		self.headers.get_mut(hash).map(|x| x.clone())
+	}
+
+	/// Query hash by number.
+	pub fn block_hash(&mut self, num: &BlockNumber) -> Option<H256> {
+		self.canon_hashes.get_mut(num).map(|x| x.clone())
+	}
+
+	/// Query block body by block hash.
+	pub fn block_body(&mut self, hash: &H256) -> Option<encoded::Body> {
+		self.bodies.get_mut(hash).map(|x| x.clone())
+	}
+
+	/// Query block receipts by block hash.
+	pub fn block_receipts(&mut self, hash: &H256) -> Option<Vec<Receipt>> {
+		self.receipts.get_mut(hash).map(|x| x.clone())
+	}
+
+	/// Query chain score by block hash.
+	pub fn chain_score(&mut self, hash: &H256) -> Option<U256> {
+		self.chain_score.get_mut(hash).map(|x| x.clone())
+	}
+
+	/// Cache the given header.
+	pub fn insert_block_header(&mut self, hash: H256, hdr: encoded::Header) {
+		self.headers.insert(hash, hdr);
+	}
+
+	/// Cache the given canonical block hash.
+	pub fn insert_block_hash(&mut self, num: BlockNumber, hash: H256) {
+		self.canon_hashes.insert(num, hash);
+	}
+
+	/// Cache the given block body.
+	pub fn insert_block_body(&mut self, hash: H256, body: encoded::Body) {
+		self.bodies.insert(hash, body);
+	}
+
+	/// Cache the given block receipts.
+	pub fn insert_block_receipts(&mut self, hash: H256, receipts: Vec<Receipt>) {
+		self.receipts.insert(hash, receipts);
+	}
+
+	/// Cache the given chain scoring.
+	pub fn insert_chain_score(&mut self, hash: H256, score: U256) {
+		self.chain_score.insert(hash, score);
+	}
+
+	/// Get gas price corpus, if recent enough.
+	pub fn gas_price_corpus(&self) -> Option<Corpus<U256>> {
+		let now = SteadyTime::now();
+
+		self.corpus.as_ref().and_then(|&(ref corpus, ref tm)| {
+			if *tm + self.corpus_expiration >= now {
+				Some(corpus.clone())
+			} else {
+				None
+			}
+		})
+	}
+
+	/// Set the cached gas price corpus.
+	pub fn set_gas_price_corpus(&mut self, corpus: Corpus<U256>) {
+		self.corpus = Some((corpus, SteadyTime::now()))
+	}
+}
+
+#[cfg(test)]
+mod tests {
+	use super::Cache;
+	use time::Duration;
+
+	#[test]
+	fn corpus_inaccessible() {
+		let mut cache = Cache::new(Default::default(), Duration::hours(5));
+
+		cache.set_gas_price_corpus(vec![].into());
+		assert_eq!(cache.gas_price_corpus(), Some(vec![].into()));
+
+		{
+			let corpus_time = &mut cache.corpus.as_mut().unwrap().1;
+			*corpus_time = *corpus_time - Duration::hours(6);
+		}
+		assert!(cache.gas_price_corpus().is_none());
+	}
+}

ethcore/light/src/client/header_chain.rs

@@ -241,6 +241,14 @@ impl HeaderChain {
 		self.block_header(BlockId::Latest).expect("Header for best block always stored; qed")
 	}
 
+	/// Get an iterator over a block and its ancestry.
+	pub fn ancestry_iter(&self, start: BlockId) -> AncestryIter {
+		AncestryIter {
+			next: self.block_header(start),
+			chain: self,
+		}
+	}
+
 	/// Get the nth CHT root, if it's been computed.
 	///
 	/// CHT root 0 is from block `1..2048`.
@@ -295,6 +303,25 @@ impl HeapSizeOf for HeaderChain {
 	}
 }
 
+/// Iterator over a block's ancestry.
+pub struct AncestryIter<'a> {
+	next: Option<encoded::Header>,
+	chain: &'a HeaderChain,
+}
+
+impl<'a> Iterator for AncestryIter<'a> {
+	type Item = encoded::Header;
+
+	fn next(&mut self) -> Option<encoded::Header> {
+		let next = self.next.take();
+		if let Some(p_hash) = next.as_ref().map(|hdr| hdr.parent_hash()) {
+			self.next = self.chain.block_header(BlockId::Hash(p_hash));
+		}
+
+		next
+	}
+}
+
 #[cfg(test)]
 mod tests {
 	use super::HeaderChain;

ethcore/light/src/client/mod.rs

@@ -20,7 +20,7 @@ use std::sync::Arc;
 
 use ethcore::block_import_error::BlockImportError;
 use ethcore::block_status::BlockStatus;
-use ethcore::client::ClientReport;
+use ethcore::client::{ClientReport, EnvInfo};
 use ethcore::engines::Engine;
 use ethcore::ids::BlockId;
 use ethcore::header::Header;
@@ -33,7 +33,7 @@ use io::IoChannel;
 
 use util::{Bytes, H256, Mutex, RwLock};
 
-use self::header_chain::HeaderChain;
+use self::header_chain::{AncestryIter, HeaderChain};
 
 pub use self::service::Service;
 
@@ -62,6 +62,12 @@ pub trait LightChainClient: Send + Sync {
 	/// Get the best block header.
 	fn best_block_header(&self) -> encoded::Header;
 
+	/// Get an iterator over a block and its ancestry.
+	fn ancestry_iter<'a>(&'a self, start: BlockId) -> Box<Iterator<Item=encoded::Header> + 'a>;
+
+	/// Get the signing network ID.
+	fn signing_network_id(&self) -> Option<u64>;
+
 	/// Query whether a block is known.
 	fn is_known(&self, hash: &H256) -> bool;
 
@@ -164,6 +170,16 @@ impl Client {
 		self.chain.best_header()
 	}
 
+	/// Get an iterator over a block and its ancestry.
+	pub fn ancestry_iter(&self, start: BlockId) -> AncestryIter {
+		self.chain.ancestry_iter(start)
+	}
+
+	/// Get the signing network id.
+	pub fn signing_network_id(&self) -> Option<u64> {
+		self.engine.signing_network_id(&self.latest_env_info())
+	}
+
 	/// Flush the header queue.
 	pub fn flush_queue(&self) {
 		self.queue.flush()
@@ -217,6 +233,33 @@ impl Client {
 	pub fn engine(&self) -> &Engine {
 		&*self.engine
 	}
+
+	fn latest_env_info(&self) -> EnvInfo {
+		let header = self.best_block_header();
+
+		EnvInfo {
+			number: header.number(),
+			author: header.author(),
+			timestamp: header.timestamp(),
+			difficulty: header.difficulty(),
+			last_hashes: self.build_last_hashes(header.hash()),
+			gas_used: Default::default(),
+			gas_limit: header.gas_limit(),
+		}
+	}
+
+	fn build_last_hashes(&self, mut parent_hash: H256) -> Arc<Vec<H256>> {
+		let mut v = Vec::with_capacity(256);
+		for _ in 0..255 {
+			v.push(parent_hash);
+			match self.block_header(BlockId::Hash(parent_hash)) {
+				Some(header) => parent_hash = header.hash(),
+				None => break,
+			}
+		}
+
+		Arc::new(v)
+	}
 }
 
 impl LightChainClient for Client {
@@ -234,6 +277,14 @@ impl LightChainClient for Client {
 		Client::best_block_header(self)
 	}
 
+	fn ancestry_iter<'a>(&'a self, start: BlockId) -> Box<Iterator<Item=encoded::Header> + 'a> {
+		Box::new(Client::ancestry_iter(self, start))
+	}
+
+	fn signing_network_id(&self) -> Option<u64> {
+		Client::signing_network_id(self)
+	}
+
 	fn is_known(&self, hash: &H256) -> bool {
 		self.status(hash) == BlockStatus::InChain
 	}

ethcore/light/src/lib.rs

@@ -37,6 +37,7 @@ pub mod cht;
 pub mod net;
 pub mod on_demand;
 pub mod transaction_queue;
+pub mod cache;
 
 #[cfg(not(feature = "ipc"))]
 pub mod provider;
@@ -71,6 +72,7 @@ extern crate time;
 extern crate futures;
 extern crate rand;
 extern crate itertools;
+extern crate stats;
 
 #[cfg(feature = "ipc")]
 extern crate ethcore_ipc as ipc;

ethcore/light/src/net/mod.rs

@@ -322,6 +322,16 @@ impl LightProtocol {
 			.map(|peer| peer.lock().status.clone())
 	}
 
+	/// Get number of (connected, active) peers.
+	pub fn peer_count(&self) -> (usize, usize) {
+		let num_pending = self.pending_peers.read().len();
+		let peers = self.peers.read();
+		(
+			num_pending + peers.len(),
+			peers.values().filter(|p| !p.lock().pending_requests.is_empty()).count(),
+		)
+	}
+
 	/// Check the maximum amount of requests of a specific type
 	/// which a peer would be able to serve. Returns zero if the
 	/// peer is unknown or has no buffer flow parameters.

ethcore/light/src/net/request_set.rs

@@ -110,6 +110,14 @@ impl RequestSet {
 	pub fn collect_ids<F>(&self) -> F where F: FromIterator<ReqId> {
 		self.ids.keys().cloned().collect()
 	}
+
+	/// Number of requests in the set.
+	pub fn len(&self) -> usize {
+		self.ids.len()
+	}
+
+	/// Whether the set is empty.
+	pub fn is_empty(&self) -> bool { self.len() == 0 }
 }
 
 #[cfg(test)]

ethcore/light/src/on_demand/mod.rs

@@ -19,6 +19,7 @@
 //! will take the raw data received here and extract meaningful results from it.
 
 use std::collections::HashMap;
+use std::sync::Arc;
 
 use ethcore::basic_account::BasicAccount;
 use ethcore::encoded;
@@ -28,10 +29,11 @@ use futures::{Async, Poll, Future};
 use futures::sync::oneshot::{self, Sender, Receiver};
 use network::PeerId;
 use rlp::{RlpStream, Stream};
-use util::{Bytes, RwLock, U256};
+use util::{Bytes, RwLock, Mutex, U256};
 use util::sha3::{SHA3_NULL_RLP, SHA3_EMPTY_LIST_RLP};
 
 use net::{Handler, Status, Capabilities, Announcement, EventContext, BasicContext, ReqId};
+use cache::Cache;
 use types::les_request::{self as les_request, Request as LesRequest};
 
 pub mod request;
@@ -42,9 +44,16 @@ struct Peer {
 	capabilities: Capabilities,
 }
 
+// Which portions of a CHT proof should be sent.
+enum ChtProofSender {
+	Both(Sender<(encoded::Header, U256)>),
+	Header(Sender<encoded::Header>),
+	ChainScore(Sender<U256>),
+}
+
 // Attempted request info and sender to put received value.
 enum Pending {
-	HeaderByNumber(request::HeaderByNumber, Sender<(encoded::Header, U256)>), // num + CHT root
+	HeaderByNumber(request::HeaderByNumber, ChtProofSender),
 	HeaderByHash(request::HeaderByHash, Sender<encoded::Header>),
 	Block(request::Body, Sender<encoded::Block>),
 	BlockReceipts(request::BlockReceipts, Sender<Vec<Receipt>>),
@@ -58,30 +67,77 @@ enum Pending {
 pub struct OnDemand {
 	peers: RwLock<HashMap<PeerId, Peer>>,
 	pending_requests: RwLock<HashMap<ReqId, Pending>>,
+	cache: Arc<Mutex<Cache>>,
 	orphaned_requests: RwLock<Vec<Pending>>,
 }
 
-impl Default for OnDemand {
-	fn default() -> Self {
+impl OnDemand {
+	/// Create a new `OnDemand` service with the given cache.
+	pub fn new(cache: Arc<Mutex<Cache>>) -> Self {
 		OnDemand {
 			peers: RwLock::new(HashMap::new()),
 			pending_requests: RwLock::new(HashMap::new()),
+			cache: cache,
 			orphaned_requests: RwLock::new(Vec::new()),
 		}
 	}
-}
 
-impl OnDemand {
 	/// Request a header by block number and CHT root hash.
-	/// Returns the header and the total difficulty.
-	pub fn header_by_number(&self, ctx: &BasicContext, req: request::HeaderByNumber) -> Receiver<(encoded::Header, U256)> {
+	/// Returns the header.
+	pub fn header_by_number(&self, ctx: &BasicContext, req: request::HeaderByNumber) -> Receiver<encoded::Header> {
 		let (sender, receiver) = oneshot::channel();
-		self.dispatch_header_by_number(ctx, req, sender);
+		let cached = {
+			let mut cache = self.cache.lock();
+			cache.block_hash(&req.num()).and_then(|hash| cache.block_header(&hash))
+		};
+
+		match cached {
+			Some(hdr) => sender.complete(hdr),
+			None => self.dispatch_header_by_number(ctx, req, ChtProofSender::Header(sender)),
+		}
+		receiver
+	}
+
+	/// Request a canonical block's chain score.
+	/// Returns the chain score.
+	pub fn chain_score_by_number(&self, ctx: &BasicContext, req: request::HeaderByNumber) -> Receiver<U256> {
+		let (sender, receiver) = oneshot::channel();
+		let cached = {
+			let mut cache = self.cache.lock();
+			cache.block_hash(&req.num()).and_then(|hash| cache.chain_score(&hash))
+		};
+
+		match cached {
+			Some(score) => sender.complete(score),
+			None => self.dispatch_header_by_number(ctx, req, ChtProofSender::ChainScore(sender)),
+		}
+
+		receiver
+	}
+
+	/// Request a canonical block's chain score.
+	/// Returns the header and chain score.
+	pub fn header_and_score_by_number(&self, ctx: &BasicContext, req: request::HeaderByNumber) -> Receiver<(encoded::Header, U256)> {
+		let (sender, receiver) = oneshot::channel();
+		let cached = {
+			let mut cache = self.cache.lock();
+			let hash = cache.block_hash(&req.num());
+			(
+				hash.clone().and_then(|hash| cache.block_header(&hash)),
+				hash.and_then(|hash| cache.chain_score(&hash)),
+			)
+		};
+
+		match cached {
+			(Some(hdr), Some(score)) => sender.complete((hdr, score)),
+			_ => self.dispatch_header_by_number(ctx, req, ChtProofSender::Both(sender)),
+		}
+
 		receiver
 	}
 
 	// dispatch the request, completing the request if no peers available.
-	fn dispatch_header_by_number(&self, ctx: &BasicContext, req: request::HeaderByNumber, sender: Sender<(encoded::Header, U256)>) {
+	fn dispatch_header_by_number(&self, ctx: &BasicContext, req: request::HeaderByNumber, sender: ChtProofSender) {
 		let num = req.num();
 		let cht_num = req.cht_num();
 
@@ -123,7 +179,10 @@ impl OnDemand {
 	/// it as easily.
 	pub fn header_by_hash(&self, ctx: &BasicContext, req: request::HeaderByHash) -> Receiver<encoded::Header> {
 		let (sender, receiver) = oneshot::channel();
-		self.dispatch_header_by_hash(ctx, req, sender);
+		match self.cache.lock().block_header(&req.0) {
+			Some(hdr) => sender.complete(hdr),
+			None => self.dispatch_header_by_hash(ctx, req, sender),
+		}
 		receiver
 	}
 
@@ -181,7 +240,16 @@ impl OnDemand {
 
 			sender.complete(encoded::Block::new(stream.out()))
 		} else {
-			self.dispatch_block(ctx, req, sender);
+			match self.cache.lock().block_body(&req.hash) {
+				Some(body) => {
+					let mut stream = RlpStream::new_list(3);
+					stream.append_raw(&req.header.into_inner(), 1);
+					stream.append_raw(&body.into_inner(), 2);
+
+					sender.complete(encoded::Block::new(stream.out()));
+				}
+				None => self.dispatch_block(ctx, req, sender),
+			}
 		}
 		receiver
 	}
@@ -224,7 +292,10 @@ impl OnDemand {
 		if req.0.receipts_root() == SHA3_NULL_RLP {
 			sender.complete(Vec::new())
 		} else {
-			self.dispatch_block_receipts(ctx, req, sender);
+			match self.cache.lock().block_receipts(&req.0.hash()) {
+				Some(receipts) => sender.complete(receipts),
+				None => self.dispatch_block_receipts(ctx, req, sender),
+			}
 		}
 
 		receiver
@@ -378,8 +449,15 @@ impl OnDemand {
 
 		for orphaned in to_dispatch {
 			match orphaned {
-				Pending::HeaderByNumber(req, mut sender) =>
-					if !check_hangup(&mut sender) { self.dispatch_header_by_number(ctx, req, sender) },
+				Pending::HeaderByNumber(req, mut sender) => {
+					let hangup = match sender {
+						ChtProofSender::Both(ref mut s) => check_hangup(s),
+						ChtProofSender::Header(ref mut s) => check_hangup(s),
+						ChtProofSender::ChainScore(ref mut s) => check_hangup(s),
+					};
+
+					if !hangup { self.dispatch_header_by_number(ctx, req, sender) }
+				}
 				Pending::HeaderByHash(req, mut sender) =>
 					if !check_hangup(&mut sender) {	self.dispatch_header_by_hash(ctx, req, sender) },
 				Pending::Block(req, mut sender) =>
@@ -439,8 +517,19 @@ impl Handler for OnDemand {
 			Pending::HeaderByNumber(req, sender) => {
 				if let Some(&(ref header, ref proof)) = proofs.get(0) {
 					match req.check_response(header, proof) {
-						Ok(header) => {
-							sender.complete(header);
+						Ok((header, score)) => {
+							let mut cache = self.cache.lock();
+							let hash = header.hash();
+							cache.insert_block_header(hash, header.clone());
+							cache.insert_block_hash(header.number(), hash);
+							cache.insert_chain_score(hash, score);
+
+							match sender {
+								ChtProofSender::Both(sender) => sender.complete((header, score)),
+								ChtProofSender::Header(sender) => sender.complete(header),
+								ChtProofSender::ChainScore(sender) => sender.complete(score),
+							}
+
 							return
 						}
 						Err(e) => {
@@ -468,6 +557,7 @@ impl Handler for OnDemand {
 				if let Some(ref header) = headers.get(0) {
 					match req.check_response(header) {
 						Ok(header) => {
+							self.cache.lock().insert_block_header(req.0, header.clone());
 							sender.complete(header);
 							return
 						}
@@ -493,9 +583,11 @@ impl Handler for OnDemand {
 
 		match req {
 			Pending::Block(req, sender) => {
-				if let Some(ref block) = bodies.get(0) {
-					match req.check_response(block) {
+				if let Some(ref body) = bodies.get(0) {
+					match req.check_response(body) {
 						Ok(block) => {
+							let body = encoded::Body::new(body.to_vec());
+							self.cache.lock().insert_block_body(req.hash, body);
 							sender.complete(block);
 							return
 						}
@@ -524,6 +616,8 @@ impl Handler for OnDemand {
 				if let Some(ref receipts) = receipts.get(0) {
 					match req.check_response(receipts) {
 						Ok(receipts) => {
+							let hash = req.0.hash();
+							self.cache.lock().insert_block_receipts(hash, receipts.clone());
 							sender.complete(receipts);
 							return
 						}
@@ -604,10 +698,16 @@ impl Handler for OnDemand {
 #[cfg(test)]
 mod tests {
 	use super::*;
+
+	use std::sync::Arc;
+
+	use cache::Cache;
 	use net::{Announcement, BasicContext, ReqId, Error as LesError};
 	use request::{Request as LesRequest, Kind as LesRequestKind};
+
 	use network::{PeerId, NodeId};
-	use util::H256;
+	use time::Duration;
+	use util::{H256, Mutex};
 
 	struct FakeContext;
 
@@ -624,7 +724,8 @@ mod tests {
 
 	#[test]
 	fn detects_hangup() {
-		let on_demand = OnDemand::default();
+		let cache = Arc::new(Mutex::new(Cache::new(Default::default(), Duration::hours(6))));
+		let on_demand = OnDemand::new(cache);
 		let result = on_demand.header_by_hash(&FakeContext, request::HeaderByHash(H256::default()));
 
 		assert!(on_demand.orphaned_requests.read().len() == 1);

ethcore/light/src/transaction_queue.rs

@@ -245,6 +245,31 @@ impl TransactionQueue {
 			.collect()
 	}
 
+	/// Get all transactions not ready to be propagated.
+	/// `best_block_number` and `best_block_timestamp` are used to filter out conditionally
+	/// propagated transactions.
+	///
+	/// Returned transactions are batched by sender, in order of ascending nonce.
+	pub fn future_transactions(&self, best_block_number: u64, best_block_timestamp: u64) -> Vec<PendingTransaction> {
+		self.by_account.values()
+			.flat_map(|acct_txs| {
+				acct_txs.current.iter().skip_while(|tx| match tx.condition {
+					None => true,
+					Some(Condition::Number(blk_num)) => blk_num <= best_block_number,
+					Some(Condition::Timestamp(time)) => time <= best_block_timestamp,
+				}).chain(acct_txs.future.values()).map(|info| info.hash)
+			})
+			.filter_map(|hash| match self.by_hash.get(&hash) {
+				Some(tx) => Some(tx.clone()),
+				None => {
+					warn!(target: "txqueue", "Inconsistency detected between `by_hash` and `by_account`: {} not stored.",
+						hash);
+					None
+				}
+			})
+			.collect()
+	}
+
 	/// Addresses for which we store transactions.
 	pub fn queued_senders(&self) -> Vec<Address> {
 		self.by_account.keys().cloned().collect()
@@ -471,4 +496,22 @@ mod tests {
 
 		assert!(txq.transaction(&hash).is_none());
 	}
+
+	#[test]
+	fn future_transactions() {
+		let sender = Address::default();
+		let mut txq = TransactionQueue::default();
+
+		for i in (0..1).chain(3..10) {
+			let mut tx = Transaction::default();
+			tx.nonce = i.into();
+
+			let tx = tx.fake_sign(sender);
+
+			txq.import(tx.into()).unwrap();
+		}
+
+		assert_eq!(txq.future_transactions(0, 0).len(), 7);
+		assert_eq!(txq.next_nonce(&sender).unwrap(), 1.into());
+	}
 }

ethcore/src/client/traits.rs

@@ -16,7 +16,6 @@
 
 use std::collections::BTreeMap;
 use util::{U256, Address, H256, H2048, Bytes, Itertools};
-use util::stats::Histogram;
 use blockchain::TreeRoute;
 use verification::queue::QueueInfo as BlockQueueInfo;
 use block::{OpenBlock, SealedBlock};
@@ -212,38 +211,24 @@ pub trait BlockChainClient : Sync + Send {
 	fn ready_transactions(&self) -> Vec<PendingTransaction>;
 
 	/// Sorted list of transaction gas prices from at least last sample_size blocks.
-	fn gas_price_corpus(&self, sample_size: usize) -> Vec<U256> {
+	fn gas_price_corpus(&self, sample_size: usize) -> ::stats::Corpus<U256> {
 		let mut h = self.chain_info().best_block_hash;
 		let mut corpus = Vec::new();
 		while corpus.is_empty() {
 			for _ in 0..sample_size {
-				let block = self.block(BlockId::Hash(h)).expect("h is either the best_block_hash or an ancestor; qed");
-				let header = block.header_view();
-				if header.number() == 0 {
-					corpus.sort();
-					return corpus;
+				let block = match self.block(BlockId::Hash(h)) {
+					Some(block) => block,
+					None => return corpus.into(),
+				};
+
+				if block.number() == 0 {
+					return corpus.into();
 				}
 				block.transaction_views().iter().foreach(|t| corpus.push(t.gas_price()));
-				h = header.parent_hash().clone();
+				h = block.parent_hash().clone();
 			}
 		}
-		corpus.sort();
-		corpus
-	}
-
-	/// Calculate median gas price from recent blocks if they have any transactions.
-	fn gas_price_median(&self, sample_size: usize) -> Option<U256> {
-		let corpus = self.gas_price_corpus(sample_size);
-		corpus.get(corpus.len() / 2).cloned()
-	}
-
-	/// Get the gas price distribution based on recent blocks if they have any transactions.
-	fn gas_price_histogram(&self, sample_size: usize, bucket_number: usize) -> Option<Histogram> {
-		let raw_corpus = self.gas_price_corpus(sample_size);
-		let raw_len = raw_corpus.len();
-		// Throw out outliers.
-		let (corpus, _) = raw_corpus.split_at(raw_len - raw_len / 40);
-		Histogram::new(corpus, bucket_number)
+		corpus.into()
 	}
 
 	/// Get the preferred network ID to sign on

ethcore/src/lib.rs

@@ -106,6 +106,7 @@ extern crate lru_cache;
 extern crate ethcore_stratum;
 extern crate ethabi;
 extern crate hardware_wallet;
+extern crate stats;
 
 #[macro_use]
 extern crate log;

ethcore/src/tests/client.rs

@@ -27,7 +27,6 @@ use miner::Miner;
 use rlp::View;
 use spec::Spec;
 use views::BlockView;
-use util::stats::Histogram;
 use ethkey::{KeyPair, Secret};
 use transaction::{PendingTransaction, Transaction, Action, Condition};
 use miner::MinerService;
@@ -208,11 +207,11 @@ fn can_collect_garbage() {
 fn can_generate_gas_price_median() {
 	let client_result = generate_dummy_client_with_data(3, 1, slice_into![1, 2, 3]);
 	let client = client_result.reference();
-	assert_eq!(Some(U256::from(2)), client.gas_price_median(3));
+	assert_eq!(Some(&U256::from(2)), client.gas_price_corpus(3).median());
 
 	let client_result = generate_dummy_client_with_data(4, 1, slice_into![1, 4, 3, 2]);
 	let client = client_result.reference();
-	assert_eq!(Some(U256::from(3)), client.gas_price_median(4));
+	assert_eq!(Some(&U256::from(3)), client.gas_price_corpus(3).median());
 }
 
 #[test]
@@ -220,8 +219,8 @@ fn can_generate_gas_price_histogram() {
 	let client_result = generate_dummy_client_with_data(20, 1, slice_into![6354,8593,6065,4842,7845,7002,689,4958,4250,6098,5804,4320,643,8895,2296,8589,7145,2000,2512,1408]);
 	let client = client_result.reference();
 
-	let hist = client.gas_price_histogram(20, 5).unwrap();
-	let correct_hist = Histogram { bucket_bounds: vec_into![643, 2294, 3945, 5596, 7247, 8898], counts: vec![4,2,4,6,4] };
+	let hist = client.gas_price_corpus(20).histogram(5).unwrap();
+	let correct_hist = ::stats::Histogram { bucket_bounds: vec_into![643, 2294, 3945, 5596, 7247, 8898], counts: vec![4,2,4,6,4] };
 	assert_eq!(hist, correct_hist);
 }
 
@@ -230,7 +229,7 @@ fn empty_gas_price_histogram() {
 	let client_result = generate_dummy_client_with_data(20, 0, slice_into![]);
 	let client = client_result.reference();
 
-	assert!(client.gas_price_histogram(20, 5).is_none());
+	assert!(client.gas_price_corpus(20).histogram(5).is_none());
 }
 
 #[test]