Fix slow balances

ethcore/Cargo.toml

@@ -49,6 +49,7 @@ num = "0.1"
 num_cpus = "1.2"
 parking_lot = "0.4"
 price-info = { path = "../price-info" }
+rayon = "0.8"
 rand = "0.3"
 rlp = { path = "../util/rlp" }
 rlp_derive = { path = "../util/rlp_derive" }

ethcore/src/blockchain/blockchain.rs

@@ -44,6 +44,7 @@ use db::{self, Writable, Readable, CacheUpdatePolicy};
 use cache_manager::CacheManager;
 use encoded;
 use engines::epoch::{Transition as EpochTransition, PendingTransition as PendingEpochTransition};
+use rayon::prelude::*;
 use ansi_term::Colour;
 
 const LOG_BLOOMS_LEVELS: usize = 3;
@@ -152,7 +153,7 @@ pub trait BlockProvider {
 
 	/// Returns logs matching given filter.
 	fn logs<F>(&self, blocks: Vec<BlockNumber>, matches: F, limit: Option<usize>) -> Vec<LocalizedLogEntry>
-		where F: Fn(&LogEntry) -> bool, Self: Sized;
+		where F: Fn(&LogEntry) -> bool + Send + Sync, Self: Sized;
 }
 
 macro_rules! otry {
@@ -363,50 +364,56 @@ impl BlockProvider for BlockChain {
 	}
 
 	fn logs<F>(&self, mut blocks: Vec<BlockNumber>, matches: F, limit: Option<usize>) -> Vec<LocalizedLogEntry>
-		where F: Fn(&LogEntry) -> bool, Self: Sized {
+		where F: Fn(&LogEntry) -> bool + Send + Sync, Self: Sized {
 		// sort in reverse order
 		blocks.sort_by(|a, b| b.cmp(a));
 
-		let mut log_index = 0;
-		let mut logs = blocks.into_iter()
-			.filter_map(|number| self.block_hash(number).map(|hash| (number, hash)))
-			.filter_map(|(number, hash)| self.block_receipts(&hash).map(|r| (number, hash, r.receipts)))
-			.filter_map(|(number, hash, receipts)| self.block_body(&hash).map(|ref b| (number, hash, receipts, b.transaction_hashes())))
-			.flat_map(|(number, hash, mut receipts, mut hashes)| {
-				if receipts.len() != hashes.len() {
-					warn!("Block {} ({}) has different number of receipts ({}) to transactions ({}). Database corrupt?", number, hash, receipts.len(), hashes.len());
-					assert!(false);
-				}
-				log_index = receipts.iter().fold(0, |sum, receipt| sum + receipt.logs.len());
-
-				let receipts_len = receipts.len();
-				hashes.reverse();
-				receipts.reverse();
-				receipts.into_iter()
-					.map(|receipt| receipt.logs)
-					.zip(hashes)
-					.enumerate()
-					.flat_map(move |(index, (mut logs, tx_hash))| {
-						let current_log_index = log_index;
-						let no_of_logs = logs.len();
-						log_index -= no_of_logs;
-
-						logs.reverse();
-						logs.into_iter()
+		let mut logs = blocks
+			.chunks(128)
+			.flat_map(move |blocks_chunk| {
+				blocks_chunk.into_par_iter()
+					.filter_map(|number| self.block_hash(*number).map(|hash| (*number, hash)))
+					.filter_map(|(number, hash)| self.block_receipts(&hash).map(|r| (number, hash, r.receipts)))
+					.filter_map(|(number, hash, receipts)| self.block_body(&hash).map(|ref b| (number, hash, receipts, b.transaction_hashes())))
+					.flat_map(|(number, hash, mut receipts, mut hashes)| {
+						if receipts.len() != hashes.len() {
+							warn!("Block {} ({}) has different number of receipts ({}) to transactions ({}). Database corrupt?", number, hash, receipts.len(), hashes.len());
+							assert!(false);
+						}
+						let mut log_index = receipts.iter().fold(0, |sum, receipt| sum + receipt.logs.len());
+
+						let receipts_len = receipts.len();
+						hashes.reverse();
+						receipts.reverse();
+						receipts.into_iter()
+							.map(|receipt| receipt.logs)
+							.zip(hashes)
 							.enumerate()
-							.map(move |(i, log)| LocalizedLogEntry {
-								entry: log,
-								block_hash: hash,
-								block_number: number,
-								transaction_hash: tx_hash,
-								// iterating in reverse order
-								transaction_index: receipts_len - index - 1,
-								transaction_log_index: no_of_logs - i - 1,
-								log_index: current_log_index - i - 1,
+							.flat_map(move |(index, (mut logs, tx_hash))| {
+								let current_log_index = log_index;
+								let no_of_logs = logs.len();
+								log_index -= no_of_logs;
+
+								logs.reverse();
+								logs.into_iter()
+									.enumerate()
+									.map(move |(i, log)| LocalizedLogEntry {
+										entry: log,
+										block_hash: hash,
+										block_number: number,
+										transaction_hash: tx_hash,
+										// iterating in reverse order
+										transaction_index: receipts_len - index - 1,
+										transaction_log_index: no_of_logs - i - 1,
+										log_index: current_log_index - i - 1,
+									})
 							})
+							.filter(|log_entry| matches(&log_entry.entry))
+							.take(limit.unwrap_or(::std::usize::MAX))
+							.collect::<Vec<_>>()
 					})
+					.collect::<Vec<_>>()
 			})
-			.filter(|log_entry| matches(&log_entry.entry))
 			.take(limit.unwrap_or(::std::usize::MAX))
 			.collect::<Vec<LocalizedLogEntry>>();
 		logs.reverse();

ethcore/src/client/client.rs

@@ -918,7 +918,7 @@ impl Client {
 			_ => {},
 		}
 
-		let block_number = match self.block_number(id.clone()) {
+		let block_number = match self.block_number(id) {
 			Some(num) => num,
 			None => return None,
 		};
@@ -1155,6 +1155,16 @@ impl Client {
 			(false, false) => call(state, env_info, engine, state_diff, t, TransactOptions::with_no_tracing()),
 		}
 	}
+
+	fn block_number_ref(&self, id: &BlockId) -> Option<BlockNumber> {
+		match *id {
+			BlockId::Number(number) => Some(number),
+			BlockId::Hash(ref hash) => self.chain.read().block_number(hash),
+			BlockId::Earliest => Some(0),
+			BlockId::Latest => Some(self.chain.read().best_block_number()),
+			BlockId::Pending => Some(self.chain.read().best_block_number() + 1),
+		}
+	}
 }
 
 impl snapshot::DatabaseRestore for Client {
@@ -1364,13 +1374,7 @@ impl BlockChainClient for Client {
 	}
 
 	fn block_number(&self, id: BlockId) -> Option<BlockNumber> {
-		match id {
-			BlockId::Number(number) => Some(number),
-			BlockId::Hash(ref hash) => self.chain.read().block_number(hash),
-			BlockId::Earliest => Some(0),
-			BlockId::Latest => Some(self.chain.read().best_block_number()),
-			BlockId::Pending => Some(self.chain.read().best_block_number() + 1),
-		}
+		self.block_number_ref(&id)
 	}
 
 	fn block_body(&self, id: BlockId) -> Option<encoded::Body> {
@@ -1651,16 +1655,17 @@ impl BlockChainClient for Client {
 		self.engine.additional_params().into_iter().collect()
 	}
 
-	fn blocks_with_bloom(&self, bloom: &H2048, from_block: BlockId, to_block: BlockId) -> Option<Vec<BlockNumber>> {
-		match (self.block_number(from_block), self.block_number(to_block)) {
-			(Some(from), Some(to)) => Some(self.chain.read().blocks_with_bloom(bloom, from, to)),
-			_ => None
-		}
-	}
-
 	fn logs(&self, filter: Filter) -> Vec<LocalizedLogEntry> {
+		let (from, to) = match (self.block_number_ref(&filter.from_block), self.block_number_ref(&filter.to_block)) {
+			(Some(from), Some(to)) => (from, to),
+			_ => return Vec::new(),
+		};
+
+		let chain = self.chain.read();
 		let blocks = filter.bloom_possibilities().iter()
-			.filter_map(|bloom| self.blocks_with_bloom(bloom, filter.from_block.clone(), filter.to_block.clone()))
+			.map(move |bloom| {
+				chain.blocks_with_bloom(bloom, from, to)
+			})
 			.flat_map(|m| m)
 			// remove duplicate elements
 			.collect::<HashSet<u64>>()

ethcore/src/client/test_client.rs

@@ -24,7 +24,7 @@ use itertools::Itertools;
 use rustc_hex::FromHex;
 use hash::keccak;
 use bigint::prelude::U256;
-use bigint::hash::{H256, H2048};
+use bigint::hash::H256;
 use parking_lot::RwLock;
 use util::*;
 use rlp::*;
@@ -508,10 +508,6 @@ impl BlockChainClient for TestBlockChainClient {
 		self.receipts.read().get(&id).cloned()
 	}
 
-	fn blocks_with_bloom(&self, _bloom: &H2048, _from_block: BlockId, _to_block: BlockId) -> Option<Vec<BlockNumber>> {
-		unimplemented!();
-	}
-
 	fn logs(&self, filter: Filter) -> Vec<LocalizedLogEntry> {
 		let mut logs = self.logs.read().clone();
 		let len = logs.len();

ethcore/src/client/traits.rs

@@ -35,7 +35,7 @@ use transaction::{LocalizedTransaction, PendingTransaction, SignedTransaction};
 use verification::queue::QueueInfo as BlockQueueInfo;
 
 use bigint::prelude::U256;
-use bigint::hash::{H256, H2048};
+use bigint::hash::H256;
 use util::{Address, Bytes};
 use util::hashdb::DBValue;
 
@@ -181,9 +181,6 @@ pub trait BlockChainClient : Sync + Send {
 	/// Get the best block header.
 	fn best_block_header(&self) -> encoded::Header;
 
-	/// Returns numbers of blocks containing given bloom.
-	fn blocks_with_bloom(&self, bloom: &H2048, from_block: BlockId, to_block: BlockId) -> Option<Vec<BlockNumber>>;
-
 	/// Returns logs matching given filter.
 	fn logs(&self, filter: Filter) -> Vec<LocalizedLogEntry>;
 

ethcore/src/lib.rs

@@ -102,6 +102,7 @@ extern crate num;
 extern crate parking_lot;
 extern crate price_info;
 extern crate rand;
+extern crate rayon;
 extern crate rlp;
 extern crate hash;
 extern crate heapsize;

ethcore/src/verification/verification.rs

@@ -392,15 +392,14 @@ mod tests {
 			self.numbers.get(&index).cloned()
 		}
 
-		fn blocks_with_bloom(&self, _bloom: &H2048, _from_block: BlockNumber, _to_block: BlockNumber) -> Vec<BlockNumber> {
+		fn block_receipts(&self, _hash: &H256) -> Option<BlockReceipts> {
 			unimplemented!()
 		}
 
-		fn block_receipts(&self, _hash: &H256) -> Option<BlockReceipts> {
+		fn blocks_with_bloom(&self, _bloom: &H2048, _from_block: BlockNumber, _to_block: BlockNumber) -> Vec<BlockNumber> {
 			unimplemented!()
 		}
 
-
 		fn logs<F>(&self, _blocks: Vec<BlockNumber>, _matches: F, _limit: Option<usize>) -> Vec<LocalizedLogEntry>
 			where F: Fn(&LogEntry) -> bool, Self: Sized {
 			unimplemented!()