use crate::{error, error::Error};
use codec::{Decode, IoReader as CodecIoReader};
use futures::{future, prelude::*};
use futures_timer::Delay;
use log::{info, warn};
use sc_chain_spec::ChainSpec;
use sc_client_api::HeaderBackend;
use sc_consensus::import_queue::{
BlockImportError, BlockImportStatus, ImportQueue, IncomingBlock, Link,
};
use serde_json::{de::IoRead as JsonIoRead, Deserializer, StreamDeserializer};
use sp_consensus::BlockOrigin;
use sp_runtime::{
generic::SignedBlock,
traits::{
Block as BlockT, CheckedDiv, Header, MaybeSerializeDeserialize, NumberFor, Saturating, Zero,
},
};
use std::{
io::Read,
pin::Pin,
task::Poll,
time::{Duration, Instant},
};
const MAX_PENDING_BLOCKS: u64 = 10_000;
const DELAY_TIME: u64 = 200;
const TIME_BETWEEN_UPDATES: u64 = 3_000;
use std::sync::Arc;
pub fn build_spec(spec: &dyn ChainSpec, raw: bool) -> error::Result<String> {
spec.as_json(raw).map_err(Into::into)
}
enum BlockIter<R, B>
where
R: std::io::Read,
{
Binary {
num_expected_blocks: u64,
read_block_count: u64,
reader: CodecIoReader<R>,
},
Json {
read_block_count: u64,
reader: StreamDeserializer<'static, JsonIoRead<R>, SignedBlock<B>>,
},
}
impl<R, B> BlockIter<R, B>
where
R: Read + 'static,
B: BlockT + MaybeSerializeDeserialize,
{
fn new(input: R, binary: bool) -> Result<Self, String> {
if binary {
let mut reader = CodecIoReader(input);
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 })
}
}
fn read_block_count(&self) -> u64 {
match self {
BlockIter::Binary { read_block_count, .. } |
BlockIter::Json { read_block_count, .. } => *read_block_count,
}
}
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 + '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 {
None
}
},
BlockIter::Json { reader, read_block_count } => {
let res = Some(reader.next()?.map_err(|e| e.to_string()));
*read_block_count += 1;
res
},
}
}
}
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();
queue.service_ref().import_blocks(
BlockOrigin::File,
vec![IncomingBlock::<TBl> {
hash,
header: Some(header),
body: Some(extrinsics),
indexed_body: None,
justifications: signed_block.justifications,
origin: None,
allow_missing_state: false,
import_existing: force,
state: None,
skip_execution: 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
}
}
struct Speedometer<B: BlockT> {
best_number: NumberFor<B>,
last_number: Option<NumberFor<B>>,
last_update: Instant,
}
impl<B: BlockT> Speedometer<B> {
fn new() -> Self {
Self {
best_number: NumberFor::<B>::from(0u32),
last_number: None,
last_update: Instant::now(),
}
}
fn display_speed(&self) {
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
};
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) {
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 {
let one_thousand = NumberFor::<B>::from(1_000u32);
let elapsed =
NumberFor::<B>::from(<u32 as TryFrom<_>>::try_from(elapsed_ms).unwrap_or(u32::MAX));
let speed = diff
.saturating_mul(one_thousand)
.checked_div(&elapsed)
.unwrap_or_else(Zero::zero);
info!("📦 Current best block: {} ({} bps)", self.best_number, speed)
}
}
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();
}
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);
}
}
}
enum ImportState<R, B>
where
R: Read + 'static,
B: BlockT + MaybeSerializeDeserialize,
{
Reading { block_iter: BlockIter<R, B> },
WaitingForImportQueueToCatchUp {
block_iter: BlockIter<R, B>,
delay: Delay,
block: SignedBlock<B>,
},
WaitingForImportQueueToFinish {
num_expected_blocks: Option<u64>,
read_block_count: u64,
delay: Delay,
},
}
pub fn import_blocks<B, IQ, C>(
client: Arc<C>,
mut import_queue: IQ,
input: impl Read + Send + 'static,
force: bool,
binary: bool,
) -> Pin<Box<dyn Future<Output = Result<(), Error>> + Send>>
where
C: HeaderBackend<B> + Send + Sync + 'static,
B: BlockT + for<'de> serde::Deserialize<'de>,
IQ: ImportQueue<B> + 'static,
{
struct WaitLink {
imported_blocks: u64,
has_error: bool,
}
impl WaitLink {
fn new() -> WaitLink {
WaitLink { imported_blocks: 0, has_error: false }
}
}
impl<B: BlockT> Link<B> for WaitLink {
fn blocks_processed(
&mut self,
imported: usize,
_num_expected_blocks: usize,
results: Vec<(Result<BlockImportStatus<NumberFor<B>>, BlockImportError>, B::Hash)>,
) {
self.imported_blocks += imported as u64;
for result in results {
if let (Err(err), hash) = result {
warn!("There was an error importing block with hash {:?}: {}", hash, err);
self.has_error = true;
break
}
}
}
}
let mut link = WaitLink::new();
let block_iter_res: Result<BlockIter<_, B>, String> = BlockIter::new(input, binary);
let block_iter = match block_iter_res {
Ok(block_iter) => block_iter,
Err(e) => {
return future::ready(Err(Error::Other(e))).boxed()
},
};
let mut state = Some(ImportState::Reading { block_iter });
let mut speedometer = Speedometer::<B>::new();
let import = future::poll_fn(move |cx| {
let client = &client;
let queue = &mut import_queue;
match state.take().expect("state should never be None; qed") {
ImportState::Reading { mut block_iter } => {
match block_iter.next() {
None => {
let num_expected_blocks = block_iter.num_expected_blocks();
let read_block_count = block_iter.read_block_count();
let delay = Delay::new(Duration::from_millis(DELAY_TIME));
state = Some(ImportState::WaitingForImportQueueToFinish {
num_expected_blocks,
read_block_count,
delay,
});
},
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 {
let delay = Delay::new(Duration::from_millis(DELAY_TIME));
state = Some(ImportState::WaitingForImportQueueToCatchUp {
block_iter,
delay,
block,
});
} else {
import_block_to_queue(block, queue, force);
state = Some(ImportState::Reading { block_iter });
}
},
Err(e) =>
return Poll::Ready(Err(Error::Other(format!(
"Error reading block #{}: {}",
read_block_count, e
)))),
}
},
}
},
ImportState::WaitingForImportQueueToCatchUp { block_iter, mut delay, block } => {
let read_block_count = block_iter.read_block_count();
if read_block_count - link.imported_blocks >= MAX_PENDING_BLOCKS {
match Pin::new(&mut delay).poll(cx) {
Poll::Pending => {
state = Some(ImportState::WaitingForImportQueueToCatchUp {
block_iter,
delay,
block,
});
return Poll::Pending
},
Poll::Ready(_) => {
delay.reset(Duration::from_millis(DELAY_TIME));
},
}
state = Some(ImportState::WaitingForImportQueueToCatchUp {
block_iter,
delay,
block,
});
} else {
import_block_to_queue(block, queue, force);
state = Some(ImportState::Reading { block_iter });
}
},
ImportState::WaitingForImportQueueToFinish {
num_expected_blocks,
read_block_count,
mut delay,
} => {
if importing_is_done(num_expected_blocks, read_block_count, link.imported_blocks) {
info!(
"🎉 Imported {} blocks. Best: #{}",
read_block_count,
client.info().best_number
);
return Poll::Ready(Ok(()))
} else {
match Pin::new(&mut delay).poll(cx) {
Poll::Pending => {
state = Some(ImportState::WaitingForImportQueueToFinish {
num_expected_blocks,
read_block_count,
delay,
});
return Poll::Pending
},
Poll::Ready(_) => {
delay.reset(Duration::from_millis(DELAY_TIME));
},
}
state = Some(ImportState::WaitingForImportQueueToFinish {
num_expected_blocks,
read_block_count,
delay,
});
}
},
}
queue.poll_actions(cx, &mut link);
let best_number = client.info().best_number;
speedometer.notify_user(best_number);
if link.has_error {
return Poll::Ready(Err(Error::Other(format!(
"Stopping after #{} blocks because of an error",
link.imported_blocks
))))
}
cx.waker().wake_by_ref();
Poll::Pending
});
Box::pin(import)
}