diff --git a/substrate/Cargo.lock b/substrate/Cargo.lock index 30d75077d1..383e2cdce3 100644 --- a/substrate/Cargo.lock +++ b/substrate/Cargo.lock @@ -6487,6 +6487,7 @@ dependencies = [ "derive_more", "futures 0.3.4", "futures-diagnose", + "futures-timer 2.0.2", "log 0.4.8", "parity-scale-codec", "parity-util-mem", diff --git a/substrate/bin/node-template/node/src/service.rs b/substrate/bin/node-template/node/src/service.rs index cf5cb361fc..5466f3e919 100644 --- a/substrate/bin/node-template/node/src/service.rs +++ b/substrate/bin/node-template/node/src/service.rs @@ -41,8 +41,7 @@ macro_rules! new_full_start { })? .with_transaction_pool(|config, client, _fetcher| { let pool_api = sc_transaction_pool::FullChainApi::new(client.clone()); - let pool = sc_transaction_pool::BasicPool::new(config, std::sync::Arc::new(pool_api)); - Ok(pool) + Ok(sc_transaction_pool::BasicPool::new(config, std::sync::Arc::new(pool_api))) })? .with_import_queue(|_config, client, mut select_chain, transaction_pool| { let select_chain = select_chain.take() diff --git a/substrate/bin/node/cli/src/service.rs b/substrate/bin/node/cli/src/service.rs index dd2684d50e..ff53b9aa3a 100644 --- a/substrate/bin/node/cli/src/service.rs +++ b/substrate/bin/node/cli/src/service.rs @@ -63,8 +63,7 @@ macro_rules! new_full_start { })? .with_transaction_pool(|config, client, _fetcher| { let pool_api = sc_transaction_pool::FullChainApi::new(client.clone()); - let pool = sc_transaction_pool::BasicPool::new(config, std::sync::Arc::new(pool_api)); - Ok(pool) + Ok(sc_transaction_pool::BasicPool::new(config, std::sync::Arc::new(pool_api))) })? .with_import_queue(|_config, client, mut select_chain, _transaction_pool| { let select_chain = select_chain.take() diff --git a/substrate/client/basic-authorship/src/basic_authorship.rs b/substrate/client/basic-authorship/src/basic_authorship.rs index a9e3d28e21..cab1231e87 100644 --- a/substrate/client/basic-authorship/src/basic_authorship.rs +++ b/substrate/client/basic-authorship/src/basic_authorship.rs @@ -308,7 +308,7 @@ mod tests { // given let client = Arc::new(substrate_test_runtime_client::new()); let txpool = Arc::new( - BasicPool::new(Default::default(), Arc::new(FullChainApi::new(client.clone()))) + BasicPool::new(Default::default(), Arc::new(FullChainApi::new(client.clone()))).0 ); futures::executor::block_on( @@ -350,7 +350,7 @@ mod tests { .build_with_backend(); let client = Arc::new(client); let txpool = Arc::new( - BasicPool::new(Default::default(), Arc::new(FullChainApi::new(client.clone()))) + BasicPool::new(Default::default(), Arc::new(FullChainApi::new(client.clone()))).0 ); let genesis_hash = client.info().best_hash; let block_id = BlockId::Hash(genesis_hash); diff --git a/substrate/client/basic-authorship/src/lib.rs b/substrate/client/basic-authorship/src/lib.rs index 9f2bc6a761..e9087c89e0 100644 --- a/substrate/client/basic-authorship/src/lib.rs +++ b/substrate/client/basic-authorship/src/lib.rs @@ -26,7 +26,7 @@ //! # use substrate_test_runtime_client::{self, runtime::{Extrinsic, Transfer}, AccountKeyring}; //! # use sc_transaction_pool::{BasicPool, FullChainApi}; //! # let client = Arc::new(substrate_test_runtime_client::new()); -//! # let txpool = Arc::new(BasicPool::new(Default::default(), Arc::new(FullChainApi::new(client.clone())))); +//! # let txpool = Arc::new(BasicPool::new(Default::default(), Arc::new(FullChainApi::new(client.clone()))).0); //! // The first step is to create a `ProposerFactory`. //! let mut proposer_factory = ProposerFactory { //! client: client.clone(), diff --git a/substrate/client/consensus/manual-seal/src/lib.rs b/substrate/client/consensus/manual-seal/src/lib.rs index a334e7c72f..c4336485a1 100644 --- a/substrate/client/consensus/manual-seal/src/lib.rs +++ b/substrate/client/consensus/manual-seal/src/lib.rs @@ -243,7 +243,7 @@ mod tests { let client = Arc::new(builder.build()); let select_chain = LongestChain::new(backend.clone()); let inherent_data_providers = InherentDataProviders::new(); - let pool = Arc::new(BasicPool::new(Options::default(), api())); + let pool = Arc::new(BasicPool::new(Options::default(), api()).0); let env = ProposerFactory { transaction_pool: pool.clone(), client: client.clone(), @@ -308,7 +308,7 @@ mod tests { let client = Arc::new(builder.build()); let select_chain = LongestChain::new(backend.clone()); let inherent_data_providers = InherentDataProviders::new(); - let pool = Arc::new(BasicPool::new(Options::default(), api())); + let pool = Arc::new(BasicPool::new(Options::default(), api()).0); let env = ProposerFactory { transaction_pool: pool.clone(), client: client.clone(), @@ -377,7 +377,7 @@ mod tests { let select_chain = LongestChain::new(backend.clone()); let inherent_data_providers = InherentDataProviders::new(); let pool_api = api(); - let pool = Arc::new(BasicPool::new(Options::default(), pool_api.clone())); + let pool = Arc::new(BasicPool::new(Options::default(), pool_api.clone()).0); let env = ProposerFactory { transaction_pool: pool.clone(), client: client.clone(), diff --git a/substrate/client/offchain/src/lib.rs b/substrate/client/offchain/src/lib.rs index fadfaa0134..ac8cc14861 100644 --- a/substrate/client/offchain/src/lib.rs +++ b/substrate/client/offchain/src/lib.rs @@ -204,7 +204,7 @@ mod tests { let pool = Arc::new(TestPool(BasicPool::new( Default::default(), Arc::new(FullChainApi::new(client.clone())), - ))); + ).0)); client.execution_extensions() .register_transaction_pool(Arc::downgrade(&pool.clone()) as _); let db = sc_client_db::offchain::LocalStorage::new_test(); diff --git a/substrate/client/rpc/src/author/tests.rs b/substrate/client/rpc/src/author/tests.rs index ba9b9d344c..41bfc46d38 100644 --- a/substrate/client/rpc/src/author/tests.rs +++ b/substrate/client/rpc/src/author/tests.rs @@ -64,7 +64,7 @@ impl Default for TestSetup { let pool = Arc::new(BasicPool::new( Default::default(), Arc::new(FullChainApi::new(client.clone())), - )); + ).0); TestSetup { runtime: runtime::Runtime::new().expect("Failed to create runtime in test setup"), client, diff --git a/substrate/client/service/src/builder.rs b/substrate/client/service/src/builder.rs index 5ca39856dc..2ab212646a 100644 --- a/substrate/client/service/src/builder.rs +++ b/substrate/client/service/src/builder.rs @@ -55,6 +55,8 @@ use sp_transaction_pool::{MaintainedTransactionPool, ChainEvent}; use sp_blockchain; use grafana_data_source::{self, record_metrics}; +pub type BackgroundTask = Pin + Send>>; + /// Aggregator for the components required to build a service. /// /// # Usage @@ -90,6 +92,7 @@ pub struct ServiceBuilder>>, marker: PhantomData<(TBl, TRtApi)>, + background_tasks: Vec<(&'static str, BackgroundTask)>, } /// Full client type. @@ -265,6 +268,7 @@ where TGen: RuntimeGenesis, TCSExt: Extension { transaction_pool: Arc::new(()), rpc_extensions: Default::default(), remote_backend: None, + background_tasks: Default::default(), marker: PhantomData, }) } @@ -350,6 +354,7 @@ where TGen: RuntimeGenesis, TCSExt: Extension { transaction_pool: Arc::new(()), rpc_extensions: Default::default(), remote_backend: Some(remote_blockchain), + background_tasks: Default::default(), marker: PhantomData, }) } @@ -398,6 +403,7 @@ impl( - self, + mut self, transaction_pool_builder: impl FnOnce( sc_transaction_pool::txpool::Options, Arc, Option, - ) -> Result + ) -> Result<(UExPool, Option), Error> ) -> Result, Error> where TSc: Clone, TFchr: Clone { - let transaction_pool = transaction_pool_builder( + let (transaction_pool, background_task) = transaction_pool_builder( self.config.transaction_pool.clone(), self.client.clone(), self.fetcher.clone(), )?; + if let Some(background_task) = background_task{ + self.background_tasks.push(("txpool-background", background_task)); + } + Ok(ServiceBuilder { config: self.config, client: self.client, @@ -625,6 +639,7 @@ impl None, }; + // Spawn background tasks which were stacked during the + // service building. + for (title, background_task) in background_tasks { + let _ = to_spawn_tx.unbounded_send(( + background_task, + title.into(), + )); + } + { // block notifications let txpool = Arc::downgrade(&transaction_pool); @@ -930,9 +956,10 @@ ServiceBuilder< ready(()) }); + let _ = to_spawn_tx.unbounded_send(( Box::pin(select(events, exit.clone()).map(drop)), - From::from("txpool-and-offchain-notif") + From::from("txpool-and-offchain-notif"), )); } @@ -955,7 +982,7 @@ ServiceBuilder< let _ = to_spawn_tx.unbounded_send(( Box::pin(select(events, exit.clone()).map(drop)), - From::from("telemetry-on-block") + From::from("telemetry-on-block"), )); } @@ -1028,7 +1055,7 @@ ServiceBuilder< let _ = to_spawn_tx.unbounded_send(( Box::pin(select(tel_task, exit.clone()).map(drop)), - From::from("telemetry-periodic-send") + From::from("telemetry-periodic-send"), )); // Periodically send the network state to the telemetry. @@ -1044,7 +1071,7 @@ ServiceBuilder< }); let _ = to_spawn_tx.unbounded_send(( Box::pin(select(tel_task_2, exit.clone()).map(drop)), - From::from("telemetry-periodic-network-state") + From::from("telemetry-periodic-network-state"), )); // RPC @@ -1130,7 +1157,7 @@ ServiceBuilder< system_rpc_rx, has_bootnodes, ), exit.clone()).map(drop)), - From::from("network-worker") + From::from("network-worker"), )); let telemetry_connection_sinks: Arc>>> = Default::default(); diff --git a/substrate/client/service/src/lib.rs b/substrate/client/service/src/lib.rs index 8c2f57dd7c..55fdba706b 100644 --- a/substrate/client/service/src/lib.rs +++ b/substrate/client/service/src/lib.rs @@ -716,7 +716,7 @@ mod tests { let pool = Arc::new(BasicPool::new( Default::default(), Arc::new(FullChainApi::new(client.clone())), - )); + ).0); let best = longest_chain.best_chain().unwrap(); let transaction = Transfer { amount: 5, diff --git a/substrate/client/transaction-pool/Cargo.toml b/substrate/client/transaction-pool/Cargo.toml index 27aec1b92e..7ccc98a9c0 100644 --- a/substrate/client/transaction-pool/Cargo.toml +++ b/substrate/client/transaction-pool/Cargo.toml @@ -20,6 +20,7 @@ sc-transaction-graph = { version = "2.0.0", path = "./graph" } sp-transaction-pool = { version = "2.0.0", path = "../../primitives/transaction-pool" } sc-client-api = { version = "2.0.0", path = "../api" } sp-blockchain = { version = "2.0.0", path = "../../primitives/blockchain" } +futures-timer = "2.0" parity-util-mem = { version = "0.5.1", default-features = false, features = ["primitive-types"] } [dev-dependencies] diff --git a/substrate/client/transaction-pool/graph/src/lib.rs b/substrate/client/transaction-pool/graph/src/lib.rs index 23970ba9b8..ed10ef38d2 100644 --- a/substrate/client/transaction-pool/graph/src/lib.rs +++ b/substrate/client/transaction-pool/graph/src/lib.rs @@ -39,4 +39,5 @@ pub use self::pool::{ Pool, Options, ChainApi, EventStream, ExtrinsicFor, BlockHash, ExHash, NumberFor, TransactionFor, + ValidatedTransaction, }; diff --git a/substrate/client/transaction-pool/graph/src/pool.rs b/substrate/client/transaction-pool/graph/src/pool.rs index 392abdca48..103f556d0e 100644 --- a/substrate/client/transaction-pool/graph/src/pool.rs +++ b/substrate/client/transaction-pool/graph/src/pool.rs @@ -36,7 +36,8 @@ use sp_runtime::{ use sp_transaction_pool::error; use wasm_timer::Instant; -use crate::validated_pool::{ValidatedPool, ValidatedTransaction}; +use crate::validated_pool::ValidatedPool; +pub use crate::validated_pool::ValidatedTransaction; /// Modification notification event stream type; pub type EventStream = mpsc::UnboundedReceiver; @@ -182,39 +183,19 @@ impl Pool { self.validated_pool.submit_and_watch(tx) } - /// Revalidate all ready transactions. - /// - /// Returns future that performs validation of all ready transactions and - /// then resubmits all transactions back to the pool. - pub async fn revalidate_ready( + /// Resubmit some transaction that were validated elsewhere. + pub fn resubmit( &self, - at: &BlockId, - max: Option, - ) -> Result<(), B::Error> { - log::debug!(target: "txpool", - "Fetching ready transactions (up to: {})", - max.map(|x| format!("{}", x)).unwrap_or_else(|| "all".into()) - ); - let validated_pool = self.validated_pool.clone(); - let ready = self.validated_pool.ready() - .map(|tx| tx.data.clone()) - .take(max.unwrap_or_else(usize::max_value)); - - let now = Instant::now(); - let revalidated_transactions = self.verify(at, ready, false).await?; - log::debug!(target: "txpool", - "Re-verified transactions, took {} ms. Resubmitting.", - now.elapsed().as_millis() - ); + revalidated_transactions: HashMap, ValidatedTransactionFor>, + ) { let now = Instant::now(); self.validated_pool.resubmit(revalidated_transactions); log::debug!(target: "txpool", "Resubmitted. Took {} ms. Status: {:?}", now.elapsed().as_millis(), - validated_pool.status() + self.validated_pool.status() ); - Ok(()) } /// Prunes known ready transactions. @@ -402,21 +383,15 @@ impl Pool { if validity.provides.is_empty() { ValidatedTransaction::Invalid(hash.clone(), error::Error::NoTagsProvided.into()) } else { - ValidatedTransaction::Valid(base::Transaction { - data: xt, + ValidatedTransaction::valid_at( + block_number.saturated_into::(), + hash.clone(), + xt, bytes, - hash: hash.clone(), - priority: validity.priority, - requires: validity.requires, - provides: validity.provides, - propagate: validity.propagate, - valid_till: block_number - .saturated_into::() - .saturating_add(validity.longevity), - }) + validity, + ) } }, - Err(TransactionValidityError::Invalid(e)) => ValidatedTransaction::Invalid(hash.clone(), error::Error::InvalidTransaction(e).into()), Err(TransactionValidityError::Unknown(e)) => @@ -988,80 +963,4 @@ mod tests { assert_eq!(pool.validated_pool().status().future, 0); } } - - #[test] - fn should_revalidate_ready_transactions() { - fn transfer(nonce: u64) -> Extrinsic { - uxt(Transfer { - from: AccountId::from_h256(H256::from_low_u64_be(1)), - to: AccountId::from_h256(H256::from_low_u64_be(2)), - amount: 5, - nonce, - }) - } - - // given - let pool = pool(); - let tx0 = transfer(0); - let hash0 = pool.validated_pool.api().hash_and_length(&tx0).0; - let watcher0 = block_on(pool.submit_and_watch(&BlockId::Number(0), tx0)).unwrap(); - let tx1 = transfer(1); - let hash1 = pool.validated_pool.api().hash_and_length(&tx1).0; - let watcher1 = block_on(pool.submit_and_watch(&BlockId::Number(0), tx1)).unwrap(); - let tx2 = transfer(2); - let hash2 = pool.validated_pool.api().hash_and_length(&tx2).0; - let watcher2 = block_on(pool.submit_and_watch(&BlockId::Number(0), tx2)).unwrap(); - let tx3 = transfer(3); - let hash3 = pool.validated_pool.api().hash_and_length(&tx3).0; - let watcher3 = block_on(pool.submit_and_watch(&BlockId::Number(0), tx3)).unwrap(); - let tx4 = transfer(4); - let hash4 = pool.validated_pool.api().hash_and_length(&tx4).0; - let watcher4 = block_on(pool.submit_and_watch(&BlockId::Number(0), tx4)).unwrap(); - assert_eq!(pool.validated_pool().status().ready, 5); - - // when - pool.validated_pool.api().invalidate.lock().insert(hash3); - pool.validated_pool.api().clear_requirements.lock().insert(hash1); - pool.validated_pool.api().add_requirements.lock().insert(hash0); - block_on(pool.revalidate_ready(&BlockId::Number(0), None)).unwrap(); - - // then - // hash0 now has unsatisfied requirements => it is moved to the future queue - // hash1 is now independent of hash0 => it is in ready queue - // hash2 still depends on hash1 => it is in ready queue - // hash3 is now invalid => it is removed from the pool - // hash4 now depends on invalidated hash3 => it is moved to the future queue - // - // events for hash3 are: Ready, Invalid - // events for hash4 are: Ready, Invalid - assert_eq!(pool.validated_pool().status().ready, 2); - assert_eq!( - futures::executor::block_on_stream(watcher3.into_stream()).collect::>(), - vec![TransactionStatus::Ready, TransactionStatus::Invalid], - ); - - // when - pool.validated_pool.remove_invalid(&[hash0, hash1, hash2, hash4]); - - // then - // events for hash0 are: Ready, Future, Invalid - // events for hash1 are: Ready, Invalid - // events for hash2 are: Ready, Invalid - assert_eq!( - futures::executor::block_on_stream(watcher0.into_stream()).collect::>(), - vec![TransactionStatus::Ready, TransactionStatus::Future, TransactionStatus::Invalid], - ); - assert_eq!( - futures::executor::block_on_stream(watcher1.into_stream()).collect::>(), - vec![TransactionStatus::Ready, TransactionStatus::Invalid], - ); - assert_eq!( - futures::executor::block_on_stream(watcher2.into_stream()).collect::>(), - vec![TransactionStatus::Ready, TransactionStatus::Invalid], - ); - assert_eq!( - futures::executor::block_on_stream(watcher4.into_stream()).collect::>(), - vec![TransactionStatus::Ready, TransactionStatus::Future, TransactionStatus::Invalid], - ); - } } diff --git a/substrate/client/transaction-pool/graph/src/validated_pool.rs b/substrate/client/transaction-pool/graph/src/validated_pool.rs index 8752414637..a62822a918 100644 --- a/substrate/client/transaction-pool/graph/src/validated_pool.rs +++ b/substrate/client/transaction-pool/graph/src/validated_pool.rs @@ -32,7 +32,7 @@ use parking_lot::{Mutex, RwLock}; use sp_runtime::{ generic::BlockId, traits::{self, SaturatedConversion}, - transaction_validity::TransactionTag as Tag, + transaction_validity::{TransactionTag as Tag, ValidTransaction}, }; use sp_transaction_pool::{error, PoolStatus}; use wasm_timer::Instant; @@ -53,6 +53,30 @@ pub enum ValidatedTransaction { Unknown(Hash, Error), } +impl ValidatedTransaction { + /// Consume validity result, transaction data and produce ValidTransaction. + pub fn valid_at( + at: u64, + hash: Hash, + data: Ex, + bytes: usize, + validity: ValidTransaction, + ) -> Self { + Self::Valid(base::Transaction { + data, + bytes, + hash, + priority: validity.priority, + requires: validity.requires, + provides: validity.provides, + propagate: validity.propagate, + valid_till: at + .saturated_into::() + .saturating_add(validity.longevity), + }) + } +} + /// A type of validated transaction stored in the pool. pub type ValidatedTransactionFor = ValidatedTransaction< ExHash, diff --git a/substrate/client/transaction-pool/src/lib.rs b/substrate/client/transaction-pool/src/lib.rs index fbfc6a24e6..a69323553f 100644 --- a/substrate/client/transaction-pool/src/lib.rs +++ b/substrate/client/transaction-pool/src/lib.rs @@ -21,6 +21,7 @@ mod api; pub mod error; +mod revalidation; #[cfg(any(feature = "test-helpers", test))] pub mod testing; @@ -51,6 +52,7 @@ pub struct BasicPool pool: Arc>, api: Arc, revalidation_strategy: Arc>>>, + revalidation_queue: Arc>, } #[cfg(not(target_os = "unknown"))] @@ -86,13 +88,16 @@ pub enum RevalidationType { impl BasicPool where Block: BlockT, - PoolApi: sc_transaction_graph::ChainApi, + PoolApi: sc_transaction_graph::ChainApi + 'static, { /// Create new basic transaction pool with provided api. + /// + /// It will also optionally return background task that might be started by the + /// caller. pub fn new( options: sc_transaction_graph::Options, pool_api: Arc, - ) -> Self { + ) -> (Self, Option + Send>>>) { Self::with_revalidation_type(options, pool_api, RevalidationType::Full) } @@ -102,18 +107,29 @@ impl BasicPool options: sc_transaction_graph::Options, pool_api: Arc, revalidation_type: RevalidationType, - ) -> Self { - let cloned_api = pool_api.clone(); - BasicPool { - api: cloned_api, - pool: Arc::new(sc_transaction_graph::Pool::new(options, pool_api)), - revalidation_strategy: Arc::new(Mutex::new( - match revalidation_type { - RevalidationType::Light => RevalidationStrategy::Light(RevalidationStatus::NotScheduled), - RevalidationType::Full => RevalidationStrategy::Always, - } - )), - } + ) -> (Self, Option + Send>>>) { + let pool = Arc::new(sc_transaction_graph::Pool::new(options, pool_api.clone())); + let (revalidation_queue, background_task) = match revalidation_type { + RevalidationType::Light => (revalidation::RevalidationQueue::new(pool_api.clone(), pool.clone()), None), + RevalidationType::Full => { + let (queue, background) = revalidation::RevalidationQueue::new_background(pool_api.clone(), pool.clone()); + (queue, Some(background)) + }, + }; + ( + BasicPool { + api: pool_api, + pool, + revalidation_queue: Arc::new(revalidation_queue), + revalidation_strategy: Arc::new(Mutex::new( + match revalidation_type { + RevalidationType::Light => RevalidationStrategy::Light(RevalidationStatus::NotScheduled), + RevalidationType::Full => RevalidationStrategy::Always, + } + )), + }, + background_task, + ) } /// Gets shared reference to the underlying pool. @@ -218,7 +234,6 @@ enum RevalidationStrategy { struct RevalidationAction { revalidate: bool, resubmit: bool, - revalidate_amount: Option, } impl RevalidationStrategy { @@ -242,12 +257,10 @@ impl RevalidationStrategy { revalidate_block_period, ), resubmit: false, - revalidate_amount: None, }, Self::Always => RevalidationAction { revalidate: true, resubmit: true, - revalidate_amount: Some(16), } } } @@ -314,6 +327,7 @@ impl MaintainedTransactionPool for BasicPool ); let revalidation_strategy = self.revalidation_strategy.clone(); let retracted = retracted.clone(); + let revalidation_queue = self.revalidation_queue.clone(); async move { // We don't query block if we won't prune anything @@ -360,9 +374,8 @@ impl MaintainedTransactionPool for BasicPool } if next_action.revalidate { - if let Err(e) = pool.revalidate_ready(&id, next_action.revalidate_amount).await { - log::warn!("Revalidate ready failed {:?}", e); - } + let hashes = pool.validated_pool().ready().map(|tx| tx.hash.clone()).collect(); + revalidation_queue.revalidate_later(block_number, hashes).await; } revalidation_strategy.lock().clear(); diff --git a/substrate/client/transaction-pool/src/revalidation.rs b/substrate/client/transaction-pool/src/revalidation.rs new file mode 100644 index 0000000000..dbf8a29354 --- /dev/null +++ b/substrate/client/transaction-pool/src/revalidation.rs @@ -0,0 +1,313 @@ +// Copyright 2018-2020 Parity Technologies (UK) Ltd. +// This file is part of Substrate. + +// Substrate 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. + +// Substrate 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 Substrate. If not, see . + +//! Pool periodic revalidation. + +use std::{sync::Arc, pin::Pin, collections::{HashMap, HashSet, BTreeMap}}; + +use sc_transaction_graph::{ChainApi, Pool, ExHash, NumberFor, ValidatedTransaction}; +use sp_runtime::traits::{Zero, SaturatedConversion}; +use sp_runtime::generic::BlockId; +use sp_runtime::transaction_validity::TransactionValidityError; + +use futures::{prelude::*, channel::mpsc, stream::unfold}; +use std::time::Duration; +use futures_timer::Delay; + +#[cfg(not(test))] +const BACKGROUND_REVALIDATION_INTERVAL: Duration = Duration::from_millis(200); +#[cfg(test)] +pub const BACKGROUND_REVALIDATION_INTERVAL: Duration = Duration::from_millis(5); + +const BACKGROUND_REVALIDATION_BATCH_SIZE: usize = 20; + +/// Payload from queue to worker. +struct WorkerPayload { + at: NumberFor, + transactions: Vec>, +} + +/// Async revalidation worker. +/// +/// Implements future and can be spawned in place or in background. +struct RevalidationWorker { + api: Arc, + pool: Arc>, + best_block: NumberFor, + block_ordered: BTreeMap, HashSet>>, + members: HashMap, NumberFor>, +} + +impl Unpin for RevalidationWorker {} + +fn interval(duration: Duration) -> impl Stream + Unpin { + unfold((), move |_| { + Delay::new(duration).map(|_| Some(((), ()))) + }).map(drop) +} + +/// Revalidate batch of transaction. +/// +/// Each transaction is validated against chain, and invalid are +/// removed from the `pool`, while valid are resubmitted. +async fn batch_revalidate( + pool: Arc>, + api: Arc, + at: NumberFor, + batch: impl IntoIterator>, +) { + let mut invalid_hashes = Vec::new(); + let mut revalidated = HashMap::new(); + + for ext_hash in batch { + let ext = match pool.validated_pool().ready_by_hash(&ext_hash) { + Some(ext) => ext, + None => continue, + }; + + match api.validate_transaction(&BlockId::Number(at), ext.data.clone()).await { + Ok(Err(TransactionValidityError::Invalid(err))) => { + log::debug!(target: "txpool", "[{:?}]: Revalidation: invalid {:?}", ext_hash, err); + invalid_hashes.push(ext_hash); + }, + Ok(Err(TransactionValidityError::Unknown(err))) => { + // skipping unknown, they might be pushed by valid or invalid transaction + // when latter resubmitted. + log::trace!(target: "txpool", "[{:?}]: Unknown during revalidation: {:?}", ext_hash, err); + }, + Ok(Ok(validity)) => { + revalidated.insert( + ext_hash.clone(), + ValidatedTransaction::valid_at( + at.saturated_into::(), + ext_hash, + ext.data.clone(), + api.hash_and_length(&ext.data).1, + validity, + ) + ); + }, + Err(validation_err) => { + log::debug!( + target: "txpool", + "[{:?}]: Error during revalidation: {:?}. Removing.", + ext_hash, + validation_err + ); + invalid_hashes.push(ext_hash); + } + } + } + + pool.validated_pool().remove_invalid(&invalid_hashes); + pool.resubmit(revalidated); +} + +impl RevalidationWorker { + fn new( + api: Arc, + pool: Arc>, + ) -> Self { + Self { + api, + pool, + block_ordered: Default::default(), + members: Default::default(), + best_block: Zero::zero(), + } + } + + fn prepare_batch(&mut self) -> Vec> { + let mut queued_exts = Vec::new(); + let mut left = BACKGROUND_REVALIDATION_BATCH_SIZE; + + // Take maximum of count transaction by order + // which they got into the pool + while left > 0 { + let first_block = match self.block_ordered.keys().next().cloned() { + Some(bn) => bn, + None => break, + }; + let mut block_drained = false; + if let Some(extrinsics) = self.block_ordered.get_mut(&first_block) { + let to_queue = extrinsics.iter().take(left).cloned().collect::>(); + if to_queue.len() == extrinsics.len() { + block_drained = true; + } else { + for xt in &to_queue { + extrinsics.remove(xt); + } + } + left -= to_queue.len(); + queued_exts.extend(to_queue); + } + + if block_drained { + self.block_ordered.remove(&first_block); + } + } + + queued_exts + } + + fn push(&mut self, worker_payload: WorkerPayload) { + // we don't add something that already scheduled for revalidation + let transactions = worker_payload.transactions; + let block_number = worker_payload.at; + + for ext_hash in transactions { + // we don't add something that already scheduled for revalidation + if self.members.contains_key(&ext_hash) { continue; } + + self.block_ordered.entry(block_number) + .and_modify(|value| { value.insert(ext_hash.clone()); }) + .or_insert_with(|| { + let mut bt = HashSet::new(); + bt.insert(ext_hash.clone()); + bt + }); + self.members.insert(ext_hash.clone(), block_number); + } + } + + /// Background worker main loop. + /// + /// It does two things: periodically tries to process some transactions + /// from the queue and also accepts messages to enqueue some more + /// transactions from the pool. + pub async fn run(mut self, from_queue: mpsc::UnboundedReceiver>) { + let interval = interval(BACKGROUND_REVALIDATION_INTERVAL).fuse(); + let from_queue = from_queue.fuse(); + futures::pin_mut!(interval, from_queue); + let this = &mut self; + + loop { + futures::select! { + _ = interval.next() => { + let next_batch = this.prepare_batch(); + batch_revalidate(this.pool.clone(), this.api.clone(), this.best_block, next_batch).await; + }, + workload = from_queue.next() => { + match workload { + Some(worker_payload) => { + this.best_block = worker_payload.at; + this.push(worker_payload); + continue; + }, + // R.I.P. worker! + None => break, + } + } + } + } + } +} + + +/// Revalidation queue. +/// +/// Can be configured background (`new_background`) +/// or immediate (just `new`). +pub struct RevalidationQueue { + pool: Arc>, + api: Arc, + background: Option>>, +} + +impl RevalidationQueue +where + Api: 'static, +{ + /// New revalidation queue without background worker. + pub fn new(api: Arc, pool: Arc>) -> Self { + Self { + api, + pool, + background: None, + } + } + + /// New revalidation queue with background worker. + pub fn new_background(api: Arc, pool: Arc>) -> + (Self, Pin + Send>>) + { + let (to_worker, from_queue) = mpsc::unbounded(); + + let worker = RevalidationWorker::new(api.clone(), pool.clone()); + + let queue = + Self { + api, + pool, + background: Some(to_worker), + }; + + (queue, worker.run(from_queue).boxed()) + } + + /// Queue some transaction for later revalidation. + /// + /// If queue configured with background worker, this will return immediately. + /// If queue configured without background worker, this will resolve after + /// revalidation is actually done. + pub async fn revalidate_later(&self, at: NumberFor, transactions: Vec>) { + if let Some(ref to_worker) = self.background { + if let Err(e) = to_worker.unbounded_send(WorkerPayload { at, transactions }) { + log::warn!(target: "txpool", "Failed to update background worker: {:?}", e); + } + return; + } else { + let pool = self.pool.clone(); + let api = self.api.clone(); + batch_revalidate(pool, api, at, transactions).await + } + } +} + +#[cfg(test)] +mod tests { + + use super::*; + use sc_transaction_graph::Pool; + use substrate_test_runtime_transaction_pool::{TestApi, uxt}; + use futures::executor::block_on; + use substrate_test_runtime_client::{ + AccountKeyring::*, + }; + + fn setup() -> (Arc, Pool) { + let test_api = Arc::new(TestApi::empty()); + let pool = Pool::new(Default::default(), test_api.clone()); + (test_api, pool) + } + + #[test] + fn smoky() { + let (api, pool) = setup(); + let pool = Arc::new(pool); + let queue = Arc::new(RevalidationQueue::new(api.clone(), pool.clone())); + + let uxt = uxt(Alice, 0); + let uxt_hash = block_on(pool.submit_one(&BlockId::number(0), uxt.clone())).expect("Should be valid"); + + block_on(queue.revalidate_later(0, vec![uxt_hash])); + + // revalidated in sync offload 2nd time + assert_eq!(api.validation_requests().len(), 2); + // number of ready + assert_eq!(pool.validated_pool().status().ready, 1); + } +} \ No newline at end of file diff --git a/substrate/client/transaction-pool/src/testing/pool.rs b/substrate/client/transaction-pool/src/testing/pool.rs index 4a4f4638df..ed7c3e2d7a 100644 --- a/substrate/client/transaction-pool/src/testing/pool.rs +++ b/substrate/client/transaction-pool/src/testing/pool.rs @@ -15,6 +15,7 @@ // along with Substrate. If not, see . use crate::*; +use sp_transaction_pool::TransactionStatus; use futures::executor::block_on; use txpool::{self, Pool}; use sp_runtime::{ @@ -22,18 +23,22 @@ use sp_runtime::{ transaction_validity::ValidTransaction, }; use substrate_test_runtime_client::{ - runtime::{Block, Hash, Index, Header}, + runtime::{Block, Hash, Index, Header, Extrinsic}, AccountKeyring::*, }; use substrate_test_runtime_transaction_pool::{TestApi, uxt}; -use sp_transaction_pool::TransactionStatus; +use crate::revalidation::BACKGROUND_REVALIDATION_INTERVAL; fn pool() -> Pool { Pool::new(Default::default(), TestApi::with_alice_nonce(209).into()) } -fn maintained_pool() -> BasicPool { - BasicPool::new(Default::default(), std::sync::Arc::new(TestApi::with_alice_nonce(209))) +fn maintained_pool() -> (BasicPool, futures::executor::ThreadPool) { + let (pool, background_task) = BasicPool::new(Default::default(), std::sync::Arc::new(TestApi::with_alice_nonce(209))); + + let thread_pool = futures::executor::ThreadPool::new().unwrap(); + thread_pool.spawn_ok(background_task.expect("basic pool have background task")); + (pool, thread_pool) } fn header(number: u64) -> Header { @@ -158,25 +163,37 @@ fn should_correctly_prune_transactions_providing_more_than_one_tag() { assert_eq!(pool.validated_pool().status().future, 2); } +fn block_event(id: u64) -> ChainEvent { + ChainEvent::NewBlock { + id: BlockId::number(id), + is_new_best: true, + retracted: vec![], + header: header(id), + } +} + +fn block_event_with_retracted(id: u64, retracted: Vec) -> ChainEvent { + ChainEvent::NewBlock { + id: BlockId::number(id), + is_new_best: true, + retracted: retracted, + header: header(id), + } +} + + #[test] fn should_prune_old_during_maintenance() { let xt = uxt(Alice, 209); - let pool = maintained_pool(); + let (pool, _guard) = maintained_pool(); block_on(pool.submit_one(&BlockId::number(0), xt.clone())).expect("1. Imported"); assert_eq!(pool.status().ready, 1); pool.api.push_block(1, vec![xt.clone()]); - let event = ChainEvent::NewBlock { - id: BlockId::number(1), - is_new_best: true, - retracted: vec![], - header: header(1), - }; - - block_on(pool.maintain(event)); + block_on(pool.maintain(block_event(1))); assert_eq!(pool.status().ready, 0); } @@ -185,21 +202,20 @@ fn should_revalidate_during_maintenance() { let xt1 = uxt(Alice, 209); let xt2 = uxt(Alice, 210); - let pool = maintained_pool(); + let (pool, _guard) = maintained_pool(); block_on(pool.submit_one(&BlockId::number(0), xt1.clone())).expect("1. Imported"); block_on(pool.submit_one(&BlockId::number(0), xt2.clone())).expect("2. Imported"); assert_eq!(pool.status().ready, 2); assert_eq!(pool.api.validation_requests().len(), 2); pool.api.push_block(1, vec![xt1.clone()]); - let event = ChainEvent::NewBlock { - id: BlockId::number(1), - is_new_best: true, - retracted: vec![], - header: header(1), - }; - block_on(pool.maintain(event)); + block_on(pool.maintain(block_event(1))); + + // maintaince is in background + block_on(futures_timer::Delay::new(BACKGROUND_REVALIDATION_INTERVAL*2)); + + block_on(pool.maintain(block_event(1))); assert_eq!(pool.status().ready, 1); // test that pool revalidated transaction that left ready and not included in the block assert_eq!(pool.api.validation_requests().len(), 3); @@ -210,19 +226,15 @@ fn should_resubmit_from_retracted_during_maintaince() { let xt = uxt(Alice, 209); let retracted_hash = Hash::random(); - let pool = maintained_pool(); + let (pool, _guard) = maintained_pool(); block_on(pool.submit_one(&BlockId::number(0), xt.clone())).expect("1. Imported"); assert_eq!(pool.status().ready, 1); pool.api.push_block(1, vec![]); pool.api.push_fork_block(retracted_hash, vec![xt.clone()]); - let event = ChainEvent::NewBlock { - id: BlockId::Number(1), - is_new_best: true, - header: header(1), - retracted: vec![retracted_hash] - }; + + let event = block_event_with_retracted(1, vec![retracted_hash]); block_on(pool.maintain(event)); assert_eq!(pool.status().ready, 1); @@ -233,7 +245,7 @@ fn should_not_retain_invalid_hashes_from_retracted() { let xt = uxt(Alice, 209); let retracted_hash = Hash::random(); - let pool = maintained_pool(); + let (pool, _guard) = maintained_pool(); block_on(pool.submit_one(&BlockId::number(0), xt.clone())).expect("1. Imported"); assert_eq!(pool.status().ready, 1); @@ -242,20 +254,90 @@ fn should_not_retain_invalid_hashes_from_retracted() { pool.api.push_fork_block(retracted_hash, vec![xt.clone()]); pool.api.add_invalid(&xt); - let event = ChainEvent::NewBlock { - id: BlockId::Number(1), - is_new_best: true, - header: header(1), - retracted: vec![retracted_hash] - }; + let event = block_event_with_retracted(1, vec![retracted_hash]); + + block_on(pool.maintain(event)); + + // maintenance is in background + block_on(futures_timer::Delay::new(BACKGROUND_REVALIDATION_INTERVAL*2)); + + let event = block_event_with_retracted(1, vec![retracted_hash]); block_on(pool.maintain(event)); assert_eq!(pool.status().ready, 0); } +#[test] +fn should_push_watchers_during_maintaince() { + fn alice_uxt(nonce: u64) -> Extrinsic { + uxt(Alice, 209 + nonce) + } + + // given + let (pool, _guard) = maintained_pool(); + + let tx0 = alice_uxt(0); + let watcher0 = block_on(pool.submit_and_watch(&BlockId::Number(0), tx0.clone())).unwrap(); + let tx1 = alice_uxt(1); + let watcher1 = block_on(pool.submit_and_watch(&BlockId::Number(0), tx1.clone())).unwrap(); + let tx2 = alice_uxt(2); + let watcher2 = block_on(pool.submit_and_watch(&BlockId::Number(0), tx2.clone())).unwrap(); + let tx3 = alice_uxt(3); + let watcher3 = block_on(pool.submit_and_watch(&BlockId::Number(0), tx3.clone())).unwrap(); + let tx4 = alice_uxt(4); + let watcher4 = block_on(pool.submit_and_watch(&BlockId::Number(0), tx4.clone())).unwrap(); + assert_eq!(pool.status().ready, 5); + + // when + pool.api.add_invalid(&tx3); + pool.api.add_invalid(&tx4); + block_on(pool.maintain(block_event(0))); + + // revalidation is in background + block_on(futures_timer::Delay::new(BACKGROUND_REVALIDATION_INTERVAL*2)); + + // then + // hash3 is now invalid + // hash4 is now invalid + + assert_eq!(pool.status().ready, 3); + assert_eq!( + futures::executor::block_on_stream(watcher3).collect::>(), + vec![TransactionStatus::Ready, TransactionStatus::Invalid], + ); + assert_eq!( + futures::executor::block_on_stream(watcher4).collect::>(), + vec![TransactionStatus::Ready, TransactionStatus::Invalid], + ); + + // when + let header_hash = pool.api.push_block(1, vec![tx0, tx1, tx2]).hash(); + block_on(pool.maintain(block_event(1))); + + let event = ChainEvent::Finalized { hash: header_hash.clone() }; + block_on(pool.maintain(event)); + + // then + // events for hash0 are: Ready, InBlock + // events for hash1 are: Ready, InBlock + // events for hash2 are: Ready, InBlock + assert_eq!( + futures::executor::block_on_stream(watcher0).collect::>(), + vec![TransactionStatus::Ready, TransactionStatus::InBlock(header_hash.clone()), TransactionStatus::Finalized(header_hash.clone())], + ); + assert_eq!( + futures::executor::block_on_stream(watcher1).collect::>(), + vec![TransactionStatus::Ready, TransactionStatus::InBlock(header_hash.clone()), TransactionStatus::Finalized(header_hash.clone())], + ); + assert_eq!( + futures::executor::block_on_stream(watcher2).collect::>(), + vec![TransactionStatus::Ready, TransactionStatus::InBlock(header_hash.clone()), TransactionStatus::Finalized(header_hash.clone())], + ); +} + #[test] fn can_track_heap_size() { - let pool = maintained_pool(); + let (pool, _guard) = maintained_pool(); block_on(pool.submit_one(&BlockId::number(0), uxt(Alice, 209))).expect("1. Imported"); block_on(pool.submit_one(&BlockId::number(0), uxt(Alice, 210))).expect("1. Imported"); block_on(pool.submit_one(&BlockId::number(0), uxt(Alice, 211))).expect("1. Imported"); @@ -269,7 +351,7 @@ fn finalization() { let xt = uxt(Alice, 209); let api = TestApi::with_alice_nonce(209); api.push_block(1, vec![]); - let pool = BasicPool::new(Default::default(), api.into()); + let (pool, _background) = BasicPool::new(Default::default(), api.into()); let watcher = block_on(pool.submit_and_watch(&BlockId::number(1), xt.clone())).expect("1. Imported"); pool.api.push_block(2, vec![xt.clone()]); @@ -298,7 +380,7 @@ fn fork_aware_finalization() { // starting block A1 (last finalized.) api.push_block(1, vec![]); - let pool = BasicPool::new(Default::default(), api.into()); + let (pool, _background) = BasicPool::new(Default::default(), api.into()); let mut canon_watchers = vec![]; let from_alice = uxt(Alice, 1); diff --git a/substrate/utils/frame/rpc/system/src/lib.rs b/substrate/utils/frame/rpc/system/src/lib.rs index 9cc01fb6ec..46204082be 100644 --- a/substrate/utils/frame/rpc/system/src/lib.rs +++ b/substrate/utils/frame/rpc/system/src/lib.rs @@ -236,7 +236,7 @@ mod tests { let _ = env_logger::try_init(); let client = Arc::new(substrate_test_runtime_client::new()); let pool = Arc::new( - BasicPool::new(Default::default(), Arc::new(FullChainApi::new(client.clone()))) + BasicPool::new(Default::default(), Arc::new(FullChainApi::new(client.clone()))).0 ); let new_transaction = |nonce: u64| {