diff --git a/Cargo.lock b/Cargo.lock index 50274857f..cf747f3cc 100644 --- a/Cargo.lock +++ b/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)", diff --git a/Cargo.toml b/Cargo.toml index f28829180..8bc94a3a3 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -32,4 +32,4 @@ path = "parity/main.rs" name = "parity" [profile.release] -debug = true +debug = false diff --git a/ethcore/src/block_queue.rs b/ethcore/src/block_queue.rs index a51a1e900..ba9867966 100644 --- a/ethcore/src/block_queue.rs +++ b/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>, verified: Mutex>, verifying: Mutex>, @@ -123,7 +124,7 @@ impl BlockQueue { let panic_handler = PanicHandler::new_in_arc(); let mut verifiers: Vec> = 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); diff --git a/ethcore/src/blockchain.rs b/ethcore/src/blockchain.rs index cc9ff56fd..22d409e8e 100644 --- a/ethcore/src/blockchain.rs +++ b/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, - max_cache_size: usize, + // All locks must be captured in the order declared here. + pref_cache_size: AtomicUsize, + max_cache_size: AtomicUsize, best_block: RwLock, @@ -166,6 +168,7 @@ pub struct BlockChain { blocks_db: DB, cache_man: RwLock, + 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, - max_cache_size: 1 << 20, + pref_cache_size: AtomicUsize::new(1 << 14), + 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) { - self.pref_cache_size = pref_cache_size; - self.max_cache_size = max_cache_size; + pub fn configure_cache(&self, pref_cache_size: usize, max_cache_size: usize) { + self.pref_cache_size.store(pref_cache_size, AtomicOrder::Relaxed); + 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(); - if let Some(b) = new_best { - *best_block = b; + { + // update best block + let mut best_block = self.best_block.write().unwrap(); + if let Some(b) = new_best { + *best_block = b; + } } - // update caches - let mut write = self.block_details.write().unwrap(); - write.remove(&header.parent_hash()); - write.insert(hash.clone(), details); - self.note_used(CacheID::Block(hash)); - + { + // update caches + let mut write = self.block_details.write().unwrap(); + write.remove(&header.parent_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(); diff --git a/ethcore/src/client.rs b/ethcore/src/client.rs index 0c8580117..68801520c 100644 --- a/ethcore/src/client.rs +++ b/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>, + chain: Arc, engine: Arc>, state_db: Mutex, 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 { @@ -361,14 +361,12 @@ impl Client { impl BlockChainClient for Client { fn block_header(&self, id: BlockId) -> Option { - let chain = self.chain.read().unwrap(); - Self::block_hash(&chain, id).and_then(|hash| chain.block(&hash).map(|bytes| BlockView::new(&bytes).rlp().at(0).as_raw().to_vec())) + 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())) } fn block_body(&self, id: BlockId) -> Option { - let chain = self.chain.read().unwrap(); - Self::block_hash(&chain, id).and_then(|hash| { - chain.block(&hash).map(|bytes| { + Self::block_hash(&self.chain, id).and_then(|hash| { + self.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 { - let chain = self.chain.read().unwrap(); - Self::block_hash(&chain, id).and_then(|hash| { - chain.block(&hash) + Self::block_hash(&self.chain, id).and_then(|hash| { + self.chain.block(&hash) }) } fn block_status(&self, id: BlockId) -> BlockStatus { - let chain = self.chain.read().unwrap(); - match Self::block_hash(&chain, id) { - Some(ref hash) if chain.is_known(hash) => BlockStatus::InChain, + match Self::block_hash(&self.chain, id) { + Some(ref hash) if self.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 { - let chain = self.chain.read().unwrap(); - Self::block_hash(&chain, id).and_then(|hash| chain.block_details(&hash)).map(|d| d.total_difficulty) + Self::block_hash(&self.chain, id).and_then(|hash| self.chain.block_details(&hash)).map(|d| d.total_difficulty) } fn code(&self, address: &Address) -> Option { @@ -404,18 +399,17 @@ impl BlockChainClient for Client { } fn transaction(&self, id: TransactionId) -> Option { - let chain = self.chain.read().unwrap(); match id { - TransactionId::Hash(ref hash) => chain.transaction_address(hash), - TransactionId::Location(id, index) => Self::block_hash(&chain, id).map(|hash| TransactionAddress { + TransactionId::Hash(ref hash) => self.chain.transaction_address(hash), + 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 { - 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 { @@ -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(), - pending_total_difficulty: chain.best_block_total_difficulty(), - genesis_hash: chain.genesis_hash(), - best_block_hash: chain.best_block_hash(), - best_block_number: From::from(chain.best_block_number()) + total_difficulty: self.chain.best_block_total_difficulty(), + pending_total_difficulty: self.chain.best_block_total_difficulty(), + genesis_hash: self.chain.genesis_hash(), + best_block_hash: self.chain.best_block_hash(), + best_block_number: From::from(self.chain.best_block_number()) } } } diff --git a/ethcore/src/verification.rs b/ethcore/src/verification.rs index fa9467e95..c7d5e265f 100644 --- a/ethcore/src/verification.rs +++ b/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 { + try!(engine.verify_block_unordered(&header, Some(&bytes))); + for u in Rlp::new(&bytes).at(2).iter().map(|rlp| rlp.as_val::
()) { + try!(engine.verify_block_unordered(&u, None)); + } // Verify transactions. let mut transactions = Vec::new(); - let v = BlockView::new(&bytes); - for t in v.transactions() { - try!(engine.verify_transaction(&t, &header)); - transactions.push(t); + { + let v = BlockView::new(&bytes); + for t in v.transactions() { + try!(engine.verify_transaction(&t, &header)); + transactions.push(t); + } } Ok(PreVerifiedBlock { header: header, diff --git a/util/src/lib.rs b/util/src/lib.rs index 5c8bd4fb0..2b7438cf3 100644 --- a/util/src/lib.rs +++ b/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::*; diff --git a/util/src/thread.rs b/util/src/thread.rs deleted file mode 100644 index b86ca3e86..000000000 --- a/util/src/thread.rs +++ /dev/null @@ -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 . - -//! 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); } -}