Unverified Commit 1696982d authored by asynchronous rob's avatar asynchronous rob Committed by GitHub
Browse files

Scheduler Module (#1162)



* scheduler module skeleton

* update scheduler skeleton to match latest version of guide

* better session change notification

* add mock randomness and fix test compilation

* shuffle validators into groups

* finish implementing session change logic for scheduler

* tweak core assignment type to track retries of parathread

* reframe queued parathread core as offset

* implement initialzation and finalization routines

* implement parathread claim queuing

* implement core_para

* implement the group_validators routine and fix errors

* add a reason for freeing cores

* implement `schedule` function

* add some docs to the scheduled function

* implement `occupied` helper

* implement availability predicate

* fix some warnings

* integrate scheduler into initializer

* integrate scheduler into mock module

* avoid conflict with Substrate's scheduler storage

* add parathreads index to paras module

* implement parathreads map in paras module

* add is_parathread to paras

* test adding parathread claim

* test that you cannot add claims when no parathread cores exist

* check session change parathread queue pruning

* test validator shuffling

* add allow_unused to scheduler items

* add test for scheduling

* add some more tests for scheduling logic

* test core rotation

* check parathread claim pruning after retries

* add bound notes

* Apply suggestions from code review
Co-authored-by: default avatarPeter Goodspeed-Niklaus <coriolinus@users.noreply.github.com>
Co-authored-by: default avatarBernhard Schuster <bernhard@ahoi.io>

* more suggestions from review

* test availability predicate, add box to please compiler

* add changes to guide
Co-authored-by: default avatarPeter Goodspeed-Niklaus <coriolinus@users.noreply.github.com>
Co-authored-by: default avatarBernhard Schuster <bernhard@ahoi.io>
parent e5c7a3e6
Pipeline #96492 canceled with stages
in 10 minutes and 43 seconds
......@@ -4442,6 +4442,8 @@ dependencies = [
"pallet-vesting",
"parity-scale-codec",
"polkadot-primitives",
"rand 0.7.3",
"rand_chacha 0.2.2",
"rustc-hex",
"serde",
"serde_derive",
......
......@@ -17,9 +17,8 @@ Included: Option<()>,
## Entry Points
* `inclusion`: This entry-point accepts two parameters: [`Bitfields`](/type-definitions.html#signed-availability-bitfield) and [`BackedCandidates`](/type-definitions.html#backed-candidate).
1. The `Bitfields` are first forwarded to the `process_bitfields` routine, returning a set of freed cores. Provide a `Scheduler::core_para` as a core-lookup to the `process_bitfields` routine.
1. If `Scheduler::availability_timeout_predicate` is `Some`, invoke `Inclusion::collect_pending` using it, and add timed-out cores to the free cores.
1. The `Bitfields` are first forwarded to the `process_bitfields` routine, returning a set of freed cores. Provide a `Scheduler::core_para` as a core-lookup to the `process_bitfields` routine. Annotate each of these freed cores with `FreedReason::Concluded`.
1. If `Scheduler::availability_timeout_predicate` is `Some`, invoke `Inclusion::collect_pending` using it, and add timed-out cores to the free cores, annotated with `FreedReason::TimedOut`.
1. Invoke `Scheduler::schedule(freed)`
1. Pass the `BackedCandidates` along with the output of `Scheduler::scheduled` to the `Inclusion::process_candidates` routine, getting a list of all newly-occupied cores.
1. Call `Scheduler::occupied` for all scheduled cores where a backed candidate was submitted.
1. If all of the above succeeds, set `Included` to `Some(())`.
......@@ -56,6 +56,8 @@ Storage layout:
```rust
/// All parachains. Ordered ascending by ParaId. Parathreads are not included.
Parachains: Vec<ParaId>,
/// All parathreads.
Parathreads: map ParaId => Option<()>,
/// The head-data of every registered para.
Heads: map ParaId => Option<HeadData>;
/// The validation code of every live para.
......@@ -94,6 +96,7 @@ OutgoingParas: Vec<ParaId>;
1. Clean up outgoing paras. This means removing the entries under `Heads`, `ValidationCode`, `FutureCodeUpgrades`, and `FutureCode`. An according entry should be added to `PastCode`, `PastCodeMeta`, and `PastCodePruning` using the outgoing `ParaId` and removed `ValidationCode` value. This is because any outdated validation code must remain available on-chain for a determined amount of blocks, and validation code outdated by de-registering the para is still subject to that invariant.
1. Apply all incoming paras by initializing the `Heads` and `ValidationCode` using the genesis parameters.
1. Amend the `Parachains` list to reflect changes in registered parachains.
1. Amend the `Parathreads` set to reflect changes in registered parathreads.
## Initialization
......@@ -106,6 +109,7 @@ OutgoingParas: Vec<ParaId>;
* `schedule_code_upgrade(ParaId, ValidationCode, expected_at: BlockNumber)`: Schedule a future code upgrade of the given parachain, to be applied after inclusion of a block of the same parachain executed in the context of a relay-chain block with number >= `expected_at`.
* `note_new_head(ParaId, HeadData, BlockNumber)`: note that a para has progressed to a new head, where the new head was executed in the context of a relay-chain block with given number. This will apply pending code upgrades based on the block number provided.
* `validation_code_at(ParaId, at: BlockNumber, assume_intermediate: Option<BlockNumber>)`: Fetches the validation code to be used when validating a block in the context of the given relay-chain height. A second block number parameter may be used to tell the lookup to proceed as if an intermediate parablock has been included at the given relay-chain height. This may return past, current, or (with certain choices of `assume_intermediate`) future code. `assume_intermediate`, if provided, must be before `at`. If the validation code has been pruned, this will return `None`.
* `is_parathread(ParaId) -> bool`: Returns true if the para ID references any live parathread.
## Finalization
......
......@@ -90,14 +90,15 @@ struct ParathreadEntry {
// A queued parathread entry, pre-assigned to a core.
struct QueuedParathread {
claim: ParathreadEntry,
core: CoreIndex,
claim: ParathreadEntry,
/// offset within the set of para-threads ranged `0..config.parathread_cores`.
core_offset: u32,
}
struct ParathreadQueue {
queue: Vec<QueuedParathread>,
// this value is between 0 and config.parathread_cores
next_core: CoreIndex,
queue: Vec<QueuedParathread>,
/// offset within the set of para-threads ranged `0..config.parathread_cores`.
next_core_offset: u32,
}
enum CoreOccupied {
......@@ -105,12 +106,22 @@ enum CoreOccupied {
Parachain,
}
enum AssignmentKind {
Parachain,
Parathread(CollatorId, u32),
}
struct CoreAssignment {
core: CoreIndex,
para_id: ParaId,
collator: Option<CollatorId>,
kind: AssignmentKind,
group_idx: GroupIndex,
}
// reasons a core might be freed.
enum FreedReason {
Concluded,
TimedOut,
}
```
Storage layout:
......@@ -150,6 +161,7 @@ Actions:
- First, we obtain "shuffled validators" `SV` by shuffling the validators using the `SessionChangeNotification`'s random seed.
- The groups are selected by partitioning `SV`. The first V % N groups will have (V / N) + 1 members, while the remaining groups will have (V / N) members each.
1. Prune the parathread queue to remove all retries beyond `configuration.parathread_retries`.
- Also prune all parathread claims corresponding to de-registered parathreads.
- all pruned claims should have their entry removed from the parathread index.
- assign all non-pruned claims to new cores if the number of parathread cores has changed between the `new_config` and `old_config` of the `SessionChangeNotification`.
- Assign claims in equal balance across all cores if rebalancing, and set the `next_core` of the `ParathreadQueue` by incrementing the relative index of the last assigned core and taking it modulo the number of parathread cores.
......@@ -172,15 +184,16 @@ Actions:
- The core used for the parathread claim is the `next_core` field of the `ParathreadQueue` and adding `Paras::parachains().len()` to it.
- `next_core` is then updated by adding 1 and taking it modulo `config.parathread_cores`.
- The claim is then added to the claim index.
- `schedule(Vec<CoreIndex>)`: schedule new core assignments, with a parameter indicating previously-occupied cores which are to be considered returned.
- `schedule(Vec<(CoreIndex, FreedReason)>)`: schedule new core assignments, with a parameter indicating previously-occupied cores which are to be considered returned and why they are being returned.
- All freed parachain cores should be assigned to their respective parachain
- All freed parathread cores should have the claim removed from the claim index.
- All freed parathread cores whose reason for freeing was `FreedReason::Concluded` should have the claim removed from the claim index.
- All freed parathread cores whose reason for freeing was `FreedReason::TimedOut` should have the claim added to the parathread queue again without retries incremented
- All freed parathread cores should take the next parathread entry from the queue.
- The i'th validator group will be assigned to the `(i+k)%n`'th core at any point in time, where `k` is the number of rotations that have occurred in the session, and `n` is the total number of cores. This makes upcoming rotations within the same session predictable.
- `scheduled() -> Vec<CoreAssignment>`: Get currently scheduled core assignments.
- `occupied(Vec<CoreIndex>)`. Note that the given cores have become occupied.
- Fails if any given cores were not scheduled.
- Fails if the given cores are not sorted ascending by core index
- Behavior undefined if any given cores were not scheduled.
- Behavior undefined if the given cores are not sorted ascending by core index
- This clears them from `Scheduled` and marks each corresponding `core` in the `AvailabilityCores` as occupied.
- Since both the availability cores and the newly-occupied cores lists are sorted ascending, this method can be implemented efficiently.
- `core_para(CoreIndex) -> ParaId`: return the currently-scheduled or occupied ParaId for the given core.
......
......@@ -35,6 +35,9 @@ frame-benchmarking = { git = "https://github.com/paritytech/substrate", branch =
primitives = { package = "polkadot-primitives", path = "../../primitives", default-features = false }
libsecp256k1 = { version = "0.3.2", default-features = false, optional = true }
rand = { version = "0.7", default-features = false }
rand_chacha = { version = "0.2.2", default-features = false }
[dev-dependencies]
hex-literal = "0.2.1"
keyring = { package = "sp-keyring", git = "https://github.com/paritytech/substrate", branch = "master" }
......
......@@ -33,7 +33,7 @@ use system::ensure_root;
/// All configuration of the runtime with respect to parachains and parathreads.
#[derive(Clone, Encode, Decode, PartialEq, Default)]
#[cfg_attr(test, derive(Debug))]
pub struct HostConfiguration<BlockNumber: Default> {
pub struct HostConfiguration<BlockNumber> {
/// The minimum frequency at which parachains can update their validation code.
pub validation_upgrade_frequency: BlockNumber,
/// The delay, in blocks, before a validation upgrade is applied.
......@@ -49,7 +49,7 @@ pub struct HostConfiguration<BlockNumber: Default> {
pub parathread_cores: u32,
/// The number of retries that a parathread author has to submit their block.
pub parathread_retries: u32,
/// How often parachain groups should be rotated across parachains.
/// How often parachain groups should be rotated across parachains. Must be non-zero.
pub parachain_rotation_frequency: BlockNumber,
/// The availability period, in blocks, for parachains. This is the amount of blocks
/// after inclusion that validators have to make the block available and signal its availability to
......
......@@ -25,11 +25,29 @@ use primitives::{
parachain::{ValidatorId},
};
use frame_support::{
decl_storage, decl_module, decl_error,
decl_storage, decl_module, decl_error, traits::Randomness,
};
use crate::{configuration, paras};
use crate::{configuration::{self, HostConfiguration}, paras, scheduler};
/// Information about a session change that has just occurred.
#[derive(Default, Clone)]
pub struct SessionChangeNotification<BlockNumber> {
/// The new validators in the session.
pub validators: Vec<ValidatorId>,
/// The qeueud validators for the following session.
pub queued: Vec<ValidatorId>,
/// The configuration before handling the session change
pub prev_config: HostConfiguration<BlockNumber>,
/// The configuration after handling the session change.
pub new_config: HostConfiguration<BlockNumber>,
/// A secure random seed for the session, gathered from BABE.
pub random_seed: [u8; 32],
}
pub trait Trait: system::Trait + configuration::Trait + paras::Trait { }
pub trait Trait: system::Trait + configuration::Trait + paras::Trait + scheduler::Trait {
/// A randomness beacon.
type Randomness: Randomness<Self::Hash>;
}
decl_storage! {
trait Store for Module<T: Trait> as Initializer {
......@@ -62,7 +80,8 @@ decl_module! {
// - Inclusion
// - Validity
let total_weight = configuration::Module::<T>::initializer_initialize(now) +
paras::Module::<T>::initializer_initialize(now);
paras::Module::<T>::initializer_initialize(now) +
scheduler::Module::<T>::initializer_initialize(now);
HasInitialized::set(Some(()));
......@@ -70,6 +89,9 @@ decl_module! {
}
fn on_finalize() {
// reverse initialization order.
scheduler::Module::<T>::initializer_finalize();
paras::Module::<T>::initializer_finalize();
configuration::Module::<T>::initializer_finalize();
HasInitialized::take();
......@@ -90,8 +112,32 @@ impl<T: Trait> Module<T> {
let validators: Vec<_> = validators.map(|(_, v)| v).collect();
let queued: Vec<_> = queued.map(|(_, v)| v).collect();
let prev_config = <configuration::Module<T>>::config();
let random_seed = {
let mut buf = [0u8; 32];
let random_hash = T::Randomness::random(&b"paras"[..]);
let len = sp_std::cmp::min(32, random_hash.as_ref().len());
buf[..len].copy_from_slice(&random_hash.as_ref()[..len]);
buf
};
// We can't pass the new config into the thing that determines the new config,
// so we don't pass the `SessionChangeNotification` into this module.
configuration::Module::<T>::initializer_on_new_session(&validators, &queued);
paras::Module::<T>::initializer_on_new_session(&validators, &queued);
let new_config = <configuration::Module<T>>::config();
let notification = SessionChangeNotification {
validators,
queued,
prev_config,
new_config,
random_seed,
};
paras::Module::<T>::initializer_on_new_session(&notification);
scheduler::Module::<T>::initializer_on_new_session(&notification);
}
}
......
......@@ -30,7 +30,7 @@ use primitives::{
};
use frame_support::{
impl_outer_origin, impl_outer_dispatch, parameter_types,
weights::Weight,
weights::Weight, traits::Randomness as RandomnessT,
};
/// A test runtime struct.
......@@ -47,6 +47,14 @@ impl_outer_dispatch! {
}
}
pub struct TestRandomness;
impl RandomnessT<H256> for TestRandomness {
fn random(_subject: &[u8]) -> H256 {
Default::default()
}
}
parameter_types! {
pub const BlockHashCount: u32 = 250;
pub const MaximumBlockWeight: Weight = 4 * 1024 * 1024;
......@@ -80,12 +88,16 @@ impl system::Trait for Test {
type OnKilledAccount = ();
}
impl crate::initializer::Trait for Test { }
impl crate::initializer::Trait for Test {
type Randomness = TestRandomness;
}
impl crate::configuration::Trait for Test { }
impl crate::paras::Trait for Test { }
impl crate::scheduler::Trait for Test { }
pub type System = system::Module<Test>;
/// Mocked initializer.
......@@ -97,6 +109,9 @@ pub type Configuration = crate::configuration::Module<Test>;
/// Mocked paras.
pub type Paras = crate::paras::Module<Test>;
/// Mocked scheduler.
pub type Scheduler = crate::scheduler::Module<Test>;
/// Create a new set of test externalities.
pub fn new_test_ext(state: GenesisConfig) -> TestExternalities {
let mut t = state.system.build_storage::<Test>().unwrap();
......
......@@ -27,7 +27,7 @@ use sp_std::prelude::*;
use sp_std::marker::PhantomData;
use sp_runtime::traits::One;
use primitives::{
parachain::{ValidatorId, Id as ParaId, ValidationCode, HeadData},
parachain::{Id as ParaId, ValidationCode, HeadData},
};
use frame_support::{
decl_storage, decl_module, decl_error,
......@@ -35,7 +35,7 @@ use frame_support::{
weights::Weight,
};
use codec::{Encode, Decode};
use crate::configuration;
use crate::{configuration, initializer::SessionChangeNotification};
#[cfg(feature = "std")]
use serde::{Serialize, Deserialize};
......@@ -159,11 +159,11 @@ impl<N: Ord + Copy> ParaPastCodeMeta<N> {
#[cfg_attr(feature = "std", derive(Serialize, Deserialize))]
pub struct ParaGenesisArgs {
/// The initial head data to use.
genesis_head: HeadData,
pub genesis_head: HeadData,
/// The initial validation code to use.
validation_code: ValidationCode,
pub validation_code: ValidationCode,
/// True if parachain, false if parathread.
parachain: bool,
pub parachain: bool,
}
......@@ -171,6 +171,8 @@ decl_storage! {
trait Store for Module<T: Trait> as Paras {
/// All parachains. Ordered ascending by ParaId. Parathreads are not included.
Parachains get(fn parachains): Vec<ParaId>;
/// All parathreads.
Parathreads: map hasher(twox_64_concat) ParaId => Option<()>;
/// The head-data of every registered para.
Heads get(fn parachain_head): map hasher(twox_64_concat) ParaId => Option<HeadData>;
/// The validation code of every live para.
......@@ -253,7 +255,7 @@ impl<T: Trait> Module<T> {
pub(crate) fn initializer_finalize() { }
/// Called by the initializer to note that a new session has started.
pub(crate) fn initializer_on_new_session(_validators: &[ValidatorId], _queued: &[ValidatorId]) {
pub(crate) fn initializer_on_new_session(_notification: &SessionChangeNotification<T::BlockNumber>) {
let now = <system::Module<T>>::block_number();
let mut parachains = Self::clean_up_outgoing(now);
Self::apply_incoming(&mut parachains);
......@@ -268,6 +270,8 @@ impl<T: Trait> Module<T> {
for outgoing_para in outgoing {
if let Ok(i) = parachains.binary_search(&outgoing_para) {
parachains.remove(i);
} else {
<Self as Store>::Parathreads::remove(&outgoing_para);
}
<Self as Store>::Heads::remove(&outgoing_para);
......@@ -296,6 +300,8 @@ impl<T: Trait> Module<T> {
if let Err(i) = parachains.binary_search(&upcoming_para) {
parachains.insert(i, upcoming_para);
}
} else {
<Self as Store>::Parathreads::insert(&upcoming_para, ());
}
<Self as Store>::Heads::insert(&upcoming_para, genesis_data.genesis_head);
......@@ -515,6 +521,11 @@ impl<T: Trait> Module<T> {
}
}
}
/// Whether a para ID corresponds to any live parathread.
pub(crate) fn is_parathread(id: ParaId) -> bool {
Parathreads::get(&id).is_some()
}
}
#[cfg(test)]
......@@ -536,7 +547,7 @@ mod tests {
System::set_block_number(b + 1);
if new_session.as_ref().map_or(false, |v| v.contains(&(b + 1))) {
Paras::initializer_on_new_session(&[], &[]);
Paras::initializer_on_new_session(&Default::default());
}
Paras::initializer_initialize(b + 1);
}
......@@ -1040,18 +1051,24 @@ mod tests {
);
assert_eq!(<Paras as Store>::UpcomingParas::get(), vec![c, b, a]);
assert!(<Paras as Store>::Parathreads::get(&a).is_none());
// run to block without session change.
run_to_block(2, None);
assert_eq!(Paras::parachains(), Vec::new());
assert_eq!(<Paras as Store>::UpcomingParas::get(), vec![c, b, a]);
assert!(<Paras as Store>::Parathreads::get(&a).is_none());
run_to_block(3, Some(vec![3]));
assert_eq!(Paras::parachains(), vec![c, b]);
assert_eq!(<Paras as Store>::UpcomingParas::get(), Vec::new());
assert!(<Paras as Store>::Parathreads::get(&a).is_some());
assert_eq!(Paras::current_code(&a), Some(vec![2].into()));
assert_eq!(Paras::current_code(&b), Some(vec![1].into()));
assert_eq!(Paras::current_code(&c), Some(vec![3].into()));
......
This diff is collapsed.
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment