pezkuwichain/pezkuwi-subxt
1
0
Fork 0
mirror of https://github.com/pezkuwichain/pezkuwi-subxt.git synced 2026-04-28 13:17:56 +00:00
Code Issues Packages Projects Releases Wiki Activity

don't use delays in tests (#5404)

Browse Source
This commit is contained in:
Nikolay Volf
2020-03-29 01:40:00 -07:00
committed by GitHub
parent 1242a7df3c
commit d7ffef43ce
5 changed files with 106 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files
substrate/Cargo.lock
Generated
+11 -1
View File
@@ -2345,6 +2345,16 @@ version = "0.1.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "141340095b15ed7491bd3d4ced9d20cebfb826174b6bb03386381f62b01e3d77"
[[package]]
name = "intervalier"
version = "0.3.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "750dc2c10615a0aa0d38a5adf9d4e62651c178109f40253cb6235b3f638af6a9"
dependencies = [
"futures 0.3.4",
"futures-timer 2.0.2",
]
[[package]]
name = "iovec"
version = "0.1.4"
@@ -6690,8 +6700,8 @@ dependencies = [
"derive_more",
"futures 0.3.4",
"futures-diagnose",
"futures-timer 2.0.2",
"hex",
"intervalier",
"log 0.4.8",
"parity-scale-codec",
"parity-util-mem",
substrate/client/transaction-pool/Cargo.toml
+1 -1
View File
@@ -23,7 +23,7 @@ sc-transaction-graph = { version = "2.0.0-alpha.5", path = "./graph" }
sp-transaction-pool = { version = "2.0.0-alpha.5", path = "../../primitives/transaction-pool" }
sc-client-api = { version = "2.0.0-alpha.5", path = "../api" }
sp-blockchain = { version = "2.0.0-alpha.5", path = "../../primitives/blockchain" }
futures-timer = "2.0"
intervalier = "0.3"
parity-util-mem = { version = "0.6.0", default-features = false, features = ["primitive-types"] }
[dev-dependencies]
substrate/client/transaction-pool/src/lib.rs
+22 -1
View File
@@ -31,7 +31,7 @@ pub use sc_transaction_graph as txpool;
pub use crate::api::{FullChainApi, LightChainApi};
use std::{collections::HashMap, sync::Arc, pin::Pin};
use futures::{Future, FutureExt, future::ready, channel::oneshot};
use futures::{prelude::*, future::ready, channel::oneshot};
use parking_lot::Mutex;
use sp_runtime::{
@@ -151,6 +151,27 @@ impl<PoolApi, Block> BasicPool<PoolApi, Block>
Self::with_revalidation_type(options, pool_api, RevalidationType::Full)
}
/// Create new basic transaction pool with provided api, for tests.
#[cfg(test)]
pub fn new_test(
pool_api: Arc<PoolApi>,
) -> (Self, Pin<Box<dyn Future<Output=()> + Send>>, intervalier::BackSignalControl) {
let pool = Arc::new(sc_transaction_graph::Pool::new(Default::default(), pool_api.clone()));
let (revalidation_queue, background_task, notifier) =
revalidation::RevalidationQueue::new_test(pool_api.clone(), pool.clone());
(
BasicPool {
api: pool_api,
pool,
revalidation_queue: Arc::new(revalidation_queue),
revalidation_strategy: Arc::new(Mutex::new(RevalidationStrategy::Always)),
ready_poll: Default::default(),
},
background_task,
notifier,
)
}
/// Create new basic transaction pool with provided api and custom
/// revalidation type.
pub fn with_revalidation_type(
substrate/client/transaction-pool/src/revalidation.rs
+33 -14
View File
@@ -23,9 +23,8 @@ 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 futures::{prelude::*, channel::mpsc};
use std::time::Duration;
use futures_timer::Delay;
#[cfg(not(test))]
const BACKGROUND_REVALIDATION_INTERVAL: Duration = Duration::from_millis(200);
@@ -53,12 +52,6 @@ struct RevalidationWorker<Api: ChainApi> {
impl<Api: ChainApi> Unpin for RevalidationWorker<Api> {}
fn interval(duration: Duration) -> impl Stream<Item=()> + Unpin {
unfold((), move |_| {
Delay::new(duration).map(|_| Some(((), ())))
}).map(drop)
}
/// Revalidate batch of transaction.
///
/// Each transaction is validated against chain, and invalid are
@@ -207,8 +200,13 @@ impl<Api: ChainApi> RevalidationWorker<Api> {
/// 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<WorkerPayload<Api>>) {
let interval = interval(BACKGROUND_REVALIDATION_INTERVAL).fuse();
pub async fn run<R: intervalier::IntoStream>(
mut self,
from_queue: mpsc::UnboundedReceiver<WorkerPayload<Api>>,
interval: R,
) where R: Send, R::Guard: Send
{
let interval = interval.into_stream().fuse();
let from_queue = from_queue.fuse();
futures::pin_mut!(interval, from_queue);
let this = &mut self;
@@ -270,9 +268,12 @@ where
}
}
/// New revalidation queue with background worker.
pub fn new_background(api: Arc<Api>, pool: Arc<Pool<Api>>) ->
(Self, Pin<Box<dyn Future<Output=()> + Send>>)
pub fn new_with_interval<R: intervalier::IntoStream>(
api: Arc<Api>,
pool: Arc<Pool<Api>>,
interval: R,
) -> (Self, Pin<Box<dyn Future<Output=()> + Send>>)
where R: Send + 'static, R::Guard: Send
{
let (to_worker, from_queue) = mpsc::unbounded();
@@ -285,7 +286,25 @@ where
background: Some(to_worker),
};
(queue, worker.run(from_queue).boxed())
(queue, worker.run(from_queue, interval).boxed())
}
/// New revalidation queue with background worker.
pub fn new_background(api: Arc<Api>, pool: Arc<Pool<Api>>) ->
(Self, Pin<Box<dyn Future<Output=()> + Send>>)
{
Self::new_with_interval(api, pool, intervalier::Interval::new(BACKGROUND_REVALIDATION_INTERVAL))
}
/// New revalidation queue with background worker and test signal.
#[cfg(test)]
pub fn new_test(api: Arc<Api>, pool: Arc<Pool<Api>>) ->
(Self, Pin<Box<dyn Future<Output=()> + Send>>, intervalier::BackSignalControl)
{
let (interval, notifier) = intervalier::BackSignalInterval::new(BACKGROUND_REVALIDATION_INTERVAL);
let (queue, background) = Self::new_with_interval(api, pool, interval);
(queue, background, notifier)
}
/// Queue some transaction for later revalidation.
substrate/client/transaction-pool/src/testing/pool.rs
+39 -35
View File
@@ -27,20 +27,25 @@ use substrate_test_runtime_client::{
AccountKeyring::*,
};
use substrate_test_runtime_transaction_pool::{TestApi, uxt};
use crate::revalidation::BACKGROUND_REVALIDATION_INTERVAL;
use futures::task::Poll;
use futures::{prelude::*, task::Poll};
use codec::Encode;
fn pool() -> Pool<TestApi> {
Pool::new(Default::default(), TestApi::with_alice_nonce(209).into())
}
fn maintained_pool() -> (BasicPool<TestApi, Block>, futures::executor::ThreadPool) {
let (pool, background_task) = BasicPool::new(Default::default(), std::sync::Arc::new(TestApi::with_alice_nonce(209)));
fn maintained_pool() -> (
BasicPool<TestApi, Block>,
futures::executor::ThreadPool,
intervalier::BackSignalControl,
) {
let (pool, background_task, notifier) = BasicPool::new_test(
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)
thread_pool.spawn_ok(background_task);
(pool, thread_pool, notifier)
}
fn header(number: u64) -> Header {
@@ -190,7 +195,7 @@ fn block_event_with_retracted(id: u64, retracted: Vec<Hash>) -> ChainEvent<Block
fn should_prune_old_during_maintenance() {
let xt = uxt(Alice, 209);
let (pool, _guard) = maintained_pool();
let (pool, _guard, _notifier) = maintained_pool();
block_on(pool.submit_one(&BlockId::number(0), SOURCE, xt.clone())).expect("1. Imported");
assert_eq!(pool.status().ready, 1);
@@ -206,7 +211,7 @@ fn should_revalidate_during_maintenance() {
let xt1 = uxt(Alice, 209);
let xt2 = uxt(Alice, 210);
let (pool, _guard) = maintained_pool();
let (pool, _guard, mut notifier) = maintained_pool();
block_on(pool.submit_one(&BlockId::number(0), SOURCE, xt1.clone())).expect("1. Imported");
block_on(pool.submit_one(&BlockId::number(0), SOURCE, xt2.clone())).expect("2. Imported");
assert_eq!(pool.status().ready, 2);
@@ -214,13 +219,11 @@ fn should_revalidate_during_maintenance() {
pool.api.push_block(1, vec![xt1.clone()]);
block_on(pool.maintain(block_event(1)));
// maintaince is in background
block_on(futures_timer::Delay::new(BACKGROUND_REVALIDATION_INTERVAL*2));
notifier.clear();
block_on(pool.maintain(block_event(1)));
assert_eq!(pool.status().ready, 1);
block_on(notifier.next());
// test that pool revalidated transaction that left ready and not included in the block
assert_eq!(pool.api.validation_requests().len(), 3);
}
@@ -230,7 +233,7 @@ fn should_resubmit_from_retracted_during_maintenance() {
let xt = uxt(Alice, 209);
let retracted_hash = Hash::random();
let (pool, _guard) = maintained_pool();
let (pool, _guard, _notifier) = maintained_pool();
block_on(pool.submit_one(&BlockId::number(0), SOURCE, xt.clone())).expect("1. Imported");
assert_eq!(pool.status().ready, 1);
@@ -249,7 +252,7 @@ fn should_not_retain_invalid_hashes_from_retracted() {
let xt = uxt(Alice, 209);
let retracted_hash = Hash::random();
let (pool, _guard) = maintained_pool();
let (pool, _guard, mut notifier) = maintained_pool();
block_on(pool.submit_one(&BlockId::number(0), SOURCE, xt.clone())).expect("1. Imported");
assert_eq!(pool.status().ready, 1);
@@ -260,10 +263,10 @@ fn should_not_retain_invalid_hashes_from_retracted() {
let event = block_event_with_retracted(1, vec![retracted_hash]);
notifier.clear();
block_on(pool.maintain(event));
// maintenance is in background
block_on(futures_timer::Delay::new(BACKGROUND_REVALIDATION_INTERVAL*2));
block_on(notifier.next());
let event = block_event_with_retracted(1, vec![retracted_hash]);
@@ -275,16 +278,16 @@ fn should_not_retain_invalid_hashes_from_retracted() {
fn should_revalidate_transaction_multiple_times() {
let xt = uxt(Alice, 209);
let (pool, _guard) = maintained_pool();
let (pool, _guard, mut notifier) = maintained_pool();
block_on(pool.submit_one(&BlockId::number(0), SOURCE, xt.clone())).expect("1. Imported");
assert_eq!(pool.status().ready, 1);
pool.api.push_block(1, vec![xt.clone()]);
// maintenance is in background
notifier.clear();
block_on(pool.maintain(block_event(1)));
block_on(futures_timer::Delay::new(BACKGROUND_REVALIDATION_INTERVAL*2));
block_on(notifier.next());
block_on(pool.submit_one(&BlockId::number(0), SOURCE, xt.clone())).expect("1. Imported");
assert_eq!(pool.status().ready, 1);
@@ -292,9 +295,9 @@ fn should_revalidate_transaction_multiple_times() {
pool.api.push_block(2, vec![]);
pool.api.add_invalid(&xt);
// maintenance is in background
notifier.clear();
block_on(pool.maintain(block_event(2)));
block_on(futures_timer::Delay::new(BACKGROUND_REVALIDATION_INTERVAL*2));
block_on(notifier.next());
assert_eq!(pool.status().ready, 0);
}
@@ -305,23 +308,24 @@ fn should_revalidate_across_many_blocks() {
let xt2 = uxt(Alice, 210);
let xt3 = uxt(Alice, 211);
let (pool, _guard) = maintained_pool();
let (pool, _guard, mut notifier) = maintained_pool();
block_on(pool.submit_one(&BlockId::number(1), SOURCE, xt1.clone())).expect("1. Imported");
block_on(pool.submit_one(&BlockId::number(1), SOURCE, xt2.clone())).expect("1. Imported");
assert_eq!(pool.status().ready, 2);
pool.api.push_block(1, vec![]);
notifier.clear();
block_on(pool.maintain(block_event(1)));
block_on(futures_timer::Delay::new(BACKGROUND_REVALIDATION_INTERVAL*2));
block_on(notifier.next());
block_on(pool.submit_one(&BlockId::number(2), SOURCE, xt3.clone())).expect("1. Imported");
assert_eq!(pool.status().ready, 3);
pool.api.push_block(2, vec![xt1.clone()]);
notifier.clear();
block_on(pool.maintain(block_event(2)));
block_on(futures_timer::Delay::new(BACKGROUND_REVALIDATION_INTERVAL*2));
block_on(notifier.next());
assert_eq!(pool.status().ready, 2);
// xt1 and xt2 validated twice, then xt3 once, then xt2 and xt3 again
@@ -336,7 +340,7 @@ fn should_push_watchers_during_maintaince() {
}
// given
let (pool, _guard) = maintained_pool();
let (pool, _guard, mut notifier) = maintained_pool();
let tx0 = alice_uxt(0);
let watcher0 = block_on(
@@ -363,15 +367,15 @@ fn should_push_watchers_during_maintaince() {
// 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));
// clear timer events if any
notifier.clear();
block_on(pool.maintain(block_event(0)));
block_on(notifier.next());
// 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<_>>(),
@@ -409,7 +413,7 @@ fn should_push_watchers_during_maintaince() {
#[test]
fn can_track_heap_size() {
let (pool, _guard) = maintained_pool();
let (pool, _guard, _notifier) = maintained_pool();
block_on(pool.submit_one(&BlockId::number(0), SOURCE, uxt(Alice, 209))).expect("1. Imported");
block_on(pool.submit_one(&BlockId::number(0), SOURCE, uxt(Alice, 210))).expect("1. Imported");
block_on(pool.submit_one(&BlockId::number(0), SOURCE, uxt(Alice, 211))).expect("1. Imported");
@@ -629,7 +633,7 @@ fn fork_aware_finalization() {
#[test]
fn ready_set_should_not_resolve_before_block_update() {
let (pool, _guard) = maintained_pool();
let (pool, _guard, _notifier) = maintained_pool();
let xt1 = uxt(Alice, 209);
block_on(pool.submit_one(&BlockId::number(1), SOURCE, xt1.clone())).expect("1. Imported");
@@ -638,7 +642,7 @@ fn ready_set_should_not_resolve_before_block_update() {
#[test]
fn ready_set_should_resolve_after_block_update() {
let (pool, _guard) = maintained_pool();
let (pool, _guard, _notifier) = maintained_pool();
pool.api.push_block(1, vec![]);
let xt1 = uxt(Alice, 209);
@@ -651,7 +655,7 @@ fn ready_set_should_resolve_after_block_update() {
#[test]
fn ready_set_should_eventually_resolve_when_block_update_arrives() {
let (pool, _guard) = maintained_pool();
let (pool, _guard, _notifier) = maintained_pool();
pool.api.push_block(1, vec![]);
let xt1 = uxt(Alice, 209);