From 3fd58bdcbd1bcecfcb513d3206dcbf8e14900531 Mon Sep 17 00:00:00 2001 From: s3krit Date: Tue, 25 Jun 2019 13:38:29 +0000 Subject: [PATCH] Beta 2.5.3 (#10776) * ethcore/res: activate atlantis classic hf on block 8772000 (#10766) * fix docker tags for publishing (#10741) * fix: aura don't add `SystemTime::now()` (#10720) This commit does the following: - Prevent overflow in `verify_timestamp()` by not adding `now` to found faulty timestamp - Use explicit `CheckedSystemTime::checked_add` to prevent potential consensus issues because SystemTime is platform depedent - remove `#[cfg(not(time_checked_add))]` conditional compilation * Update version * Treat empty account the same as non-exist accounts in EIP-1052 (#10775) * DevP2p: Get node IP address and udp port from Socket, if not included in PING packet (#10705) * get node IP address and udp port from Socket, if not included in PING packet * prevent bootnodes from being added to host nodes * code corrections * code corrections * code corrections * code corrections * docs * code corrections * code corrections * Apply suggestions from code review Co-Authored-By: David * Add a way to signal shutdown to snapshotting threads (#10744) * Add a way to signal shutdown to snapshotting threads * Pass Progress to fat_rlps() so we can abort from there too. * Checking for abort in a single spot * Remove nightly-only weak/strong counts * fix warning * Fix tests * Add dummy impl to abort snapshots * Add another dummy impl for TestSnapshotService * Remove debugging code * Return error instead of the odd Ok(()) Switch to AtomicU64 * revert .as_bytes() change * fix build * fix build maybe --- Cargo.lock | 12 +-- Cargo.toml | 2 +- ethcore/res/ethereum/classic.json | 24 +++--- ethcore/service/src/service.rs | 11 ++- ethcore/src/client/client.rs | 27 ++++-- ethcore/src/engines/authority_round/mod.rs | 14 ++- ethcore/src/engines/clique/block_state.rs | 6 +- ethcore/src/engines/clique/mod.rs | 12 ++- ethcore/src/externalities.rs | 6 +- ethcore/src/lib.rs | 5 +- ethcore/src/snapshot/account.rs | 44 +++++++--- ethcore/src/snapshot/error.rs | 3 + ethcore/src/snapshot/io.rs | 5 +- ethcore/src/snapshot/mod.rs | 90 ++++++++++++-------- ethcore/src/snapshot/service.rs | 27 ++++-- ethcore/src/snapshot/tests/service.rs | 9 +- ethcore/src/snapshot/tests/state.rs | 12 +-- ethcore/src/snapshot/traits.rs | 3 + ethcore/src/verification/verification.rs | 7 +- ethcore/sync/src/tests/snapshot.rs | 2 + parity/configuration.rs | 2 +- parity/lib.rs | 1 - parity/run.rs | 30 +++++-- parity/snapshot.rs | 2 +- rpc/src/v1/tests/helpers/snapshot_service.rs | 1 + scripts/docker/hub/publish-docker.sh | 10 +++ util/network-devp2p/src/discovery.rs | 29 +++++-- util/network-devp2p/src/node_table.rs | 14 ++- util/version/Cargo.toml | 2 +- 29 files changed, 263 insertions(+), 149 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index c0f79d9da..d36117e7c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2457,7 +2457,7 @@ dependencies = [ "futures 0.1.25 (registry+https://github.com/rust-lang/crates.io-index)", "jni 0.11.0 (registry+https://github.com/rust-lang/crates.io-index)", "panic_hook 0.1.0", - "parity-ethereum 2.5.2", + "parity-ethereum 2.5.3", "tokio 0.1.11 (registry+https://github.com/rust-lang/crates.io-index)", "tokio-current-thread 0.1.3 (registry+https://github.com/rust-lang/crates.io-index)", ] @@ -2487,7 +2487,7 @@ dependencies = [ [[package]] name = "parity-ethereum" -version = "2.5.2" +version = "2.5.3" dependencies = [ "ansi_term 0.10.2 (registry+https://github.com/rust-lang/crates.io-index)", "atty 0.2.11 (registry+https://github.com/rust-lang/crates.io-index)", @@ -2540,7 +2540,7 @@ dependencies = [ "parity-rpc 1.12.0", "parity-runtime 0.1.0", "parity-updater 1.12.0", - "parity-version 2.5.2", + "parity-version 2.5.3", "parity-whisper 0.1.0", "parking_lot 0.7.1 (registry+https://github.com/rust-lang/crates.io-index)", "pretty_assertions 0.1.2 (registry+https://github.com/rust-lang/crates.io-index)", @@ -2683,7 +2683,7 @@ dependencies = [ "parity-crypto 0.3.0 (registry+https://github.com/rust-lang/crates.io-index)", "parity-runtime 0.1.0", "parity-updater 1.12.0", - "parity-version 2.5.2", + "parity-version 2.5.3", "parking_lot 0.7.1 (registry+https://github.com/rust-lang/crates.io-index)", "pretty_assertions 0.1.2 (registry+https://github.com/rust-lang/crates.io-index)", "rand 0.4.3 (registry+https://github.com/rust-lang/crates.io-index)", @@ -2781,7 +2781,7 @@ dependencies = [ "parity-bytes 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)", "parity-hash-fetch 1.12.0", "parity-path 0.1.1 (registry+https://github.com/rust-lang/crates.io-index)", - "parity-version 2.5.2", + "parity-version 2.5.3", "parking_lot 0.7.1 (registry+https://github.com/rust-lang/crates.io-index)", "rand 0.4.3 (registry+https://github.com/rust-lang/crates.io-index)", "semver 0.9.0 (registry+https://github.com/rust-lang/crates.io-index)", @@ -2791,7 +2791,7 @@ dependencies = [ [[package]] name = "parity-version" -version = "2.5.2" +version = "2.5.3" dependencies = [ "parity-bytes 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)", "rlp 0.3.0 (registry+https://github.com/rust-lang/crates.io-index)", diff --git a/Cargo.toml b/Cargo.toml index ed34b6862..162dc27a2 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -2,7 +2,7 @@ description = "Parity Ethereum client" name = "parity-ethereum" # NOTE Make sure to update util/version/Cargo.toml as well -version = "2.5.2" +version = "2.5.3" license = "GPL-3.0" authors = ["Parity Technologies "] diff --git a/ethcore/res/ethereum/classic.json b/ethcore/res/ethereum/classic.json index 682b75299..2750f6f63 100644 --- a/ethcore/res/ethereum/classic.json +++ b/ethcore/res/ethereum/classic.json @@ -12,7 +12,7 @@ "ecip1010PauseTransition": "0x2dc6c0", "ecip1010ContinueTransition": "0x4c4b40", "ecip1017EraRounds": "0x4c4b40", - "eip100bTransition": "0x7fffffffffffffff", + "eip100bTransition": "0x85d9a0", "bombDefuseTransition": "0x5a06e0" } } @@ -29,15 +29,15 @@ "forkCanonHash": "0x94365e3a8c0b35089c1d1195081fe7489b528a84b22199c916180db8b28ade7f", "eip150Transition": "0x2625a0", "eip160Transition": "0x2dc6c0", - "eip161abcTransition": "0x7fffffffffffffff", - "eip161dTransition": "0x7fffffffffffffff", + "eip161abcTransition": "0x85d9a0", + "eip161dTransition": "0x85d9a0", "eip155Transition": "0x2dc6c0", "maxCodeSize": "0x6000", - "maxCodeSizeTransition": "0x7fffffffffffffff", - "eip140Transition": "0x7fffffffffffffff", - "eip211Transition": "0x7fffffffffffffff", - "eip214Transition": "0x7fffffffffffffff", - "eip658Transition": "0x7fffffffffffffff" + "maxCodeSizeTransition": "0x85d9a0", + "eip140Transition": "0x85d9a0", + "eip211Transition": "0x85d9a0", + "eip214Transition": "0x85d9a0", + "eip658Transition": "0x85d9a0" }, "genesis": { "seal": { @@ -3905,7 +3905,7 @@ "0x0000000000000000000000000000000000000005": { "builtin": { "name": "modexp", - "activate_at": "0x7fffffffffffffff", + "activate_at": "0x85d9a0", "pricing": { "modexp": { "divisor": 20 @@ -3916,7 +3916,7 @@ "0x0000000000000000000000000000000000000006": { "builtin": { "name": "alt_bn128_add", - "activate_at": "0x7fffffffffffffff", + "activate_at": "0x85d9a0", "pricing": { "linear": { "base": 500, @@ -3928,7 +3928,7 @@ "0x0000000000000000000000000000000000000007": { "builtin": { "name": "alt_bn128_mul", - "activate_at": "0x7fffffffffffffff", + "activate_at": "0x85d9a0", "pricing": { "linear": { "base": 40000, @@ -3940,7 +3940,7 @@ "0x0000000000000000000000000000000000000008": { "builtin": { "name": "alt_bn128_pairing", - "activate_at": "0x7fffffffffffffff", + "activate_at": "0x85d9a0", "pricing": { "alt_bn128_pairing": { "base": 100000, diff --git a/ethcore/service/src/service.rs b/ethcore/service/src/service.rs index c16a07189..095397037 100644 --- a/ethcore/service/src/service.rs +++ b/ethcore/service/src/service.rs @@ -30,8 +30,10 @@ use blockchain::{BlockChainDB, BlockChainDBHandler}; use ethcore::client::{Client, ClientConfig, ChainNotify, ClientIoMessage}; use ethcore::miner::Miner; use ethcore::snapshot::service::{Service as SnapshotService, ServiceParams as SnapServiceParams}; -use ethcore::snapshot::{SnapshotService as _SnapshotService, RestorationStatus}; +use ethcore::snapshot::{SnapshotService as _SnapshotService, RestorationStatus, Error as SnapshotError}; use ethcore::spec::Spec; +use ethcore::error::{Error as EthcoreError, ErrorKind}; + use ethcore_private_tx::{self, Importer, Signer}; use Error; @@ -197,6 +199,7 @@ impl ClientService { /// Shutdown the Client Service pub fn shutdown(&self) { + trace!(target: "shutdown", "Shutting down Client Service"); self.snapshot.shutdown(); } } @@ -257,7 +260,11 @@ impl IoHandler for ClientIoHandler { let res = thread::Builder::new().name("Periodic Snapshot".into()).spawn(move || { if let Err(e) = snapshot.take_snapshot(&*client, num) { - warn!("Failed to take snapshot at block #{}: {}", num, e); + match e { + EthcoreError(ErrorKind::Snapshot(SnapshotError::SnapshotAborted), _) => info!("Snapshot aborted"), + _ => warn!("Failed to take snapshot at block #{}: {}", num, e), + } + } }); diff --git a/ethcore/src/client/client.rs b/ethcore/src/client/client.rs index 4ec65b85e..047de5580 100644 --- a/ethcore/src/client/client.rs +++ b/ethcore/src/client/client.rs @@ -764,8 +764,8 @@ impl Client { liveness: AtomicBool::new(awake), mode: Mutex::new(config.mode.clone()), chain: RwLock::new(chain), - tracedb: tracedb, - engine: engine, + tracedb, + engine, pruning: config.pruning.clone(), db: RwLock::new(db.clone()), state_db: RwLock::new(state_db), @@ -778,8 +778,8 @@ impl Client { ancient_blocks_import_lock: Default::default(), queue_consensus_message: IoChannelQueue::new(usize::max_value()), last_hashes: RwLock::new(VecDeque::new()), - factories: factories, - history: history, + factories, + history, on_user_defaults_change: Mutex::new(None), registrar_address, exit_handler: Mutex::new(None), @@ -1138,7 +1138,12 @@ impl Client { /// Take a snapshot at the given block. /// If the ID given is "latest", this will default to 1000 blocks behind. - pub fn take_snapshot(&self, writer: W, at: BlockId, p: &snapshot::Progress) -> Result<(), EthcoreError> { + pub fn take_snapshot( + &self, + writer: W, + at: BlockId, + p: &snapshot::Progress, + ) -> Result<(), EthcoreError> { let db = self.state_db.read().journal_db().boxed_clone(); let best_block_number = self.chain_info().best_block_number; let block_number = self.block_number(at).ok_or_else(|| snapshot::Error::InvalidStartingBlock(at))?; @@ -1168,8 +1173,16 @@ impl Client { }; let processing_threads = self.config.snapshot.processing_threads; - snapshot::take_snapshot(&*self.engine, &self.chain.read(), start_hash, db.as_hash_db(), writer, p, processing_threads)?; - + let chunker = self.engine.snapshot_components().ok_or(snapshot::Error::SnapshotsUnsupported)?; + snapshot::take_snapshot( + chunker, + &self.chain.read(), + start_hash, + db.as_hash_db(), + writer, + p, + processing_threads, + )?; Ok(()) } diff --git a/ethcore/src/engines/authority_round/mod.rs b/ethcore/src/engines/authority_round/mod.rs index 36f25144f..312b58e61 100644 --- a/ethcore/src/engines/authority_round/mod.rs +++ b/ethcore/src/engines/authority_round/mod.rs @@ -22,7 +22,7 @@ use std::iter::FromIterator; use std::ops::Deref; use std::sync::atomic::{AtomicUsize, AtomicBool, Ordering as AtomicOrdering}; use std::sync::{Weak, Arc}; -use std::time::{UNIX_EPOCH, SystemTime, Duration}; +use std::time::{UNIX_EPOCH, Duration}; use block::*; use client::EngineClient; @@ -42,14 +42,12 @@ use itertools::{self, Itertools}; use rlp::{encode, Decodable, DecoderError, Encodable, RlpStream, Rlp}; use ethereum_types::{H256, H520, Address, U128, U256}; use parking_lot::{Mutex, RwLock}; +use time_utils::CheckedSystemTime; use types::BlockNumber; use types::header::{Header, ExtendedHeader}; use types::ancestry_action::AncestryAction; use unexpected::{Mismatch, OutOfBounds}; -#[cfg(not(time_checked_add))] -use time_utils::CheckedSystemTime; - mod finality; /// `AuthorityRound` params. @@ -578,10 +576,10 @@ fn verify_timestamp(step: &Step, header_step: u64) -> Result<(), BlockError> { // Returning it further won't recover the sync process. trace!(target: "engine", "verify_timestamp: block too early"); - let now = SystemTime::now(); - let found = now.checked_add(Duration::from_secs(oob.found)).ok_or(BlockError::TimestampOverflow)?; - let max = oob.max.and_then(|m| now.checked_add(Duration::from_secs(m))); - let min = oob.min.and_then(|m| now.checked_add(Duration::from_secs(m))); + let found = CheckedSystemTime::checked_add(UNIX_EPOCH, Duration::from_secs(oob.found)) + .ok_or(BlockError::TimestampOverflow)?; + let max = oob.max.and_then(|m| CheckedSystemTime::checked_add(UNIX_EPOCH, Duration::from_secs(m))); + let min = oob.min.and_then(|m| CheckedSystemTime::checked_add(UNIX_EPOCH, Duration::from_secs(m))); let new_oob = OutOfBounds { min, max, found }; diff --git a/ethcore/src/engines/clique/block_state.rs b/ethcore/src/engines/clique/block_state.rs index 4257076c0..6518d8ac2 100644 --- a/ethcore/src/engines/clique/block_state.rs +++ b/ethcore/src/engines/clique/block_state.rs @@ -24,13 +24,11 @@ use engines::clique::{VoteType, DIFF_INTURN, DIFF_NOTURN, NULL_AUTHOR, SIGNING_D use error::{Error, BlockError}; use ethereum_types::{Address, H64}; use rand::Rng; +use time_utils::CheckedSystemTime; use types::BlockNumber; use types::header::Header; use unexpected::Mismatch; -#[cfg(not(feature = "time_checked_add"))] -use time_utils::CheckedSystemTime; - /// Type that keeps track of the state for a given vote // Votes that go against the proposal aren't counted since it's equivalent to not voting #[derive(Copy, Clone, Debug, PartialEq, PartialOrd)] @@ -268,7 +266,7 @@ impl CliqueBlockState { // This is a quite bad API because we must mutate both variables even when already `inturn` fails // That's why we can't return early and must have the `if-else` in the end pub fn calc_next_timestamp(&mut self, timestamp: u64, period: u64) -> Result<(), Error> { - let inturn = UNIX_EPOCH.checked_add(Duration::from_secs(timestamp.saturating_add(period))); + let inturn = CheckedSystemTime::checked_add(UNIX_EPOCH, Duration::from_secs(timestamp.saturating_add(period))); self.next_timestamp_inturn = inturn; diff --git a/ethcore/src/engines/clique/mod.rs b/ethcore/src/engines/clique/mod.rs index 742a7aec4..8be54da26 100644 --- a/ethcore/src/engines/clique/mod.rs +++ b/ethcore/src/engines/clique/mod.rs @@ -82,12 +82,10 @@ use parking_lot::RwLock; use rand::Rng; use super::signer::EngineSigner; use unexpected::{Mismatch, OutOfBounds}; +use time_utils::CheckedSystemTime; use types::BlockNumber; use types::header::{ExtendedHeader, Header}; -#[cfg(not(feature = "time_checked_add"))] -use time_utils::CheckedSystemTime; - use self::block_state::CliqueBlockState; use self::params::CliqueParams; use self::step_service::StepService; @@ -536,7 +534,7 @@ impl Engine for Clique { // Don't waste time checking blocks from the future { - let limit = SystemTime::now().checked_add(Duration::from_secs(self.period)) + let limit = CheckedSystemTime::checked_add(SystemTime::now(), Duration::from_secs(self.period)) .ok_or(BlockError::TimestampOverflow)?; // This should succeed under the contraints that the system clock works @@ -546,7 +544,7 @@ impl Engine for Clique { let hdr = Duration::from_secs(header.timestamp()); if hdr > limit_as_dur { - let found = UNIX_EPOCH.checked_add(hdr).ok_or(BlockError::TimestampOverflow)?; + let found = CheckedSystemTime::checked_add(UNIX_EPOCH, hdr).ok_or(BlockError::TimestampOverflow)?; Err(BlockError::TemporarilyInvalid(OutOfBounds { min: None, @@ -657,8 +655,8 @@ impl Engine for Clique { // Ensure that the block's timestamp isn't too close to it's parent let limit = parent.timestamp().saturating_add(self.period); if limit > header.timestamp() { - let max = UNIX_EPOCH.checked_add(Duration::from_secs(header.timestamp())); - let found = UNIX_EPOCH.checked_add(Duration::from_secs(limit)) + let max = CheckedSystemTime::checked_add(UNIX_EPOCH, Duration::from_secs(header.timestamp())); + let found = CheckedSystemTime::checked_add(UNIX_EPOCH, Duration::from_secs(limit)) .ok_or(BlockError::TimestampOverflow)?; Err(BlockError::InvalidTimestamp(OutOfBounds { diff --git a/ethcore/src/externalities.rs b/ethcore/src/externalities.rs index 23a4a83c3..41122afc1 100644 --- a/ethcore/src/externalities.rs +++ b/ethcore/src/externalities.rs @@ -314,7 +314,11 @@ impl<'a, T: 'a, V: 'a, B: 'a> Ext for Externalities<'a, T, V, B> } fn extcodehash(&self, address: &Address) -> vm::Result> { - Ok(self.state.code_hash(address)?) + if self.state.exists_and_not_null(address)? { + Ok(self.state.code_hash(address)?) + } else { + Ok(None) + } } fn extcodesize(&self, address: &Address) -> vm::Result> { diff --git a/ethcore/src/lib.rs b/ethcore/src/lib.rs index eee076b32..6578d3607 100644 --- a/ethcore/src/lib.rs +++ b/ethcore/src/lib.rs @@ -15,7 +15,6 @@ // along with Parity Ethereum. If not, see . #![warn(missing_docs, unused_extern_crates)] -#![cfg_attr(feature = "time_checked_add", feature(time_checked_add))] //! Ethcore library //! @@ -100,6 +99,7 @@ extern crate rlp; extern crate rustc_hex; extern crate serde; extern crate stats; +extern crate time_utils; extern crate triehash_ethereum as triehash; extern crate unexpected; extern crate using_queue; @@ -149,9 +149,6 @@ extern crate fetch; #[cfg(all(test, feature = "price-info"))] extern crate parity_runtime; -#[cfg(not(time_checked_add))] -extern crate time_utils; - pub mod block; pub mod builtin; pub mod client; diff --git a/ethcore/src/snapshot/account.rs b/ethcore/src/snapshot/account.rs index ed56e2435..2a9ac911f 100644 --- a/ethcore/src/snapshot/account.rs +++ b/ethcore/src/snapshot/account.rs @@ -24,9 +24,10 @@ use ethtrie::{TrieDB, TrieDBMut}; use hash::{KECCAK_EMPTY, KECCAK_NULL_RLP}; use hash_db::HashDB; use rlp::{RlpStream, Rlp}; -use snapshot::Error; +use snapshot::{Error, Progress}; use std::collections::HashSet; use trie::{Trie, TrieMut}; +use std::sync::atomic::Ordering; // An empty account -- these were replaced with RLP null data for a space optimization in v1. const ACC_EMPTY: BasicAccount = BasicAccount { @@ -65,8 +66,16 @@ impl CodeState { // walk the account's storage trie, returning a vector of RLP items containing the // account address hash, account properties and the storage. Each item contains at most `max_storage_items` // storage records split according to snapshot format definition. -pub fn to_fat_rlps(account_hash: &H256, acc: &BasicAccount, acct_db: &AccountDB, used_code: &mut HashSet, first_chunk_size: usize, max_chunk_size: usize) -> Result, Error> { - let db = &(acct_db as &HashDB<_,_>); +pub fn to_fat_rlps( + account_hash: &H256, + acc: &BasicAccount, + acct_db: &AccountDB, + used_code: &mut HashSet, + first_chunk_size: usize, + max_chunk_size: usize, + p: &Progress, +) -> Result, Error> { + let db = &(acct_db as &dyn HashDB<_,_>); let db = TrieDB::new(db, &acc.storage_root)?; let mut chunks = Vec::new(); let mut db_iter = db.iter()?; @@ -112,6 +121,10 @@ pub fn to_fat_rlps(account_hash: &H256, acc: &BasicAccount, acct_db: &AccountDB, } loop { + if p.abort.load(Ordering::SeqCst) { + trace!(target: "snapshot", "to_fat_rlps: aborting snapshot"); + return Err(Error::SnapshotAborted); + } match db_iter.next() { Some(Ok((k, v))) => { let pair = { @@ -211,6 +224,7 @@ mod tests { use types::basic_account::BasicAccount; use test_helpers::get_temp_state_db; use snapshot::tests::helpers::fill_storage; + use snapshot::Progress; use hash::{KECCAK_EMPTY, KECCAK_NULL_RLP, keccak}; use ethereum_types::{H256, Address}; @@ -236,8 +250,8 @@ mod tests { let thin_rlp = ::rlp::encode(&account); assert_eq!(::rlp::decode::(&thin_rlp).unwrap(), account); - - let fat_rlps = to_fat_rlps(&keccak(&addr), &account, &AccountDB::new(db.as_hash_db(), &addr), &mut Default::default(), usize::max_value(), usize::max_value()).unwrap(); + let p = Progress::default(); + let fat_rlps = to_fat_rlps(&keccak(&addr), &account, &AccountDB::new(db.as_hash_db(), &addr), &mut Default::default(), usize::max_value(), usize::max_value(), &p).unwrap(); let fat_rlp = Rlp::new(&fat_rlps[0]).at(1).unwrap(); assert_eq!(from_fat_rlp(&mut AccountDBMut::new(db.as_hash_db_mut(), &addr), fat_rlp, H256::zero()).unwrap().0, account); } @@ -262,7 +276,9 @@ mod tests { let thin_rlp = ::rlp::encode(&account); assert_eq!(::rlp::decode::(&thin_rlp).unwrap(), account); - let fat_rlp = to_fat_rlps(&keccak(&addr), &account, &AccountDB::new(db.as_hash_db(), &addr), &mut Default::default(), usize::max_value(), usize::max_value()).unwrap(); + let p = Progress::default(); + + let fat_rlp = to_fat_rlps(&keccak(&addr), &account, &AccountDB::new(db.as_hash_db(), &addr), &mut Default::default(), usize::max_value(), usize::max_value(), &p).unwrap(); let fat_rlp = Rlp::new(&fat_rlp[0]).at(1).unwrap(); assert_eq!(from_fat_rlp(&mut AccountDBMut::new(db.as_hash_db_mut(), &addr), fat_rlp, H256::zero()).unwrap().0, account); } @@ -287,7 +303,8 @@ mod tests { let thin_rlp = ::rlp::encode(&account); assert_eq!(::rlp::decode::(&thin_rlp).unwrap(), account); - let fat_rlps = to_fat_rlps(&keccak(addr), &account, &AccountDB::new(db.as_hash_db(), &addr), &mut Default::default(), 500, 1000).unwrap(); + let p = Progress::default(); + let fat_rlps = to_fat_rlps(&keccak(addr), &account, &AccountDB::new(db.as_hash_db(), &addr), &mut Default::default(), 500, 1000, &p).unwrap(); let mut root = KECCAK_NULL_RLP; let mut restored_account = None; for rlp in fat_rlps { @@ -319,20 +336,21 @@ mod tests { nonce: 50.into(), balance: 123456789.into(), storage_root: KECCAK_NULL_RLP, - code_hash: code_hash, + code_hash, }; let account2 = BasicAccount { nonce: 400.into(), balance: 98765432123456789usize.into(), storage_root: KECCAK_NULL_RLP, - code_hash: code_hash, + code_hash, }; let mut used_code = HashSet::new(); - - let fat_rlp1 = to_fat_rlps(&keccak(&addr1), &account1, &AccountDB::new(db.as_hash_db(), &addr1), &mut used_code, usize::max_value(), usize::max_value()).unwrap(); - let fat_rlp2 = to_fat_rlps(&keccak(&addr2), &account2, &AccountDB::new(db.as_hash_db(), &addr2), &mut used_code, usize::max_value(), usize::max_value()).unwrap(); + let p1 = Progress::default(); + let p2 = Progress::default(); + let fat_rlp1 = to_fat_rlps(&keccak(&addr1), &account1, &AccountDB::new(db.as_hash_db(), &addr1), &mut used_code, usize::max_value(), usize::max_value(), &p1).unwrap(); + let fat_rlp2 = to_fat_rlps(&keccak(&addr2), &account2, &AccountDB::new(db.as_hash_db(), &addr2), &mut used_code, usize::max_value(), usize::max_value(), &p2).unwrap(); assert_eq!(used_code.len(), 1); let fat_rlp1 = Rlp::new(&fat_rlp1[0]).at(1).unwrap(); @@ -350,6 +368,6 @@ mod tests { #[test] fn encoding_empty_acc() { let mut db = get_temp_state_db(); - assert_eq!(from_fat_rlp(&mut AccountDBMut::new(db.as_hash_db_mut(), &Address::default()), Rlp::new(&::rlp::NULL_RLP), H256::zero()).unwrap(), (ACC_EMPTY, None)); + assert_eq!(from_fat_rlp(&mut AccountDBMut::new(db.as_hash_db_mut(), &Address::zero()), Rlp::new(&::rlp::NULL_RLP), H256::zero()).unwrap(), (ACC_EMPTY, None)); } } diff --git a/ethcore/src/snapshot/error.rs b/ethcore/src/snapshot/error.rs index b71f79f80..6faa19da2 100644 --- a/ethcore/src/snapshot/error.rs +++ b/ethcore/src/snapshot/error.rs @@ -61,6 +61,8 @@ pub enum Error { ChunkTooLarge, /// Snapshots not supported by the consensus engine. SnapshotsUnsupported, + /// Aborted snapshot + SnapshotAborted, /// Bad epoch transition. BadEpochProof(u64), /// Wrong chunk format. @@ -91,6 +93,7 @@ impl fmt::Display for Error { Error::ChunkTooSmall => write!(f, "Chunk size is too small."), Error::ChunkTooLarge => write!(f, "Chunk size is too large."), Error::SnapshotsUnsupported => write!(f, "Snapshots unsupported by consensus engine."), + Error::SnapshotAborted => write!(f, "Snapshot was aborted."), Error::BadEpochProof(i) => write!(f, "Bad epoch proof for transition to epoch {}", i), Error::WrongChunkFormat(ref msg) => write!(f, "Wrong chunk format: {}", msg), Error::UnlinkedAncientBlockChain => write!(f, "Unlinked ancient blocks chain"), diff --git a/ethcore/src/snapshot/io.rs b/ethcore/src/snapshot/io.rs index c5f178cd3..536862e7b 100644 --- a/ethcore/src/snapshot/io.rs +++ b/ethcore/src/snapshot/io.rs @@ -310,10 +310,7 @@ impl LooseReader { dir.pop(); - Ok(LooseReader { - dir: dir, - manifest: manifest, - }) + Ok(LooseReader { dir, manifest }) } } diff --git a/ethcore/src/snapshot/mod.rs b/ethcore/src/snapshot/mod.rs index 19a5f8ce6..ae7a21709 100644 --- a/ethcore/src/snapshot/mod.rs +++ b/ethcore/src/snapshot/mod.rs @@ -22,7 +22,7 @@ use std::collections::{HashMap, HashSet}; use std::cmp; use std::sync::Arc; -use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering}; use hash::{keccak, KECCAK_NULL_RLP, KECCAK_EMPTY}; use account_db::{AccountDB, AccountDBMut}; @@ -107,7 +107,7 @@ impl Default for SnapshotConfiguration { fn default() -> Self { SnapshotConfiguration { no_periodic: false, - processing_threads: ::std::cmp::max(1, num_cpus::get() / 2), + processing_threads: ::std::cmp::max(1, num_cpus::get_physical() / 2), } } } @@ -117,8 +117,9 @@ impl Default for SnapshotConfiguration { pub struct Progress { accounts: AtomicUsize, blocks: AtomicUsize, - size: AtomicUsize, // Todo [rob] use Atomicu64 when it stabilizes. + size: AtomicU64, done: AtomicBool, + abort: AtomicBool, } impl Progress { @@ -127,6 +128,7 @@ impl Progress { self.accounts.store(0, Ordering::Release); self.blocks.store(0, Ordering::Release); self.size.store(0, Ordering::Release); + self.abort.store(false, Ordering::Release); // atomic fence here to ensure the others are written first? // logs might very rarely get polluted if not. @@ -140,7 +142,7 @@ impl Progress { pub fn blocks(&self) -> usize { self.blocks.load(Ordering::Acquire) } /// Get the written size of the snapshot in bytes. - pub fn size(&self) -> usize { self.size.load(Ordering::Acquire) } + pub fn size(&self) -> u64 { self.size.load(Ordering::Acquire) } /// Whether the snapshot is complete. pub fn done(&self) -> bool { self.done.load(Ordering::Acquire) } @@ -148,27 +150,28 @@ impl Progress { } /// Take a snapshot using the given blockchain, starting block hash, and database, writing into the given writer. pub fn take_snapshot( - engine: &EthEngine, + chunker: Box, chain: &BlockChain, - block_at: H256, - state_db: &HashDB, + block_hash: H256, + state_db: &dyn HashDB, writer: W, p: &Progress, processing_threads: usize, ) -> Result<(), Error> { - let start_header = chain.block_header_data(&block_at) - .ok_or_else(|| Error::InvalidStartingBlock(BlockId::Hash(block_at)))?; + let start_header = chain.block_header_data(&block_hash) + .ok_or_else(|| Error::InvalidStartingBlock(BlockId::Hash(block_hash)))?; let state_root = start_header.state_root(); - let number = start_header.number(); + let block_number = start_header.number(); - info!("Taking snapshot starting at block {}", number); + info!("Taking snapshot starting at block {}", block_number); + let version = chunker.current_version(); let writer = Mutex::new(writer); - let chunker = engine.snapshot_components().ok_or(Error::SnapshotsUnsupported)?; - let snapshot_version = chunker.current_version(); let (state_hashes, block_hashes) = scope(|scope| -> Result<(Vec, Vec), Error> { let writer = &writer; - let block_guard = scope.spawn(move || chunk_secondary(chunker, chain, block_at, writer, p)); + let block_guard = scope.spawn(move || { + chunk_secondary(chunker, chain, block_hash, writer, p) + }); // The number of threads must be between 1 and SNAPSHOT_SUBPARTS assert!(processing_threads >= 1, "Cannot use less than 1 threads for creating snapshots"); @@ -183,7 +186,7 @@ pub fn take_snapshot( for part in (thread_idx..SNAPSHOT_SUBPARTS).step_by(num_threads) { debug!(target: "snapshot", "Chunking part {} in thread {}", part, thread_idx); - let mut hashes = chunk_state(state_db, &state_root, writer, p, Some(part))?; + let mut hashes = chunk_state(state_db, &state_root, writer, p, Some(part), thread_idx)?; chunk_hashes.append(&mut hashes); } @@ -207,12 +210,12 @@ pub fn take_snapshot( info!(target: "snapshot", "produced {} state chunks and {} block chunks.", state_hashes.len(), block_hashes.len()); let manifest_data = ManifestData { - version: snapshot_version, - state_hashes: state_hashes, - block_hashes: block_hashes, - state_root: state_root, - block_number: number, - block_hash: block_at, + version, + state_hashes, + block_hashes, + state_root, + block_number, + block_hash, }; writer.into_inner().finish(manifest_data)?; @@ -228,7 +231,13 @@ pub fn take_snapshot( /// Secondary chunks are engine-specific, but they intend to corroborate the state data /// in the state chunks. /// Returns a list of chunk hashes, with the first having the blocks furthest from the genesis. -pub fn chunk_secondary<'a>(mut chunker: Box, chain: &'a BlockChain, start_hash: H256, writer: &Mutex, progress: &'a Progress) -> Result, Error> { +pub fn chunk_secondary<'a>( + mut chunker: Box, + chain: &'a BlockChain, + start_hash: H256, + writer: &Mutex, + progress: &'a Progress +) -> Result, Error> { let mut chunk_hashes = Vec::new(); let mut snappy_buffer = vec![0; snappy::max_compressed_len(PREFERRED_CHUNK_SIZE)]; @@ -243,7 +252,7 @@ pub fn chunk_secondary<'a>(mut chunker: Box, chain: &'a Bloc trace!(target: "snapshot", "wrote secondary chunk. hash: {:x}, size: {}, uncompressed size: {}", hash, size, raw_data.len()); - progress.size.fetch_add(size, Ordering::SeqCst); + progress.size.fetch_add(size as u64, Ordering::SeqCst); chunk_hashes.push(hash); Ok(()) }; @@ -266,8 +275,9 @@ struct StateChunker<'a> { rlps: Vec, cur_size: usize, snappy_buffer: Vec, - writer: &'a Mutex, + writer: &'a Mutex, progress: &'a Progress, + thread_idx: usize, } impl<'a> StateChunker<'a> { @@ -297,10 +307,10 @@ impl<'a> StateChunker<'a> { let hash = keccak(&compressed); self.writer.lock().write_state_chunk(hash, compressed)?; - trace!(target: "snapshot", "wrote state chunk. size: {}, uncompressed size: {}", compressed_size, raw_data.len()); + trace!(target: "snapshot", "Thread {} wrote state chunk. size: {}, uncompressed size: {}", self.thread_idx, compressed_size, raw_data.len()); self.progress.accounts.fetch_add(num_entries, Ordering::SeqCst); - self.progress.size.fetch_add(compressed_size, Ordering::SeqCst); + self.progress.size.fetch_add(compressed_size as u64, Ordering::SeqCst); self.hashes.push(hash); self.cur_size = 0; @@ -321,7 +331,14 @@ impl<'a> StateChunker<'a> { /// /// Returns a list of hashes of chunks created, or any error it may /// have encountered. -pub fn chunk_state<'a>(db: &HashDB, root: &H256, writer: &Mutex, progress: &'a Progress, part: Option) -> Result, Error> { +pub fn chunk_state<'a>( + db: &dyn HashDB, + root: &H256, + writer: &Mutex, + progress: &'a Progress, + part: Option, + thread_idx: usize, +) -> Result, Error> { let account_trie = TrieDB::new(&db, &root)?; let mut chunker = StateChunker { @@ -329,8 +346,9 @@ pub fn chunk_state<'a>(db: &HashDB, root: &H256, writer: rlps: Vec::new(), cur_size: 0, snappy_buffer: vec![0; snappy::max_compressed_len(PREFERRED_CHUNK_SIZE)], - writer: writer, - progress: progress, + writer, + progress, + thread_idx, }; let mut used_code = HashSet::new(); @@ -365,7 +383,7 @@ pub fn chunk_state<'a>(db: &HashDB, root: &H256, writer: let account = ::rlp::decode(&*account_data)?; let account_db = AccountDB::from_hash(db, account_key_hash); - let fat_rlps = account::to_fat_rlps(&account_key_hash, &account, &account_db, &mut used_code, PREFERRED_CHUNK_SIZE - chunker.chunk_size(), PREFERRED_CHUNK_SIZE)?; + let fat_rlps = account::to_fat_rlps(&account_key_hash, &account, &account_db, &mut used_code, PREFERRED_CHUNK_SIZE - chunker.chunk_size(), PREFERRED_CHUNK_SIZE, progress)?; for (i, fat_rlp) in fat_rlps.into_iter().enumerate() { if i > 0 { chunker.write_chunk()?; @@ -383,7 +401,7 @@ pub fn chunk_state<'a>(db: &HashDB, root: &H256, writer: /// Used to rebuild the state trie piece by piece. pub struct StateRebuilder { - db: Box, + db: Box, state_root: H256, known_code: HashMap, // code hashes mapped to first account with this code. missing_code: HashMap>, // maps code hashes to lists of accounts missing that code. @@ -393,7 +411,7 @@ pub struct StateRebuilder { impl StateRebuilder { /// Create a new state rebuilder to write into the given backing DB. - pub fn new(db: Arc, pruning: Algorithm) -> Self { + pub fn new(db: Arc, pruning: Algorithm) -> Self { StateRebuilder { db: journaldb::new(db.clone(), pruning, ::db::COL_STATE), state_root: KECCAK_NULL_RLP, @@ -411,7 +429,7 @@ impl StateRebuilder { let mut pairs = Vec::with_capacity(rlp.item_count()?); // initialize the pairs vector with empty values so we have slots to write into. - pairs.resize(rlp.item_count()?, (H256::new(), Vec::new())); + pairs.resize(rlp.item_count()?, (H256::zero(), Vec::new())); let status = rebuild_accounts( self.db.as_hash_db_mut(), @@ -468,7 +486,7 @@ impl StateRebuilder { /// Finalize the restoration. Check for accounts missing code and make a dummy /// journal entry. /// Once all chunks have been fed, there should be nothing missing. - pub fn finalize(mut self, era: u64, id: H256) -> Result, ::error::Error> { + pub fn finalize(mut self, era: u64, id: H256) -> Result, ::error::Error> { let missing = self.missing_code.keys().cloned().collect::>(); if !missing.is_empty() { return Err(Error::MissingCode(missing).into()) } @@ -493,7 +511,7 @@ struct RebuiltStatus { // rebuild a set of accounts and their storage. // returns a status detailing newly-loaded code and accounts missing code. fn rebuild_accounts( - db: &mut HashDB, + db: &mut dyn HashDB, account_fat_rlps: Rlp, out_chunk: &mut [(H256, Bytes)], known_code: &HashMap, @@ -560,7 +578,7 @@ const POW_VERIFY_RATE: f32 = 0.02; /// Verify an old block with the given header, engine, blockchain, body. If `always` is set, it will perform /// the fullest verification possible. If not, it will take a random sample to determine whether it will /// do heavy or light verification. -pub fn verify_old_block(rng: &mut OsRng, header: &Header, engine: &EthEngine, chain: &BlockChain, always: bool) -> Result<(), ::error::Error> { +pub fn verify_old_block(rng: &mut OsRng, header: &Header, engine: &dyn EthEngine, chain: &BlockChain, always: bool) -> Result<(), ::error::Error> { engine.verify_block_basic(header)?; if always || rng.gen::() <= POW_VERIFY_RATE { diff --git a/ethcore/src/snapshot/service.rs b/ethcore/src/snapshot/service.rs index 4b3f196cb..ddae76a00 100644 --- a/ethcore/src/snapshot/service.rs +++ b/ethcore/src/snapshot/service.rs @@ -415,7 +415,7 @@ impl Service { _ => break, } - // Writting changes to DB and logging every now and then + // Writing changes to DB and logging every now and then if block_number % 1_000 == 0 { next_db.key_value().write_buffered(batch); next_chain.commit(); @@ -479,16 +479,12 @@ impl Service { let guard = Guard::new(temp_dir.clone()); let res = client.take_snapshot(writer, BlockId::Number(num), &self.progress); - self.taking_snapshot.store(false, Ordering::SeqCst); if let Err(e) = res { if client.chain_info().best_block_number >= num + client.pruning_history() { - // "Cancelled" is mincing words a bit -- what really happened - // is that the state we were snapshotting got pruned out - // before we could finish. - info!("Periodic snapshot failed: block state pruned.\ - Run with a longer `--pruning-history` or with `--no-periodic-snapshot`"); - return Ok(()) + // The state we were snapshotting was pruned before we could finish. + info!("Periodic snapshot failed: block state pruned. Run with a longer `--pruning-history` or with `--no-periodic-snapshot`"); + return Err(e); } else { return Err(e); } @@ -846,14 +842,29 @@ impl SnapshotService for Service { } } + fn abort_snapshot(&self) { + if self.taking_snapshot.load(Ordering::SeqCst) { + trace!(target: "snapshot", "Aborting snapshot – Snapshot under way"); + self.progress.abort.store(true, Ordering::SeqCst); + } + } + fn shutdown(&self) { + trace!(target: "snapshot", "Shut down SnapshotService"); self.abort_restore(); + trace!(target: "snapshot", "Shut down SnapshotService - restore aborted"); + self.abort_snapshot(); + trace!(target: "snapshot", "Shut down SnapshotService - snapshot aborted"); } } impl Drop for Service { fn drop(&mut self) { + trace!(target: "shutdown", "Dropping Service"); self.abort_restore(); + trace!(target: "shutdown", "Dropping Service - restore aborted"); + self.abort_snapshot(); + trace!(target: "shutdown", "Dropping Service - snapshot aborted"); } } diff --git a/ethcore/src/snapshot/tests/service.rs b/ethcore/src/snapshot/tests/service.rs index 37a10048a..515e5992f 100644 --- a/ethcore/src/snapshot/tests/service.rs +++ b/ethcore/src/snapshot/tests/service.rs @@ -188,14 +188,15 @@ fn keep_ancient_blocks() { &state_root, &writer, &Progress::default(), - None + None, + 0 ).unwrap(); let manifest = ::snapshot::ManifestData { version: 2, - state_hashes: state_hashes, - state_root: state_root, - block_hashes: block_hashes, + state_hashes, + state_root, + block_hashes, block_number: NUM_BLOCKS, block_hash: best_hash, }; diff --git a/ethcore/src/snapshot/tests/state.rs b/ethcore/src/snapshot/tests/state.rs index fa7df6b61..0d9760332 100644 --- a/ethcore/src/snapshot/tests/state.rs +++ b/ethcore/src/snapshot/tests/state.rs @@ -55,7 +55,7 @@ fn snap_and_restore() { let mut state_hashes = Vec::new(); for part in 0..SNAPSHOT_SUBPARTS { - let mut hashes = chunk_state(&old_db, &state_root, &writer, &Progress::default(), Some(part)).unwrap(); + let mut hashes = chunk_state(&old_db, &state_root, &writer, &Progress::default(), Some(part), 0).unwrap(); state_hashes.append(&mut hashes); } @@ -126,8 +126,8 @@ fn get_code_from_prev_chunk() { let mut make_chunk = |acc, hash| { let mut db = journaldb::new_memory_db(); AccountDBMut::from_hash(&mut db, hash).insert(&code[..]); - - let fat_rlp = account::to_fat_rlps(&hash, &acc, &AccountDB::from_hash(&db, hash), &mut used_code, usize::max_value(), usize::max_value()).unwrap(); + let p = Progress::default(); + let fat_rlp = account::to_fat_rlps(&hash, &acc, &AccountDB::from_hash(&db, hash), &mut used_code, usize::max_value(), usize::max_value(), &p).unwrap(); let mut stream = RlpStream::new_list(1); stream.append_raw(&fat_rlp[0], 1); stream.out() @@ -171,13 +171,13 @@ fn checks_flag() { let state_root = producer.state_root(); let writer = Mutex::new(PackedWriter::new(&snap_file).unwrap()); - let state_hashes = chunk_state(&old_db, &state_root, &writer, &Progress::default(), None).unwrap(); + let state_hashes = chunk_state(&old_db, &state_root, &writer, &Progress::default(), None, 0).unwrap(); writer.into_inner().finish(::snapshot::ManifestData { version: 2, - state_hashes: state_hashes, + state_hashes, block_hashes: Vec::new(), - state_root: state_root, + state_root, block_number: 0, block_hash: H256::default(), }).unwrap(); diff --git a/ethcore/src/snapshot/traits.rs b/ethcore/src/snapshot/traits.rs index bb4ab3b39..aa61b595b 100644 --- a/ethcore/src/snapshot/traits.rs +++ b/ethcore/src/snapshot/traits.rs @@ -55,6 +55,9 @@ pub trait SnapshotService : Sync + Send { /// no-op if currently restoring. fn restore_block_chunk(&self, hash: H256, chunk: Bytes); + /// Abort in-progress snapshotting if there is one. + fn abort_snapshot(&self); + /// Shutdown the Snapshot Service by aborting any ongoing restore fn shutdown(&self); } diff --git a/ethcore/src/verification/verification.rs b/ethcore/src/verification/verification.rs index 36a847f64..61d711924 100644 --- a/ethcore/src/verification/verification.rs +++ b/ethcore/src/verification/verification.rs @@ -40,7 +40,6 @@ use types::{BlockNumber, header::Header}; use types::transaction::SignedTransaction; use verification::queue::kind::blocks::Unverified; -#[cfg(not(time_checked_add))] use time_utils::CheckedSystemTime; /// Preprocessed block data gathered in `verify_block_unordered` call @@ -310,7 +309,7 @@ pub fn verify_header_params(header: &Header, engine: &EthEngine, is_full: bool, // this will resist overflow until `year 2037` let max_time = SystemTime::now() + ACCEPTABLE_DRIFT; let invalid_threshold = max_time + ACCEPTABLE_DRIFT * 9; - let timestamp = UNIX_EPOCH.checked_add(Duration::from_secs(header.timestamp())) + let timestamp = CheckedSystemTime::checked_add(UNIX_EPOCH, Duration::from_secs(header.timestamp())) .ok_or(BlockError::TimestampOverflow)?; if timestamp > invalid_threshold { @@ -334,9 +333,9 @@ fn verify_parent(header: &Header, parent: &Header, engine: &EthEngine) -> Result if !engine.is_timestamp_valid(header.timestamp(), parent.timestamp()) { let now = SystemTime::now(); - let min = now.checked_add(Duration::from_secs(parent.timestamp().saturating_add(1))) + let min = CheckedSystemTime::checked_add(now, Duration::from_secs(parent.timestamp().saturating_add(1))) .ok_or(BlockError::TimestampOverflow)?; - let found = now.checked_add(Duration::from_secs(header.timestamp())) + let found = CheckedSystemTime::checked_add(now, Duration::from_secs(header.timestamp())) .ok_or(BlockError::TimestampOverflow)?; return Err(From::from(BlockError::InvalidTimestamp(OutOfBounds { max: None, min: Some(min), found }))) } diff --git a/ethcore/sync/src/tests/snapshot.rs b/ethcore/sync/src/tests/snapshot.rs index a3aa77d36..d865adc2a 100644 --- a/ethcore/sync/src/tests/snapshot.rs +++ b/ethcore/sync/src/tests/snapshot.rs @@ -122,6 +122,8 @@ impl SnapshotService for TestSnapshotService { self.block_restoration_chunks.lock().clear(); } + fn abort_snapshot(&self) {} + fn restore_state_chunk(&self, hash: H256, chunk: Bytes) { if self.restoration_manifest.lock().as_ref().map_or(false, |m| m.state_hashes.iter().any(|h| h == &hash)) { self.state_restoration_chunks.lock().insert(hash, chunk); diff --git a/parity/configuration.rs b/parity/configuration.rs index 198a58c3f..1a8338637 100644 --- a/parity/configuration.rs +++ b/parity/configuration.rs @@ -932,7 +932,7 @@ impl Configuration { no_periodic: self.args.flag_no_periodic_snapshot, processing_threads: match self.args.arg_snapshot_threads { Some(threads) if threads > 0 => threads, - _ => ::std::cmp::max(1, num_cpus::get() / 2), + _ => ::std::cmp::max(1, num_cpus::get_physical() / 2), }, }; diff --git a/parity/lib.rs b/parity/lib.rs index 9141bdcad..a88a68947 100644 --- a/parity/lib.rs +++ b/parity/lib.rs @@ -15,7 +15,6 @@ // along with Parity Ethereum. If not, see . //! Ethcore client application. - #![warn(missing_docs)] extern crate ansi_term; diff --git a/parity/run.rs b/parity/run.rs index a23f3d032..6ddeadbe0 100644 --- a/parity/run.rs +++ b/parity/run.rs @@ -893,17 +893,27 @@ impl RunningClient { // Create a weak reference to the client so that we can wait on shutdown // until it is dropped let weak_client = Arc::downgrade(&client); - // Shutdown and drop the ServiceClient + // Shutdown and drop the ClientService client_service.shutdown(); + trace!(target: "shutdown", "ClientService shut down"); drop(client_service); + trace!(target: "shutdown", "ClientService dropped"); // drop this stuff as soon as exit detected. drop(rpc); + trace!(target: "shutdown", "RPC dropped"); drop(keep_alive); + trace!(target: "shutdown", "KeepAlive dropped"); // to make sure timer does not spawn requests while shutdown is in progress informant.shutdown(); + trace!(target: "shutdown", "Informant shut down"); // just Arc is dropping here, to allow other reference release in its default time drop(informant); + trace!(target: "shutdown", "Informant dropped"); drop(client); + trace!(target: "shutdown", "Client dropped"); + // This may help when debugging ref cycles. Requires nightly-only `#![feature(weak_counts)]` + // trace!(target: "shutdown", "Waiting for refs to Client to shutdown, strong_count={:?}, weak_count={:?}", weak_client.strong_count(), weak_client.weak_count()); + trace!(target: "shutdown", "Waiting for refs to Client to shutdown"); wait_for_drop(weak_client); } } @@ -937,24 +947,30 @@ fn print_running_environment(data_dir: &str, dirs: &Directories, db_dirs: &Datab } fn wait_for_drop(w: Weak) { - let sleep_duration = Duration::from_secs(1); - let warn_timeout = Duration::from_secs(60); - let max_timeout = Duration::from_secs(300); + const SLEEP_DURATION: Duration = Duration::from_secs(1); + const WARN_TIMEOUT: Duration = Duration::from_secs(60); + const MAX_TIMEOUT: Duration = Duration::from_secs(300); let instant = Instant::now(); let mut warned = false; - while instant.elapsed() < max_timeout { + while instant.elapsed() < MAX_TIMEOUT { if w.upgrade().is_none() { return; } - if !warned && instant.elapsed() > warn_timeout { + if !warned && instant.elapsed() > WARN_TIMEOUT { warned = true; warn!("Shutdown is taking longer than expected."); } - thread::sleep(sleep_duration); + thread::sleep(SLEEP_DURATION); + + // When debugging shutdown issues on a nightly build it can help to enable this with the + // `#![feature(weak_counts)]` added to lib.rs (TODO: enable when + // https://github.com/rust-lang/rust/issues/57977 is stable) + // trace!(target: "shutdown", "Waiting for client to drop, strong_count={:?}, weak_count={:?}", w.strong_count(), w.weak_count()); + trace!(target: "shutdown", "Waiting for client to drop"); } warn!("Shutdown timeout reached, exiting uncleanly."); diff --git a/parity/snapshot.rs b/parity/snapshot.rs index 70957762f..269965c33 100644 --- a/parity/snapshot.rs +++ b/parity/snapshot.rs @@ -261,7 +261,7 @@ impl SnapshotCommand { let cur_size = p.size(); if cur_size != last_size { last_size = cur_size; - let bytes = ::informant::format_bytes(p.size()); + let bytes = ::informant::format_bytes(cur_size as usize); info!("Snapshot: {} accounts {} blocks {}", p.accounts(), p.blocks(), bytes); } diff --git a/rpc/src/v1/tests/helpers/snapshot_service.rs b/rpc/src/v1/tests/helpers/snapshot_service.rs index 5450886bb..881c434e1 100644 --- a/rpc/src/v1/tests/helpers/snapshot_service.rs +++ b/rpc/src/v1/tests/helpers/snapshot_service.rs @@ -48,6 +48,7 @@ impl SnapshotService for TestSnapshotService { fn status(&self) -> RestorationStatus { self.status.lock().clone() } fn begin_restore(&self, _manifest: ManifestData) { } fn abort_restore(&self) { } + fn abort_snapshot(&self) {} fn restore_state_chunk(&self, _hash: H256, _chunk: Bytes) { } fn restore_block_chunk(&self, _hash: H256, _chunk: Bytes) { } fn shutdown(&self) { } diff --git a/scripts/docker/hub/publish-docker.sh b/scripts/docker/hub/publish-docker.sh index 6602d55c2..84feedb28 100755 --- a/scripts/docker/hub/publish-docker.sh +++ b/scripts/docker/hub/publish-docker.sh @@ -3,7 +3,9 @@ set -e # fail on any error VERSION=$(cat ./tools/VERSION) +TRACK=$(cat ./tools/TRACK) echo "Parity Ethereum version = ${VERSION}" +echo "Parity Ethereum track = ${TRACK}" test "$Docker_Hub_User_Parity" -a "$Docker_Hub_Pass_Parity" \ || ( echo "no docker credentials provided"; exit 1 ) @@ -44,6 +46,14 @@ case "${SCHEDULE_TAG:-${CI_COMMIT_REF_NAME}}" in --file tools/Dockerfile .; docker push "parity/parity:${VERSION}-${CI_COMMIT_REF_NAME}"; docker push "parity/parity:stable";; + v[0-9]*.[0-9]*) + echo "Docker TAG - 'parity/parity:${VERSION}-${TRACK}'" + docker build --no-cache \ + --build-arg VCS_REF="${CI_COMMIT_SHA}" \ + --build-arg BUILD_DATE="$(date -u '+%Y-%m-%dT%H:%M:%SZ')" \ + --tag "parity/parity:${VERSION}-${TRACK}" \ + --file tools/Dockerfile .; + docker push "parity/parity:${VERSION}-${TRACK}";; *) echo "Docker TAG - 'parity/parity:${VERSION}-${CI_COMMIT_REF_NAME}'" docker build --no-cache \ diff --git a/util/network-devp2p/src/discovery.rs b/util/network-devp2p/src/discovery.rs index 7bf8dc62e..f18469e16 100644 --- a/util/network-devp2p/src/discovery.rs +++ b/util/network-devp2p/src/discovery.rs @@ -168,7 +168,6 @@ pub struct Discovery<'a> { discovery_id: NodeId, discovery_nodes: HashSet, node_buckets: Vec, - // Sometimes we don't want to add nodes to the NodeTable, but still want to // keep track of them to avoid excessive pinging (happens when an unknown node sends // a discovery request to us -- the node might be on a different net). @@ -257,7 +256,7 @@ impl<'a> Discovery<'a> { Ok(()) => None, Err(BucketError::Ourselves) => None, Err(BucketError::NotInTheBucket{node_entry, bucket_distance}) => Some((node_entry, bucket_distance)) - }.map(|(node_entry, bucket_distance)| { + }.and_then(|(node_entry, bucket_distance)| { trace!(target: "discovery", "Adding a new node {:?} into our bucket {}", &node_entry, bucket_distance); let mut added = HashMap::with_capacity(1); @@ -265,7 +264,7 @@ impl<'a> Discovery<'a> { let node_to_ping = { let bucket = &mut self.node_buckets[bucket_distance]; - bucket.nodes.push_front(BucketEntry::new(node_entry)); + bucket.nodes.push_front(BucketEntry::new(node_entry.clone())); if bucket.nodes.len() > BUCKET_SIZE { select_bucket_ping(bucket.nodes.iter()) } else { @@ -275,7 +274,12 @@ impl<'a> Discovery<'a> { if let Some(node) = node_to_ping { self.try_ping(node, PingReason::Default); }; - TableUpdates{added, removed: HashSet::new()} + + if node_entry.endpoint.is_valid_sync_node() { + Some(TableUpdates { added, removed: HashSet::new() }) + } else { + None + } }) } @@ -518,7 +522,18 @@ impl<'a> Discovery<'a> { fn on_ping(&mut self, rlp: &Rlp, node_id: &NodeId, from: &SocketAddr, echo_hash: &[u8]) -> Result, Error> { trace!(target: "discovery", "Got Ping from {:?}", &from); - let ping_from = NodeEndpoint::from_rlp(&rlp.at(1)?)?; + let ping_from = if let Ok(node_endpoint) = NodeEndpoint::from_rlp(&rlp.at(1)?) { + node_endpoint + } else { + let mut address = from.clone(); + // address here is the node's tcp port. If we are unable to get the `NodeEndpoint` from the `ping_from` + // rlp field then this is most likely a BootNode, set the tcp port to 0 because it can not be used for syncing. + address.set_port(0); + NodeEndpoint { + address, + udp_port: from.port() + } + }; let ping_to = NodeEndpoint::from_rlp(&rlp.at(2)?)?; let timestamp: u64 = rlp.val_at(3)?; self.check_timestamp(timestamp)?; @@ -540,7 +555,7 @@ impl<'a> Discovery<'a> { self.send_packet(PACKET_PONG, from, &response.drain())?; let entry = NodeEntry { id: *node_id, endpoint: pong_to.clone() }; - if !entry.endpoint.is_valid() { + if !entry.endpoint.is_valid_discovery_node() { debug!(target: "discovery", "Got bad address: {:?}", entry); } else if !self.is_allowed(&entry) { debug!(target: "discovery", "Address not allowed: {:?}", entry); @@ -728,7 +743,7 @@ impl<'a> Discovery<'a> { trace!(target: "discovery", "Got {} Neighbours from {:?}", results_count, &from); for r in rlp.at(0)?.iter() { let endpoint = NodeEndpoint::from_rlp(&r)?; - if !endpoint.is_valid() { + if !endpoint.is_valid_discovery_node() { debug!(target: "discovery", "Bad address: {:?}", endpoint); continue; } diff --git a/util/network-devp2p/src/node_table.rs b/util/network-devp2p/src/node_table.rs index 3cee93fd9..db001bfe7 100644 --- a/util/network-devp2p/src/node_table.rs +++ b/util/network-devp2p/src/node_table.rs @@ -103,10 +103,16 @@ impl NodeEndpoint { self.to_rlp(rlp); } - /// Validates that the port is not 0 and address IP is specified - pub fn is_valid(&self) -> bool { - self.udp_port != 0 && self.address.port() != 0 && - match self.address { + /// Validates that the tcp port is not 0 and that the node is a valid discovery node (i.e. `is_valid_discovery_node()` is true). + /// Sync happens over tcp. + pub fn is_valid_sync_node(&self) -> bool { + self.is_valid_discovery_node() && self.address.port() != 0 + } + + /// Validates that the udp port is not 0 and address IP is specified. + /// Peer discovery happens over udp. + pub fn is_valid_discovery_node(&self) -> bool { + self.udp_port != 0 && match self.address { SocketAddr::V4(a) => !a.ip().is_unspecified(), SocketAddr::V6(a) => !a.ip().is_unspecified() } diff --git a/util/version/Cargo.toml b/util/version/Cargo.toml index 978c60490..64239cb46 100644 --- a/util/version/Cargo.toml +++ b/util/version/Cargo.toml @@ -3,7 +3,7 @@ [package] name = "parity-version" # NOTE: this value is used for Parity Ethereum version string (via env CARGO_PKG_VERSION) -version = "2.5.2" +version = "2.5.3" authors = ["Parity Technologies "] build = "build.rs"