Backporting to beta (#3980)

* v1.4.7

* Allow sync reorg up to pruning history size

* Peer difficulty tracking

* Abort downloading block if received with NewBlock

* Fixed test


Former-commit-id: f2058bdd2feefe3651134aa07e6ce9e3e041fbec

ethcore/src/client/client.rs

@@ -46,6 +46,7 @@ use transaction::{LocalizedTransaction, SignedTransaction, Action};
 use blockchain::extras::TransactionAddress;
 use types::filter::Filter;
 use types::mode::Mode as IpcMode;
+use types::pruning_info::PruningInfo;
 use log_entry::LocalizedLogEntry;
 use verification::queue::BlockQueue;
 use blockchain::{BlockChain, BlockProvider, TreeRoute, ImportRoute};
@@ -1081,7 +1082,7 @@ impl BlockChainClient for Client {
 	}
 
 	fn import_block(&self, bytes: Bytes) -> Result<H256, BlockImportError> {
-		use verification::queue::kind::HasHash;
+		use verification::queue::kind::BlockLike;
 		use verification::queue::kind::blocks::Unverified;
 
 		// create unverified block here so the `sha3` calculation can be cached.
@@ -1121,7 +1122,9 @@ impl BlockChainClient for Client {
 	}
 
 	fn chain_info(&self) -> BlockChainInfo {
-		self.chain.read().chain_info()
+		let mut chain_info = self.chain.read().chain_info();
+		chain_info.pending_total_difficulty = chain_info.total_difficulty + self.block_queue.total_difficulty();
+		chain_info
 	}
 
 	fn additional_params(&self) -> BTreeMap<String, String> {
@@ -1226,6 +1229,12 @@ impl BlockChainClient for Client {
 		self.uncle(id)
 			.map(|header| self.engine.extra_info(&decode(&header)))
 	}
+
+	fn pruning_info(&self) -> PruningInfo {
+		PruningInfo {
+			history_size: Some(self.history),
+		}
+	}
 }
 
 impl MiningBlockChainClient for Client {

ethcore/src/client/test_client.rs

@@ -38,6 +38,7 @@ use evm::{Factory as EvmFactory, VMType, Schedule};
 use miner::{Miner, MinerService, TransactionImportResult};
 use spec::Spec;
 use types::mode::Mode;
+use types::pruning_info::PruningInfo;
 use views::BlockView;
 
 use verification::queue::QueueInfo;
@@ -89,6 +90,8 @@ pub struct TestBlockChainClient {
 	pub ancient_block: RwLock<Option<(H256, u64)>>,
 	/// First block info.
 	pub first_block: RwLock<Option<(H256, u64)>>,
+	/// Pruning history size to report.
+	pub history: RwLock<Option<u64>>,
 }
 
 #[derive(Clone)]
@@ -140,6 +143,7 @@ impl TestBlockChainClient {
 			latest_block_timestamp: RwLock::new(10_000_000),
 			ancient_block: RwLock::new(None),
 			first_block: RwLock::new(None),
+			history: RwLock::new(None),
 		};
 		client.add_blocks(1, EachBlockWith::Nothing); // add genesis block
 		client.genesis_hash = client.last_hash.read().clone();
@@ -300,6 +304,11 @@ impl TestBlockChainClient {
 		let res = res.into_iter().next().unwrap().expect("Successful import");
 		assert_eq!(res, TransactionImportResult::Current);
 	}
+
+	/// Set reported history size.
+	pub fn set_history(&self, h: Option<u64>) {
+		*self.history.write() = h;
+	}
 }
 
 pub fn get_temp_state_db() -> GuardedTempResult<StateDB> {
@@ -650,4 +659,10 @@ impl BlockChainClient for TestBlockChainClient {
 	fn mode(&self) -> Mode { Mode::Active }
 
 	fn set_mode(&self, _: Mode) { unimplemented!(); }
+
+	fn pruning_info(&self) -> PruningInfo {
+		PruningInfo {
+			history_size: *self.history.read(),
+		}
+	}
 }

ethcore/src/client/traits.rs

@@ -39,6 +39,7 @@ use types::call_analytics::CallAnalytics;
 use types::blockchain_info::BlockChainInfo;
 use types::block_status::BlockStatus;
 use types::mode::Mode;
+use types::pruning_info::PruningInfo;
 
 #[ipc(client_ident="RemoteClient")]
 /// Blockchain database client. Owns and manages a blockchain and a block queue.
@@ -241,6 +242,9 @@ pub trait BlockChainClient : Sync + Send {
 
 	/// Returns engine-related extra info for `UncleID`.
 	fn uncle_extra_info(&self, id: UncleID) -> Option<BTreeMap<String, String>>;
+
+	/// Get pruning settings.
+	fn pruning_info(&self) -> PruningInfo;
 }
 
 /// Extended client interface used for mining

ethcore/src/types/mod.rs.in

@@ -33,4 +33,5 @@ pub mod transaction_import;
 pub mod block_import_error;
 pub mod restoration_status;
 pub mod snapshot_manifest;
-pub mod mode;
+pub mod mode;
+pub mod pruning_info;

ethcore/src/types/pruning_info.rs

@@ -0,0 +1,29 @@
+// Copyright 2015, 2016 Parity Technologies (UK) Ltd.
+// This file is part of Parity.
+
+// Parity is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// Parity is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+
+// You should have received a copy of the GNU General Public License
+// along with Parity.  If not, see <http://www.gnu.org/licenses/>.
+
+//! Information about portions of the state and chain which the client may serve.
+//!
+//! Currently assumes that a client will store everything past a certain point
+//! or everything. Will be extended in the future to support a definition
+//! of which portions of the ancient chain and current state trie are stored as well.
+
+/// Client pruning info. See module-level docs for more details.
+#[derive(Debug, Clone)]
+#[cfg_attr(feature = "ipc", binary)]
+pub struct PruningInfo {
+	/// Pruning history size
+	pub history_size: Option<u64>,
+}

ethcore/src/verification/queue/kind.rs

@@ -19,18 +19,21 @@
 use engines::Engine;
 use error::Error;
 
-use util::{HeapSizeOf, H256};
+use util::{HeapSizeOf, H256, U256};
 
 pub use self::blocks::Blocks;
 pub use self::headers::Headers;
 
 /// Something which can produce a hash and a parent hash.
-pub trait HasHash {
+pub trait BlockLike {
 	/// Get the hash of this item.
 	fn hash(&self) -> H256;
 
 	/// Get the hash of this item's parent.
 	fn parent_hash(&self) -> H256;
+
+	/// Get the difficulty of this item.
+	fn difficulty(&self) -> U256;
 }
 
 /// Defines transitions between stages of verification.
@@ -45,13 +48,13 @@ pub trait HasHash {
 /// consistent.
 pub trait Kind: 'static + Sized + Send + Sync {
 	/// The first stage: completely unverified.
-	type Input: Sized + Send + HasHash + HeapSizeOf;
+	type Input: Sized + Send + BlockLike + HeapSizeOf;
 
 	/// The second stage: partially verified.
-	type Unverified: Sized + Send + HasHash + HeapSizeOf;
+	type Unverified: Sized + Send + BlockLike + HeapSizeOf;
 
 	/// The third stage: completely verified.
-	type Verified: Sized + Send + HasHash + HeapSizeOf;
+	type Verified: Sized + Send + BlockLike + HeapSizeOf;
 
 	/// Attempt to create the `Unverified` item from the input.
 	fn create(input: Self::Input, engine: &Engine) -> Result<Self::Unverified, Error>;
@@ -62,14 +65,14 @@ pub trait Kind: 'static + Sized + Send + Sync {
 
 /// The blocks verification module.
 pub mod blocks {
-	use super::{Kind, HasHash};
+	use super::{Kind, BlockLike};
 
 	use engines::Engine;
 	use error::Error;
 	use header::Header;
 	use verification::{PreverifiedBlock, verify_block_basic, verify_block_unordered};
 
-	use util::{Bytes, HeapSizeOf, H256};
+	use util::{Bytes, HeapSizeOf, H256, U256};
 
 	/// A mode for verifying blocks.
 	pub struct Blocks;
@@ -126,7 +129,7 @@ pub mod blocks {
 		}
 	}
 
-	impl HasHash for Unverified {
+	impl BlockLike for Unverified {
 		fn hash(&self) -> H256 {
 			self.header.hash()
 		}
@@ -134,9 +137,13 @@ pub mod blocks {
 		fn parent_hash(&self) -> H256 {
 			self.header.parent_hash().clone()
 		}
+
+		fn difficulty(&self) -> U256 {
+			self.header.difficulty().clone()
+		}
 	}
 
-	impl HasHash for PreverifiedBlock {
+	impl BlockLike for PreverifiedBlock {
 		fn hash(&self) -> H256 {
 			self.header.hash()
 		}
@@ -144,12 +151,16 @@ pub mod blocks {
 		fn parent_hash(&self) -> H256 {
 			self.header.parent_hash().clone()
 		}
+
+		fn difficulty(&self) -> U256 {
+			self.header.difficulty().clone()
+		}
 	}
 }
 
 /// Verification for headers.
 pub mod headers {
-	use super::{Kind, HasHash};
+	use super::{Kind, BlockLike};
 
 	use engines::Engine;
 	use error::Error;
@@ -157,10 +168,12 @@ pub mod headers {
 	use verification::verify_header_params;
 
 	use util::hash::H256;
+	use util::U256;
 
-	impl HasHash for Header {
+	impl BlockLike for Header {
 		fn hash(&self) -> H256 { self.hash() }
 		fn parent_hash(&self) -> H256 { self.parent_hash().clone() }
+		fn difficulty(&self) -> U256 { self.difficulty().clone() }
 	}
 
 	/// A mode for verifying headers.

ethcore/src/verification/queue/mod.rs

@@ -26,7 +26,7 @@ use error::*;
 use engines::Engine;
 use service::*;
 
-use self::kind::{HasHash, Kind};
+use self::kind::{BlockLike, Kind};
 
 pub use types::verification_queue_info::VerificationQueueInfo as QueueInfo;
 
@@ -101,9 +101,10 @@ pub struct VerificationQueue<K: Kind> {
 	deleting: Arc<AtomicBool>,
 	ready_signal: Arc<QueueSignal>,
 	empty: Arc<SCondvar>,
-	processing: RwLock<HashSet<H256>>,
+	processing: RwLock<HashMap<H256, U256>>, // hash to difficulty
 	max_queue_size: usize,
 	max_mem_use: usize,
+	total_difficulty: RwLock<U256>,
 }
 
 struct QueueSignal {
@@ -214,10 +215,11 @@ impl<K: Kind> VerificationQueue<K> {
 			verification: verification.clone(),
 			verifiers: verifiers,
 			deleting: deleting.clone(),
-			processing: RwLock::new(HashSet::new()),
+			processing: RwLock::new(HashMap::new()),
 			empty: empty.clone(),
 			max_queue_size: max(config.max_queue_size, MIN_QUEUE_LIMIT),
 			max_mem_use: max(config.max_mem_use, MIN_MEM_LIMIT),
+			total_difficulty: RwLock::new(0.into()),
 		}
 	}
 
@@ -335,6 +337,7 @@ impl<K: Kind> VerificationQueue<K> {
 		sizes.unverified.store(0, AtomicOrdering::Release);
 		sizes.verifying.store(0, AtomicOrdering::Release);
 		sizes.verified.store(0, AtomicOrdering::Release);
+		*self.total_difficulty.write() = 0.into();
 
 		self.processing.write().clear();
 	}
@@ -349,7 +352,7 @@ impl<K: Kind> VerificationQueue<K> {
 
 	/// Check if the item is currently in the queue
 	pub fn status(&self, hash: &H256) -> Status {
-		if self.processing.read().contains(hash) {
+		if self.processing.read().contains_key(hash) {
 			return Status::Queued;
 		}
 		if self.verification.bad.lock().contains(hash) {
@@ -362,7 +365,7 @@ impl<K: Kind> VerificationQueue<K> {
 	pub fn import(&self, input: K::Input) -> ImportResult {
 		let h = input.hash();
 		{
-			if self.processing.read().contains(&h) {
+			if self.processing.read().contains_key(&h) {
 				return Err(ImportError::AlreadyQueued.into());
 			}
 
@@ -381,7 +384,11 @@ impl<K: Kind> VerificationQueue<K> {
 			Ok(item) => {
 				self.verification.sizes.unverified.fetch_add(item.heap_size_of_children(), AtomicOrdering::SeqCst);
 
-				self.processing.write().insert(h.clone());
+				self.processing.write().insert(h.clone(), item.difficulty());
+				{
+					let mut td = self.total_difficulty.write();
+					*td = *td + item.difficulty();
+				}
 				self.verification.unverified.lock().push_back(item);
 				self.more_to_verify.notify_all();
 				Ok(h)
@@ -406,7 +413,10 @@ impl<K: Kind> VerificationQueue<K> {
 		bad.reserve(hashes.len());
 		for hash in hashes {
 			bad.insert(hash.clone());
-			processing.remove(hash);
+			if let Some(difficulty) = processing.remove(hash) {
+				let mut td = self.total_difficulty.write();
+				*td = *td - difficulty;
+			}
 		}
 
 		let mut new_verified = VecDeque::new();
@@ -415,7 +425,10 @@ impl<K: Kind> VerificationQueue<K> {
 			if bad.contains(&output.parent_hash()) {
 				removed_size += output.heap_size_of_children();
 				bad.insert(output.hash());
-				processing.remove(&output.hash());
+				if let Some(difficulty) = processing.remove(&output.hash()) {
+					let mut td = self.total_difficulty.write();
+					*td = *td - difficulty;
+				}
 			} else {
 				new_verified.push_back(output);
 			}
@@ -433,7 +446,10 @@ impl<K: Kind> VerificationQueue<K> {
 		}
 		let mut processing = self.processing.write();
 		for hash in hashes {
-			processing.remove(hash);
+			if let Some(difficulty) = processing.remove(hash) {
+				let mut td = self.total_difficulty.write();
+				*td = *td - difficulty;
+			}
 		}
 		processing.is_empty()
 	}
@@ -487,7 +503,13 @@ impl<K: Kind> VerificationQueue<K> {
 		}
 	}
 
-	/// Optimise memory footprint of the heap fields.
+	/// Get the total difficulty of all the blocks in the queue.
+	pub fn total_difficulty(&self) -> U256 {
+		self.total_difficulty.read().clone()
+	}
+
+	/// Optimise memory footprint of the heap fields, and adjust the number of threads
+	/// to better suit the workload.
 	pub fn collect_garbage(&self) {
 		{
 			self.verification.unverified.lock().shrink_to_fit();
@@ -569,6 +591,22 @@ mod tests {
 		}
 	}
 
+	#[test]
+	fn returns_total_difficulty() {
+		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(Unverified::new(block)) {
+			panic!("error importing block that is valid by definition({:?})", e);
+		}
+		queue.flush();
+		assert_eq!(queue.total_difficulty(), 131072.into());
+		queue.drain(10);
+		assert_eq!(queue.total_difficulty(), 131072.into());
+		queue.mark_as_good(&[ hash ]);
+		assert_eq!(queue.total_difficulty(), 0.into());
+	}
+
 	#[test]
 	fn returns_ok_for_drained_duplicates() {
 		let queue = get_test_queue();