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: Peter Goodspeed-Niklaus <coriolinus@users.noreply.github.com>
Co-authored-by: Bernhard Schuster <bernhard@ahoi.io>

* more suggestions from review

* test availability predicate, add box to please compiler

* add changes to guide

Co-authored-by: Peter Goodspeed-Niklaus <coriolinus@users.noreply.github.com>
Co-authored-by: Bernhard Schuster <bernhard@ahoi.io>

polkadot/Cargo.lock

@@ -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",

polkadot/roadmap/implementors-guide/src/runtime/inclusioninherent.md

@@ -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. 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.
+    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(())`.

polkadot/roadmap/implementors-guide/src/runtime/paras.md

@@ -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
 

polkadot/roadmap/implementors-guide/src/runtime/scheduler.md

@@ -91,13 +91,14 @@ struct ParathreadEntry {
 // A queued parathread entry, pre-assigned to a core.
 struct QueuedParathread {
   claim: ParathreadEntry,
- core: CoreIndex,
+  /// 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
+  /// offset within the set of para-threads ranged `0..config.parathread_cores`.
- next_core: CoreIndex,
+  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.
+  - Behavior undefined if any given cores were not scheduled.
-  - Fails if the given cores are not sorted ascending by core index
+  - 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.

polkadot/runtime/parachains/Cargo.toml

@@ -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" }

polkadot/runtime/parachains/src/configuration.rs

@@ -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

polkadot/runtime/parachains/src/initializer.rs

@@ -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};
 
-pub trait Trait: system::Trait + configuration::Trait + paras::Trait { }
+/// 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 + 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);
 	}
 }
 

polkadot/runtime/parachains/src/mock.rs

@@ -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();

polkadot/runtime/parachains/src/paras.rs

@@ -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()));