[beta] Backports (#8785)

* Fix light sync with initial validator-set contract (#8528) * Fix #8468 * Use U256::max_value() instead * Fix again * Also change initial transaction gas * Resumable warp-sync / Seed downloaded snapshots (#8544) * Start dividing sync chain : first supplier method * WIP - updated chain sync supplier * Finish refactoring the Chain Sync Supplier * Create Chain Sync Requester * Add Propagator for Chain Sync * Add the Chain Sync Handler * Move tests from mod -> handler * Move tests to propagator * Refactor SyncRequester arguments * Refactoring peer fork header handler * Fix wrong highest block number in snapshot sync * Small refactor... * Resume warp-sync downloaded chunks * Add comments * Refactoring the previous chunks import * Fix tests * Address PR grumbles * Fix not seeding current snapshot * Address PR Grumbles * Address PR grumble * Retry failed CI job * Update SnapshotService readiness check Fix restoration locking issue for previous chunks restoration * Fix tests * Fix tests * Fix test * Early abort importing previous chunks * PR Grumbles * Update Gitlab CI config * SyncState back to Waiting when Manifest peers disconnect * Move fix * Better fix * Revert GitLab CI changes * Fix Warning * Refactor resuming snapshots * Fix string construction * Revert "Refactor resuming snapshots" This reverts commit 75fd4b553a38e4a49dc5d6a878c70e830ff382eb. * Update informant log * Fix string construction * Refactor resuming snapshots * Fix informant * PR Grumbles * Update informant message : show chunks done * PR Grumbles * Fix * Fix Warning * PR Grumbles * Don't open Browser post-install on Mac (#8641) Since we start parity with the UI disabled per default now, opening the browser post installation will show an annoying error message, confusing the user. This patch removes opening the browser to prevent that annoyance. fixes #8194 * Fix not downloading old blocks (#8642) * Fix PoW blockchains sealing notifications in chain_new_blocks (#8656) * Shutdown the Snapshot Service early (#8658) * Shutdown the Snapshot Service when shutting down the runner * Rename `service` to `client_service` * Fix tests * Fix cli signer (#8682) * Update ethereum-types so `{:#x}` applies 0x prefix * Set the request index to that of the current request (#8683) * Set the request index to that of the current request When setting up the chain of (two) requests to look up a block by hash, the second need to refer to the first. This fixes an issue where the back ref was set to the subsequent request, not the current one. When the requests are executed we loop through them in order and ensure the requests that should produce headers all match up. We do this by index so they better be right. In other words: off by one. * network-devp2p: handle UselessPeer disconnect (#8686) * Fix local transactions policy. (#8691) * CI: Fixes for Android Pipeline (#8745) * ci: Remove check for shared libraries in gitlab script * ci: allow android arm build to fail * Custom Error Messages on ENFILE and EMFILE IO Errors (#8744) * Custom Error Messages on ENFILE and EMFILE IO Errors Add custom mapping of ENFILE and EMFILE IO Errors (Failure because of missing system resource) right when chaining ioError into ::util::Network::Error to improve Error Messages given to user Note: Adds libc as a dependency to util/network * Use assert-matches for more readable tests * Fix Wording and consistency * ethcore-sync: fix connection to peers behind chain fork block (#8710)
2018-06-04 21:29:20 +01:00
parent 69d6d8239f
commit 7bf16cce1b
30 changed files with 706 additions and 265 deletions
--- a/ethcore/src/client/client.rs
+++ b/ethcore/src/client/client.rs
@@ -18,7 +18,7 @@ use std::collections::{HashSet, HashMap, BTreeMap, BTreeSet, VecDeque};
 use std::fmt;
 use std::str::FromStr;
 use std::sync::{Arc, Weak};
-use std::sync::atomic::{AtomicUsize, AtomicBool, Ordering as AtomicOrdering};
+use std::sync::atomic::{AtomicBool, Ordering as AtomicOrdering};
 use std::time::{Instant};

 // util
@@ -208,6 +208,8 @@ pub struct Client {
 	queue_transactions: IoChannelQueue,
 	/// Ancient blocks import queue
 	queue_ancient_blocks: IoChannelQueue,
+	/// Hashes of pending ancient block wainting to be included
+	pending_ancient_blocks: RwLock<HashSet<H256>>,
 	/// Consensus messages import queue
 	queue_consensus_message: IoChannelQueue,

@@ -461,6 +463,7 @@ impl Importer {
 		let hash = header.hash();
 		let _import_lock = self.import_lock.lock();

+		trace!(target: "client", "Trying to import old block #{}", header.number());
 		{
 			trace_time!("import_old_block");
 			// verify the block, passing the chain for updating the epoch verifier.
@@ -741,6 +744,7 @@ impl Client {
 			notify: RwLock::new(Vec::new()),
 			queue_transactions: IoChannelQueue::new(MAX_TX_QUEUE_SIZE),
 			queue_ancient_blocks: IoChannelQueue::new(MAX_ANCIENT_BLOCKS_QUEUE_SIZE),
+			pending_ancient_blocks: RwLock::new(HashSet::new()),
 			queue_consensus_message: IoChannelQueue::new(usize::max_value()),
 			last_hashes: RwLock::new(VecDeque::new()),
 			factories: factories,
@@ -1972,7 +1976,7 @@ impl BlockChainClient for Client {
 impl IoClient for Client {
 	fn queue_transactions(&self, transactions: Vec<Bytes>, peer_id: usize) {
 		let len = transactions.len();
-		self.queue_transactions.queue(&mut self.io_channel.lock(), len, move |client| {
+		self.queue_transactions.queue(&mut self.io_channel.lock(), move |client| {
 			trace_time!("import_queued_transactions");

 			let txs: Vec<UnverifiedTransaction> = transactions
@@ -1996,23 +2000,32 @@ impl IoClient for Client {

 		{
 			// check block order
-			if self.chain.read().is_known(&header.hash()) {
+			if self.chain.read().is_known(&hash) {
 				bail!(BlockImportErrorKind::Import(ImportErrorKind::AlreadyInChain));
 			}
-			let status = self.block_status(BlockId::Hash(*header.parent_hash()));
-			if  status == BlockStatus::Unknown || status == BlockStatus::Pending {
-				bail!(BlockImportErrorKind::Block(BlockError::UnknownParent(*header.parent_hash())));
+
+			let parent_hash = *header.parent_hash();
+			let parent_pending = self.pending_ancient_blocks.read().contains(&parent_hash);
+			let status = self.block_status(BlockId::Hash(parent_hash));
+			if !parent_pending && (status == BlockStatus::Unknown || status == BlockStatus::Pending) {
+				bail!(BlockImportErrorKind::Block(BlockError::UnknownParent(parent_hash)));
 			}
 		}

-		match self.queue_ancient_blocks.queue(&mut self.io_channel.lock(), 1, move |client| {
-			client.importer.import_old_block(
+		self.pending_ancient_blocks.write().insert(hash);
+
+		trace!(target: "client", "Queuing old block #{}", header.number());
+		match self.queue_ancient_blocks.queue(&mut self.io_channel.lock(), move |client| {
+			let result = client.importer.import_old_block(
 				&header,
 				&block_bytes,
 				&receipts_bytes,
 				&**client.db.read(),
 				&*client.chain.read()
-			).map(|_| ()).unwrap_or_else(|e| {
+			);
+
+			client.pending_ancient_blocks.write().remove(&hash);
+			result.map(|_| ()).unwrap_or_else(|e| {
 				error!(target: "client", "Error importing ancient block: {}", e);
 			});
 		}) {
@@ -2022,7 +2035,7 @@ impl IoClient for Client {
 	}

 	fn queue_consensus_message(&self, message: Bytes) {
-		match self.queue_consensus_message.queue(&mut self.io_channel.lock(), 1, move |client| {
+		match self.queue_consensus_message.queue(&mut self.io_channel.lock(), move |client| {
 			if let Err(e) = client.engine().handle_message(&message) {
 				debug!(target: "poa", "Invalid message received: {}", e);
 			}
@@ -2131,7 +2144,7 @@ impl ImportSealedBlock for Client {
 			route
 		};
 		let (enacted, retracted) = self.importer.calculate_enacted_retracted(&[route]);
-		self.importer.miner.chain_new_blocks(self, &[h.clone()], &[], &enacted, &retracted, true);
+		self.importer.miner.chain_new_blocks(self, &[h.clone()], &[], &enacted, &retracted, self.engine.seals_internally().is_some());
 		self.notify(|notify| {
 			notify.new_blocks(
 				vec![h.clone()],
@@ -2433,35 +2446,38 @@ impl fmt::Display for QueueError {

 /// Queue some items to be processed by IO client.
 struct IoChannelQueue {
-	currently_queued: Arc<AtomicUsize>,
+	queue: Arc<Mutex<VecDeque<Box<Fn(&Client) + Send>>>>,
 	limit: usize,
 }

 impl IoChannelQueue {
 	pub fn new(limit: usize) -> Self {
 		IoChannelQueue {
-			currently_queued: Default::default(),
+			queue: Default::default(),
 			limit,
 		}
 	}

-	pub fn queue<F>(&self, channel: &mut IoChannel<ClientIoMessage>, count: usize, fun: F) -> Result<(), QueueError> where
-		F: Fn(&Client) + Send + Sync + 'static,
+	pub fn queue<F>(&self, channel: &mut IoChannel<ClientIoMessage>, fun: F) -> Result<(), QueueError>
+		where F: Fn(&Client) + Send + Sync + 'static
 	{
-		let queue_size = self.currently_queued.load(AtomicOrdering::Relaxed);
-		ensure!(queue_size < self.limit, QueueError::Full(self.limit));
+		{
+			let mut queue = self.queue.lock();
+			let queue_size = queue.len();
+			ensure!(queue_size < self.limit, QueueError::Full(self.limit));

-		let currently_queued = self.currently_queued.clone();
+			queue.push_back(Box::new(fun));
+		}
+
+		let queue = self.queue.clone();
 		let result = channel.send(ClientIoMessage::execute(move |client| {
-			currently_queued.fetch_sub(count, AtomicOrdering::SeqCst);
-			fun(client);
+			while let Some(fun) = queue.lock().pop_front() {
+				fun(client);
+			}
 		}));

 		match result {
-			Ok(_) => {
-				self.currently_queued.fetch_add(count, AtomicOrdering::SeqCst);
-				Ok(())
-			},
+			Ok(_) => Ok(()),
 			Err(e) => Err(QueueError::Channel(e)),
 		}
 	}
--- a/ethcore/src/snapshot/service.rs
+++ b/ethcore/src/snapshot/service.rs
@@ -17,8 +17,8 @@
 //! Snapshot network service implementation.

 use std::collections::HashSet;
-use std::io::ErrorKind;
-use std::fs;
+use std::io::{self, Read, ErrorKind};
+use std::fs::{self, File};
 use std::path::PathBuf;
 use std::sync::Arc;
 use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
@@ -30,6 +30,7 @@ use blockchain::BlockChain;
 use client::{Client, ChainInfo, ClientIoMessage};
 use engines::EthEngine;
 use error::Error;
+use hash::keccak;
 use ids::BlockId;

 use io::IoChannel;
@@ -270,8 +271,8 @@ impl Service {
 			}
 		}

-		// delete the temporary restoration dir if it does exist.
-		if let Err(e) = fs::remove_dir_all(service.restoration_dir()) {
+		// delete the temporary restoration DB dir if it does exist.
+		if let Err(e) = fs::remove_dir_all(service.restoration_db()) {
 			if e.kind() != ErrorKind::NotFound {
 				return Err(e.into())
 			}
@@ -325,6 +326,13 @@ impl Service {
 		dir
 	}

+	// previous snapshot chunks path.
+	fn prev_chunks_dir(&self) -> PathBuf {
+		let mut dir = self.snapshot_root.clone();
+		dir.push("prev_chunks");
+		dir
+	}
+
 	// replace one the client's database with our own.
 	fn replace_client_db(&self) -> Result<(), Error> {
 		let our_db = self.restoration_db();
@@ -406,10 +414,27 @@ impl Service {
 	/// Initialize the restoration synchronously.
 	/// The recover flag indicates whether to recover the restored snapshot.
 	pub fn init_restore(&self, manifest: ManifestData, recover: bool) -> Result<(), Error> {
-		let rest_dir = self.restoration_dir();
-
 		let mut res = self.restoration.lock();

+		let rest_dir = self.restoration_dir();
+		let rest_db = self.restoration_db();
+		let recovery_temp = self.temp_recovery_dir();
+		let prev_chunks = self.prev_chunks_dir();
+
+		// delete and restore the restoration dir.
+		if let Err(e) = fs::remove_dir_all(&prev_chunks) {
+			match e.kind() {
+				ErrorKind::NotFound => {},
+				_ => return Err(e.into()),
+			}
+		}
+
+		// Move the previous recovery temp directory
+		// to `prev_chunks` to be able to restart restoring
+		// with previously downloaded blocks
+		// This step is optional, so don't fail on error
+		fs::rename(&recovery_temp, &prev_chunks).ok();
+
 		self.state_chunks.store(0, Ordering::SeqCst);
 		self.block_chunks.store(0, Ordering::SeqCst);

@@ -424,29 +449,38 @@ impl Service {
 			}
 		}

+		*self.status.lock() = RestorationStatus::Initializing {
+			chunks_done: 0,
+		};
+
 		fs::create_dir_all(&rest_dir)?;

 		// make new restoration.
 		let writer = match recover {
-			true => Some(LooseWriter::new(self.temp_recovery_dir())?),
+			true => Some(LooseWriter::new(recovery_temp)?),
 			false => None
 		};

 		let params = RestorationParams {
-			manifest: manifest,
+			manifest: manifest.clone(),
 			pruning: self.pruning,
-			db: self.restoration_db_handler.open(&self.restoration_db())?,
+			db: self.restoration_db_handler.open(&rest_db)?,
 			writer: writer,
 			genesis: &self.genesis_block,
-			guard: Guard::new(rest_dir),
+			guard: Guard::new(rest_db),
 			engine: &*self.engine,
 		};

-		let state_chunks = params.manifest.state_hashes.len();
-		let block_chunks = params.manifest.block_hashes.len();
+		let state_chunks = manifest.state_hashes.len();
+		let block_chunks = manifest.block_hashes.len();

 		*res = Some(Restoration::new(params)?);

+		self.restoring_snapshot.store(true, Ordering::SeqCst);
+
+		// Import previous chunks, continue if it fails
+		self.import_prev_chunks(&mut res, manifest).ok();
+
 		*self.status.lock() = RestorationStatus::Ongoing {
 			state_chunks: state_chunks as u32,
 			block_chunks: block_chunks as u32,
@@ -454,10 +488,65 @@ impl Service {
 			block_chunks_done: self.block_chunks.load(Ordering::SeqCst) as u32,
 		};

-		self.restoring_snapshot.store(true, Ordering::SeqCst);
 		Ok(())
 	}

+	/// Import the previous chunks into the current restoration
+	fn import_prev_chunks(&self, restoration: &mut Option<Restoration>, manifest: ManifestData) -> Result<(), Error> {
+		let prev_chunks = self.prev_chunks_dir();
+
+		// Restore previous snapshot chunks
+		let files = fs::read_dir(prev_chunks.as_path())?;
+		let mut num_temp_chunks = 0;
+
+		for prev_chunk_file in files {
+			// Don't go over all the files if the restoration has been aborted
+			if !self.restoring_snapshot.load(Ordering::SeqCst) {
+				trace!(target:"snapshot", "Aborting importing previous chunks");
+				return Ok(());
+			}
+			// Import the chunk, don't fail and continue if one fails
+			match self.import_prev_chunk(restoration, &manifest, prev_chunk_file) {
+				Ok(true) => num_temp_chunks += 1,
+				Err(e) => trace!(target: "snapshot", "Error importing chunk: {:?}", e),
+				_ => (),
+			}
+		}
+
+		trace!(target:"snapshot", "Imported {} previous chunks", num_temp_chunks);
+
+		// Remove the prev temp directory
+		fs::remove_dir_all(&prev_chunks)?;
+
+		Ok(())
+	}
+
+	/// Import a previous chunk at the given path. Returns whether the block was imported or not
+	fn import_prev_chunk(&self, restoration: &mut Option<Restoration>, manifest: &ManifestData, file: io::Result<fs::DirEntry>) -> Result<bool, Error> {
+		let file = file?;
+		let path = file.path();
+
+		let mut file = File::open(path.clone())?;
+		let mut buffer = Vec::new();
+		file.read_to_end(&mut buffer)?;
+
+		let hash = keccak(&buffer);
+
+		let is_state = if manifest.block_hashes.contains(&hash) {
+			false
+		} else if manifest.state_hashes.contains(&hash) {
+			true
+		} else {
+			return Ok(false);
+		};
+
+		self.feed_chunk_with_restoration(restoration, hash, &buffer, is_state)?;
+
+		trace!(target: "snapshot", "Fed chunk {:?}", hash);
+
+		Ok(true)
+	}
+
 	// finalize the restoration. this accepts an already-locked
 	// restoration as an argument -- so acquiring it again _will_
 	// lead to deadlock.
@@ -499,12 +588,19 @@ impl Service {
 	/// Feed a chunk of either kind. no-op if no restoration or status is wrong.
 	fn feed_chunk(&self, hash: H256, chunk: &[u8], is_state: bool) -> Result<(), Error> {
 		// TODO: be able to process block chunks and state chunks at same time?
-		let (result, db) = {
-			let mut restoration = self.restoration.lock();
+		let mut restoration = self.restoration.lock();
+		self.feed_chunk_with_restoration(&mut restoration, hash, chunk, is_state)
+	}

+	/// Feed a chunk with the Restoration
+	fn feed_chunk_with_restoration(&self, restoration: &mut Option<Restoration>, hash: H256, chunk: &[u8], is_state: bool) -> Result<(), Error> {
+		let (result, db) = {
 			match self.status() {
-				RestorationStatus::Inactive | RestorationStatus::Failed => return Ok(()),
-				RestorationStatus::Ongoing { .. } => {
+				RestorationStatus::Inactive | RestorationStatus::Failed => {
+					trace!(target: "snapshot", "Tried to restore chunk {:x} while inactive or failed", hash);
+					return Ok(());
+				},
+				RestorationStatus::Ongoing { .. } | RestorationStatus::Initializing { .. } => {
 					let (res, db) = {
 						let rest = match *restoration {
 							Some(ref mut r) => r,
@@ -583,11 +679,41 @@ impl SnapshotService for Service {
 		self.reader.read().as_ref().and_then(|r| r.chunk(hash).ok())
 	}

+	fn completed_chunks(&self) -> Option<Vec<H256>> {
+		let restoration = self.restoration.lock();
+
+		match *restoration {
+			Some(ref restoration) => {
+				let completed_chunks = restoration.manifest.block_hashes
+					.iter()
+					.filter(|h| !restoration.block_chunks_left.contains(h))
+					.chain(
+						restoration.manifest.state_hashes
+							.iter()
+							.filter(|h| !restoration.state_chunks_left.contains(h))
+					)
+					.map(|h| *h)
+					.collect();
+
+				Some(completed_chunks)
+			},
+			None => None,
+		}
+	}
+
 	fn status(&self) -> RestorationStatus {
 		let mut cur_status = self.status.lock();
-		if let RestorationStatus::Ongoing { ref mut state_chunks_done, ref mut block_chunks_done, .. } = *cur_status {
-			*state_chunks_done = self.state_chunks.load(Ordering::SeqCst) as u32;
-			*block_chunks_done = self.block_chunks.load(Ordering::SeqCst) as u32;
+
+		match *cur_status {
+			RestorationStatus::Initializing { ref mut chunks_done } => {
+				*chunks_done = self.state_chunks.load(Ordering::SeqCst) as u32 +
+					self.block_chunks.load(Ordering::SeqCst) as u32;
+			}
+			RestorationStatus::Ongoing { ref mut state_chunks_done, ref mut block_chunks_done, .. } => {
+				*state_chunks_done = self.state_chunks.load(Ordering::SeqCst) as u32;
+				*block_chunks_done = self.block_chunks.load(Ordering::SeqCst) as u32;
+			},
+			_ => (),
 		}

 		cur_status.clone()
@@ -600,6 +726,7 @@ impl SnapshotService for Service {
 	}

 	fn abort_restore(&self) {
+		trace!(target: "snapshot", "Aborting restore");
 		self.restoring_snapshot.store(false, Ordering::SeqCst);
 		*self.restoration.lock() = None;
 		*self.status.lock() = RestorationStatus::Inactive;
@@ -616,6 +743,10 @@ impl SnapshotService for Service {
 			trace!("Error sending snapshot service message: {:?}", e);
 		}
 	}
+
+	fn shutdown(&self) {
+		self.abort_restore();
+	}
 }

 impl Drop for Service {
--- a/ethcore/src/snapshot/tests/service.rs
+++ b/ethcore/src/snapshot/tests/service.rs
@@ -130,12 +130,16 @@ fn guards_delete_folders() {
 	service.init_restore(manifest.clone(), true).unwrap();
 	assert!(path.exists());

+	// The `db` folder should have been deleted,
+	// while the `temp` one kept
 	service.abort_restore();
-	assert!(!path.exists());
+	assert!(!path.join("db").exists());
+	assert!(path.join("temp").exists());

 	service.init_restore(manifest.clone(), true).unwrap();
 	assert!(path.exists());

 	drop(service);
-	assert!(!path.exists());
+	assert!(!path.join("db").exists());
+	assert!(path.join("temp").exists());
 }
--- a/ethcore/src/snapshot/traits.rs
+++ b/ethcore/src/snapshot/traits.rs
@@ -30,6 +30,9 @@ pub trait SnapshotService : Sync + Send {
 	/// `None` indicates warp sync isn't supported by the consensus engine.
 	fn supported_versions(&self) -> Option<(u64, u64)>;

+	/// Returns a list of the completed chunks
+	fn completed_chunks(&self) -> Option<Vec<H256>>;
+
 	/// Get raw chunk for a given hash.
 	fn chunk(&self, hash: H256) -> Option<Bytes>;

@@ -51,4 +54,7 @@ pub trait SnapshotService : Sync + Send {
 	/// Feed a raw block chunk to the service to be processed asynchronously.
 	/// no-op if currently restoring.
 	fn restore_block_chunk(&self, hash: H256, chunk: Bytes);
+
+	/// Shutdown the Snapshot Service by aborting any ongoing restore
+	fn shutdown(&self);
 }
--- a/ethcore/src/spec/spec.rs
+++ b/ethcore/src/spec/spec.rs
@@ -776,7 +776,7 @@ impl Spec {
 				author: *genesis.author(),
 				timestamp: genesis.timestamp(),
 				difficulty: *genesis.difficulty(),
-				gas_limit: *genesis.gas_limit(),
+				gas_limit: U256::max_value(),
 				last_hashes: Arc::new(Vec::new()),
 				gas_used: 0.into(),
 			};
@@ -785,7 +785,7 @@ impl Spec {
 			let tx = Transaction {
 				nonce: self.engine.account_start_nonce(0),
 				action: Action::Call(a),
-				gas: U256::from(50_000_000), // TODO: share with client.
+				gas: U256::max_value(),
 				gas_price: U256::default(),
 				value: U256::default(),
 				data: d,