use super::*;
use alloc::collections::btree_map::BTreeMap;
use pezframe_support::assert_ok;
use pezkuwi_primitives::{BlockNumber, SchedulerParams, SessionIndex, ValidationCode, ValidatorId};
use pezsp_keyring::Sr25519Keyring;
use crate::{
configuration::HostConfiguration,
initializer::SessionChangeNotification,
mock::{
new_test_ext, Configuration, MockAssigner, MockGenesisConfig, Paras, ParasShared,
RuntimeOrigin, Scheduler, System, Test,
},
paras::{ParaGenesisArgs, ParaKind},
scheduler::{self, common::Assignment, ClaimQueue},
};
fn register_para(id: ParaId) {
let validation_code: ValidationCode = vec![1, 2, 3].into();
assert_ok!(Paras::schedule_para_initialize(
id,
ParaGenesisArgs {
genesis_head: Vec::new().into(),
validation_code: validation_code.clone(),
para_kind: ParaKind::Parathread, }
));
assert_ok!(Paras::add_trusted_validation_code(RuntimeOrigin::root(), validation_code));
}
fn run_to_block(
to: BlockNumber,
new_session: impl Fn(BlockNumber) -> Option<SessionChangeNotification<BlockNumber>>,
) {
while System::block_number() < to {
let b = System::block_number();
Scheduler::initializer_finalize();
Paras::initializer_finalize(b);
if let Some(mut notification) = new_session(b + 1) {
if notification.session_index == SessionIndex::default() {
notification.session_index = ParasShared::scheduled_session();
}
Configuration::force_set_active_config(notification.new_config.clone());
Paras::initializer_on_new_session(¬ification);
Scheduler::initializer_on_new_session(¬ification);
}
System::on_finalize(b);
System::on_initialize(b + 1);
System::set_block_number(b + 1);
Paras::initializer_initialize(b + 1);
Scheduler::initializer_initialize(b + 1);
Scheduler::advance_claim_queue(&Default::default());
}
}
fn default_config() -> HostConfiguration<BlockNumber> {
HostConfiguration {
minimum_validation_upgrade_delay: 6,
#[allow(deprecated)]
scheduler_params: SchedulerParams {
group_rotation_frequency: 10,
paras_availability_period: 3,
lookahead: 2,
num_cores: 3,
max_availability_timeouts: 1,
..Default::default()
},
..Default::default()
}
}
fn genesis_config(config: &HostConfiguration<BlockNumber>) -> MockGenesisConfig {
MockGenesisConfig {
configuration: crate::configuration::GenesisConfig { config: config.clone() },
..Default::default()
}
}
fn next_assignments() -> impl Iterator<Item = (CoreIndex, Assignment)> {
let claim_queue = ClaimQueue::<Test>::get();
claim_queue
.into_iter()
.filter_map(|(core_idx, v)| v.front().map(|a| (core_idx, a.clone())))
}
#[test]
fn session_change_shuffles_validators() {
let mut config = default_config();
config.scheduler_params.num_cores = 5;
let genesis_config = genesis_config(&config);
new_test_ext(genesis_config).execute_with(|| {
assert!(ValidatorGroups::<Test>::get().is_empty());
run_to_block(1, |number| match number {
1 => Some(SessionChangeNotification {
new_config: config.clone(),
validators: vec![
ValidatorId::from(Sr25519Keyring::Alice.public()),
ValidatorId::from(Sr25519Keyring::Bob.public()),
ValidatorId::from(Sr25519Keyring::Charlie.public()),
ValidatorId::from(Sr25519Keyring::Dave.public()),
ValidatorId::from(Sr25519Keyring::Eve.public()),
ValidatorId::from(Sr25519Keyring::Ferdie.public()),
ValidatorId::from(Sr25519Keyring::One.public()),
],
random_seed: [99; 32],
..Default::default()
}),
_ => None,
});
let groups = ValidatorGroups::<Test>::get();
assert_eq!(groups.len(), 5);
for i in 0..2 {
assert_eq!(groups[i].len(), 2);
}
for i in 2..5 {
assert_eq!(groups[i].len(), 1);
}
});
}
#[test]
fn session_change_takes_only_max_per_core() {
let config = {
let mut config = default_config();
config.scheduler_params.num_cores = 2;
config.scheduler_params.max_validators_per_core = Some(1);
config
};
let genesis_config = genesis_config(&config);
new_test_ext(genesis_config).execute_with(|| {
run_to_block(1, |number| match number {
1 => Some(SessionChangeNotification {
new_config: config.clone(),
validators: vec![
ValidatorId::from(Sr25519Keyring::Alice.public()),
ValidatorId::from(Sr25519Keyring::Bob.public()),
ValidatorId::from(Sr25519Keyring::Charlie.public()),
ValidatorId::from(Sr25519Keyring::Dave.public()),
ValidatorId::from(Sr25519Keyring::Eve.public()),
ValidatorId::from(Sr25519Keyring::Ferdie.public()),
ValidatorId::from(Sr25519Keyring::One.public()),
],
random_seed: [99; 32],
..Default::default()
}),
_ => None,
});
let groups = ValidatorGroups::<Test>::get();
assert_eq!(groups.len(), 7);
for i in 0..7 {
assert_eq!(groups[i].len(), 1);
}
});
}
#[test]
fn advance_claim_queue_doubles_assignment_only_if_empty() {
let mut config = default_config();
config.scheduler_params.lookahead = 3;
config.scheduler_params.num_cores = 2;
let genesis_config = genesis_config(&config);
let para_a = ParaId::from(3_u32);
let para_b = ParaId::from(4_u32);
let para_c = ParaId::from(5_u32);
let assignment_a = Assignment::Bulk(para_a);
let assignment_b = Assignment::Bulk(para_b);
let assignment_c = Assignment::Bulk(para_c);
new_test_ext(genesis_config).execute_with(|| {
register_para(para_a);
register_para(para_b);
register_para(para_c);
run_to_block(1, |number| match number {
1 => Some(SessionChangeNotification {
new_config: config.clone(),
validators: vec![
ValidatorId::from(Sr25519Keyring::Alice.public()),
ValidatorId::from(Sr25519Keyring::Bob.public()),
],
..Default::default()
}),
_ => None,
});
MockAssigner::add_test_assignment(assignment_a.clone());
MockAssigner::add_test_assignment(assignment_b.clone());
MockAssigner::add_test_assignment(assignment_c.clone());
run_to_block(2, |_| None);
{
assert_eq!(Scheduler::claim_queue_len(), 5);
let mut claim_queue = scheduler::ClaimQueue::<Test>::get();
assert_eq!(
claim_queue.remove(&CoreIndex(0)).unwrap(),
[assignment_a.clone(), assignment_a, assignment_b]
.into_iter()
.collect::<VecDeque<_>>()
);
assert_eq!(
claim_queue.remove(&CoreIndex(1)).unwrap(),
[assignment_c.clone(), assignment_c].into_iter().collect::<VecDeque<_>>()
);
}
});
}
#[test]
fn advance_claim_queue_no_entry_if_empty() {
let mut config = default_config();
config.scheduler_params.lookahead = 3;
config.scheduler_params.num_cores = 2;
let genesis_config = genesis_config(&config);
let para_a = ParaId::from(3_u32);
let assignment_a = Assignment::Bulk(para_a);
new_test_ext(genesis_config).execute_with(|| {
register_para(para_a);
run_to_block(1, |number| match number {
1 => Some(SessionChangeNotification {
new_config: config.clone(),
validators: vec![
ValidatorId::from(Sr25519Keyring::Alice.public()),
ValidatorId::from(Sr25519Keyring::Bob.public()),
],
..Default::default()
}),
_ => None,
});
MockAssigner::add_test_assignment(assignment_a.clone());
run_to_block(3, |_| None);
{
let mut claim_queue = scheduler::ClaimQueue::<Test>::get();
assert_eq!(
claim_queue.remove(&CoreIndex(0)).unwrap(),
[assignment_a].into_iter().collect::<VecDeque<_>>()
);
assert!(claim_queue.remove(&CoreIndex(1)).is_none());
}
});
}
#[test]
fn advance_claim_queue_except_for() {
let mut config = default_config();
config.scheduler_params.lookahead = 1;
config.scheduler_params.num_cores = 3;
let genesis_config = genesis_config(&config);
let para_a = ParaId::from(1_u32);
let para_b = ParaId::from(2_u32);
let para_c = ParaId::from(3_u32);
let para_d = ParaId::from(4_u32);
let para_e = ParaId::from(5_u32);
let assignment_a = Assignment::Bulk(para_a);
let assignment_b = Assignment::Bulk(para_b);
let assignment_c = Assignment::Bulk(para_c);
let assignment_d = Assignment::Bulk(para_d);
let assignment_e = Assignment::Bulk(para_e);
new_test_ext(genesis_config).execute_with(|| {
register_para(para_a);
register_para(para_b);
register_para(para_c);
register_para(para_d);
register_para(para_e);
run_to_block(1, |number| match number {
1 => Some(SessionChangeNotification {
new_config: config.clone(),
validators: vec![
ValidatorId::from(Sr25519Keyring::Alice.public()),
ValidatorId::from(Sr25519Keyring::Bob.public()),
ValidatorId::from(Sr25519Keyring::Charlie.public()),
],
..Default::default()
}),
_ => None,
});
MockAssigner::add_test_assignment(assignment_a.clone());
MockAssigner::add_test_assignment(assignment_c.clone());
run_to_block(2, |_| None);
Scheduler::advance_claim_queue(&Default::default());
assert_eq!(Scheduler::claim_queue_len(), 0);
MockAssigner::add_test_assignment(assignment_a.clone());
MockAssigner::add_test_assignment(assignment_c.clone());
MockAssigner::add_test_assignment(assignment_b.clone());
MockAssigner::add_test_assignment(assignment_d.clone());
MockAssigner::add_test_assignment(assignment_e.clone());
run_to_block(3, |_| None);
{
let scheduled: BTreeMap<_, _> = next_assignments().collect();
assert_eq!(scheduled.len(), 3);
assert_eq!(scheduled.get(&CoreIndex(0)).unwrap(), &Assignment::Bulk(para_a));
assert_eq!(scheduled.get(&CoreIndex(1)).unwrap(), &Assignment::Bulk(para_c));
assert_eq!(scheduled.get(&CoreIndex(2)).unwrap(), &Assignment::Bulk(para_b));
}
Scheduler::advance_claim_queue(&std::iter::once(CoreIndex(2)).collect());
{
let scheduled: BTreeMap<_, _> = next_assignments().collect();
assert_eq!(scheduled.len(), 3);
assert_eq!(scheduled.get(&CoreIndex(0)).unwrap(), &Assignment::Bulk(para_d));
assert_eq!(scheduled.get(&CoreIndex(1)).unwrap(), &Assignment::Bulk(para_e));
assert_eq!(scheduled.get(&CoreIndex(2)).unwrap(), &Assignment::Bulk(para_b));
}
});
}
#[test]
fn schedule_rotates_groups() {
let on_demand_cores = 2;
let config = {
let mut config = default_config();
config.scheduler_params.lookahead = 1;
config.scheduler_params.num_cores = on_demand_cores;
config
};
let rotation_frequency = config.scheduler_params.group_rotation_frequency;
let genesis_config = genesis_config(&config);
let para_a = ParaId::from(1_u32);
let para_b = ParaId::from(2_u32);
new_test_ext(genesis_config).execute_with(|| {
register_para(para_a);
register_para(para_b);
run_to_block(1, |number| match number {
1 => Some(SessionChangeNotification {
new_config: config.clone(),
validators: vec![
ValidatorId::from(Sr25519Keyring::Alice.public()),
ValidatorId::from(Sr25519Keyring::Eve.public()),
],
..Default::default()
}),
_ => None,
});
let session_start_block = scheduler::SessionStartBlock::<Test>::get();
assert_eq!(session_start_block, 1);
let mut now = 2;
run_to_block(now, |_| None);
let assert_groups_rotated = |rotations: u32, now: &BlockNumberFor<Test>| {
assert_eq!(
Scheduler::group_assigned_to_core(CoreIndex(0), *now).unwrap(),
GroupIndex((0u32 + rotations) % on_demand_cores)
);
assert_eq!(
Scheduler::group_assigned_to_core(CoreIndex(1), *now).unwrap(),
GroupIndex((1u32 + rotations) % on_demand_cores)
);
};
assert_groups_rotated(0, &now);
now = rotation_frequency;
run_to_block(now, |_| None);
assert_groups_rotated(0, &now);
now = now + 1;
run_to_block(now, |_| None);
assert_groups_rotated(1, &now);
now = rotation_frequency * 2;
run_to_block(now, |_| None);
assert_groups_rotated(1, &now);
now = now + 1;
run_to_block(now, |_| None);
assert_groups_rotated(2, &now);
});
}
#[test]
fn availability_predicate_works() {
let genesis_config = genesis_config(&default_config());
let SchedulerParams { group_rotation_frequency, paras_availability_period, .. } =
default_config().scheduler_params;
new_test_ext(genesis_config).execute_with(|| {
run_to_block(1 + paras_availability_period, |_| None);
assert!(!Scheduler::availability_timeout_check_required());
run_to_block(1 + group_rotation_frequency, |_| None);
{
let now = System::block_number();
assert!(Scheduler::availability_timeout_check_required());
let pred = Scheduler::availability_timeout_predicate();
let last_rotation = Scheduler::group_rotation_info(now).last_rotation_at();
let would_be_timed_out = now - paras_availability_period;
let should_not_be_timed_out = last_rotation;
assert!(pred(would_be_timed_out).timed_out);
assert!(!pred(should_not_be_timed_out).timed_out);
assert!(!pred(now).timed_out);
assert!(!pred(would_be_timed_out + 1).timed_out);
}
});
}
#[test]
fn next_up_on_available_uses_next_scheduled_or_none() {
let mut config = default_config();
config.scheduler_params.num_cores = 1;
let genesis_config = genesis_config(&config);
let para_a = ParaId::from(1_u32);
let para_b = ParaId::from(2_u32);
new_test_ext(genesis_config).execute_with(|| {
register_para(para_a);
register_para(para_b);
run_to_block(1, |number| match number {
1 => Some(SessionChangeNotification {
new_config: config.clone(),
validators: vec![
ValidatorId::from(Sr25519Keyring::Alice.public()),
ValidatorId::from(Sr25519Keyring::Eve.public()),
],
..Default::default()
}),
_ => None,
});
MockAssigner::add_test_assignment(Assignment::Bulk(para_a));
run_to_block(2, |_| None);
{
assert_eq!(Scheduler::claim_queue_len(), 2);
assert!(Scheduler::next_up_on_available(CoreIndex(1)).is_none());
assert_eq!(
Scheduler::next_up_on_available(CoreIndex(0)).unwrap(),
ScheduledCore { para_id: para_a, collator: None }
);
Scheduler::advance_claim_queue(&Default::default());
assert_eq!(
Scheduler::next_up_on_available(CoreIndex(0)).unwrap(),
ScheduledCore { para_id: para_a, collator: None }
);
Scheduler::advance_claim_queue(&Default::default());
assert!(Scheduler::next_up_on_available(CoreIndex(0)).is_none());
}
});
}
#[test]
fn session_change_increasing_number_of_cores() {
let mut config = default_config();
config.scheduler_params.num_cores = 2;
let genesis_config = genesis_config(&config);
let para_a = ParaId::from(3_u32);
let para_b = ParaId::from(4_u32);
let assignment_a = Assignment::Bulk(para_a);
let assignment_b = Assignment::Bulk(para_b);
new_test_ext(genesis_config).execute_with(|| {
register_para(para_a);
register_para(para_b);
run_to_block(1, |number| match number {
1 => Some(SessionChangeNotification {
new_config: config.clone(),
validators: vec![
ValidatorId::from(Sr25519Keyring::Alice.public()),
ValidatorId::from(Sr25519Keyring::Bob.public()),
],
..Default::default()
}),
_ => None,
});
MockAssigner::add_test_assignment(assignment_a.clone());
MockAssigner::add_test_assignment(assignment_b.clone());
run_to_block(2, |_| None);
{
let mut claim_queue = scheduler::ClaimQueue::<Test>::get();
assert_eq!(Scheduler::claim_queue_len(), 4);
assert_eq!(
claim_queue.remove(&CoreIndex(0)).unwrap(),
[assignment_a.clone(), assignment_a.clone()]
.into_iter()
.collect::<VecDeque<_>>()
);
assert_eq!(
claim_queue.remove(&CoreIndex(1)).unwrap(),
[assignment_b.clone(), assignment_b.clone()]
.into_iter()
.collect::<VecDeque<_>>()
);
}
let old_config = config;
let mut new_config = old_config.clone();
new_config.scheduler_params.num_cores = 4;
MockAssigner::add_test_assignment(assignment_b.clone());
run_to_block(3, |number| match number {
3 => Some(SessionChangeNotification {
new_config: new_config.clone(),
prev_config: old_config.clone(),
validators: vec![
ValidatorId::from(Sr25519Keyring::Alice.public()),
ValidatorId::from(Sr25519Keyring::Bob.public()),
ValidatorId::from(Sr25519Keyring::Charlie.public()),
ValidatorId::from(Sr25519Keyring::Dave.public()),
],
..Default::default()
}),
_ => None,
});
{
let mut claim_queue = scheduler::ClaimQueue::<Test>::get();
assert_eq!(Scheduler::claim_queue_len(), 3);
assert_eq!(
claim_queue.remove(&CoreIndex(0)).unwrap(),
[assignment_a].into_iter().collect::<VecDeque<_>>()
);
assert_eq!(
claim_queue.remove(&CoreIndex(1)).unwrap(),
[assignment_b.clone()].into_iter().collect::<VecDeque<_>>()
);
assert_eq!(
claim_queue.remove(&CoreIndex(2)).unwrap(),
[assignment_b.clone()].into_iter().collect::<VecDeque<_>>()
);
}
});
}
#[test]
fn session_change_decreasing_number_of_cores() {
let mut config = default_config();
config.scheduler_params.num_cores = 3;
let genesis_config = genesis_config(&config);
let para_a = ParaId::from(3_u32);
let para_b = ParaId::from(4_u32);
let assignment_a = Assignment::Bulk(para_a);
let assignment_b = Assignment::Bulk(para_b);
new_test_ext(genesis_config).execute_with(|| {
register_para(para_a);
register_para(para_b);
run_to_block(1, |number| match number {
1 => Some(SessionChangeNotification {
new_config: config.clone(),
validators: vec![
ValidatorId::from(Sr25519Keyring::Alice.public()),
ValidatorId::from(Sr25519Keyring::Bob.public()),
],
..Default::default()
}),
_ => None,
});
scheduler::Pezpallet::<Test>::set_claim_queue(BTreeMap::from([
(CoreIndex::from(0), VecDeque::from([assignment_a.clone()])),
(CoreIndex::from(2), VecDeque::from([assignment_b.clone(), assignment_b.clone()])),
]));
let old_config = config;
let mut new_config = old_config.clone();
new_config.scheduler_params.num_cores = 1;
run_to_block(3, |number| match number {
3 => Some(SessionChangeNotification {
new_config: new_config.clone(),
prev_config: old_config.clone(),
validators: vec![ValidatorId::from(Sr25519Keyring::Alice.public())],
..Default::default()
}),
_ => None,
});
let mut claim_queue = scheduler::ClaimQueue::<Test>::get();
assert_eq!(Scheduler::claim_queue_len(), 1);
assert_eq!(
claim_queue.remove(&CoreIndex(0)).unwrap(),
[assignment_b.clone()].into_iter().collect::<VecDeque<_>>()
);
Scheduler::advance_claim_queue(&Default::default());
assert_eq!(Scheduler::claim_queue_len(), 0);
MockAssigner::add_test_assignment(assignment_b.clone());
run_to_block(4, |number| match number {
4 => Some(SessionChangeNotification {
new_config: new_config.clone(),
prev_config: new_config.clone(),
validators: vec![],
..Default::default()
}),
_ => None,
});
assert_eq!(Scheduler::claim_queue_len(), 0);
});
}
#[test]
fn session_change_increasing_lookahead() {
let mut config = default_config();
config.scheduler_params.num_cores = 2;
config.scheduler_params.lookahead = 2;
let genesis_config = genesis_config(&config);
let para_a = ParaId::from(3_u32);
let para_b = ParaId::from(4_u32);
let assignment_a = Assignment::Bulk(para_a);
let assignment_b = Assignment::Bulk(para_b);
new_test_ext(genesis_config).execute_with(|| {
register_para(para_a);
register_para(para_b);
run_to_block(1, |number| match number {
1 => Some(SessionChangeNotification {
new_config: config.clone(),
validators: vec![
ValidatorId::from(Sr25519Keyring::Alice.public()),
ValidatorId::from(Sr25519Keyring::Bob.public()),
],
..Default::default()
}),
_ => None,
});
MockAssigner::add_test_assignment(assignment_a.clone());
MockAssigner::add_test_assignment(assignment_a.clone());
MockAssigner::add_test_assignment(assignment_a.clone());
MockAssigner::add_test_assignment(assignment_b.clone());
MockAssigner::add_test_assignment(assignment_b.clone());
MockAssigner::add_test_assignment(assignment_b.clone());
run_to_block(2, |_| None);
{
let mut claim_queue = scheduler::ClaimQueue::<Test>::get();
assert_eq!(Scheduler::claim_queue_len(), 4);
assert_eq!(
claim_queue.remove(&CoreIndex(0)).unwrap(),
[assignment_a.clone(), assignment_a.clone()]
.into_iter()
.collect::<VecDeque<_>>()
);
assert_eq!(
claim_queue.remove(&CoreIndex(1)).unwrap(),
[assignment_a.clone(), assignment_a.clone()]
.into_iter()
.collect::<VecDeque<_>>()
);
}
let old_config = config;
let mut new_config = old_config.clone();
new_config.scheduler_params.lookahead = 4;
run_to_block(3, |number| match number {
3 => Some(SessionChangeNotification {
new_config: new_config.clone(),
prev_config: old_config.clone(),
validators: vec![
ValidatorId::from(Sr25519Keyring::Alice.public()),
ValidatorId::from(Sr25519Keyring::Bob.public()),
],
..Default::default()
}),
_ => None,
});
{
let mut claim_queue = scheduler::ClaimQueue::<Test>::get();
assert_eq!(Scheduler::claim_queue_len(), 6);
assert_eq!(
claim_queue.remove(&CoreIndex(0)).unwrap(),
[assignment_a.clone(), assignment_a.clone(), assignment_b.clone()]
.into_iter()
.collect::<VecDeque<_>>()
);
assert_eq!(
claim_queue.remove(&CoreIndex(1)).unwrap(),
[assignment_a.clone(), assignment_b.clone(), assignment_b.clone()]
.into_iter()
.collect::<VecDeque<_>>()
);
}
});
}