grandpa: remove light-client specific block import pipeline (#7546)

* grandpa: remove light-client specific block import

* consensus, network: remove finality proofs
This commit is contained in:
André Silva
2020-11-23 14:28:55 +00:00
committed by GitHub
parent cd2490f56d
commit 1871a95088
44 changed files with 96 additions and 2512 deletions
@@ -26,7 +26,7 @@ use std::sync::Arc;
use std::any::Any;
use crate::Error;
use crate::import_queue::{Verifier, CacheKeyId};
use crate::import_queue::CacheKeyId;
/// Block import result.
#[derive(Debug, PartialEq, Eq)]
@@ -54,8 +54,6 @@ pub struct ImportedAux {
pub needs_justification: bool,
/// Received a bad justification.
pub bad_justification: bool,
/// Request a finality proof for the given block.
pub needs_finality_proof: bool,
/// Whether the block that was imported is the new best block.
pub is_new_best: bool,
}
@@ -63,7 +61,7 @@ pub struct ImportedAux {
impl ImportResult {
/// Returns default value for `ImportResult::Imported` with
/// `clear_justification_requests`, `needs_justification`,
/// `bad_justification` and `needs_finality_proof` set to false.
/// `bad_justification` set to false.
pub fn imported(is_new_best: bool) -> ImportResult {
let mut aux = ImportedAux::default();
aux.is_new_best = is_new_best;
@@ -345,21 +343,3 @@ pub trait JustificationImport<B: BlockT> {
justification: Justification,
) -> Result<(), Self::Error>;
}
/// Finality proof import trait.
pub trait FinalityProofImport<B: BlockT> {
type Error: std::error::Error + Send + 'static;
/// Called by the import queue when it is started. Returns a list of finality proofs to request
/// from the network.
fn on_start(&mut self) -> Vec<(B::Hash, NumberFor<B>)> { Vec::new() }
/// Import a Block justification and finalize the given block. Returns finalized block or error.
fn import_finality_proof(
&mut self,
hash: B::Hash,
number: NumberFor<B>,
finality_proof: Vec<u8>,
verifier: &mut dyn Verifier<B>,
) -> Result<(B::Hash, NumberFor<B>), Self::Error>;
}
@@ -34,7 +34,7 @@ use crate::{
error::Error as ConsensusError,
block_import::{
BlockImport, BlockOrigin, BlockImportParams, ImportedAux, JustificationImport, ImportResult,
BlockCheckParams, FinalityProofImport,
BlockCheckParams,
},
metrics::Metrics,
};
@@ -56,11 +56,6 @@ pub type BoxBlockImport<B, Transaction> = Box<
/// Shared justification import struct used by the queue.
pub type BoxJustificationImport<B> = Box<dyn JustificationImport<B, Error=ConsensusError> + Send + Sync>;
/// Shared finality proof import struct used by the queue.
pub type BoxFinalityProofImport<B> = Box<
dyn FinalityProofImport<B, Error = ConsensusError> + Send + Sync
>;
/// Maps to the Origin used by the network.
pub type Origin = libp2p::PeerId;
@@ -115,15 +110,6 @@ pub trait ImportQueue<B: BlockT>: Send {
number: NumberFor<B>,
justification: Justification
);
/// Import block finality proof.
fn import_finality_proof(
&mut self,
who: Origin,
hash: B::Hash,
number: NumberFor<B>,
finality_proof: Vec<u8>
);
/// Polls for actions to perform on the network.
///
/// This method should behave in a way similar to `Future::poll`. It can register the current
@@ -146,19 +132,6 @@ pub trait Link<B: BlockT>: Send {
fn justification_imported(&mut self, _who: Origin, _hash: &B::Hash, _number: NumberFor<B>, _success: bool) {}
/// Request a justification for the given block.
fn request_justification(&mut self, _hash: &B::Hash, _number: NumberFor<B>) {}
/// Finality proof import result.
///
/// Even though we have asked for finality proof of block A, provider could return proof of
/// some earlier block B, if the proof for A was too large. The sync module should continue
/// asking for proof of A in this case.
fn finality_proof_imported(
&mut self,
_who: Origin,
_request_block: (B::Hash, NumberFor<B>),
_finalization_result: Result<(B::Hash, NumberFor<B>), ()>,
) {}
/// Request a finality proof for the given block.
fn request_finality_proof(&mut self, _hash: &B::Hash, _number: NumberFor<B>) {}
}
/// Block import successful result.
@@ -25,7 +25,7 @@ use prometheus_endpoint::Registry;
use crate::{
block_import::BlockOrigin,
import_queue::{
BlockImportResult, BlockImportError, Verifier, BoxBlockImport, BoxFinalityProofImport,
BlockImportResult, BlockImportError, Verifier, BoxBlockImport,
BoxJustificationImport, ImportQueue, Link, Origin,
IncomingBlock, import_single_block_metered,
buffered_link::{self, BufferedLinkSender, BufferedLinkReceiver},
@@ -36,8 +36,8 @@ use crate::{
/// Interface to a basic block import queue that is importing blocks sequentially in a separate
/// task, with plugable verification.
pub struct BasicQueue<B: BlockT, Transaction> {
/// Channel to send finality work messages to the background task.
finality_sender: TracingUnboundedSender<worker_messages::Finality<B>>,
/// Channel to send justifcation import messages to the background task.
justification_sender: TracingUnboundedSender<worker_messages::ImportJustification<B>>,
/// Channel to send block import messages to the background task.
block_import_sender: TracingUnboundedSender<worker_messages::ImportBlocks<B>>,
/// Results coming from the worker task.
@@ -48,7 +48,7 @@ pub struct BasicQueue<B: BlockT, Transaction> {
impl<B: BlockT, Transaction> Drop for BasicQueue<B, Transaction> {
fn drop(&mut self) {
// Flush the queue and close the receiver to terminate the future.
self.finality_sender.close_channel();
self.justification_sender.close_channel();
self.block_import_sender.close_channel();
self.result_port.close();
}
@@ -57,13 +57,11 @@ impl<B: BlockT, Transaction> Drop for BasicQueue<B, Transaction> {
impl<B: BlockT, Transaction: Send + 'static> BasicQueue<B, Transaction> {
/// Instantiate a new basic queue, with given verifier.
///
/// This creates a background task, and calls `on_start` on the justification importer and
/// finality proof importer.
/// This creates a background task, and calls `on_start` on the justification importer.
pub fn new<V: 'static + Verifier<B>>(
verifier: V,
block_import: BoxBlockImport<B, Transaction>,
justification_import: Option<BoxJustificationImport<B>>,
finality_proof_import: Option<BoxFinalityProofImport<B>>,
spawner: &impl sp_core::traits::SpawnNamed,
prometheus_registry: Option<&Registry>,
) -> Self {
@@ -77,19 +75,18 @@ impl<B: BlockT, Transaction: Send + 'static> BasicQueue<B, Transaction> {
.ok()
});
let (future, finality_sender, block_import_sender) = BlockImportWorker::new(
let (future, justification_sender, block_import_sender) = BlockImportWorker::new(
result_sender,
verifier,
block_import,
justification_import,
finality_proof_import,
metrics,
);
spawner.spawn_blocking("basic-block-import-worker", future.boxed());
Self {
finality_sender,
justification_sender,
block_import_sender,
result_port,
_phantom: PhantomData,
@@ -122,8 +119,8 @@ impl<B: BlockT, Transaction: Send> ImportQueue<B> for BasicQueue<B, Transaction>
number: NumberFor<B>,
justification: Justification,
) {
let res = self.finality_sender.unbounded_send(
worker_messages::Finality::ImportJustification(who, hash, number, justification),
let res = self.justification_sender.unbounded_send(
worker_messages::ImportJustification(who, hash, number, justification),
);
if res.is_err() {
@@ -134,26 +131,6 @@ impl<B: BlockT, Transaction: Send> ImportQueue<B> for BasicQueue<B, Transaction>
}
}
fn import_finality_proof(
&mut self,
who: Origin,
hash: B::Hash,
number: NumberFor<B>,
finality_proof: Vec<u8>,
) {
trace!(target: "sync", "Scheduling finality proof of {}/{} for import", number, hash);
let res = self.finality_sender.unbounded_send(
worker_messages::Finality::ImportFinalityProof(who, hash, number, finality_proof),
);
if res.is_err() {
log::error!(
target: "sync",
"import_finality_proof: Background import task is no longer alive"
);
}
}
fn poll_actions(&mut self, cx: &mut Context, link: &mut dyn Link<B>) {
if self.result_port.poll_actions(cx, link).is_err() {
log::error!(target: "sync", "poll_actions: Background import task is no longer alive");
@@ -166,17 +143,12 @@ mod worker_messages {
use super::*;
pub struct ImportBlocks<B: BlockT>(pub BlockOrigin, pub Vec<IncomingBlock<B>>);
pub enum Finality<B: BlockT> {
ImportJustification(Origin, B::Hash, NumberFor<B>, Justification),
ImportFinalityProof(Origin, B::Hash, NumberFor<B>, Vec<u8>),
}
pub struct ImportJustification<B: BlockT>(pub Origin, pub B::Hash, pub NumberFor<B>, pub Justification);
}
struct BlockImportWorker<B: BlockT, Transaction> {
result_sender: BufferedLinkSender<B>,
justification_import: Option<BoxJustificationImport<B>>,
finality_proof_import: Option<BoxFinalityProofImport<B>>,
delay_between_blocks: Duration,
metrics: Option<Metrics>,
_phantom: PhantomData<Transaction>,
@@ -188,17 +160,16 @@ impl<B: BlockT, Transaction: Send> BlockImportWorker<B, Transaction> {
verifier: V,
block_import: BoxBlockImport<B, Transaction>,
justification_import: Option<BoxJustificationImport<B>>,
finality_proof_import: Option<BoxFinalityProofImport<B>>,
metrics: Option<Metrics>,
) -> (
impl Future<Output = ()> + Send,
TracingUnboundedSender<worker_messages::Finality<B>>,
TracingUnboundedSender<worker_messages::ImportJustification<B>>,
TracingUnboundedSender<worker_messages::ImportBlocks<B>>,
) {
use worker_messages::*;
let (finality_sender, mut finality_port) =
tracing_unbounded("mpsc_import_queue_worker_finality");
let (justification_sender, mut justification_port) =
tracing_unbounded("mpsc_import_queue_worker_justification");
let (block_import_sender, mut block_import_port) =
tracing_unbounded("mpsc_import_queue_worker_blocks");
@@ -206,23 +177,17 @@ impl<B: BlockT, Transaction: Send> BlockImportWorker<B, Transaction> {
let mut worker = BlockImportWorker {
result_sender,
justification_import,
finality_proof_import,
delay_between_blocks: Duration::new(0, 0),
metrics,
_phantom: PhantomData,
};
// Let's initialize `justification_import` and `finality_proof_import`.
// Let's initialize `justification_import`
if let Some(justification_import) = worker.justification_import.as_mut() {
for (hash, number) in justification_import.on_start() {
worker.result_sender.request_justification(&hash, number);
}
}
if let Some(finality_proof_import) = worker.finality_proof_import.as_mut() {
for (hash, number) in finality_proof_import.on_start() {
worker.result_sender.request_finality_proof(&hash, number);
}
}
// The future below has two possible states:
//
@@ -230,7 +195,7 @@ impl<B: BlockT, Transaction: Send> BlockImportWorker<B, Transaction> {
// `Future`, and `block_import` is `None`.
// - Something else, in which case `block_import` is `Some` and `importing` is None.
//
// Additionally, the task will prioritize processing of finality work messages over
// Additionally, the task will prioritize processing of justification import messages over
// block import messages, hence why two distinct channels are used.
let mut block_import_verifier = Some((block_import, verifier));
let mut importing = None;
@@ -243,28 +208,15 @@ impl<B: BlockT, Transaction: Send> BlockImportWorker<B, Transaction> {
return Poll::Ready(())
}
// Grab the next finality action request sent to the import queue.
let finality_work = match Stream::poll_next(Pin::new(&mut finality_port), cx) {
Poll::Ready(Some(msg)) => Some(msg),
Poll::Ready(None) => return Poll::Ready(()),
Poll::Pending => None,
};
match finality_work {
Some(Finality::ImportFinalityProof(who, hash, number, proof)) => {
let (_, verif) = block_import_verifier
.as_mut()
.expect("block_import_verifier is always Some; qed");
worker.import_finality_proof(verif, who, hash, number, proof);
continue;
}
Some(Finality::ImportJustification(who, hash, number, justification)) => {
// Grab the next justification import request sent to the import queue.
match Stream::poll_next(Pin::new(&mut justification_port), cx) {
Poll::Ready(Some(ImportJustification(who, hash, number, justification))) => {
worker.import_justification(who, hash, number, justification);
continue;
}
None => {}
}
},
Poll::Ready(None) => return Poll::Ready(()),
Poll::Pending => {},
};
// If we are in the process of importing a bunch of blocks, let's resume this
// process before doing anything more.
@@ -299,7 +251,7 @@ impl<B: BlockT, Transaction: Send> BlockImportWorker<B, Transaction> {
}
});
(future, finality_sender, block_import_sender)
(future, justification_sender, block_import_sender)
}
/// Returns a `Future` that imports the given blocks and sends the results on
@@ -324,36 +276,6 @@ impl<B: BlockT, Transaction: Send> BlockImportWorker<B, Transaction> {
})
}
fn import_finality_proof<V: 'static + Verifier<B>>(
&mut self,
verifier: &mut V,
who: Origin,
hash: B::Hash,
number: NumberFor<B>,
finality_proof: Vec<u8>
) {
let started = wasm_timer::Instant::now();
let result = self.finality_proof_import.as_mut().map(|finality_proof_import| {
finality_proof_import.import_finality_proof(hash, number, finality_proof, verifier)
.map_err(|e| {
debug!(
"Finality proof import failed with {:?} for hash: {:?} number: {:?} coming from node: {:?}",
e,
hash,
number,
who,
);
})
}).unwrap_or(Err(()));
if let Some(metrics) = self.metrics.as_ref() {
metrics.finality_proof_import_time.observe(started.elapsed().as_secs_f64());
}
trace!(target: "sync", "Imported finality proof for {}/{}", number, hash);
self.result_sender.finality_proof_imported(who, (hash, number), result);
}
fn import_justification(
&mut self,
who: Origin,
@@ -596,7 +518,7 @@ mod tests {
let (result_sender, mut result_port) = buffered_link::buffered_link();
let (mut worker, mut finality_sender, mut block_import_sender) =
BlockImportWorker::new(result_sender, (), Box::new(()), Some(Box::new(())), None, None);
BlockImportWorker::new(result_sender, (), Box::new(()), Some(Box::new(())), None);
let mut import_block = |n| {
let header = Header {
@@ -629,7 +551,7 @@ mod tests {
let mut import_justification = || {
let hash = Hash::random();
block_on(finality_sender.send(worker_messages::Finality::ImportJustification(
block_on(finality_sender.send(worker_messages::ImportJustification(
libp2p::PeerId::random(),
hash,
1,
@@ -81,8 +81,6 @@ enum BlockImportWorkerMsg<B: BlockT> {
BlocksProcessed(usize, usize, Vec<(Result<BlockImportResult<NumberFor<B>>, BlockImportError>, B::Hash)>),
JustificationImported(Origin, B::Hash, NumberFor<B>, bool),
RequestJustification(B::Hash, NumberFor<B>),
FinalityProofImported(Origin, (B::Hash, NumberFor<B>), Result<(B::Hash, NumberFor<B>), ()>),
RequestFinalityProof(B::Hash, NumberFor<B>),
}
impl<B: BlockT> Link<B> for BufferedLinkSender<B> {
@@ -109,20 +107,6 @@ impl<B: BlockT> Link<B> for BufferedLinkSender<B> {
fn request_justification(&mut self, hash: &B::Hash, number: NumberFor<B>) {
let _ = self.tx.unbounded_send(BlockImportWorkerMsg::RequestJustification(hash.clone(), number));
}
fn finality_proof_imported(
&mut self,
who: Origin,
request_block: (B::Hash, NumberFor<B>),
finalization_result: Result<(B::Hash, NumberFor<B>), ()>,
) {
let msg = BlockImportWorkerMsg::FinalityProofImported(who, request_block, finalization_result);
let _ = self.tx.unbounded_send(msg);
}
fn request_finality_proof(&mut self, hash: &B::Hash, number: NumberFor<B>) {
let _ = self.tx.unbounded_send(BlockImportWorkerMsg::RequestFinalityProof(hash.clone(), number));
}
}
/// See [`buffered_link`].
@@ -154,10 +138,6 @@ impl<B: BlockT> BufferedLinkReceiver<B> {
link.justification_imported(who, &hash, number, success),
BlockImportWorkerMsg::RequestJustification(hash, number) =>
link.request_justification(&hash, number),
BlockImportWorkerMsg::FinalityProofImported(who, block, result) =>
link.finality_proof_imported(who, block, result),
BlockImportWorkerMsg::RequestFinalityProof(hash, number) =>
link.request_finality_proof(&hash, number),
}
}
}
@@ -49,7 +49,7 @@ mod metrics;
pub use self::error::Error;
pub use block_import::{
BlockImport, BlockOrigin, ForkChoiceStrategy, ImportedAux, BlockImportParams, BlockCheckParams,
ImportResult, JustificationImport, FinalityProofImport,
ImportResult, JustificationImport,
};
pub use select_chain::SelectChain;
pub use sp_state_machine::Backend as StateBackend;
@@ -30,7 +30,6 @@ pub(crate) struct Metrics {
pub import_queue_processed: CounterVec<U64>,
pub block_verification_time: HistogramVec,
pub block_verification_and_import_time: Histogram,
pub finality_proof_import_time: Histogram,
pub justification_import_time: Histogram,
}
@@ -63,15 +62,6 @@ impl Metrics {
)?,
registry,
)?,
finality_proof_import_time: register(
Histogram::with_opts(
HistogramOpts::new(
"finality_proof_import_time",
"Time taken to import finality proofs",
),
)?,
registry,
)?,
justification_import_time: register(
Histogram::with_opts(
HistogramOpts::new(