(
- &self,
- wrapped_block_import: I,
- client: Arc,
- spawner: impl Spawn,
- keystore: KeyStorePtr,
- ) -> ClientResult>
- where
- P: ProvideRuntimeApi + BlockchainEvents + BlockBackend + Send + Sync + 'static,
- P::Api: ParachainHost,
- P::Api: ApiExt,
- // Rust bug: https://github.com/rust-lang/rust/issues/24159
- sp_api::StateBackendFor: sp_api::StateBackend>,
- {
- let to_worker = self.to_worker.clone();
-
- let import = AvailabilityBlockImport::new(
- client,
- wrapped_block_import,
- spawner,
- keystore,
- to_worker,
- );
-
- Ok(import)
- }
-
- /// Make some data available provisionally.
- ///
- /// Validators with the responsibility of maintaining availability
- /// for a block or collators collating a block will call this function
- /// in order to persist that data to disk and so it can be queried and provided
- /// to other nodes in the network.
- ///
- /// Determination of invalidity is beyond the scope of this function.
- ///
- /// This method will send the data to the background worker, allowing the caller to
- /// asynchronously wait for the result.
- pub async fn make_available(&self, candidate_hash: Hash, available_data: AvailableData)
- -> io::Result<()>
- {
- let (s, r) = oneshot::channel();
- let msg = WorkerMsg::MakeAvailable(MakeAvailable {
- candidate_hash,
- available_data,
- result: s,
- });
-
- let _ = self.to_worker.unbounded_send(msg);
-
- if let Ok(Ok(())) = r.await {
- Ok(())
- } else {
- Err(io::Error::new(io::ErrorKind::Other, format!("adding erasure chunks failed")))
- }
-
- }
-
- /// Get a set of all chunks we are waiting for.
- pub fn awaited_chunks(&self) -> Option> {
- self.inner.awaited_chunks()
- }
-
- /// Adds an erasure chunk to storage.
- ///
- /// The chunk should be checked for validity against the root of encoding
- /// and its proof prior to calling this.
- ///
- /// This method will send the chunk to the background worker, allowing the caller to
- /// asynchronously wait for the result.
- pub async fn add_erasure_chunk(
- &self,
- candidate: AbridgedCandidateReceipt,
- n_validators: u32,
- chunk: ErasureChunk,
- ) -> io::Result<()> {
- self.add_erasure_chunks(candidate, n_validators, std::iter::once(chunk)).await
- }
-
- /// Adds a set of erasure chunks to storage.
- ///
- /// The chunks should be checked for validity against the root of encoding
- /// and its proof prior to calling this.
- ///
- /// This method will send the chunks to the background worker, allowing the caller to
- /// asynchronously wait for the result.
- pub async fn add_erasure_chunks(
- &self,
- candidate: AbridgedCandidateReceipt,
- n_validators: u32,
- chunks: I,
- ) -> io::Result<()>
- where I: IntoIterator-
- {
- let candidate_hash = candidate.hash();
-
- self.add_candidate(candidate).await?;
-
- let (s, r) = oneshot::channel();
- let chunks = chunks.into_iter().collect();
-
- let msg = WorkerMsg::Chunks(Chunks {
- candidate_hash,
- chunks,
- n_validators,
- result: s,
- });
-
- let _ = self.to_worker.unbounded_send(msg);
-
- if let Ok(Ok(())) = r.await {
- Ok(())
- } else {
- Err(io::Error::new(io::ErrorKind::Other, format!("adding erasure chunks failed")))
- }
- }
-
- /// Queries an erasure chunk by the candidate hash and validator index.
- pub fn get_erasure_chunk(
- &self,
- candidate_hash: &Hash,
- validator_index: usize,
- ) -> Option {
- self.inner.get_erasure_chunk(candidate_hash, validator_index)
- }
-
- /// Note a validator's index and a number of validators at a relay parent in the
- /// store.
- ///
- /// This should be done before adding erasure chunks with this relay parent.
- pub fn note_validator_index_and_n_validators(
- &self,
- relay_parent: &Hash,
- validator_index: u32,
- n_validators: u32,
- ) -> io::Result<()> {
- self.inner.note_validator_index_and_n_validators(
- relay_parent,
- validator_index,
- n_validators,
- )
- }
-
- // Stores a candidate receipt.
- async fn add_candidate(
- &self,
- candidate: AbridgedCandidateReceipt,
- ) -> io::Result<()> {
- let (s, r) = oneshot::channel();
-
- let msg = WorkerMsg::IncludedParachainBlocks(IncludedParachainBlocks {
- blocks: vec![crate::worker::IncludedParachainBlock {
- candidate,
- available_data: None,
- }],
- result: s,
- });
-
- let _ = self.to_worker.unbounded_send(msg);
-
- if let Ok(Ok(())) = r.await {
- Ok(())
- } else {
- Err(io::Error::new(io::ErrorKind::Other, format!("adding erasure chunks failed")))
- }
- }
-
- /// Queries a candidate receipt by its hash.
- pub fn get_candidate(&self, candidate_hash: &Hash)
- -> Option
- {
- self.inner.get_candidate(candidate_hash)
- }
-
- /// Query execution data by pov-block hash.
- pub fn execution_data(&self, candidate_hash: &Hash)
- -> Option
- {
- self.inner.execution_data(candidate_hash)
- }
-}
diff --git a/availability-store/src/store.rs b/availability-store/src/store.rs
deleted file mode 100644
index e3b1e35929795211130d09e84454c28a34faeea4..0000000000000000000000000000000000000000
--- a/availability-store/src/store.rs
+++ /dev/null
@@ -1,623 +0,0 @@
-// Copyright 2018-2020 Parity Technologies (UK) Ltd.
-// This file is part of Polkadot.
-
-// Polkadot 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.
-
-// Polkadot 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 Polkadot. If not, see .
-
-#[cfg(not(target_os = "unknown"))]
-use kvdb_rocksdb::{Database, DatabaseConfig};
-use kvdb::{KeyValueDB, DBTransaction};
-use codec::{Encode, Decode};
-use polkadot_erasure_coding::{self as erasure};
-use polkadot_primitives::{
- Hash,
- parachain::{
- ErasureChunk, AvailableData, AbridgedCandidateReceipt,
- },
-};
-use parking_lot::Mutex;
-
-use log::{trace, warn};
-use std::collections::HashSet;
-use std::sync::Arc;
-use std::iter::FromIterator;
-use std::io;
-
-use crate::{LOG_TARGET, Config, ExecutionData};
-
-mod columns {
- pub const DATA: u32 = 0;
- pub const META: u32 = 1;
- pub const NUM_COLUMNS: u32 = 2;
-}
-
-#[derive(Clone)]
-pub struct Store {
- inner: Arc,
- candidate_descendents_lock: Arc>
-}
-
-// data keys
-fn execution_data_key(candidate_hash: &Hash) -> Vec {
- (candidate_hash, 0i8).encode()
-}
-
-fn erasure_chunks_key(candidate_hash: &Hash) -> Vec {
- (candidate_hash, 1i8).encode()
-}
-
-fn candidate_key(candidate_hash: &Hash) -> Vec {
- (candidate_hash, 2i8).encode()
-}
-
-fn candidates_with_relay_parent_key(relay_block: &Hash) -> Vec {
- (relay_block, 4i8).encode()
-}
-
-// meta keys
-const AWAITED_CHUNKS_KEY: [u8; 14] = *b"awaited_chunks";
-
-fn validator_index_and_n_validators_key(relay_parent: &Hash) -> Vec {
- (relay_parent, 1i8).encode()
-}
-
-fn available_chunks_key(candidate_hash: &Hash) -> Vec {
- (candidate_hash, 2i8).encode()
-}
-
-/// An entry in the awaited frontier of chunks we are interested in.
-#[derive(Encode, Decode, Debug, Hash, PartialEq, Eq, Clone)]
-pub struct AwaitedFrontierEntry {
- /// The hash of the candidate for which we want to fetch a chunk for.
- /// There will be duplicate entries in the case of multiple candidates with
- /// the same erasure-root, but this is unlikely.
- pub candidate_hash: Hash,
- /// Although the relay-parent is implicitly referenced by the candidate hash,
- /// we include it here as well for convenience in pruning the set.
- pub relay_parent: Hash,
- /// The index of the validator we represent.
- pub validator_index: u32,
-}
-
-impl Store {
- /// Create a new `Store` with given condig on disk.
- #[cfg(not(target_os = "unknown"))]
- pub(super) fn new(config: Config) -> io::Result {
- let mut db_config = DatabaseConfig::with_columns(columns::NUM_COLUMNS);
-
- if let Some(cache_size) = config.cache_size {
- let mut memory_budget = std::collections::HashMap::new();
- for i in 0..columns::NUM_COLUMNS {
- memory_budget.insert(i, cache_size / columns::NUM_COLUMNS as usize);
- }
-
- db_config.memory_budget = memory_budget;
- }
-
- let path = config.path.to_str().ok_or_else(|| io::Error::new(
- io::ErrorKind::Other,
- format!("Bad database path: {:?}", config.path),
- ))?;
-
- let db = Database::open(&db_config, &path)?;
-
- Ok(Store {
- inner: Arc::new(db),
- candidate_descendents_lock: Arc::new(Mutex::new(())),
- })
- }
-
- /// Create a new `Store` in-memory. Useful for tests.
- pub(super) fn new_in_memory() -> Self {
- Store {
- inner: Arc::new(::kvdb_memorydb::create(columns::NUM_COLUMNS)),
- candidate_descendents_lock: Arc::new(Mutex::new(())),
- }
- }
-
- /// Make some data available provisionally.
- pub(crate) fn make_available(&self, candidate_hash: Hash, available_data: AvailableData)
- -> io::Result<()>
- {
- let mut tx = DBTransaction::new();
-
- // at the moment, these structs are identical. later, we will also
- // keep outgoing message queues available, and these are not needed
- // for execution.
- let AvailableData { pov_block, omitted_validation } = available_data;
- let execution_data = ExecutionData {
- pov_block,
- omitted_validation,
- };
-
- tx.put_vec(
- columns::DATA,
- execution_data_key(&candidate_hash).as_slice(),
- execution_data.encode(),
- );
-
- self.inner.write(tx)
- }
-
- /// Get a set of all chunks we are waiting for.
- pub fn awaited_chunks(&self) -> Option> {
- self.query_inner(columns::META, &AWAITED_CHUNKS_KEY).map(|vec: Vec| {
- HashSet::from_iter(vec.into_iter())
- })
- }
-
- /// Adds a set of candidates hashes that were included in a relay block by the block's parent.
- ///
- /// If we already possess the receipts for these candidates _and_ our position at the specified
- /// relay chain the awaited frontier of the erasure chunks will also be extended.
- ///
- /// This method modifies the erasure chunks awaited frontier by adding this validator's
- /// chunks from `candidates` to it. In order to do so the information about this validator's
- /// position at parent `relay_parent` should be known to the store prior to calling this
- /// method, in other words `note_validator_index_and_n_validators` should be called for
- /// the given `relay_parent` before calling this function.
- pub(crate) fn note_candidates_with_relay_parent(
- &self,
- relay_parent: &Hash,
- candidates: &[Hash],
- ) -> io::Result<()> {
- let mut tx = DBTransaction::new();
- let dbkey = candidates_with_relay_parent_key(relay_parent);
-
- // This call can race against another call to `note_candidates_with_relay_parent`
- // with a different set of descendents.
- let _lock = self.candidate_descendents_lock.lock();
-
- if let Some((validator_index, _)) = self.get_validator_index_and_n_validators(relay_parent) {
- let candidates = candidates.clone();
- let awaited_frontier: Vec = self
- .query_inner(columns::META, &AWAITED_CHUNKS_KEY)
- .unwrap_or_else(|| Vec::new());
-
- let mut awaited_frontier: HashSet =
- HashSet::from_iter(awaited_frontier.into_iter());
-
- awaited_frontier.extend(candidates.iter().cloned().map(|candidate_hash| {
- AwaitedFrontierEntry {
- relay_parent: relay_parent.clone(),
- candidate_hash,
- validator_index,
- }
- }));
- let awaited_frontier = Vec::from_iter(awaited_frontier.into_iter());
- tx.put_vec(columns::META, &AWAITED_CHUNKS_KEY, awaited_frontier.encode());
- }
-
- let mut descendent_candidates = self.get_candidates_with_relay_parent(relay_parent);
- descendent_candidates.extend(candidates.iter().cloned());
- tx.put_vec(columns::DATA, &dbkey, descendent_candidates.encode());
-
- self.inner.write(tx)
- }
-
- /// Make a validator's index and a number of validators at a relay parent available.
- pub(crate) fn note_validator_index_and_n_validators(
- &self,
- relay_parent: &Hash,
- validator_index: u32,
- n_validators: u32,
- ) -> io::Result<()> {
- let mut tx = DBTransaction::new();
- let dbkey = validator_index_and_n_validators_key(relay_parent);
-
- tx.put_vec(columns::META, &dbkey, (validator_index, n_validators).encode());
-
- self.inner.write(tx)
- }
-
- /// Query a validator's index and n_validators by relay parent.
- pub(crate) fn get_validator_index_and_n_validators(&self, relay_parent: &Hash) -> Option<(u32, u32)> {
- let dbkey = validator_index_and_n_validators_key(relay_parent);
-
- self.query_inner(columns::META, &dbkey)
- }
-
- /// Add a set of chunks.
- ///
- /// The same as `add_erasure_chunk` but adds a set of chunks in one atomic transaction.
- pub fn add_erasure_chunks(
- &self,
- n_validators: u32,
- candidate_hash: &Hash,
- chunks: I,
- ) -> io::Result<()>
- where I: IntoIterator
-
- {
- if let Some(receipt) = self.get_candidate(candidate_hash) {
- let mut tx = DBTransaction::new();
- let dbkey = erasure_chunks_key(candidate_hash);
-
- let mut v = self.query_inner(columns::DATA, &dbkey).unwrap_or(Vec::new());
-
- let av_chunks_key = available_chunks_key(candidate_hash);
- let mut have_chunks = self.query_inner(columns::META, &av_chunks_key).unwrap_or(Vec::new());
-
- let awaited_frontier: Option> = self.query_inner(
- columns::META,
- &AWAITED_CHUNKS_KEY,
- );
-
- for chunk in chunks.into_iter() {
- if !have_chunks.contains(&chunk.index) {
- have_chunks.push(chunk.index);
- }
- v.push(chunk);
- }
-
- if let Some(mut awaited_frontier) = awaited_frontier {
- awaited_frontier.retain(|entry| {
- !(
- entry.relay_parent == receipt.relay_parent &&
- &entry.candidate_hash == candidate_hash &&
- have_chunks.contains(&entry.validator_index)
- )
- });
- tx.put_vec(columns::META, &AWAITED_CHUNKS_KEY, awaited_frontier.encode());
- }
-
- // If there are no block data in the store at this point,
- // check that they can be reconstructed now and add them to store if they can.
- if self.execution_data(&candidate_hash).is_none() {
- if let Ok(available_data) = erasure::reconstruct(
- n_validators as usize,
- v.iter().map(|chunk| (chunk.chunk.as_ref(), chunk.index as usize)),
- )
- {
- self.make_available(*candidate_hash, available_data)?;
- }
- }
-
- tx.put_vec(columns::DATA, &dbkey, v.encode());
- tx.put_vec(columns::META, &av_chunks_key, have_chunks.encode());
-
- self.inner.write(tx)
- } else {
- trace!(target: LOG_TARGET, "Candidate with hash {} not found", candidate_hash);
- Ok(())
- }
- }
-
- /// Queries an erasure chunk by its block's relay-parent, the candidate hash, and index.
- pub fn get_erasure_chunk(
- &self,
- candidate_hash: &Hash,
- index: usize,
- ) -> Option {
- self.query_inner(columns::DATA, &erasure_chunks_key(candidate_hash))
- .and_then(|chunks: Vec| {
- chunks.iter()
- .find(|chunk: &&ErasureChunk| chunk.index == index as u32)
- .map(|chunk| chunk.clone())
- })
- }
-
- /// Stores a candidate receipt.
- pub fn add_candidate(
- &self,
- receipt: &AbridgedCandidateReceipt,
- ) -> io::Result<()> {
- let candidate_hash = receipt.hash();
- let dbkey = candidate_key(&candidate_hash);
- let mut tx = DBTransaction::new();
-
- tx.put_vec(columns::DATA, &dbkey, receipt.encode());
-
- self.inner.write(tx)
- }
-
- /// Queries a candidate receipt by the relay parent hash and its hash.
- pub(crate) fn get_candidate(&self, candidate_hash: &Hash)
- -> Option
- {
- self.query_inner(columns::DATA, &candidate_key(candidate_hash))
- }
-
- /// Note that a set of candidates have been included in a finalized block with given hash and parent hash.
- pub(crate) fn candidates_finalized(
- &self,
- relay_parent: Hash,
- finalized_candidates: HashSet,
- ) -> io::Result<()> {
- let mut tx = DBTransaction::new();
-
- let awaited_frontier: Option> = self
- .query_inner(columns::META, &AWAITED_CHUNKS_KEY);
-
- if let Some(mut awaited_frontier) = awaited_frontier {
- awaited_frontier.retain(|entry| entry.relay_parent != relay_parent);
- tx.put_vec(columns::META, &AWAITED_CHUNKS_KEY, awaited_frontier.encode());
- }
-
- let candidates = self.get_candidates_with_relay_parent(&relay_parent);
-
- for candidate in candidates.into_iter().filter(|c| !finalized_candidates.contains(c)) {
- // we only delete this data for candidates which were not finalized.
- // we keep all data for the finalized chain forever at the moment.
- tx.delete(columns::DATA, execution_data_key(&candidate).as_slice());
- tx.delete(columns::DATA, &erasure_chunks_key(&candidate));
- tx.delete(columns::DATA, &candidate_key(&candidate));
-
- tx.delete(columns::META, &available_chunks_key(&candidate));
- }
-
- self.inner.write(tx)
- }
-
- /// Query execution data by relay parent and candidate hash.
- pub(crate) fn execution_data(&self, candidate_hash: &Hash) -> Option {
- self.query_inner(columns::DATA, &execution_data_key(candidate_hash))
- }
-
- /// Get candidates which pinned to the environment of the given relay parent.
- /// Note that this is not necessarily the same as candidates that were included in a direct
- /// descendent of the given relay-parent.
- fn get_candidates_with_relay_parent(&self, relay_parent: &Hash) -> Vec {
- let key = candidates_with_relay_parent_key(relay_parent);
- self.query_inner(columns::DATA, &key[..]).unwrap_or_default()
- }
-
- fn query_inner(&self, column: u32, key: &[u8]) -> Option {
- match self.inner.get(column, key) {
- Ok(Some(raw)) => {
- let res = T::decode(&mut &raw[..]).expect("all stored data serialized correctly; qed");
- Some(res)
- }
- Ok(None) => None,
- Err(e) => {
- warn!(target: LOG_TARGET, "Error reading from the availability store: {:?}", e);
- None
- }
- }
- }
-}
-
-#[cfg(test)]
-mod tests {
- use super::*;
- use polkadot_erasure_coding::{self as erasure};
- use polkadot_primitives::parachain::{
- Id as ParaId, BlockData, AvailableData, PoVBlock, OmittedValidationData,
- };
-
- fn available_data(block_data: &[u8]) -> AvailableData {
- AvailableData {
- pov_block: PoVBlock {
- block_data: BlockData(block_data.to_vec()),
- },
- omitted_validation: OmittedValidationData {
- global_validation: Default::default(),
- local_validation: Default::default(),
- }
- }
- }
-
- fn execution_data(available: &AvailableData) -> ExecutionData {
- let AvailableData { pov_block, omitted_validation } = available.clone();
- ExecutionData { pov_block, omitted_validation }
- }
-
- #[test]
- fn finalization_removes_unneeded() {
- let relay_parent = [1; 32].into();
-
- let para_id_1 = 5.into();
- let para_id_2 = 6.into();
-
- let mut candidate_1 = AbridgedCandidateReceipt::default();
- let mut candidate_2 = AbridgedCandidateReceipt::default();
-
- candidate_1.parachain_index = para_id_1;
- candidate_1.commitments.erasure_root = [6; 32].into();
- candidate_1.relay_parent = relay_parent;
-
- candidate_2.parachain_index = para_id_2;
- candidate_2.commitments.erasure_root = [6; 32].into();
- candidate_2.relay_parent = relay_parent;
-
-
- let candidate_1_hash = candidate_1.hash();
- let candidate_2_hash = candidate_2.hash();
-
- let available_data_1 = available_data(&[1, 2, 3]);
- let available_data_2 = available_data(&[4, 5, 6]);
-
- let erasure_chunk_1 = ErasureChunk {
- chunk: vec![10, 20, 30],
- index: 1,
- proof: vec![],
- };
-
- let erasure_chunk_2 = ErasureChunk {
- chunk: vec![40, 50, 60],
- index: 1,
- proof: vec![],
- };
-
- let store = Store::new_in_memory();
- store.make_available(candidate_1_hash, available_data_1.clone()).unwrap();
-
- store.make_available(candidate_2_hash, available_data_2.clone()).unwrap();
-
- store.add_candidate(&candidate_1).unwrap();
- store.add_candidate(&candidate_2).unwrap();
-
- store.note_candidates_with_relay_parent(&relay_parent, &[candidate_1_hash, candidate_2_hash]).unwrap();
-
- assert!(store.add_erasure_chunks(3, &candidate_1_hash, vec![erasure_chunk_1.clone()]).is_ok());
- assert!(store.add_erasure_chunks(3, &candidate_2_hash, vec![erasure_chunk_2.clone()]).is_ok());
-
- assert_eq!(store.execution_data(&candidate_1_hash).unwrap(), execution_data(&available_data_1));
- assert_eq!(store.execution_data(&candidate_2_hash).unwrap(), execution_data(&available_data_2));
-
- assert_eq!(store.get_erasure_chunk(&candidate_1_hash, 1).as_ref(), Some(&erasure_chunk_1));
- assert_eq!(store.get_erasure_chunk(&candidate_2_hash, 1), Some(erasure_chunk_2));
-
- assert_eq!(store.get_candidate(&candidate_1_hash), Some(candidate_1.clone()));
- assert_eq!(store.get_candidate(&candidate_2_hash), Some(candidate_2.clone()));
-
- store.candidates_finalized(relay_parent, [candidate_1_hash].iter().cloned().collect()).unwrap();
-
- assert_eq!(store.get_erasure_chunk(&candidate_1_hash, 1).as_ref(), Some(&erasure_chunk_1));
- assert!(store.get_erasure_chunk(&candidate_2_hash, 1).is_none());
-
- assert_eq!(store.get_candidate(&candidate_1_hash), Some(candidate_1));
- assert_eq!(store.get_candidate(&candidate_2_hash), None);
-
- assert_eq!(store.execution_data(&candidate_1_hash).unwrap(), execution_data(&available_data_1));
- assert!(store.execution_data(&candidate_2_hash).is_none());
- }
-
- #[test]
- fn erasure_coding() {
- let relay_parent: Hash = [1; 32].into();
- let para_id: ParaId = 5.into();
- let available_data = available_data(&[42; 8]);
- let n_validators = 5;
-
- let erasure_chunks = erasure::obtain_chunks(
- n_validators,
- &available_data,
- ).unwrap();
-
- let branches = erasure::branches(erasure_chunks.as_ref());
-
- let mut candidate = AbridgedCandidateReceipt::default();
- candidate.parachain_index = para_id;
- candidate.commitments.erasure_root = [6; 32].into();
- candidate.relay_parent = relay_parent;
-
- let candidate_hash = candidate.hash();
-
- let chunks: Vec<_> = erasure_chunks
- .iter()
- .zip(branches.map(|(proof, _)| proof))
- .enumerate()
- .map(|(index, (chunk, proof))| ErasureChunk {
- chunk: chunk.clone(),
- proof,
- index: index as u32,
- })
- .collect();
-
- let store = Store::new_in_memory();
-
- store.add_candidate(&candidate).unwrap();
- store.add_erasure_chunks(n_validators as u32, &candidate_hash, vec![chunks[0].clone()]).unwrap();
- assert_eq!(store.get_erasure_chunk(&candidate_hash, 0), Some(chunks[0].clone()));
-
- assert!(store.execution_data(&candidate_hash).is_none());
-
- store.add_erasure_chunks(n_validators as u32, &candidate_hash, chunks).unwrap();
- assert_eq!(store.execution_data(&candidate_hash), Some(execution_data(&available_data)));
- }
-
- #[test]
- fn add_validator_index_works() {
- let relay_parent = [42; 32].into();
- let store = Store::new_in_memory();
-
- store.note_validator_index_and_n_validators(&relay_parent, 42, 24).unwrap();
- assert_eq!(store.get_validator_index_and_n_validators(&relay_parent).unwrap(), (42, 24));
- }
-
- #[test]
- fn add_candidates_in_relay_block_works() {
- let relay_parent = [42; 32].into();
- let store = Store::new_in_memory();
-
- let candidates = vec![[1; 32].into(), [2; 32].into(), [3; 32].into()];
-
- store.note_candidates_with_relay_parent(&relay_parent, &candidates).unwrap();
- assert_eq!(store.get_candidates_with_relay_parent(&relay_parent), candidates);
- }
-
- #[test]
- fn awaited_chunks_works() {
- use std::iter::FromIterator;
- let validator_index = 3;
- let n_validators = 10;
- let relay_parent = [42; 32].into();
- let erasure_root_1 = [11; 32].into();
- let erasure_root_2 = [12; 32].into();
- let mut receipt_1 = AbridgedCandidateReceipt::default();
- let mut receipt_2 = AbridgedCandidateReceipt::default();
-
-
- receipt_1.parachain_index = 1.into();
- receipt_1.commitments.erasure_root = erasure_root_1;
- receipt_1.relay_parent = relay_parent;
-
- receipt_2.parachain_index = 2.into();
- receipt_2.commitments.erasure_root = erasure_root_2;
- receipt_2.relay_parent = relay_parent;
-
- let receipt_1_hash = receipt_1.hash();
- let receipt_2_hash = receipt_2.hash();
-
- let chunk = ErasureChunk {
- chunk: vec![1, 2, 3],
- index: validator_index,
- proof: Vec::new(),
- };
- let candidates = vec![receipt_1_hash, receipt_2_hash];
-
- let store = Store::new_in_memory();
-
- store.note_validator_index_and_n_validators(
- &relay_parent,
- validator_index,
- n_validators
- ).unwrap();
- store.add_candidate(&receipt_1).unwrap();
- store.add_candidate(&receipt_2).unwrap();
-
- // We are waiting for chunks from two candidates.
- store.note_candidates_with_relay_parent(&relay_parent, &candidates).unwrap();
-
- let awaited_frontier = store.awaited_chunks().unwrap();
- warn!(target: "availability", "awaited {:?}", awaited_frontier);
- let expected: HashSet<_> = candidates
- .clone()
- .into_iter()
- .map(|c| AwaitedFrontierEntry {
- relay_parent,
- candidate_hash: c,
- validator_index,
- })
- .collect();
- assert_eq!(awaited_frontier, expected);
-
- // We add chunk from one of the candidates.
- store.add_erasure_chunks(n_validators, &receipt_1_hash, vec![chunk]).unwrap();
-
- let awaited_frontier = store.awaited_chunks().unwrap();
- // Now we wait for the other chunk that we haven't received yet.
- let expected: HashSet<_> = vec![AwaitedFrontierEntry {
- relay_parent,
- candidate_hash: receipt_2_hash,
- validator_index,
- }].into_iter().collect();
-
- assert_eq!(awaited_frontier, expected);
-
- // Finalizing removes awaited candidates from frontier.
- store.candidates_finalized(relay_parent, HashSet::from_iter(candidates.into_iter())).unwrap();
-
- assert_eq!(store.awaited_chunks().unwrap().len(), 0);
- }
-}
diff --git a/availability-store/src/worker.rs b/availability-store/src/worker.rs
deleted file mode 100644
index 0ff59a9fca38377bd30c1fd405184f4796cce7f1..0000000000000000000000000000000000000000
--- a/availability-store/src/worker.rs
+++ /dev/null
@@ -1,942 +0,0 @@
-// Copyright 2018-2020 Parity Technologies (UK) Ltd.
-// This file is part of Polkadot.
-
-// Polkadot 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.
-
-// Polkadot 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 Polkadot. If not, see .
-
-use std::collections::{HashMap, HashSet};
-use std::io;
-use std::sync::Arc;
-use std::thread;
-
-use log::{error, info, trace, warn};
-use sp_blockchain::{Result as ClientResult};
-use sp_runtime::traits::{Header as HeaderT, Block as BlockT, HashFor, BlakeTwo256};
-use sp_api::{ApiExt, ProvideRuntimeApi};
-use client::{
- BlockchainEvents, BlockBackend,
- blockchain::ProvideCache,
-};
-use consensus_common::{
- self, BlockImport, BlockCheckParams, BlockImportParams, Error as ConsensusError,
- ImportResult,
- import_queue::CacheKeyId,
-};
-use polkadot_primitives::{Block, BlockId, Hash};
-use polkadot_primitives::parachain::{
- ParachainHost, ValidatorId, AbridgedCandidateReceipt, AvailableData,
- ValidatorPair, ErasureChunk,
-};
-use futures::{prelude::*, future::select, channel::{mpsc, oneshot}, task::{Spawn, SpawnExt}};
-use futures::future::AbortHandle;
-use keystore::KeyStorePtr;
-
-use tokio::runtime::{Handle, Runtime as LocalRuntime};
-
-use crate::{LOG_TARGET, ErasureNetworking};
-use crate::store::Store;
-
-/// Errors that may occur.
-#[derive(Debug, derive_more::Display, derive_more::From)]
-pub(crate) enum Error {
- #[from]
- StoreError(io::Error),
- #[display(fmt = "Validator's id and number of validators at block with parent {} not found", relay_parent)]
- IdAndNValidatorsNotFound { relay_parent: Hash },
-}
-
-/// Used in testing to interact with the worker thread.
-#[cfg(test)]
-pub(crate) struct WithWorker(Box);
-
-#[cfg(test)]
-impl std::fmt::Debug for WithWorker {
- fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
- write!(f, "")
- }
-}
-
-/// Messages sent to the `Worker`.
-///
-/// Messages are sent in a number of different scenarios,
-/// for instance, when:
-/// * importing blocks in `BlockImport` implementation,
-/// * recieving finality notifications,
-/// * when the `Store` api is used by outside code.
-#[derive(Debug)]
-pub(crate) enum WorkerMsg {
- IncludedParachainBlocks(IncludedParachainBlocks),
- Chunks(Chunks),
- CandidatesFinalized(CandidatesFinalized),
- MakeAvailable(MakeAvailable),
- #[cfg(test)]
- WithWorker(WithWorker),
-}
-
-/// A notification of a parachain block included in the relay chain.
-#[derive(Debug)]
-pub(crate) struct IncludedParachainBlock {
- /// The abridged candidate receipt, extracted from a relay-chain block.
- pub candidate: AbridgedCandidateReceipt,
- /// The data to keep available from the candidate, if known.
- pub available_data: Option,
-}
-
-/// The receipts of the heads included into the block with a given parent.
-#[derive(Debug)]
-pub(crate) struct IncludedParachainBlocks {
- /// The blocks themselves.
- pub blocks: Vec,
- /// A sender to signal the result asynchronously.
- pub result: oneshot::Sender>,
-}
-
-/// We have received chunks we requested.
-#[derive(Debug)]
-pub(crate) struct Chunks {
- /// The hash of the parachain candidate these chunks belong to.
- pub candidate_hash: Hash,
- /// The chunks
- pub chunks: Vec,
- /// The number of validators present at the candidate's relay-parent.
- pub n_validators: u32,
- /// A sender to signal the result asynchronously.
- pub result: oneshot::Sender>,
-}
-
-/// These candidates have been finalized, so unneded availability may be now pruned
-#[derive(Debug)]
-pub(crate) struct CandidatesFinalized {
- /// The relay parent of the block that was finalized.
- relay_parent: Hash,
- /// The hashes of candidates that were finalized in this block.
- included_candidates: HashSet,
-}
-
-/// The message that corresponds to `make_available` call of the crate API.
-#[derive(Debug)]
-pub(crate) struct MakeAvailable {
- /// The hash of the candidate for which we are publishing data.
- pub candidate_hash: Hash,
- /// The data to make available.
- pub available_data: AvailableData,
- /// A sender to signal the result asynchronously.
- pub result: oneshot::Sender>,
-}
-
-/// Description of a chunk we are listening for.
-#[derive(Hash, Debug, PartialEq, Eq)]
-struct ListeningKey {
- candidate_hash: Hash,
- index: u32,
-}
-
-/// An availability worker with it's inner state.
-pub(super) struct Worker {
- availability_store: Store,
- listening_for: HashMap,
-
- sender: mpsc::UnboundedSender,
-}
-
-/// The handle to the `Worker`.
-pub(super) struct WorkerHandle {
- thread: Option>>,
- sender: mpsc::UnboundedSender,
- exit_signal: Option,
-}
-
-impl WorkerHandle {
- pub(crate) fn to_worker(&self) -> &mpsc::UnboundedSender {
- &self.sender
- }
-}
-
-impl Drop for WorkerHandle {
- fn drop(&mut self) {
- if let Some(signal) = self.exit_signal.take() {
- let _ = signal.fire();
- }
-
- if let Some(thread) = self.thread.take() {
- if let Err(_) = thread.join() {
- error!(target: LOG_TARGET, "Errored stopping the thread");
- }
- }
- }
-}
-
-
-fn fetch_candidates
(client: &P, extrinsics: Vec<::Extrinsic>, parent: &BlockId)
- -> ClientResult>>
-where
- P: ProvideRuntimeApi,
- P::Api: ParachainHost,
- // Rust bug: https://github.com/rust-lang/rust/issues/24159
- sp_api::StateBackendFor: sp_api::StateBackend>,
-{
- let api = client.runtime_api();
-
- let candidates = if api.has_api_with::, _>(
- parent,
- |version| version >= 2,
- ).map_err(|e| ConsensusError::ChainLookup(e.to_string()))? {
- api.get_heads(&parent, extrinsics)
- .map_err(|e| ConsensusError::ChainLookup(e.to_string()))?
- } else {
- None
- };
-
- Ok(candidates)
-}
-
-/// Creates a task to prune entries in availability store upon block finalization.
-async fn prune_unneeded_availability(client: Arc, mut sender: S)
-where
- P: ProvideRuntimeApi + BlockchainEvents + BlockBackend + Send + Sync + 'static,
- P::Api: ParachainHost + ApiExt,
- S: Sink + Clone + Send + Sync + Unpin,
- // Rust bug: https://github.com/rust-lang/rust/issues/24159
- sp_api::StateBackendFor: sp_api::StateBackend>,
-{
- let mut finality_notification_stream = client.finality_notification_stream();
-
- while let Some(notification) = finality_notification_stream.next().await {
- let hash = notification.hash;
- let parent_hash = notification.header.parent_hash;
- let extrinsics = match client.block_body(&BlockId::hash(hash)) {
- Ok(Some(extrinsics)) => extrinsics,
- Ok(None) => {
- error!(
- target: LOG_TARGET,
- "No block body found for imported block {:?}",
- hash,
- );
- continue;
- }
- Err(e) => {
- error!(
- target: LOG_TARGET,
- "Failed to get block body for imported block {:?}: {:?}",
- hash,
- e,
- );
- continue;
- }
- };
-
- let included_candidates = match fetch_candidates(
- &*client,
- extrinsics,
- &BlockId::hash(parent_hash),
- ) {
- Ok(Some(candidates)) => candidates
- .into_iter()
- .map(|c| c.hash())
- .collect(),
- Ok(None) => {
- warn!(
- target: LOG_TARGET,
- "Failed to extract candidates from block body of imported block {:?}", hash
- );
- continue;
- }
- Err(e) => {
- warn!(
- target: LOG_TARGET,
- "Failed to fetch block body for imported block {:?}: {:?}", hash, e
- );
- continue;
- }
- };
-
- let msg = WorkerMsg::CandidatesFinalized(CandidatesFinalized {
- relay_parent: parent_hash,
- included_candidates
- });
-
- if let Err(_) = sender.send(msg).await {
- break;
- }
- }
-}
-
-impl Worker {
-
- // Called on startup of the worker to initiate fetch from network for all awaited chunks.
- fn initiate_all_fetches(
- &mut self,
- runtime_handle: &Handle,
- erasure_network: &EN,
- sender: &mut mpsc::UnboundedSender,
- ) {
- if let Some(awaited_chunks) = self.availability_store.awaited_chunks() {
- for awaited_chunk in awaited_chunks {
- if let Err(e) = self.initiate_fetch(
- runtime_handle,
- erasure_network,
- sender,
- awaited_chunk.relay_parent,
- awaited_chunk.candidate_hash,
- ) {
- warn!(target: LOG_TARGET, "Failed to register network listener: {}", e);
- }
- }
- }
- }
-
- // initiates a fetch from network for the described chunk, with our local index.
- fn initiate_fetch(
- &mut self,
- runtime_handle: &Handle,
- erasure_network: &EN,
- sender: &mut mpsc::UnboundedSender,
- relay_parent: Hash,
- candidate_hash: Hash,
- ) -> Result<(), Error> {
- let (local_id, n_validators) = self.availability_store
- .get_validator_index_and_n_validators(&relay_parent)
- .ok_or(Error::IdAndNValidatorsNotFound { relay_parent })?;
-
- // fast exit for if we already have the chunk.
- if self.availability_store.get_erasure_chunk(&candidate_hash, local_id as _).is_some() {
- return Ok(())
- }
-
- trace!(
- target: LOG_TARGET,
- "Initiating fetch for erasure-chunk at parent {} with candidate-hash {}",
- relay_parent,
- candidate_hash,
- );
-
- let fut = erasure_network.fetch_erasure_chunk(&candidate_hash, local_id);
- let mut sender = sender.clone();
- let (fut, signal) = future::abortable(async move {
- let chunk = match fut.await {
- Ok(chunk) => chunk,
- Err(e) => {
- warn!(target: LOG_TARGET, "Unable to fetch erasure-chunk from network: {:?}", e);
- return
- }
- };
- let (s, _) = oneshot::channel();
- let _ = sender.send(WorkerMsg::Chunks(Chunks {
- candidate_hash,
- chunks: vec![chunk],
- n_validators,
- result: s,
- })).await;
- }.map(drop).boxed());
-
-
- let key = ListeningKey {
- candidate_hash,
- index: local_id,
- };
-
- self.listening_for.insert(key, signal);
- let _ = runtime_handle.spawn(fut);
-
- Ok(())
- }
-
- fn on_parachain_blocks_received(
- &mut self,
- runtime_handle: &Handle,
- erasure_network: &EN,
- sender: &mut mpsc::UnboundedSender,
- blocks: Vec,
- ) -> Result<(), Error> {
- // First we have to add the receipts themselves.
- for IncludedParachainBlock { candidate, available_data }
- in blocks.into_iter()
- {
- let _ = self.availability_store.add_candidate(&candidate);
-
- if let Some(_available_data) = available_data {
- // Should we be breaking block into chunks here and gossiping it and so on?
- }
-
- // This leans on the codebase-wide assumption that the `relay_parent`
- // of all candidates in a block matches the parent hash of that block.
- //
- // In the future this will not always be true.
- let candidate_hash = candidate.hash();
- let _ = self.availability_store.note_candidates_with_relay_parent(
- &candidate.relay_parent,
- &[candidate_hash],
- );
-
- if let Err(e) = self.initiate_fetch(
- runtime_handle,
- erasure_network,
- sender,
- candidate.relay_parent,
- candidate_hash,
- ) {
- warn!(target: LOG_TARGET, "Failed to register chunk listener: {}", e);
- }
- }
-
- Ok(())
- }
-
- // Handles chunks that were required.
- fn on_chunks(
- &mut self,
- candidate_hash: Hash,
- chunks: Vec,
- n_validators: u32,
- ) -> Result<(), Error> {
- for c in &chunks {
- let key = ListeningKey {
- candidate_hash,
- index: c.index,
- };
-
- // remove bookkeeping so network does not attempt to fetch
- // any longer.
- if let Some(exit_signal) = self.listening_for.remove(&key) {
- exit_signal.abort();
- }
- }
-
- self.availability_store.add_erasure_chunks(
- n_validators,
- &candidate_hash,
- chunks,
- )?;
-
- Ok(())
- }
-
- /// Starts a worker with a given availability store and a gossip messages provider.
- pub fn start(
- availability_store: Store,
- erasure_network: EN,
- ) -> WorkerHandle {
- let (sender, mut receiver) = mpsc::unbounded();
-
- let mut worker = Worker {
- availability_store,
- listening_for: HashMap::new(),
- sender: sender.clone(),
- };
-
- let sender = sender.clone();
- let (signal, exit) = exit_future::signal();
-
- let handle = thread::spawn(move || -> io::Result<()> {
- let mut runtime = LocalRuntime::new()?;
- let mut sender = worker.sender.clone();
-
- let runtime_handle = runtime.handle().clone();
-
- // On startup, initiates fetch from network for all
- // entries in the awaited frontier.
- worker.initiate_all_fetches(runtime.handle(), &erasure_network, &mut sender);
-
- let process_notification = async move {
- while let Some(msg) = receiver.next().await {
- trace!(target: LOG_TARGET, "Received message {:?}", msg);
-
- let res = match msg {
- WorkerMsg::IncludedParachainBlocks(msg) => {
- let IncludedParachainBlocks {
- blocks,
- result,
- } = msg;
-
- let res = worker.on_parachain_blocks_received(
- &runtime_handle,
- &erasure_network,
- &mut sender,
- blocks,
- );
-
- let _ = result.send(res);
- Ok(())
- }
- WorkerMsg::Chunks(msg) => {
- let Chunks {
- candidate_hash,
- chunks,
- n_validators,
- result,
- } = msg;
-
- let res = worker.on_chunks(
- candidate_hash,
- chunks,
- n_validators,
- );
-
- let _ = result.send(res);
- Ok(())
- }
- WorkerMsg::CandidatesFinalized(msg) => {
- let CandidatesFinalized { relay_parent, included_candidates } = msg;
-
- worker.availability_store.candidates_finalized(
- relay_parent,
- included_candidates,
- )
- }
- WorkerMsg::MakeAvailable(msg) => {
- let MakeAvailable { candidate_hash, available_data, result } = msg;
- let res = worker.availability_store
- .make_available(candidate_hash, available_data)
- .map_err(|e| e.into());
- let _ = result.send(res);
- Ok(())
- }
- #[cfg(test)]
- WorkerMsg::WithWorker(with_worker) => {
- (with_worker.0)(&mut worker);
- Ok(())
- }
- };
-
- if let Err(_) = res {
- warn!(target: LOG_TARGET, "An error occured while processing a message");
- }
- }
-
- };
-
- runtime.spawn(select(process_notification.boxed(), exit.clone()).map(drop));
- runtime.block_on(exit);
-
- info!(target: LOG_TARGET, "Availability worker exiting");
-
- Ok(())
- });
-
- WorkerHandle {
- thread: Some(handle),
- sender,
- exit_signal: Some(signal),
- }
- }
-}
-
-/// Implementor of the [`BlockImport`] trait.
-///
-/// Used to embed `availability-store` logic into the block imporing pipeline.
-///
-/// [`BlockImport`]: https://substrate.dev/rustdocs/v1.0/substrate_consensus_common/trait.BlockImport.html
-pub struct AvailabilityBlockImport {
- inner: I,
- client: Arc,
- keystore: KeyStorePtr,
- to_worker: mpsc::UnboundedSender,
- exit_signal: AbortHandle,
-}
-
-impl Drop for AvailabilityBlockImport {
- fn drop(&mut self) {
- self.exit_signal.abort();
- }
-}
-
-impl BlockImport for AvailabilityBlockImport where
- I: BlockImport> + Send + Sync,
- I::Error: Into,
- P: ProvideRuntimeApi + ProvideCache,
- P::Api: ParachainHost,
- // Rust bug: https://github.com/rust-lang/rust/issues/24159
- sp_api::StateBackendFor: sp_api::StateBackend
-{
- type Error = ConsensusError;
- type Transaction = sp_api::TransactionFor;
-
- fn import_block(
- &mut self,
- block: BlockImportParams,
- new_cache: HashMap>,
- ) -> Result {
- trace!(
- target: LOG_TARGET,
- "Importing block #{}, ({})",
- block.header.number(),
- block.post_hash(),
- );
-
- if let Some(ref extrinsics) = block.body {
- let parent_id = BlockId::hash(*block.header.parent_hash());
- // Extract our local position i from the validator set of the parent.
- let validators = self.client.runtime_api().validators(&parent_id)
- .map_err(|e| ConsensusError::ChainLookup(e.to_string()))?;
-
- let our_id = self.our_id(&validators);
-
- // Use a runtime API to extract all included erasure-roots from the imported block.
- let candidates = fetch_candidates(&*self.client, extrinsics.clone(), &parent_id)
- .map_err(|e| ConsensusError::ChainLookup(e.to_string()))?;
-
- match candidates {
- Some(candidates) => {
- match our_id {
- Some(our_id) => {
- trace!(
- target: LOG_TARGET,
- "Our validator id is {}, the candidates included are {:?}",
- our_id,
- candidates,
- );
-
- let (s, _) = oneshot::channel();
-
- // Inform the worker about the included parachain blocks.
- let blocks = candidates
- .into_iter()
- .map(|c| IncludedParachainBlock {
- candidate: c,
- available_data: None,
- })
- .collect();
-
- let msg = WorkerMsg::IncludedParachainBlocks(IncludedParachainBlocks {
- blocks,
- result: s,
- });
-
- let _ = self.to_worker.unbounded_send(msg);
- }
- None => (),
- }
- }
- None => {
- trace!(
- target: LOG_TARGET,
- "No parachain heads were included in block {}", block.header.hash()
- );
- },
- }
- }
-
- self.inner.import_block(block, new_cache).map_err(Into::into)
- }
-
- fn check_block(
- &mut self,
- block: BlockCheckParams,
- ) -> Result {
- self.inner.check_block(block).map_err(Into::into)
- }
-}
-
-impl AvailabilityBlockImport {
- pub(crate) fn new(
- client: Arc,
- block_import: I,
- spawner: impl Spawn,
- keystore: KeyStorePtr,
- to_worker: mpsc::UnboundedSender,
- ) -> Self
- where
- P: ProvideRuntimeApi + BlockBackend + BlockchainEvents + Send + Sync + 'static,
- P::Api: ParachainHost,
- P::Api: ApiExt,
- // Rust bug: https://github.com/rust-lang/rust/issues/24159
- sp_api::StateBackendFor: sp_api::StateBackend>,
- {
- // This is not the right place to spawn the finality future,
- // it would be more appropriate to spawn it in the `start` method of the `Worker`.
- // However, this would make the type of the `Worker` and the `Store` itself
- // dependent on the types of client and executor, which would prove
- // not not so handy in the testing code.
- let (prune_available, exit_signal) = future::abortable(prune_unneeded_availability(
- client.clone(),
- to_worker.clone(),
- ));
-
- if let Err(_) = spawner.spawn(prune_available.map(drop)) {
- error!(target: LOG_TARGET, "Failed to spawn availability pruning task");
- }
-
- AvailabilityBlockImport {
- client,
- inner: block_import,
- to_worker,
- keystore,
- exit_signal,
- }
- }
-
- fn our_id(&self, validators: &[ValidatorId]) -> Option {
- let keystore = self.keystore.read();
- validators
- .iter()
- .enumerate()
- .find_map(|(i, v)| {
- keystore.key_pair::(&v).map(|_| i as u32).ok()
- })
- }
-}
-
-
-#[cfg(test)]
-mod tests {
- use super::*;
- use futures::channel::oneshot;
- use std::sync::Arc;
- use std::pin::Pin;
- use tokio::runtime::Runtime;
- use parking_lot::Mutex;
- use crate::store::AwaitedFrontierEntry;
-
- #[derive(Default, Clone)]
- struct TestErasureNetwork {
- chunk_receivers: Arc
- >>>,
- }
-
- impl TestErasureNetwork {
- // adds a receiver. this returns a sender for the erasure-chunk
- // along with an exit future that fires when the erasure chunk has
- // been fully-processed
- fn add_receiver(&self, candidate_hash: Hash, index: u32)
- -> oneshot::Sender
- {
- let (sender, receiver) = oneshot::channel();
- self.chunk_receivers.lock().insert((candidate_hash, index), receiver);
- sender
- }
- }
-
- impl ErasureNetworking for TestErasureNetwork {
- type Error = String;
-
- fn fetch_erasure_chunk(&self, candidate_hash: &Hash, index: u32)
- -> Pin> + Send>>
- {
- match self.chunk_receivers.lock().remove(&(*candidate_hash, index)) {
- Some(receiver) => receiver.then(|x| match x {
- Ok(x) => future::ready(Ok(x)).left_future(),
- Err(_) => future::pending().right_future(),
- }).boxed(),
- None => future::pending().boxed(),
- }
- }
-
- fn distribute_erasure_chunk(
- &self,
- _candidate_hash: Hash,
- _chunk: ErasureChunk
- ) {}
- }
-
- // This test tests that as soon as the worker receives info about new parachain blocks
- // included it registers gossip listeners for it's own chunks. Upon receiving the awaited
- // chunk messages the corresponding listeners are deregistered and these chunks are removed
- // from the awaited chunks set.
- #[test]
- fn receiving_gossip_chunk_removes_from_frontier() {
- let mut runtime = Runtime::new().unwrap();
- let relay_parent = [1; 32].into();
- let local_id = 2;
- let n_validators = 4;
-
- let store = Store::new_in_memory();
-
- let mut candidate = AbridgedCandidateReceipt::default();
-
- candidate.relay_parent = relay_parent;
- let candidate_hash = candidate.hash();
-
- // Tell the store our validator's position and the number of validators at given point.
- store.note_validator_index_and_n_validators(&relay_parent, local_id, n_validators).unwrap();
-
- let network = TestErasureNetwork::default();
- let chunk_sender = network.add_receiver(candidate_hash, local_id);
-
- // At this point we shouldn't be waiting for any chunks.
- assert!(store.awaited_chunks().is_none());
-
- let (s, r) = oneshot::channel();
-
- let msg = WorkerMsg::IncludedParachainBlocks(IncludedParachainBlocks {
- blocks: vec![IncludedParachainBlock {
- candidate,
- available_data: None,
- }],
- result: s,
- });
-
- let handle = Worker::start(store.clone(), network);
-
- // Tell the worker that the new blocks have been included into the relay chain.
- // This should trigger the registration of gossip message listeners for the
- // chunk topics.
- handle.sender.unbounded_send(msg).unwrap();
-
- runtime.block_on(r).unwrap().unwrap();
-
- // Make sure that at this point we are waiting for the appropriate chunk.
- assert_eq!(
- store.awaited_chunks().unwrap(),
- vec![AwaitedFrontierEntry {
- relay_parent,
- candidate_hash,
- validator_index: local_id,
- }].into_iter().collect()
- );
-
- // Complete the chunk request.
- chunk_sender.send(ErasureChunk {
- chunk: vec![1, 2, 3],
- index: local_id as u32,
- proof: vec![],
- }).unwrap();
-
- // wait until worker thread has de-registered the listener for a
- // particular chunk.
- loop {
- let (s, r) = oneshot::channel();
- handle.sender.unbounded_send(WorkerMsg::WithWorker(WithWorker(Box::new(move |worker| {
- let key = ListeningKey {
- candidate_hash,
- index: local_id,
- };
-
- let is_waiting = worker.listening_for.contains_key(&key);
-
- s.send(!is_waiting).unwrap(); // tell the test thread `true` if we are not waiting.
- })))).unwrap();
-
- if runtime.block_on(r).unwrap() {
- break
- }
- }
-
- // The awaited chunk has been received so at this point we no longer wait for any chunks.
- assert_eq!(store.awaited_chunks().unwrap().len(), 0);
- }
-
- #[test]
- fn included_parachain_blocks_registers_listener() {
- let mut runtime = Runtime::new().unwrap();
- let relay_parent = [1; 32].into();
- let erasure_root_1 = [2; 32].into();
- let erasure_root_2 = [3; 32].into();
- let pov_block_hash_1 = [4; 32].into();
- let pov_block_hash_2 = [5; 32].into();
- let local_id = 2;
- let n_validators = 4;
-
- let mut candidate_1 = AbridgedCandidateReceipt::default();
- candidate_1.commitments.erasure_root = erasure_root_1;
- candidate_1.pov_block_hash = pov_block_hash_1;
- candidate_1.relay_parent = relay_parent;
- let candidate_1_hash = candidate_1.hash();
-
- let mut candidate_2 = AbridgedCandidateReceipt::default();
- candidate_2.commitments.erasure_root = erasure_root_2;
- candidate_2.pov_block_hash = pov_block_hash_2;
- candidate_2.relay_parent = relay_parent;
- let candidate_2_hash = candidate_2.hash();
-
- let store = Store::new_in_memory();
-
- // Tell the store our validator's position and the number of validators at given point.
- store.note_validator_index_and_n_validators(&relay_parent, local_id, n_validators).unwrap();
-
- // Let the store know about the candidates
- store.add_candidate(&candidate_1).unwrap();
- store.add_candidate(&candidate_2).unwrap();
-
- // And let the store know about the chunk from the second candidate.
- store.add_erasure_chunks(
- n_validators,
- &candidate_2_hash,
- vec![ErasureChunk {
- chunk: vec![1, 2, 3],
- index: local_id,
- proof: Vec::default(),
- }],
- ).unwrap();
-
- let network = TestErasureNetwork::default();
- let _ = network.add_receiver(candidate_1_hash, local_id);
- let _ = network.add_receiver(candidate_2_hash, local_id);
-
- let handle = Worker::start(store.clone(), network.clone());
-
- {
- let (s, r) = oneshot::channel();
- // Tell the worker to listen for chunks from candidate 2 (we alredy have a chunk from it).
- let listen_msg_2 = WorkerMsg::IncludedParachainBlocks(IncludedParachainBlocks {
- blocks: vec![IncludedParachainBlock {
- candidate: candidate_2,
- available_data: None,
- }],
- result: s,
- });
-
- handle.sender.unbounded_send(listen_msg_2).unwrap();
-
- runtime.block_on(r).unwrap().unwrap();
- // The receiver for this chunk left intact => listener not registered.
- assert!(network.chunk_receivers.lock().contains_key(&(candidate_2_hash, local_id)));
-
- // more directly:
- let (s, r) = oneshot::channel();
- handle.sender.unbounded_send(WorkerMsg::WithWorker(WithWorker(Box::new(move |worker| {
- let key = ListeningKey {
- candidate_hash: candidate_2_hash,
- index: local_id,
- };
- let _ = s.send(worker.listening_for.contains_key(&key));
- })))).unwrap();
-
- assert!(!runtime.block_on(r).unwrap());
- }
-
- {
- let (s, r) = oneshot::channel();
-
- // Tell the worker to listen for chunks from candidate 1.
- // (we don't have a chunk from it yet).
- let listen_msg_1 = WorkerMsg::IncludedParachainBlocks(IncludedParachainBlocks {
- blocks: vec![IncludedParachainBlock {
- candidate: candidate_1,
- available_data: None,
- }],
- result: s,
- });
-
- handle.sender.unbounded_send(listen_msg_1).unwrap();
- runtime.block_on(r).unwrap().unwrap();
-
- // The receiver taken => listener registered.
- assert!(!network.chunk_receivers.lock().contains_key(&(candidate_1_hash, local_id)));
-
-
- // more directly:
- let (s, r) = oneshot::channel();
- handle.sender.unbounded_send(WorkerMsg::WithWorker(WithWorker(Box::new(move |worker| {
- let key = ListeningKey {
- candidate_hash: candidate_1_hash,
- index: local_id,
- };
- let _ = s.send(worker.listening_for.contains_key(&key));
- })))).unwrap();
-
- assert!(runtime.block_on(r).unwrap());
- }
- }
-}
diff --git a/cli/Cargo.toml b/cli/Cargo.toml
index ff67b236c547a586fe52b7861d1e6dd5bb94b52b..99eb04e9dd6377a9286bf3cf918d796ad9ea36e1 100644
--- a/cli/Cargo.toml
+++ b/cli/Cargo.toml
@@ -1,6 +1,6 @@
[package]
name = "polkadot-cli"
-version = "0.8.12"
+version = "0.8.22"
authors = ["Parity Technologies "]
description = "Polkadot Relay-chain Client Node"
edition = "2018"
@@ -34,12 +34,15 @@ sc-service = { git = "https://github.com/paritytech/substrate", branch = "master
wasm-bindgen = { version = "0.2.57", optional = true }
wasm-bindgen-futures = { version = "0.4.7", optional = true }
browser-utils = { package = "substrate-browser-utils", git = "https://github.com/paritytech/substrate", branch = "master", optional = true }
+# this crate is used only to enable `trie-memory-tracker` feature
+# see https://github.com/paritytech/substrate/pull/6745
+sp-trie = { git = "https://github.com/paritytech/substrate", branch = "master", default-features = false }
[build-dependencies]
substrate-build-script-utils = { git = "https://github.com/paritytech/substrate", branch = "master" }
[features]
-default = [ "wasmtime", "db", "cli", "service-old" ]
+default = [ "wasmtime", "db", "cli", "service-old", "trie-memory-tracker" ]
wasmtime = [ "sc-cli/wasmtime" ]
db = [ "service/db" ]
cli = [
@@ -57,3 +60,4 @@ browser = [
]
runtime-benchmarks = [ "service/runtime-benchmarks" ]
service-rewr = [ "service-new/full-node" ]
+trie-memory-tracker = [ "sp-trie/memory-tracker" ]
diff --git a/cli/src/browser.rs b/cli/src/browser.rs
index 6f3a4000843ae1c2394681f32d8789469bf6cba2..d3523e92a60005ff305963cf7bff516e1a67341a 100644
--- a/cli/src/browser.rs
+++ b/cli/src/browser.rs
@@ -46,8 +46,7 @@ async fn start_inner(chain_spec: String, log_level: String) -> Result,
- #[allow(missing_docs)]
#[structopt(flatten)]
pub run: RunCmd,
}
diff --git a/cli/src/command.rs b/cli/src/command.rs
index 714e3c2ddacd14f0ab33f20cec42b52459cac8d8..964e13e6d3fd8a6e722cfc5ef342f635c30b5d18 100644
--- a/cli/src/command.rs
+++ b/cli/src/command.rs
@@ -19,8 +19,7 @@ use log::info;
use service::{IdentifyVariant, self};
#[cfg(feature = "service-rewr")]
use service_new::{IdentifyVariant, self as service};
-use sc_executor::NativeExecutionDispatch;
-use sc_cli::{SubstrateCli, Result};
+use sc_cli::{SubstrateCli, Result, RuntimeVersion, Role};
use crate::cli::{Cli, Subcommand};
fn get_exec_name() -> Option {
@@ -31,19 +30,19 @@ fn get_exec_name() -> Option {
}
impl SubstrateCli for Cli {
- fn impl_name() -> &'static str { "Parity Polkadot" }
+ fn impl_name() -> String { "Parity Polkadot".into() }
- fn impl_version() -> &'static str { env!("SUBSTRATE_CLI_IMPL_VERSION") }
+ fn impl_version() -> String { env!("SUBSTRATE_CLI_IMPL_VERSION").into() }
- fn description() -> &'static str { env!("CARGO_PKG_DESCRIPTION") }
+ fn description() -> String { env!("CARGO_PKG_DESCRIPTION").into() }
- fn author() -> &'static str { env!("CARGO_PKG_AUTHORS") }
+ fn author() -> String { env!("CARGO_PKG_AUTHORS").into() }
- fn support_url() -> &'static str { "https://github.com/paritytech/polkadot/issues/new" }
+ fn support_url() -> String { "https://github.com/paritytech/polkadot/issues/new".into() }
fn copyright_start_year() -> i32 { 2017 }
- fn executable_name() -> &'static str { "polkadot" }
+ fn executable_name() -> String { "polkadot".into() }
fn load_spec(&self, id: &str) -> std::result::Result, String> {
let id = if id == "" {
@@ -54,27 +53,47 @@ impl SubstrateCli for Cli {
.unwrap_or("polkadot")
} else { id };
Ok(match id {
- "polkadot-dev" | "dev" => Box::new(service::chain_spec::polkadot_development_config()),
- "polkadot-local" => Box::new(service::chain_spec::polkadot_local_testnet_config()),
- "polkadot-staging" => Box::new(service::chain_spec::polkadot_staging_testnet_config()),
- "kusama-dev" => Box::new(service::chain_spec::kusama_development_config()),
- "kusama-local" => Box::new(service::chain_spec::kusama_local_testnet_config()),
- "kusama-staging" => Box::new(service::chain_spec::kusama_staging_testnet_config()),
+ "polkadot-dev" | "dev" => Box::new(service::chain_spec::polkadot_development_config()?),
+ "polkadot-local" => Box::new(service::chain_spec::polkadot_local_testnet_config()?),
+ "polkadot-staging" => Box::new(service::chain_spec::polkadot_staging_testnet_config()?),
+ "kusama-dev" => Box::new(service::chain_spec::kusama_development_config()?),
+ "kusama-local" => Box::new(service::chain_spec::kusama_local_testnet_config()?),
+ "kusama-staging" => Box::new(service::chain_spec::kusama_staging_testnet_config()?),
"polkadot" => Box::new(service::chain_spec::polkadot_config()?),
"westend" => Box::new(service::chain_spec::westend_config()?),
"kusama" => Box::new(service::chain_spec::kusama_config()?),
- "westend-dev" => Box::new(service::chain_spec::westend_development_config()),
- "westend-local" => Box::new(service::chain_spec::westend_local_testnet_config()),
- "westend-staging" => Box::new(service::chain_spec::westend_staging_testnet_config()),
- path if self.run.force_kusama => {
- Box::new(service::KusamaChainSpec::from_json_file(std::path::PathBuf::from(path))?)
+ "westend-dev" => Box::new(service::chain_spec::westend_development_config()?),
+ "westend-local" => Box::new(service::chain_spec::westend_local_testnet_config()?),
+ "westend-staging" => Box::new(service::chain_spec::westend_staging_testnet_config()?),
+ path => {
+ let path = std::path::PathBuf::from(path);
+
+ let starts_with = |prefix: &str| {
+ path.file_name().map(|f| f.to_str().map(|s| s.starts_with(&prefix))).flatten().unwrap_or(false)
+ };
+
+ // When `force_*` is given or the file name starts with the name of one of the known chains,
+ // we use the chain spec for the specific chain.
+ if self.run.force_kusama || starts_with("kusama") {
+ Box::new(service::KusamaChainSpec::from_json_file(path)?)
+ } else if self.run.force_westend || starts_with("westend") {
+ Box::new(service::WestendChainSpec::from_json_file(path)?)
+ } else {
+ Box::new(service::PolkadotChainSpec::from_json_file(path)?)
+ }
},
- path if self.run.force_westend => {
- Box::new(service::WestendChainSpec::from_json_file(std::path::PathBuf::from(path))?)
- },
- path => Box::new(service::PolkadotChainSpec::from_json_file(std::path::PathBuf::from(path))?),
})
}
+
+ fn native_runtime_version(spec: &Box) -> &'static RuntimeVersion {
+ if spec.is_kusama() {
+ &service::kusama_runtime::VERSION
+ } else if spec.is_westend() {
+ &service::westend_runtime::VERSION
+ } else {
+ &service::polkadot_runtime::VERSION
+ }
+ }
}
/// Parses polkadot specific CLI arguments and run the service.
@@ -97,8 +116,8 @@ pub fn run() -> Result<()> {
match &cli.subcommand {
None => {
- let runtime = cli.create_runner(&cli.run.base)?;
- let chain_spec = &runtime.config().chain_spec;
+ let runner = cli.create_runner(&cli.run.base)?;
+ let chain_spec = &runner.config().chain_spec;
set_default_ss58_version(chain_spec);
@@ -115,87 +134,47 @@ pub fn run() -> Result<()> {
info!(" endorsed by the ");
info!(" KUSAMA FOUNDATION ");
info!("----------------------------");
-
- runtime.run_node(
- |config| {
- service::kusama_new_light(config)
- },
- |config| {
- service::kusama_new_full(
- config,
- None,
- None,
- authority_discovery_enabled,
- 6000,
- grandpa_pause,
- ).map(|(s, _, _)| s)
- },
- service::KusamaExecutor::native_version().runtime_version
- )
- } else if chain_spec.is_westend() {
- runtime.run_node(
- |config| {
- service::westend_new_light(config)
- },
- |config| {
- service::westend_new_full(
- config,
- None,
- None,
- authority_discovery_enabled,
- 6000,
- grandpa_pause,
- ).map(|(s, _, _)| s)
- },
- service::WestendExecutor::native_version().runtime_version
- )
- } else {
- runtime.run_node(
- |config| {
- service::polkadot_new_light(config)
- },
- |config| {
- service::polkadot_new_full(
- config,
- None,
- None,
- authority_discovery_enabled,
- 6000,
- grandpa_pause,
- ).map(|(s, _, _)| s)
- },
- service::PolkadotExecutor::native_version().runtime_version
- )
}
+
+ runner.run_node_until_exit(|config| {
+ let role = config.role.clone();
+
+ match role {
+ Role::Light => service::build_light(config).map(|(task_manager, _)| task_manager),
+ _ => service::build_full(
+ config,
+ None,
+ authority_discovery_enabled,
+ grandpa_pause,
+ ).map(|r| r.0),
+ }
+ })
},
Some(Subcommand::Base(subcommand)) => {
- let runtime = cli.create_runner(subcommand)?;
- let chain_spec = &runtime.config().chain_spec;
+ let runner = cli.create_runner(subcommand)?;
+ let chain_spec = &runner.config().chain_spec;
set_default_ss58_version(chain_spec);
if chain_spec.is_kusama() {
- runtime.run_subcommand(subcommand, |config|
+ runner.run_subcommand(subcommand, |config|
service::new_chain_ops::<
service::kusama_runtime::RuntimeApi,
service::KusamaExecutor,
- service::kusama_runtime::UncheckedExtrinsic,
>(config)
)
} else if chain_spec.is_westend() {
- runtime.run_subcommand(subcommand, |config|
+ runner.run_subcommand(subcommand, |config|
service::new_chain_ops::<
service::westend_runtime::RuntimeApi,
service::WestendExecutor,
- service::westend_runtime::UncheckedExtrinsic,
>(config)
)
} else {
- runtime.run_subcommand(subcommand, |config|
+ runner.run_subcommand(subcommand, |config|
service::new_chain_ops::<
service::polkadot_runtime::RuntimeApi,
service::PolkadotExecutor,
- service::polkadot_runtime::UncheckedExtrinsic,
>(config)
)
}
@@ -212,21 +191,21 @@ pub fn run() -> Result<()> {
}
},
Some(Subcommand::Benchmark(cmd)) => {
- let runtime = cli.create_runner(cmd)?;
- let chain_spec = &runtime.config().chain_spec;
+ let runner = cli.create_runner(cmd)?;
+ let chain_spec = &runner.config().chain_spec;
set_default_ss58_version(chain_spec);
if chain_spec.is_kusama() {
- runtime.sync_run(|config| {
+ runner.sync_run(|config| {
cmd.run::(config)
})
} else if chain_spec.is_westend() {
- runtime.sync_run(|config| {
+ runner.sync_run(|config| {
cmd.run::(config)
})
} else {
- runtime.sync_run(|config| {
+ runner.sync_run(|config| {
cmd.run::(config)
})
}
diff --git a/cli/src/lib.rs b/cli/src/lib.rs
index be2f3c6cd646444761e813d005ace4282ef52e19..385a24d364c8582065827054dcd6d52b88da26d8 100644
--- a/cli/src/lib.rs
+++ b/cli/src/lib.rs
@@ -28,14 +28,14 @@ mod command;
#[cfg(not(feature = "service-rewr"))]
pub use service::{
- AbstractService, ProvideRuntimeApi, CoreApi, ParachainHost, IdentifyVariant,
+ ProvideRuntimeApi, CoreApi, ParachainHost, IdentifyVariant,
Block, self, RuntimeApiCollection, TFullClient
};
#[cfg(feature = "service-rewr")]
pub use service_new::{
self as service,
- AbstractService, ProvideRuntimeApi, CoreApi, ParachainHost, IdentifyVariant,
+ ProvideRuntimeApi, CoreApi, ParachainHost, IdentifyVariant,
Block, self, RuntimeApiCollection, TFullClient
};
diff --git a/collator/Cargo.toml b/collator/Cargo.toml
deleted file mode 100644
index 2c47f20dbb8f96a2fa1fb50f2b0331011633f225..0000000000000000000000000000000000000000
--- a/collator/Cargo.toml
+++ /dev/null
@@ -1,37 +0,0 @@
-[package]
-name = "polkadot-collator"
-version = "0.8.12"
-authors = ["Parity Technologies "]
-description = "Collator node implementation"
-edition = "2018"
-
-[dependencies]
-futures = "0.3.4"
-sc-service = { git = "https://github.com/paritytech/substrate", branch = "master" }
-sc-executor = { git = "https://github.com/paritytech/substrate", branch = "master" }
-sc-cli = { git = "https://github.com/paritytech/substrate", branch = "master" }
-sc-client-api = { git = "https://github.com/paritytech/substrate", branch = "master" }
-sc-network = { git = "https://github.com/paritytech/substrate", branch = "master" }
-sp-blockchain = { git = "https://github.com/paritytech/substrate", branch = "master" }
-sp-api = { git = "https://github.com/paritytech/substrate", branch = "master" }
-sp-core = { git = "https://github.com/paritytech/substrate", branch = "master" }
-sp-runtime = { git = "https://github.com/paritytech/substrate", branch = "master" }
-consensus_common = { package = "sp-consensus", git = "https://github.com/paritytech/substrate", branch = "master" }
-polkadot-primitives = { path = "../primitives" }
-polkadot-cli = { path = "../cli" }
-polkadot-network = { path = "../network" }
-polkadot-validation = { path = "../validation" }
-polkadot-service = { path = "../service", optional = true}
-polkadot-service-new = { path = "../node/service", optional = true }
-log = "0.4.8"
-tokio = "0.2.13"
-futures-timer = "2.0"
-codec = { package = "parity-scale-codec", version = "1.3.0" }
-
-[dev-dependencies]
-keyring = { package = "sp-keyring", git = "https://github.com/paritytech/substrate", branch = "master" }
-
-[features]
-default = ["service-old"]
-service-old = [ "polkadot-service" ]
-service-rewr = [ "polkadot-service-new" ]
diff --git a/collator/README.adoc b/collator/README.adoc
deleted file mode 100644
index d302cd2af0fe4941a6e24c136bb0df9b2be07f35..0000000000000000000000000000000000000000
--- a/collator/README.adoc
+++ /dev/null
@@ -1,5 +0,0 @@
-
-= Polkadot Collator
-
-placeholder
-//TODO Write content :) (https://github.com/paritytech/polkadot/issues/159)
diff --git a/collator/src/lib.rs b/collator/src/lib.rs
deleted file mode 100644
index 063bdb1735c8acfb5037192dd0b98453eea077e2..0000000000000000000000000000000000000000
--- a/collator/src/lib.rs
+++ /dev/null
@@ -1,513 +0,0 @@
-// Copyright 2017-2020 Parity Technologies (UK) Ltd.
-// This file is part of Polkadot.
-
-// Polkadot 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.
-
-// Polkadot 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 Polkadot. If not, see .
-
-//! Collation node logic.
-//!
-//! A collator node lives on a distinct parachain and submits a proposal for
-//! a state transition, along with a proof for its validity
-//! (what we might call a witness or block data).
-//!
-//! One of collators' other roles is to route messages between chains.
-//! Each parachain produces a list of "egress" posts of messages for each other
-//! parachain on each block, for a total of N^2 lists all together.
-//!
-//! We will refer to the egress list at relay chain block X of parachain A with
-//! destination B as egress(X)[A -> B]
-//!
-//! On every block, each parachain will be intended to route messages from some
-//! subset of all the other parachains. (NOTE: in practice this is not done until PoC-3)
-//!
-//! Since the egress information is unique to every block, when routing from a
-//! parachain a collator must gather all egress posts from that parachain
-//! up to the last point in history that messages were successfully routed
-//! from that parachain, accounting for relay chain blocks where no candidate
-//! from the collator's parachain was produced.
-//!
-//! In the case that all parachains route to each other and a candidate for the
-//! collator's parachain was included in the last relay chain block, the collator
-//! only has to gather egress posts from other parachains one block back in relay
-//! chain history.
-//!
-//! This crate defines traits which provide context necessary for collation logic
-//! to be performed, as the collation logic itself.
-
-use std::collections::HashSet;
-use std::fmt;
-use std::sync::Arc;
-use std::time::Duration;
-use std::pin::Pin;
-
-use futures::{future, Future, Stream, FutureExt, StreamExt, task::Spawn};
-use log::warn;
-use sc_client_api::{StateBackend, BlockchainEvents};
-use sp_blockchain::HeaderBackend;
-use sp_core::Pair;
-use polkadot_primitives::{
- BlockId, Hash, Block,
- parachain::{
- self, BlockData, DutyRoster, HeadData, Id as ParaId,
- PoVBlock, ValidatorId, CollatorPair, LocalValidationData, GlobalValidationSchedule,
- }
-};
-use polkadot_cli::{
- ProvideRuntimeApi, AbstractService, ParachainHost, IdentifyVariant,
- service::{self, Role}
-};
-pub use polkadot_cli::service::Configuration;
-pub use polkadot_cli::Cli;
-pub use polkadot_validation::SignedStatement;
-pub use polkadot_primitives::parachain::CollatorId;
-pub use sc_network::PeerId;
-pub use service::RuntimeApiCollection;
-pub use sc_cli::SubstrateCli;
-use sp_api::{ConstructRuntimeApi, ApiExt, HashFor};
-#[cfg(not(feature = "service-rewr"))]
-use polkadot_service::{FullNodeHandles, PolkadotClient};
-#[cfg(feature = "service-rewr")]
-use polkadot_service_new::{
- self as polkadot_service,
- Error as ServiceError, FullNodeHandles, PolkadotClient,
-};
-
-const COLLATION_TIMEOUT: Duration = Duration::from_secs(30);
-
-/// An abstraction over the `Network` with useful functions for a `Collator`.
-pub trait Network: Send + Sync {
- /// Create a `Stream` of checked statements for the given `relay_parent`.
- ///
- /// The returned stream will not terminate, so it is required to make sure that the stream is
- /// dropped when it is not required anymore. Otherwise, it will stick around in memory
- /// infinitely.
- fn checked_statements(&self, relay_parent: Hash) -> Pin + Send>>;
-}
-
-impl Network for polkadot_network::protocol::Service {
- fn checked_statements(&self, relay_parent: Hash) -> Pin + Send>> {
- polkadot_network::protocol::Service::checked_statements(self, relay_parent).boxed()
- }
-}
-
-/// Collation errors.
-#[derive(Debug)]
-pub enum Error {
- /// Error on the relay-chain side of things.
- Polkadot(String),
-}
-
-impl fmt::Display for Error {
- fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
- match *self {
- Error::Polkadot(ref err) => write!(f, "Polkadot node error: {}", err),
- }
- }
-}
-
-/// Something that can build a `ParachainContext`.
-pub trait BuildParachainContext {
- /// The parachain context produced by the `build` function.
- type ParachainContext: self::ParachainContext;
-
- /// Build the `ParachainContext`.
- fn build(
- self,
- client: Arc,
- spawner: SP,
- network: impl Network + Clone + 'static,
- ) -> Result
- where
- Client: ProvideRuntimeApi + HeaderBackend + BlockchainEvents + Send + Sync + 'static,
- Client::Api: RuntimeApiCollection,
- >::StateBackend: StateBackend>,
- Extrinsic: codec::Codec + Send + Sync + 'static,
- SP: Spawn + Clone + Send + Sync + 'static;
-}
-
-/// Parachain context needed for collation.
-///
-/// This can be implemented through an externally attached service or a stub.
-/// This is expected to be a lightweight, shared type like an Arc.
-pub trait ParachainContext: Clone {
- type ProduceCandidate: Future