use crate::{configuration, initializer::SessionChangeNotification, paras};
use frame_support::pallet_prelude::*;
use frame_system::pallet_prelude::BlockNumberFor;
use primitives::{
v5::ParasEntry, CoreIndex, CoreOccupied, GroupIndex, GroupRotationInfo, Id as ParaId,
ScheduledCore, ValidatorIndex,
};
use sp_runtime::traits::{One, Saturating};
use sp_std::{
collections::{btree_map::BTreeMap, vec_deque::VecDeque},
prelude::*,
};
pub mod common;
use common::{AssignmentProvider, AssignmentProviderConfig, CoreAssignment, FreedReason};
pub use pallet::*;
#[cfg(test)]
mod tests;
const LOG_TARGET: &str = "runtime::parachains::scheduler";
pub mod migration;
#[frame_support::pallet]
pub mod pallet {
use super::*;
const STORAGE_VERSION: StorageVersion = StorageVersion::new(1);
#[pallet::pallet]
#[pallet::without_storage_info]
#[pallet::storage_version(STORAGE_VERSION)]
pub struct Pallet<T>(_);
#[pallet::config]
pub trait Config: frame_system::Config + configuration::Config + paras::Config {
type AssignmentProvider: AssignmentProvider<BlockNumberFor<Self>>;
}
#[pallet::storage]
#[pallet::getter(fn validator_groups)]
pub(crate) type ValidatorGroups<T> = StorageValue<_, Vec<Vec<ValidatorIndex>>, ValueQuery>;
#[pallet::storage]
#[pallet::getter(fn availability_cores)]
pub(crate) type AvailabilityCores<T: Config> =
StorageValue<_, Vec<CoreOccupied<BlockNumberFor<T>>>, ValueQuery>;
#[pallet::storage]
#[pallet::getter(fn session_start_block)]
pub(crate) type SessionStartBlock<T: Config> = StorageValue<_, BlockNumberFor<T>, ValueQuery>;
#[pallet::storage]
#[pallet::getter(fn claimqueue)]
pub(crate) type ClaimQueue<T: Config> = StorageValue<
_,
BTreeMap<CoreIndex, VecDeque<Option<ParasEntry<BlockNumberFor<T>>>>>,
ValueQuery,
>;
}
type PositionInClaimqueue = u32;
type TimedoutParas<T> = BTreeMap<CoreIndex, ParasEntry<BlockNumberFor<T>>>;
type ConcludedParas = BTreeMap<CoreIndex, ParaId>;
impl<T: Config> Pallet<T> {
pub(crate) fn initializer_initialize(_now: BlockNumberFor<T>) -> Weight {
Weight::zero()
}
pub(crate) fn initializer_finalize() {}
pub(crate) fn pre_new_session() {
Self::push_claimqueue_items_to_assignment_provider();
Self::push_occupied_cores_to_assignment_provider();
}
pub(crate) fn initializer_on_new_session(
notification: &SessionChangeNotification<BlockNumberFor<T>>,
) {
let SessionChangeNotification { validators, new_config, .. } = notification;
let config = new_config;
let n_cores = core::cmp::max(
T::AssignmentProvider::session_core_count(),
match config.max_validators_per_core {
Some(x) if x != 0 => validators.len() as u32 / x,
_ => 0,
},
);
AvailabilityCores::<T>::mutate(|cores| {
cores.resize(n_cores as _, CoreOccupied::Free);
});
if n_cores == 0 || validators.is_empty() {
ValidatorGroups::<T>::set(Vec::new());
} else {
let group_base_size = validators.len() / n_cores as usize;
let n_larger_groups = validators.len() % n_cores as usize;
let mut groups: Vec<Vec<ValidatorIndex>> = Vec::new();
for i in 0..n_larger_groups {
let offset = (group_base_size + 1) * i;
groups.push(
(0..group_base_size + 1)
.map(|j| offset + j)
.map(|j| ValidatorIndex(j as _))
.collect(),
);
}
for i in 0..(n_cores as usize - n_larger_groups) {
let offset = (n_larger_groups * (group_base_size + 1)) + (i * group_base_size);
groups.push(
(0..group_base_size)
.map(|j| offset + j)
.map(|j| ValidatorIndex(j as _))
.collect(),
);
}
ValidatorGroups::<T>::set(groups);
}
let now = <frame_system::Pallet<T>>::block_number() + One::one();
<SessionStartBlock<T>>::set(now);
}
fn free_cores(
just_freed_cores: impl IntoIterator<Item = (CoreIndex, FreedReason)>,
) -> (ConcludedParas, TimedoutParas<T>) {
let mut timedout_paras: BTreeMap<CoreIndex, ParasEntry<BlockNumberFor<T>>> =
BTreeMap::new();
let mut concluded_paras = BTreeMap::new();
AvailabilityCores::<T>::mutate(|cores| {
let c_len = cores.len();
just_freed_cores
.into_iter()
.filter(|(freed_index, _)| (freed_index.0 as usize) < c_len)
.for_each(|(freed_index, freed_reason)| {
match &cores[freed_index.0 as usize] {
CoreOccupied::Free => {},
CoreOccupied::Paras(entry) => {
match freed_reason {
FreedReason::Concluded => {
concluded_paras.insert(freed_index, entry.para_id());
},
FreedReason::TimedOut => {
timedout_paras.insert(freed_index, entry.clone());
},
};
},
};
cores[freed_index.0 as usize] = CoreOccupied::Free;
})
});
(concluded_paras, timedout_paras)
}
pub(crate) fn occupied(
now_occupied: BTreeMap<CoreIndex, ParaId>,
) -> BTreeMap<CoreIndex, PositionInClaimqueue> {
let mut availability_cores = AvailabilityCores::<T>::get();
log::debug!(target: LOG_TARGET, "[occupied] now_occupied {:?}", now_occupied);
let pos_mapping: BTreeMap<CoreIndex, PositionInClaimqueue> = now_occupied
.iter()
.flat_map(|(core_idx, para_id)| {
match Self::remove_from_claimqueue(*core_idx, *para_id) {
Err(e) => {
log::debug!(
target: LOG_TARGET,
"[occupied] error on remove_from_claimqueue {}",
e
);
None
},
Ok((pos_in_claimqueue, pe)) => {
availability_cores[core_idx.0 as usize] = CoreOccupied::Paras(pe);
Some((*core_idx, pos_in_claimqueue))
},
}
})
.collect();
Self::drop_expired_claims_from_claimqueue();
AvailabilityCores::<T>::set(availability_cores);
pos_mapping
}
fn drop_expired_claims_from_claimqueue() {
let now = <frame_system::Pallet<T>>::block_number();
let availability_cores = AvailabilityCores::<T>::get();
ClaimQueue::<T>::mutate(|cq| {
for (idx, _) in (0u32..).zip(availability_cores) {
let core_idx = CoreIndex(idx);
if let Some(core_claimqueue) = cq.get_mut(&core_idx) {
let mut dropped_claims: Vec<Option<ParaId>> = vec![];
core_claimqueue.retain(|maybe_entry| {
if let Some(entry) = maybe_entry {
if entry.ttl < now {
dropped_claims.push(Some(entry.para_id()));
return false
}
}
true
});
for drop in dropped_claims {
match T::AssignmentProvider::pop_assignment_for_core(core_idx, drop) {
Some(assignment) => {
let AssignmentProviderConfig { ttl, .. } =
T::AssignmentProvider::get_provider_config(core_idx);
core_claimqueue.push_back(Some(ParasEntry::new(
assignment.clone(),
now + ttl,
)));
},
None => (),
}
}
}
}
});
}
pub(crate) fn core_para(core_index: CoreIndex) -> Option<ParaId> {
let cores = AvailabilityCores::<T>::get();
match cores.get(core_index.0 as usize) {
None | Some(CoreOccupied::Free) => None,
Some(CoreOccupied::Paras(entry)) => Some(entry.para_id()),
}
}
pub(crate) fn group_validators(group_index: GroupIndex) -> Option<Vec<ValidatorIndex>> {
ValidatorGroups::<T>::get().get(group_index.0 as usize).map(|g| g.clone())
}
pub(crate) fn group_assigned_to_core(
core: CoreIndex,
at: BlockNumberFor<T>,
) -> Option<GroupIndex> {
let config = <configuration::Pallet<T>>::config();
let session_start_block = <SessionStartBlock<T>>::get();
if at < session_start_block {
return None
}
let validator_groups = ValidatorGroups::<T>::get();
if core.0 as usize >= validator_groups.len() {
return None
}
let rotations_since_session_start: BlockNumberFor<T> =
(at - session_start_block) / config.group_rotation_frequency.into();
let rotations_since_session_start =
<BlockNumberFor<T> as TryInto<u32>>::try_into(rotations_since_session_start)
.unwrap_or(0);
let group_idx =
(core.0 as usize + rotations_since_session_start as usize) % validator_groups.len();
Some(GroupIndex(group_idx as u32))
}
pub(crate) fn availability_timeout_predicate(
) -> Option<impl Fn(CoreIndex, BlockNumberFor<T>) -> bool> {
let now = <frame_system::Pallet<T>>::block_number();
let config = <configuration::Pallet<T>>::config();
let session_start = <SessionStartBlock<T>>::get();
let blocks_since_session_start = now.saturating_sub(session_start);
let blocks_since_last_rotation =
blocks_since_session_start % config.group_rotation_frequency.max(1u8.into());
if blocks_since_last_rotation >= config.paras_availability_period {
None
} else {
Some(|core_index: CoreIndex, pending_since| {
let availability_cores = AvailabilityCores::<T>::get();
let AssignmentProviderConfig { availability_period, .. } =
T::AssignmentProvider::get_provider_config(core_index);
let now = <frame_system::Pallet<T>>::block_number();
match availability_cores.get(core_index.0 as usize) {
None => true, Some(CoreOccupied::Free) => true, Some(CoreOccupied::Paras(_)) =>
now.saturating_sub(pending_since) >= availability_period,
}
})
}
}
pub(crate) fn group_rotation_info(
now: BlockNumberFor<T>,
) -> GroupRotationInfo<BlockNumberFor<T>> {
let session_start_block = Self::session_start_block();
let group_rotation_frequency =
<configuration::Pallet<T>>::config().group_rotation_frequency;
GroupRotationInfo { session_start_block, now, group_rotation_frequency }
}
pub(crate) fn next_up_on_available(core: CoreIndex) -> Option<ScheduledCore> {
ClaimQueue::<T>::get().get(&core).and_then(|a| {
a.iter()
.find_map(|e| e.as_ref())
.map(|pe| Self::paras_entry_to_scheduled_core(pe))
})
}
fn paras_entry_to_scheduled_core(pe: &ParasEntry<BlockNumberFor<T>>) -> ScheduledCore {
ScheduledCore { para_id: pe.para_id(), collator: None }
}
pub(crate) fn next_up_on_time_out(core: CoreIndex) -> Option<ScheduledCore> {
Self::next_up_on_available(core).or_else(|| {
let cores = AvailabilityCores::<T>::get();
cores.get(core.0 as usize).and_then(|c| match c {
CoreOccupied::Free => None,
CoreOccupied::Paras(pe) => {
let AssignmentProviderConfig { max_availability_timeouts, .. } =
T::AssignmentProvider::get_provider_config(core);
if pe.availability_timeouts < max_availability_timeouts {
Some(Self::paras_entry_to_scheduled_core(pe))
} else {
None
}
},
})
})
}
fn push_occupied_cores_to_assignment_provider() {
AvailabilityCores::<T>::mutate(|cores| {
for (core_idx, core) in cores.iter_mut().enumerate() {
match core {
CoreOccupied::Free => continue,
CoreOccupied::Paras(entry) => {
let core_idx = CoreIndex::from(core_idx as u32);
Self::maybe_push_assignment(core_idx, entry.clone());
},
}
*core = CoreOccupied::Free;
}
});
}
fn push_claimqueue_items_to_assignment_provider() {
for (core_idx, core_claimqueue) in ClaimQueue::<T>::take() {
for para_entry in core_claimqueue.into_iter().flatten().rev() {
Self::maybe_push_assignment(core_idx, para_entry);
}
}
}
fn maybe_push_assignment(core_idx: CoreIndex, pe: ParasEntry<BlockNumberFor<T>>) {
if pe.availability_timeouts == 0 {
T::AssignmentProvider::push_assignment_for_core(core_idx, pe.assignment);
}
}
fn claimqueue_lookahead() -> u32 {
<configuration::Pallet<T>>::config().scheduling_lookahead
}
pub(crate) fn update_claimqueue(
just_freed_cores: impl IntoIterator<Item = (CoreIndex, FreedReason)>,
now: BlockNumberFor<T>,
) -> Vec<CoreAssignment<BlockNumberFor<T>>> {
Self::move_claimqueue_forward();
Self::free_cores_and_fill_claimqueue(just_freed_cores, now)
}
fn move_claimqueue_forward() {
let mut cq = ClaimQueue::<T>::get();
for (_, core_queue) in cq.iter_mut() {
match core_queue.front() {
None => {},
Some(None) => {
core_queue.pop_front();
},
Some(_) => {},
}
}
ClaimQueue::<T>::set(cq);
}
fn free_cores_and_fill_claimqueue(
just_freed_cores: impl IntoIterator<Item = (CoreIndex, FreedReason)>,
now: BlockNumberFor<T>,
) -> Vec<CoreAssignment<BlockNumberFor<T>>> {
let (mut concluded_paras, mut timedout_paras) = Self::free_cores(just_freed_cores);
if ValidatorGroups::<T>::get().is_empty() {
vec![]
} else {
let n_lookahead = Self::claimqueue_lookahead();
let n_session_cores = T::AssignmentProvider::session_core_count();
let cq = ClaimQueue::<T>::get();
let ttl = <configuration::Pallet<T>>::config().on_demand_ttl;
for core_idx in 0..n_session_cores {
let core_idx = CoreIndex::from(core_idx);
if let Some(mut entry) = timedout_paras.remove(&core_idx) {
let AssignmentProviderConfig { max_availability_timeouts, .. } =
T::AssignmentProvider::get_provider_config(core_idx);
if entry.availability_timeouts < max_availability_timeouts {
entry.availability_timeouts += 1;
entry.ttl = now + ttl;
Self::add_to_claimqueue(core_idx, entry);
continue
} else {
let ret = concluded_paras.insert(core_idx, entry.para_id());
debug_assert!(ret.is_none());
}
}
let n_lookahead_used = cq.get(&core_idx).map_or(0, |v| v.len() as u32) +
if Self::is_core_occupied(core_idx) { 1 } else { 0 };
for _ in n_lookahead_used..n_lookahead {
let concluded_para = concluded_paras.remove(&core_idx);
if let Some(assignment) =
T::AssignmentProvider::pop_assignment_for_core(core_idx, concluded_para)
{
Self::add_to_claimqueue(core_idx, ParasEntry::new(assignment, now + ttl));
}
}
}
debug_assert!(timedout_paras.is_empty());
debug_assert!(concluded_paras.is_empty());
Self::scheduled_claimqueue()
}
}
fn is_core_occupied(core_idx: CoreIndex) -> bool {
match AvailabilityCores::<T>::get().get(core_idx.0 as usize) {
None | Some(CoreOccupied::Free) => false,
Some(CoreOccupied::Paras(_)) => true,
}
}
fn add_to_claimqueue(core_idx: CoreIndex, pe: ParasEntry<BlockNumberFor<T>>) {
ClaimQueue::<T>::mutate(|la| {
let la_deque = la.entry(core_idx).or_insert_with(|| VecDeque::new());
la_deque.push_back(Some(pe));
});
}
fn remove_from_claimqueue(
core_idx: CoreIndex,
para_id: ParaId,
) -> Result<(PositionInClaimqueue, ParasEntry<BlockNumberFor<T>>), &'static str> {
ClaimQueue::<T>::mutate(|cq| {
let core_claims = cq.get_mut(&core_idx).ok_or("core_idx not found in lookahead")?;
let pos = core_claims
.iter()
.position(|a| a.as_ref().map_or(false, |pe| pe.para_id() == para_id))
.ok_or("para id not found at core_idx lookahead")?;
let pe = core_claims
.remove(pos)
.ok_or("remove returned None")?
.ok_or("Element in Claimqueue was None.")?;
if core_claims.front() != Some(&None) {
core_claims.push_front(None);
}
Ok((pos as u32, pe))
})
}
pub(crate) fn scheduled_claimqueue() -> Vec<CoreAssignment<BlockNumberFor<T>>> {
let claimqueue = ClaimQueue::<T>::get();
claimqueue
.into_iter()
.flat_map(|(core_idx, v)| {
v.front()
.cloned()
.flatten()
.map(|pe| CoreAssignment { core: core_idx, paras_entry: pe })
})
.collect()
}
#[cfg(any(feature = "runtime-benchmarks", test))]
pub(crate) fn assignment_provider_config(
core_idx: CoreIndex,
) -> AssignmentProviderConfig<BlockNumberFor<T>> {
T::AssignmentProvider::get_provider_config(core_idx)
}
#[cfg(any(feature = "try-runtime", test))]
fn claimqueue_len() -> usize {
ClaimQueue::<T>::get().iter().map(|la_vec| la_vec.1.len()).sum()
}
#[cfg(all(not(feature = "runtime-benchmarks"), test))]
pub(crate) fn claimqueue_is_empty() -> bool {
Self::claimqueue_len() == 0
}
#[cfg(test)]
pub(crate) fn set_validator_groups(validator_groups: Vec<Vec<ValidatorIndex>>) {
ValidatorGroups::<T>::set(validator_groups);
}
}