Add JSON format to import blocks and set it as default (#5816)

* Add BlockStream Enum and utility fn * WIP: Modify import closure to work with BlockStream * Fix trait bounds * Working prototype * Revamp block importing * Add export_import_flow tests * Add comments and clean code * Add more comments in the import fn * Add link code to import function * Add condition when returning Ready(Ok(()) to make sure we've imported every block * Add check for imported blocks in JSON case * Use rest pattern * Fix compilation error for undeclared variable * Add polling and waker before pending * Print read_block_count instead of count * Simplify binary cli option with structopt * Update test to reflect changes in CLI api * Change Stream to take SignedBlock<B> instead of B * Add comments to BlockStream * Move out logic to smaller functions for clearer code * Remove result over import_blocks return type * Check for error in command output rather than simply checking command exit status * Revamp export/import/revert testing * Fix minor typos and formatting errors Co-authored-by: Bastian Köcher <bkchr@users.noreply.github.com> * Remove unnecessary if condition in terminating condition Co-authored-by: Bastian Köcher <bkchr@users.noreply.github.com> * Explicit error instead of returning it as a string Co-authored-by: Bastian Köcher <bkchr@users.noreply.github.com> * Pass BlockStream to log_importing_status_updates and simplify matching arms for block stream * Use .contains() instead of regex match * Line break in match block; return future::ready instead of poll_fn * Update Cargo.lock * Add check so that queue doesn't grow too big * Use Iterator instead of Stream * Remove allow dead_code * Remove outdated comments Co-authored-by: Bastian Köcher <bkchr@users.noreply.github.com> * Return Errors instead of logging them * Simplify match arms Co-authored-by: Bastian Köcher <bkchr@users.noreply.github.com> * Remove check before terminating block import * Apply suggestions from code review * Check that queue is not full BEFORE calling * Revert "Remove check before terminating block import" This reverts commit 377823c0a648a3eb2e61185a257a61023067893d. * Improve unit tests to make sure we actually import blocks * Remove Unpin implementation for BlockIter * Add prototype of enum for ImportStates * Add working prototype for StateMachine * Add comments for clearer code * Add sleep before calling Waker when waiting for import queue * Add Speedometer * add dbg!(&log) for test debugging * Fix lines with more than 100 cols * Fix regex capture for test * Update regexes to take to capture the whole number * Rename Cmd to Command Co-authored-by: Gavin Wood <gavin@parity.io> * Actually rename Cmd to Command * Apply suggestions from code review Co-authored-by: Gavin Wood <gavin@parity.io> * Fix compilation errors for tests * Fix compilation errors from code review suggestion * Update bin/node/cli/tests/export_import_flow.rs Co-authored-by: Bastian Köcher <bkchr@users.noreply.github.com> Co-authored-by: Gavin Wood <gavin@parity.io> Co-authored-by: Benjamin Kampmann <ben@gnunicorn.org>
2026-06-12 17:01:09 +00:00 · 2020-05-22 13:50:25 +02:00
parent 66fe846d48
commit 18d4fa10d2
10 changed files with 546 additions and 142 deletions
@@ -831,6 +831,7 @@ pub trait ServiceBuilderCommand {
 		self,
 		input: impl Read + Seek + Send + 'static,
 		force: bool,
+		binary: bool,
 	) -> Pin<Box<dyn Future<Output = Result<(), Error>> + Send>>;

 	/// Performs the blocks export.
@@ -25,10 +25,10 @@ use sc_chain_spec::ChainSpec;
 use log::{warn, info};
 use futures::{future, prelude::*};
 use sp_runtime::traits::{
-	Block as BlockT, NumberFor, One, Zero, Header, SaturatedConversion
+	Block as BlockT, NumberFor, One, Zero, Header, SaturatedConversion, MaybeSerializeDeserialize,
 };
 use sp_runtime::generic::{BlockId, SignedBlock};
-use codec::{Decode, Encode, IoReader};
+use codec::{Decode, Encode, IoReader as CodecIoReader};
 use crate::client::{Client, LocalCallExecutor};
 use sp_consensus::{
 	BlockOrigin,
@@ -39,12 +39,250 @@ use sp_core::storage::{StorageKey, well_known_keys, ChildInfo, Storage, StorageC
 use sc_client_api::{StorageProvider, BlockBackend, UsageProvider};

 use std::{io::{Read, Write, Seek}, pin::Pin, collections::HashMap};
+use std::{thread, time::{Duration, Instant}};
+use serde_json::{de::IoRead as JsonIoRead, Deserializer, StreamDeserializer};
+use std::convert::{TryFrom, TryInto};
+use sp_runtime::traits::{CheckedDiv, Saturating};
+
+/// Number of blocks we will add to the queue before waiting for the queue to catch up.
+const MAX_PENDING_BLOCKS: u64 = 1_024;
+
+/// Number of milliseconds to wait until next poll.
+const DELAY_TIME: u64 = 2_000;
+
+/// Number of milliseconds that must have passed between two updates.
+const TIME_BETWEEN_UPDATES: u64 = 3_000;

 /// Build a chain spec json
 pub fn build_spec(spec: &dyn ChainSpec, raw: bool) -> error::Result<String> {
 	spec.as_json(raw).map_err(Into::into)
 }

+
+/// Helper enum that wraps either a binary decoder (from parity-scale-codec), or a JSON decoder (from serde_json).
+/// Implements the Iterator Trait, calling `next()` will decode the next SignedBlock and return it.
+enum BlockIter<R, B> where
+	R: std::io::Read + std::io::Seek,
+{
+	Binary {
+		// Total number of blocks we are expecting to decode.
+		num_expected_blocks: u64,
+		// Number of blocks we have decoded thus far.
+		read_block_count: u64,
+		// Reader to the data, used for decoding new blocks.
+		reader: CodecIoReader<R>,
+	},
+	Json {
+		// Nubmer of blocks we have decoded thus far.
+		read_block_count: u64,
+		// Stream to the data, used for decoding new blocks.
+		reader: StreamDeserializer<'static, JsonIoRead<R>, SignedBlock<B>>,
+	},
+}
+
+impl<R, B> BlockIter<R, B> where
+	R: Read + Seek + 'static,
+	B: BlockT + MaybeSerializeDeserialize,
+{
+	fn new(input: R, binary: bool) -> Result<Self, String> {
+		if binary {
+			let mut reader = CodecIoReader(input);
+			// If the file is encoded in binary format, it is expected to first specify the number
+			// of blocks that are going to be decoded. We read it and add it to our enum struct.
+			let num_expected_blocks: u64 = Decode::decode(&mut reader)
+				.map_err(|e| format!("Failed to decode the number of blocks: {:?}", e))?;
+			Ok(BlockIter::Binary {
+				num_expected_blocks,
+				read_block_count: 0,
+				reader,
+			})
+		} else {
+			let stream_deser = Deserializer::from_reader(input)
+				.into_iter::<SignedBlock<B>>();
+			Ok(BlockIter::Json {
+				reader: stream_deser,
+				read_block_count: 0,
+			})
+		}
+	}
+
+	/// Returns the number of blocks read thus far.
+	fn read_block_count(&self) -> u64 {
+		match self {
+			BlockIter::Binary { read_block_count, .. }
+			| BlockIter::Json { read_block_count, .. }
+			=> *read_block_count,
+		}
+	}
+
+	/// Returns the total number of blocks to be imported, if possible.
+	fn num_expected_blocks(&self) -> Option<u64> {
+		match self {
+			BlockIter::Binary { num_expected_blocks, ..} => Some(*num_expected_blocks),
+			BlockIter::Json {..} => None
+		}
+	}
+}
+
+impl<R, B> Iterator for BlockIter<R, B> where
+	R: Read + Seek + 'static,
+	B: BlockT + MaybeSerializeDeserialize,
+{
+	type Item = Result<SignedBlock<B>, String>;
+
+	fn next(&mut self) -> Option<Self::Item> {
+		match self {
+			BlockIter::Binary { num_expected_blocks, read_block_count, reader } => {
+				if read_block_count < num_expected_blocks {
+					let block_result: Result<SignedBlock::<B>, _> =  SignedBlock::<B>::decode(reader)
+						.map_err(|e| e.to_string());
+					*read_block_count += 1;
+					Some(block_result)
+				} else {
+					// `read_block_count` == `num_expected_blocks` so we've read enough blocks.
+					None
+				}
+			}
+			BlockIter::Json { reader, read_block_count } => {
+				let res = Some(reader.next()?.map_err(|e| e.to_string()));
+				*read_block_count += 1;
+				res
+			}
+		}
+	}
+}
+
+/// Imports the SignedBlock to the queue.
+fn import_block_to_queue<TBl, TImpQu>(
+	signed_block: SignedBlock<TBl>,
+	queue: &mut TImpQu,
+	force: bool
+) where
+	TBl: BlockT + MaybeSerializeDeserialize,
+	TImpQu: 'static + ImportQueue<TBl>,
+{
+	let (header, extrinsics) = signed_block.block.deconstruct();
+	let hash = header.hash();
+	// import queue handles verification and importing it into the client.
+	queue.import_blocks(BlockOrigin::File, vec![
+		IncomingBlock::<TBl> {
+			hash,
+			header: Some(header),
+			body: Some(extrinsics),
+			justification: signed_block.justification,
+			origin: None,
+			allow_missing_state: false,
+			import_existing: force,
+		}
+	]);
+}
+
+/// Returns true if we have imported every block we were supposed to import, else returns false.
+fn importing_is_done(
+	num_expected_blocks: Option<u64>,
+	read_block_count: u64,
+	imported_blocks: u64
+) -> bool {
+	if let Some(num_expected_blocks) = num_expected_blocks {
+		imported_blocks >= num_expected_blocks
+	} else {
+		imported_blocks >= read_block_count
+	}
+}
+
+/// Structure used to log the block importing speed.
+struct Speedometer<B: BlockT> {
+	best_number: NumberFor<B>,
+	last_number: Option<NumberFor<B>>,
+	last_update: Instant,
+}
+
+impl<B: BlockT> Speedometer<B> {
+	/// Creates a fresh Speedometer.
+	fn new() -> Self {
+		Self {
+			best_number: NumberFor::<B>::from(0),
+			last_number: None,
+			last_update: Instant::now(),
+		}
+	}
+
+	/// Calculates `(best_number - last_number) / (now - last_update)` and 
+	/// logs the speed of import.
+	fn display_speed(&self) {
+		// Number of milliseconds elapsed since last time.
+		let elapsed_ms = {
+			let elapsed = self.last_update.elapsed();
+			let since_last_millis = elapsed.as_secs() * 1000;
+			let since_last_subsec_millis = elapsed.subsec_millis() as u64;
+			since_last_millis + since_last_subsec_millis
+		};
+
+		// Number of blocks that have been imported since last time.
+		let diff = match self.last_number {
+			None => return,
+			Some(n) => self.best_number.saturating_sub(n)
+		};
+
+		if let Ok(diff) = TryInto::<u128>::try_into(diff) {
+			// If the number of blocks can be converted to a regular integer, then it's easy: just
+			// do the math and turn it into a `f64`.
+			let speed = diff.saturating_mul(10_000).checked_div(u128::from(elapsed_ms))
+				.map_or(0.0, |s| s as f64) / 10.0;
+			info!("📦 Current best block: {} ({:4.1} bps)", self.best_number, speed);
+		} else {
+			// If the number of blocks can't be converted to a regular integer, then we need a more
+			// algebraic approach and we stay within the realm of integers.
+			let one_thousand = NumberFor::<B>::from(1_000);
+			let elapsed = NumberFor::<B>::from(
+				<u32 as TryFrom<_>>::try_from(elapsed_ms).unwrap_or(u32::max_value())
+			);
+
+			let speed = diff.saturating_mul(one_thousand).checked_div(&elapsed)
+				.unwrap_or_else(Zero::zero);
+			info!("📦 Current best block: {} ({} bps)", self.best_number, speed)
+		}
+	}
+
+	/// Updates the Speedometer.
+	fn update(&mut self, best_number: NumberFor<B>) {
+		self.last_number = Some(self.best_number);
+		self.best_number = best_number;
+		self.last_update = Instant::now();
+	}
+
+	// If more than TIME_BETWEEN_UPDATES has elapsed since last update,
+	// then print and update the speedometer.
+	fn notify_user(&mut self, best_number: NumberFor<B>) {
+		let delta = Duration::from_millis(TIME_BETWEEN_UPDATES);
+		if Instant::now().duration_since(self.last_update) >= delta {
+			self.display_speed();
+			self.update(best_number);
+		}
+	}
+}
+
+/// Different State that the `import_blocks` future could be in.
+enum ImportState<R, B> where 
+	R: Read + Seek + 'static,
+	B: BlockT + MaybeSerializeDeserialize,
+{
+	/// We are reading from the BlockIter structure, adding those blocks to the queue if possible.
+	Reading{block_iter: BlockIter<R, B>},
+	/// The queue is full (contains at least MAX_PENDING_BLOCKS blocks) and we are waiting for it to catch up.
+	WaitingForImportQueueToCatchUp{
+		block_iter: BlockIter<R, B>,
+		delay: Duration,
+		block: SignedBlock<B>
+	},
+	// We have added all the blocks to the queue but they are still being processed.
+	WaitingForImportQueueToFinish{
+		num_expected_blocks: Option<u64>, 
+		read_block_count: u64,
+		delay: Duration,
+	},
+}
+
 impl<
 	TBl, TRtApi, TBackend,
 	TExecDisp, TFchr, TSc, TImpQu, TFprb, TFpp,
@@ -54,7 +292,7 @@ impl<
 	Client<TBackend, LocalCallExecutor<TBackend, NativeExecutor<TExecDisp>>, TBl, TRtApi>,
 	TFchr, TSc, TImpQu, TFprb, TFpp, TExPool, TRpc, Backend
 > where
-	TBl: BlockT,
+	TBl: BlockT + MaybeSerializeDeserialize,
 	TBackend: 'static + sc_client_api::backend::Backend<TBl> + Send,
 	TExecDisp: 'static + NativeExecutionDispatch,
 	TImpQu: 'static + ImportQueue<TBl>,
@@ -68,6 +306,7 @@ impl<
 		mut self,
 		input: impl Read + Seek + Send + 'static,
 		force: bool,
+		binary: bool,
 	) -> Pin<Box<dyn Future<Output = Result<(), Error>> + Send>> {
 		struct WaitLink {
 			imported_blocks: u64,
@@ -87,7 +326,7 @@ impl<
 			fn blocks_processed(
 				&mut self,
 				imported: usize,
-				_count: usize,
+				_num_expected_blocks: usize,
 				results: Vec<(Result<BlockImportResult<NumberFor<B>>, BlockImportError>, B::Hash)>
 			) {
 				self.imported_blocks += imported as u64;
@@ -102,10 +341,20 @@ impl<
 			}
 		}

-		let mut io_reader_input = IoReader(input);
-		let mut count = None::<u64>;
-		let mut read_block_count = 0;
 		let mut link = WaitLink::new();
+		let block_iter_res: Result<BlockIter<_, Self::Block>, String> = BlockIter::new(input, binary);
+
+		let block_iter = match block_iter_res {
+			Ok(block_iter) => block_iter,
+			Err(e) => {
+				// We've encountered an error while creating the block iterator 
+				// so we can just return a future that returns an error.
+				return future::ready(Err(Error::Other(e))).boxed()
+			}
+		};
+
+		let mut state = Some(ImportState::Reading{block_iter});
+		let mut speedometer = Speedometer::<TBl>::new();

 		// Importing blocks is implemented as a future, because we want the operation to be
 		// interruptible.
@@ -117,85 +366,85 @@ impl<
 		let import = future::poll_fn(move |cx| {
 			let client = &self.client;
 			let queue = &mut self.import_queue;
-
-			// Start by reading the number of blocks if not done so already.
-			let count = match count {
-				Some(c) => c,
-				None => {
-					let c: u64 = match Decode::decode(&mut io_reader_input) {
-						Ok(c) => c,
-						Err(err) => {
-							let err = format!("Error reading file: {}", err);
-							return std::task::Poll::Ready(Err(From::from(err)));
+			match state.take().expect("state should never be None; qed") {
+				ImportState::Reading{mut block_iter} => {
+					match block_iter.next() {
+						None => {
+							// The iterator is over: we now need to wait for the import queue to finish.
+							let num_expected_blocks = block_iter.num_expected_blocks();
+							let read_block_count = block_iter.read_block_count();
+							let delay = Duration::from_millis(DELAY_TIME);
+							state = Some(ImportState::WaitingForImportQueueToFinish{num_expected_blocks, read_block_count, delay});
 						},
-					};
-					info!("📦 Importing {} blocks", c);
-					count = Some(c);
-					c
-				}
-			};
-
-			// Read blocks from the input.
-			if read_block_count < count {
-				match SignedBlock::<Self::Block>::decode(&mut io_reader_input) {
-					Ok(signed) => {
-						let (header, extrinsics) = signed.block.deconstruct();
-						let hash = header.hash();
-						// import queue handles verification and importing it into the client
-						queue.import_blocks(BlockOrigin::File, vec![
-							IncomingBlock::<Self::Block> {
-								hash,
-								header: Some(header),
-								body: Some(extrinsics),
-								justification: signed.justification,
-								origin: None,
-								allow_missing_state: false,
-								import_existing: force,
+						Some(block_result) => {
+							let read_block_count = block_iter.read_block_count();
+							match block_result {
+								Ok(block) => {
+									if read_block_count - link.imported_blocks >= MAX_PENDING_BLOCKS {
+										// The queue is full, so do not add this block and simply wait until
+										// the queue has made some progress.
+										let delay = Duration::from_millis(DELAY_TIME);
+										state = Some(ImportState::WaitingForImportQueueToCatchUp{block_iter, delay, block});
+									} else {
+										// Queue is not full, we can keep on adding blocks to the queue.
+										import_block_to_queue(block, queue, force);
+										state = Some(ImportState::Reading{block_iter});
+									}
+								}
+								Err(e) => {
+									return std::task::Poll::Ready(
+										Err(Error::Other(format!("Error reading block #{}: {}", read_block_count, e))))
+								}
 							}
-						]);
+						}
 					}
-					Err(e) => {
-						warn!("Error reading block data at {}: {}", read_block_count, e);
-						return std::task::Poll::Ready(Ok(()));
+				},
+				ImportState::WaitingForImportQueueToCatchUp{block_iter, delay, block} => {
+					let read_block_count = block_iter.read_block_count();
+					if read_block_count - link.imported_blocks >= MAX_PENDING_BLOCKS {
+						thread::sleep(delay);
+						// Queue is still full, so wait until there is room to insert our block.
+						state = Some(ImportState::WaitingForImportQueueToCatchUp{block_iter, delay, block});
+					} else {
+						// Queue is no longer full, so we can add our block to the queue.
+						import_block_to_queue(block, queue, force);
+						// Switch back to Reading state.
+						state = Some(ImportState::Reading{block_iter});
+					}
+				},
+				ImportState::WaitingForImportQueueToFinish{num_expected_blocks, read_block_count, delay} => {
+					// All the blocks have been added to the queue, which doesn't mean they 
+					// have all been properly imported.
+					if importing_is_done(num_expected_blocks, read_block_count, link.imported_blocks) {
+						// Importing is done, we can log the result and return.
+						info!(
+							"🎉 Imported {} blocks. Best: #{}",
+							read_block_count, client.chain_info().best_number
+						);
+						return std::task::Poll::Ready(Ok(()))
+					} else {
+						thread::sleep(delay);
+						// Importing is not done, we still have to wait for the queue to finish.
+						state = Some(ImportState::WaitingForImportQueueToFinish{num_expected_blocks, read_block_count, delay});
 					}
 				}
-
-				read_block_count += 1;
-				if read_block_count % 1000 == 0 {
-					info!("#{} blocks were added to the queue", read_block_count);
-				}
-
-				cx.waker().wake_by_ref();
-				return std::task::Poll::Pending;
 			}

-			let blocks_before = link.imported_blocks;
 			queue.poll_actions(cx, &mut link);

+			let best_number = client.chain_info().best_number;
+			speedometer.notify_user(best_number);
+
 			if link.has_error {
-				info!(
-					"Stopping after #{} blocks because of an error",
-					link.imported_blocks,
-				);
-				return std::task::Poll::Ready(Ok(()));
+				return std::task::Poll::Ready(Err(
+					Error::Other(
+						format!("Stopping after #{} blocks because of an error", link.imported_blocks)
+					)
+				))
 			}

-			if link.imported_blocks / 1000 != blocks_before / 1000 {
-				info!(
-					"#{} blocks were imported (#{} left)",
-					link.imported_blocks,
-					count - link.imported_blocks
-				);
-			}
-
-			if link.imported_blocks >= count {
-				info!("🎉 Imported {} blocks. Best: #{}", read_block_count, client.chain_info().best_number);
-				return std::task::Poll::Ready(Ok(()));
-
-			} else {
-				// Polling the import queue will re-schedule the task when ready.
-				return std::task::Poll::Pending;
-			}
+			cx.waker().wake_by_ref();
+			std::task::Poll::Pending
 		});
 		Box::pin(import)
 	}
@@ -295,7 +544,7 @@ impl<
 				1u64.encode_to(&mut buf);
 				block.encode_to(&mut buf);
 				let reader = std::io::Cursor::new(buf);
-				self.import_blocks(reader, true)
+				self.import_blocks(reader, true, true)
 			}
 			Ok(None) => Box::pin(future::err("Unknown block".into())),
 			Err(e) => Box::pin(future::err(format!("Error reading block: {:?}", e).into())),