Remove locks from the block chain

Cargo.lock

@@ -151,6 +151,7 @@ dependencies = [
 [[package]]
 name = "eth-secp256k1"
 version = "0.5.4"
+source = "git+https://github.com/arkpar/rust-secp256k1.git#45503e1de68d909b1862e3f2bdb9e1cdfdff3f1e"
 dependencies = [
  "arrayvec 0.3.15 (registry+https://github.com/rust-lang/crates.io-index)",
  "gcc 0.3.24 (registry+https://github.com/rust-lang/crates.io-index)",
@@ -222,7 +223,7 @@ dependencies = [
  "crossbeam 0.2.8 (registry+https://github.com/rust-lang/crates.io-index)",
  "elastic-array 0.4.0 (registry+https://github.com/rust-lang/crates.io-index)",
  "env_logger 0.3.2 (registry+https://github.com/rust-lang/crates.io-index)",
- "eth-secp256k1 0.5.4",
+ "eth-secp256k1 0.5.4 (git+https://github.com/arkpar/rust-secp256k1.git)",
  "ethcore-devtools 0.9.99",
  "heapsize 0.3.2 (registry+https://github.com/rust-lang/crates.io-index)",
  "igd 0.4.2 (registry+https://github.com/rust-lang/crates.io-index)",

Cargo.toml

@@ -32,4 +32,4 @@ path = "parity/main.rs"
 name = "parity"
 
 [profile.release]
-debug = true
+debug = false

ethcore/src/block_queue.rs

@@ -99,6 +99,7 @@ impl QueueSignal {
 }
 
 struct Verification {
+	// All locks must be captured in the order declared here.
 	unverified: Mutex<VecDeque<UnVerifiedBlock>>,
 	verified: Mutex<VecDeque<PreVerifiedBlock>>,
 	verifying: Mutex<VecDeque<VerifyingBlock>>,
@@ -123,7 +124,7 @@ impl BlockQueue {
 		let panic_handler = PanicHandler::new_in_arc();
 
 		let mut verifiers: Vec<JoinHandle<()>> = Vec::new();
-		let thread_count = max(::num_cpus::get(), 5) - 0;
+		let thread_count = max(::num_cpus::get(), 3) - 2;
 		for i in 0..thread_count {
 			let verification = verification.clone();
 			let engine = engine.clone();
@@ -137,7 +138,6 @@ impl BlockQueue {
 				.name(format!("Verifier #{}", i))
 				.spawn(move || {
 					panic_handler.catch_panic(move || {
-						lower_thread_priority();
 						BlockQueue::verify(verification, engine, more_to_verify, ready_signal, deleting, empty)
 					}).unwrap()
 				})
@@ -392,7 +392,7 @@ mod tests {
 
 	#[test]
 	fn can_import_blocks() {
-		let mut queue = get_test_queue();
+		let queue = get_test_queue();
 		if let Err(e) = queue.import_block(get_good_dummy_block()) {
 			panic!("error importing block that is valid by definition({:?})", e);
 		}
@@ -400,7 +400,7 @@ mod tests {
 
 	#[test]
 	fn returns_error_for_duplicates() {
-		let mut queue = get_test_queue();
+		let queue = get_test_queue();
 		if let Err(e) = queue.import_block(get_good_dummy_block()) {
 			panic!("error importing block that is valid by definition({:?})", e);
 		}
@@ -419,7 +419,7 @@ mod tests {
 
 	#[test]
 	fn returns_ok_for_drained_duplicates() {
-		let mut queue = get_test_queue();
+		let queue = get_test_queue();
 		let block = get_good_dummy_block();
 		let hash = BlockView::new(&block).header().hash().clone();
 		if let Err(e) = queue.import_block(block) {
@@ -436,7 +436,7 @@ mod tests {
 
 	#[test]
 	fn returns_empty_once_finished() {
-		let mut queue = get_test_queue();
+		let queue = get_test_queue();
 		queue.import_block(get_good_dummy_block()).expect("error importing block that is valid by definition");
 		queue.flush();
 		queue.drain(1);

ethcore/src/blockchain.rs

@@ -16,6 +16,7 @@
 
 //! Blockchain database.
 
+use std::sync::atomic::{AtomicUsize, Ordering as AtomicOrder};
 use util::*;
 use rocksdb::{DB, WriteBatch, Writable};
 use header::*;
@@ -147,8 +148,9 @@ struct CacheManager {
 ///
 /// **Does not do input data verification.**
 pub struct BlockChain {
-	pref_cache_size: usize,
+	// All locks must be captured in the order declared here.
-	max_cache_size: usize,
+	pref_cache_size: AtomicUsize,
+	max_cache_size: AtomicUsize,
 
 	best_block: RwLock<BestBlock>,
 
@@ -166,6 +168,7 @@ pub struct BlockChain {
 	blocks_db: DB,
 
 	cache_man: RwLock<CacheManager>,
+	insert_lock: Mutex<()>
 }
 
 impl BlockProvider for BlockChain {
@@ -261,8 +264,8 @@ impl BlockChain {
 		(0..COLLECTION_QUEUE_SIZE).foreach(|_| cache_man.cache_usage.push_back(HashSet::new()));
 
 		let bc = BlockChain {
-			pref_cache_size: 1 << 14,
+			pref_cache_size: AtomicUsize::new(1 << 14),
-			max_cache_size: 1 << 20,
+			max_cache_size: AtomicUsize::new(1 << 20),
 			best_block: RwLock::new(BestBlock::new()),
 			blocks: RwLock::new(HashMap::new()),
 			block_details: RwLock::new(HashMap::new()),
@@ -273,6 +276,7 @@ impl BlockChain {
 			extras_db: extras_db,
 			blocks_db: blocks_db,
 			cache_man: RwLock::new(cache_man),
+			insert_lock: Mutex::new(()),
 		};
 
 		// load best block
@@ -315,9 +319,9 @@ impl BlockChain {
 	}
 
 	/// Set the cache configuration.
-	pub fn configure_cache(&mut self, pref_cache_size: usize, max_cache_size: usize) {
+	pub fn configure_cache(&self, pref_cache_size: usize, max_cache_size: usize) {
-		self.pref_cache_size = pref_cache_size;
+		self.pref_cache_size.store(pref_cache_size, AtomicOrder::Relaxed);
-		self.max_cache_size = max_cache_size;
+		self.max_cache_size.store(max_cache_size, AtomicOrder::Relaxed);
 	}
 
 	/// Returns a tree route between `from` and `to`, which is a tuple of:
@@ -435,22 +439,26 @@ impl BlockChain {
 			return;
 		}
 
+		let _lock = self.insert_lock.lock();
 		// store block in db
 		self.blocks_db.put(&hash, &bytes).unwrap();
 		let (batch, new_best, details) = self.block_to_extras_insert_batch(bytes);
 
-		// update best block
+		{
-		let mut best_block = self.best_block.write().unwrap();
+			// update best block
-		if let Some(b) = new_best {
+			let mut best_block = self.best_block.write().unwrap();
-			*best_block = b;
+			if let Some(b) = new_best {
+				*best_block = b;
+			}
 		}
 
-		// update caches
+		{
-		let mut write = self.block_details.write().unwrap();
+			// update caches
-		write.remove(&header.parent_hash());
+			let mut write = self.block_details.write().unwrap();
-		write.insert(hash.clone(), details);
+			write.remove(&header.parent_hash());
-		self.note_used(CacheID::Block(hash));
+			write.insert(hash.clone(), details);
-
+			self.note_used(CacheID::Block(hash));
+		}
 		// update extras database
 		self.extras_db.write(batch).unwrap();
 	}
@@ -622,17 +630,17 @@ impl BlockChain {
 
 	/// Ticks our cache system and throws out any old data.
 	pub fn collect_garbage(&self) {
-		if self.cache_size().total() < self.pref_cache_size { return; }
+		if self.cache_size().total() < self.pref_cache_size.load(AtomicOrder::Relaxed) { return; }
 
 		for _ in 0..COLLECTION_QUEUE_SIZE {
 			{
-				let mut cache_man = self.cache_man.write().unwrap();
 				let mut blocks = self.blocks.write().unwrap();
 				let mut block_details = self.block_details.write().unwrap();
 				let mut block_hashes = self.block_hashes.write().unwrap();
 				let mut transaction_addresses = self.transaction_addresses.write().unwrap();
 				let mut block_logs = self.block_logs.write().unwrap();
 				let mut blocks_blooms = self.blocks_blooms.write().unwrap();
+				let mut cache_man = self.cache_man.write().unwrap();
 
 				for id in cache_man.cache_usage.pop_back().unwrap().into_iter() {
 					cache_man.in_use.remove(&id);
@@ -650,7 +658,7 @@ impl BlockChain {
 				// TODO: handle block_hashes properly.
 				block_hashes.clear();
 			}
-			if self.cache_size().total() < self.max_cache_size { break; }
+			if self.cache_size().total() < self.max_cache_size.load(AtomicOrder::Relaxed) { break; }
 		}
 
 		// TODO: m_lastCollection = chrono::system_clock::now();

ethcore/src/client.rs

@@ -169,7 +169,7 @@ impl ClientReport {
 /// Blockchain database client backed by a persistent database. Owns and manages a blockchain and a block queue.
 /// Call `import_block()` to import a block asynchronously; `flush_queue()` flushes the queue.
 pub struct Client {
-	chain: Arc<RwLock<BlockChain>>,
+	chain: Arc<BlockChain>,
 	engine: Arc<Box<Engine>>,
 	state_db: Mutex<JournalDB>,
 	block_queue: BlockQueue,
@@ -190,7 +190,7 @@ impl Client {
 		dir.push(format!("v{}-sec-pruned", CLIENT_DB_VER_STR));
 		let path = dir.as_path();
 		let gb = spec.genesis_block();
-		let chain = Arc::new(RwLock::new(BlockChain::new(&gb, path)));
+		let chain = Arc::new(BlockChain::new(&gb, path));
 		let mut opts = Options::new();
 		opts.set_max_open_files(256);
 		opts.create_if_missing(true);
@@ -258,13 +258,13 @@ impl Client {
 			}
 
 			let header = &block.header;
-			if let Err(e) = verify_block_family(&header, &block.bytes, self.engine.deref().deref(), self.chain.read().unwrap().deref()) {
+			if let Err(e) = verify_block_family(&header, &block.bytes, self.engine.deref().deref(), self.chain.deref()) {
 				warn!(target: "client", "Stage 3 block verification failed for #{} ({})\nError: {:?}", header.number(), header.hash(), e);
 				self.block_queue.mark_as_bad(&header.hash());
 				bad.insert(block.header.hash());
 				break;
 			};
-			let parent = match self.chain.read().unwrap().block_header(&header.parent_hash) {
+			let parent = match self.chain.block_header(&header.parent_hash) {
 				Some(p) => p,
 				None => {
 					warn!(target: "client", "Block import failed for #{} ({}): Parent not found ({}) ", header.number(), header.hash(), header.parent_hash);
@@ -278,7 +278,7 @@ impl Client {
 			last_hashes.resize(256, H256::new());
 			last_hashes[0] = header.parent_hash.clone();
 			for i in 0..255 {
-				match self.chain.read().unwrap().block_details(&last_hashes[i]) {
+				match self.chain.block_details(&last_hashes[i]) {
 					Some(details) => {
 						last_hashes[i + 1] = details.parent.clone();
 					},
@@ -304,9 +304,9 @@ impl Client {
 
 			good_blocks.push(header.hash().clone());
 
-			self.chain.write().unwrap().insert_block(&block.bytes); //TODO: err here?
+			self.chain.insert_block(&block.bytes); //TODO: err here?
 			let ancient = if header.number() >= HISTORY { Some(header.number() - HISTORY) } else { None };
-			match result.drain().commit(header.number(), &header.hash(), ancient.map(|n|(n, self.chain.read().unwrap().block_hash(n).unwrap()))) {
+			match result.drain().commit(header.number(), &header.hash(), ancient.map(|n|(n, self.chain.block_hash(n).unwrap()))) {
 				Ok(_) => (),
 				Err(e) => {
 					warn!(target: "client", "State DB commit failed: {:?}", e);
@@ -331,7 +331,7 @@ impl Client {
 
 	/// Get info on the cache.
 	pub fn cache_info(&self) -> CacheSize {
-		self.chain.read().unwrap().cache_size()
+		self.chain.cache_size()
 	}
 
 	/// Get the report.
@@ -341,12 +341,12 @@ impl Client {
 
 	/// Tick the client.
 	pub fn tick(&self) {
-		self.chain.read().unwrap().collect_garbage();
+		self.chain.collect_garbage();
 	}
 
 	/// Set up the cache behaviour.
 	pub fn configure_cache(&self, pref_cache_size: usize, max_cache_size: usize) {
-		self.chain.write().unwrap().configure_cache(pref_cache_size, max_cache_size);
+		self.chain.configure_cache(pref_cache_size, max_cache_size);
 	}
 
 	fn block_hash(chain: &BlockChain, id: BlockId) -> Option<H256> {
@@ -361,14 +361,12 @@ impl Client {
 
 impl BlockChainClient for Client {
 	fn block_header(&self, id: BlockId) -> Option<Bytes> {
-		let chain = self.chain.read().unwrap();
+		Self::block_hash(&self.chain, id).and_then(|hash| self.chain.block(&hash).map(|bytes| BlockView::new(&bytes).rlp().at(0).as_raw().to_vec()))
-		Self::block_hash(&chain, id).and_then(|hash| chain.block(&hash).map(|bytes| BlockView::new(&bytes).rlp().at(0).as_raw().to_vec()))
 	}
 
 	fn block_body(&self, id: BlockId) -> Option<Bytes> {
-		let chain = self.chain.read().unwrap();
+		Self::block_hash(&self.chain, id).and_then(|hash| {
-		Self::block_hash(&chain, id).and_then(|hash| {
+			self.chain.block(&hash).map(|bytes| {
-			chain.block(&hash).map(|bytes| {
 				let rlp = Rlp::new(&bytes);
 				let mut body = RlpStream::new_list(2);
 				body.append_raw(rlp.at(1).as_raw(), 1);
@@ -379,24 +377,21 @@ impl BlockChainClient for Client {
 	}
 
 	fn block(&self, id: BlockId) -> Option<Bytes> {
-		let chain = self.chain.read().unwrap();
+		Self::block_hash(&self.chain, id).and_then(|hash| {
-		Self::block_hash(&chain, id).and_then(|hash| {
+			self.chain.block(&hash)
-			chain.block(&hash)
 		})
 	}
 
 	fn block_status(&self, id: BlockId) -> BlockStatus {
-		let chain = self.chain.read().unwrap();
+		match Self::block_hash(&self.chain, id) {
-		match Self::block_hash(&chain, id) {
+			Some(ref hash) if self.chain.is_known(hash) => BlockStatus::InChain,
-			Some(ref hash) if chain.is_known(hash) => BlockStatus::InChain,
 			Some(hash) => self.block_queue.block_status(&hash),
 			None => BlockStatus::Unknown
 		}
 	}
 	
 	fn block_total_difficulty(&self, id: BlockId) -> Option<U256> {
-		let chain = self.chain.read().unwrap();
+		Self::block_hash(&self.chain, id).and_then(|hash| self.chain.block_details(&hash)).map(|d| d.total_difficulty)
-		Self::block_hash(&chain, id).and_then(|hash| chain.block_details(&hash)).map(|d| d.total_difficulty)
 	}
 
 	fn code(&self, address: &Address) -> Option<Bytes> {
@@ -404,18 +399,17 @@ impl BlockChainClient for Client {
 	}
 
 	fn transaction(&self, id: TransactionId) -> Option<LocalizedTransaction> {
-		let chain = self.chain.read().unwrap();
 		match id {
-			TransactionId::Hash(ref hash) => chain.transaction_address(hash),
+			TransactionId::Hash(ref hash) => self.chain.transaction_address(hash),
-			TransactionId::Location(id, index) => Self::block_hash(&chain, id).map(|hash| TransactionAddress {
+			TransactionId::Location(id, index) => Self::block_hash(&self.chain, id).map(|hash| TransactionAddress {
 				block_hash: hash,
 				index: index
 			})
-		}.and_then(|address| chain.transaction(&address))
+		}.and_then(|address| self.chain.transaction(&address))
 	}
 
 	fn tree_route(&self, from: &H256, to: &H256) -> Option<TreeRoute> {
-		self.chain.read().unwrap().tree_route(from.clone(), to.clone())
+		self.chain.tree_route(from.clone(), to.clone())
 	}
 
 	fn state_data(&self, _hash: &H256) -> Option<Bytes> {
@@ -428,7 +422,7 @@ impl BlockChainClient for Client {
 
 	fn import_block(&self, bytes: Bytes) -> ImportResult {
 		let header = BlockView::new(&bytes).header();
-		if self.chain.read().unwrap().is_known(&header.hash()) {
+		if self.chain.is_known(&header.hash()) {
 			return Err(ImportError::AlreadyInChain);
 		}
 		if self.block_status(BlockId::Hash(header.parent_hash)) == BlockStatus::Unknown {
@@ -446,13 +440,12 @@ impl BlockChainClient for Client {
 	}
 
 	fn chain_info(&self) -> BlockChainInfo {
-		let chain = self.chain.read().unwrap();
 		BlockChainInfo {
-			total_difficulty: chain.best_block_total_difficulty(),
+			total_difficulty: self.chain.best_block_total_difficulty(),
-			pending_total_difficulty: chain.best_block_total_difficulty(),
+			pending_total_difficulty: self.chain.best_block_total_difficulty(),
-			genesis_hash: chain.genesis_hash(),
+			genesis_hash: self.chain.genesis_hash(),
-			best_block_hash: chain.best_block_hash(),
+			best_block_hash: self.chain.best_block_hash(),
-			best_block_number: From::from(chain.best_block_number())
+			best_block_number: From::from(self.chain.best_block_number())
 		}
 	}
 }

ethcore/src/verification.rs

@@ -57,12 +57,18 @@ pub fn verify_block_basic(header: &Header, bytes: &[u8], engine: &Engine) -> Res
 /// Still operates on a individual block
 /// Returns a PreVerifiedBlock structure populated with transactions
 pub fn verify_block_unordered(header: Header, bytes: Bytes, engine: &Engine) -> Result<PreVerifiedBlock, Error> {
+	try!(engine.verify_block_unordered(&header, Some(&bytes)));
+	for u in Rlp::new(&bytes).at(2).iter().map(|rlp| rlp.as_val::<Header>()) {
+		try!(engine.verify_block_unordered(&u, None));
+	}
 	// Verify transactions. 
 	let mut transactions = Vec::new();
-	let v = BlockView::new(&bytes);
+	{
-	for t in v.transactions() {
+		let v = BlockView::new(&bytes);
-		try!(engine.verify_transaction(&t, &header));
+		for t in v.transactions() {
-		transactions.push(t);
+			try!(engine.verify_transaction(&t, &header));
+			transactions.push(t);
+		}
 	}
 	Ok(PreVerifiedBlock {
 		header: header,

util/src/lib.rs

@@ -143,7 +143,6 @@ pub mod network;
 pub mod log;
 pub mod panics;
 pub mod keys;
-mod thread;
 
 pub use common::*;
 pub use misc::*;
@@ -164,5 +163,4 @@ pub use semantic_version::*;
 pub use network::*;
 pub use io::*;
 pub use log::*;
-pub use thread::*;
 

util/src/thread.rs

@@ -1,43 +0,0 @@
-// 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/>.
-
-//! Thread management helpers
-
-use libc::{c_int, pthread_self, pthread_t};
-
-#[repr(C)]
-struct sched_param {
-    priority: c_int,
-    padding: c_int,
-}
-
-extern {
-	fn setpriority(which: c_int, who: c_int, prio: c_int) -> c_int;
-	fn pthread_setschedparam(thread: pthread_t, policy: c_int, param: *const sched_param) -> c_int;
-}
-const PRIO_DARWIN_THREAD: c_int = 3;
-const PRIO_DARWIN_BG: c_int = 0x1000;
-const SCHED_RR: c_int = 2;
-
-/// Lower thread priority and put it into background mode
-#[cfg(target_os="macos")]
-pub fn lower_thread_priority() {
-	let sp = sched_param { priority: 0, padding: 0 };
-	if unsafe { pthread_setschedparam(pthread_self(), SCHED_RR, &sp) } == -1 {
-		trace!("Could not decrease thread piority");
-	}
-	//unsafe { setpriority(PRIO_DARWIN_THREAD, 0, PRIO_DARWIN_BG); }
-}