elastic scaling: rework core selector handling

cumulus/client/collator/src/service.rs

@@ -18,7 +18,10 @@
 //! operations used in parachain consensus/authoring.
 
 use cumulus_client_network::WaitToAnnounce;
-use cumulus_primitives_core::{CollationInfo, CollectCollationInfo, ParachainBlockData};
+use cumulus_primitives_core::{
+	relay_chain::vstaging::UMP_SEPARATOR, CollationInfo, CollectCollationInfo, GetCoreSelectorApi,
+	ParachainBlockData,
+};
 
 use sc_client_api::BlockBackend;
 use sp_api::{ApiExt, ProvideRuntimeApi};
@@ -232,8 +235,27 @@ where
 			},
 		};
 
+		let runtime_api = self.runtime_api.runtime_api();
+
+		// Don't allow sending ump signals until the GetCoreSelectorApi is implemented and at
+		// version 2.
+		let allow_ump_signal = match runtime_api
+			.api_version::<dyn GetCoreSelectorApi<Block>>(block_hash)
+			.ok()
+			.flatten()
+		{
+			Some(version) => version >= 2,
+			None => {
+				tracing::error!(
+					target: LOG_TARGET,
+					"Could not fetch `GetCoreSelectorApi` runtime api version."
+				);
+				return None
+			},
+		};
+
 		// Create the parachain block data for the validators.
-		let collation_info = self
+		let mut collation_info = self
 			.fetch_collation_info(block_hash, &header)
 			.map_err(|e| {
 				tracing::error!(
@@ -251,6 +273,14 @@ where
 			block_data: BlockData(block_data.encode()),
 		});
 
+		if !allow_ump_signal {
+			collation_info.upward_messages = collation_info
+				.upward_messages
+				.into_iter()
+				.take_while(|message| *message != UMP_SEPARATOR)
+				.collect();
+		}
+
 		let upward_messages = collation_info
 			.upward_messages
 			.try_into()
@@ -262,6 +292,7 @@ where
 				)
 			})
 			.ok()?;
+
 		let horizontal_messages = collation_info
 			.horizontal_messages
 			.try_into()

cumulus/client/consensus/aura/src/collators/slot_based/block_builder_task.rs

@@ -525,13 +525,19 @@ where
 	let block_hash = parent.hash;
 	let runtime_api = para_client.runtime_api();
 
-	if runtime_api.has_api::<dyn GetCoreSelectorApi<Block>>(block_hash)? {
-		Ok(runtime_api.core_selector(block_hash)?)
-	} else {
-		let next_block_number: U256 = (*parent.header.number() + One::one()).into();
-
-		// If the runtime API does not support the core selector API, fallback to some default
-		// values.
-		Ok((CoreSelector(next_block_number.byte(0)), ClaimQueueOffset(DEFAULT_CLAIM_QUEUE_OFFSET)))
-	}
+	let next_block_number: U256 = (*parent.header.number() + One::one()).into();
+	// If the runtime API does not support the core selector API, fallback to some default
+	// values.
+	let fallback_core_selector =
+		(CoreSelector(next_block_number.byte(0)), ClaimQueueOffset(DEFAULT_CLAIM_QUEUE_OFFSET));
+
+	let maybe_api_version = runtime_api.api_version::<dyn GetCoreSelectorApi<Block>>(block_hash)?;
+
+	Ok(match maybe_api_version {
+		Some(api_version) if api_version >= 2 =>
+			runtime_api.core_selector(block_hash)?.unwrap_or(fallback_core_selector),
+		#[allow(deprecated)]
+		Some(_) => runtime_api.core_selector_before_version_2(block_hash)?,
+		None => fallback_core_selector,
+	})
 }

cumulus/pallets/parachain-system/Cargo.toml

@@ -120,5 +120,3 @@ try-runtime = [
 	"polkadot-runtime-parachains/try-runtime",
 	"sp-runtime/try-runtime",
 ]
-
-experimental-ump-signals = []

cumulus/pallets/parachain-system/src/lib.rs

@@ -194,42 +194,51 @@ pub mod ump_constants {
 /// Trait for selecting the next core to build the candidate for.
 pub trait SelectCore {
 	/// Core selector information for the current block.
-	fn selected_core() -> (CoreSelector, ClaimQueueOffset);
+	fn selected_core() -> Option<(CoreSelector, ClaimQueueOffset)>;
 	/// Core selector information for the next block.
-	fn select_next_core() -> (CoreSelector, ClaimQueueOffset);
+	fn select_next_core() -> Option<(CoreSelector, ClaimQueueOffset)>;
+}
+
+impl SelectCore for () {
+	fn selected_core() -> Option<(CoreSelector, ClaimQueueOffset)> {
+		None
+	}
+	fn select_next_core() -> Option<(CoreSelector, ClaimQueueOffset)> {
+		None
+	}
 }
 
 /// The default core selection policy.
 pub struct DefaultCoreSelector<T>(PhantomData<T>);
 
 impl<T: frame_system::Config> SelectCore for DefaultCoreSelector<T> {
-	fn selected_core() -> (CoreSelector, ClaimQueueOffset) {
+	fn selected_core() -> Option<(CoreSelector, ClaimQueueOffset)> {
 		let core_selector: U256 = frame_system::Pallet::<T>::block_number().into();
 
-		(CoreSelector(core_selector.byte(0)), ClaimQueueOffset(DEFAULT_CLAIM_QUEUE_OFFSET))
+		Some((CoreSelector(core_selector.byte(0)), ClaimQueueOffset(DEFAULT_CLAIM_QUEUE_OFFSET)))
 	}
 
-	fn select_next_core() -> (CoreSelector, ClaimQueueOffset) {
+	fn select_next_core() -> Option<(CoreSelector, ClaimQueueOffset)> {
 		let core_selector: U256 = (frame_system::Pallet::<T>::block_number() + One::one()).into();
 
-		(CoreSelector(core_selector.byte(0)), ClaimQueueOffset(DEFAULT_CLAIM_QUEUE_OFFSET))
+		Some((CoreSelector(core_selector.byte(0)), ClaimQueueOffset(DEFAULT_CLAIM_QUEUE_OFFSET)))
 	}
 }
 
 /// Core selection policy that builds on claim queue offset 1.
 pub struct LookaheadCoreSelector<T>(PhantomData<T>);
 
 impl<T: frame_system::Config> SelectCore for LookaheadCoreSelector<T> {
-	fn selected_core() -> (CoreSelector, ClaimQueueOffset) {
+	fn selected_core() -> Option<(CoreSelector, ClaimQueueOffset)> {
 		let core_selector: U256 = frame_system::Pallet::<T>::block_number().into();
 
-		(CoreSelector(core_selector.byte(0)), ClaimQueueOffset(1))
+		Some((CoreSelector(core_selector.byte(0)), ClaimQueueOffset(1)))
 	}
 
-	fn select_next_core() -> (CoreSelector, ClaimQueueOffset) {
+	fn select_next_core() -> Option<(CoreSelector, ClaimQueueOffset)> {
 		let core_selector: U256 = (frame_system::Pallet::<T>::block_number() + One::one()).into();
 
-		(CoreSelector(core_selector.byte(0)), ClaimQueueOffset(1))
+		Some((CoreSelector(core_selector.byte(0)), ClaimQueueOffset(1)))
 	}
 }
 
@@ -391,9 +400,7 @@ pub mod pallet {
 				UpwardMessages::<T>::put(&up[..num as usize]);
 				*up = up.split_off(num as usize);
 
-				// Send the core selector UMP signal. This is experimental until relay chain
-				// validators are upgraded to handle ump signals.
-				#[cfg(feature = "experimental-ump-signals")]
+				// Send the core selector UMP signal.
 				Self::send_ump_signal();
 
 				// If the total size of the pending messages is less than the threshold,
@@ -1429,7 +1436,7 @@ impl<T: Config> Pallet<T> {
 	}
 
 	/// Returns the core selector for the next block.
-	pub fn core_selector() -> (CoreSelector, ClaimQueueOffset) {
+	pub fn core_selector() -> Option<(CoreSelector, ClaimQueueOffset)> {
 		T::SelectCore::select_next_core()
 	}
 
@@ -1450,17 +1457,18 @@ impl<T: Config> Pallet<T> {
 	}
 
 	/// Send the ump signals
-	#[cfg(feature = "experimental-ump-signals")]
 	fn send_ump_signal() {
 		use cumulus_primitives_core::relay_chain::vstaging::{UMPSignal, UMP_SEPARATOR};
 
-		UpwardMessages::<T>::mutate(|up| {
-			up.push(UMP_SEPARATOR);
+		// If the runtime is configured with a core selection policy, send the core selector signal.
+		let maybe_core_selector = T::SelectCore::selected_core();
 
-			// Send the core selector signal.
-			let core_selector = T::SelectCore::selected_core();
-			up.push(UMPSignal::SelectCore(core_selector.0, core_selector.1).encode());
-		});
+		if let Some(core_selector) = maybe_core_selector {
+			UpwardMessages::<T>::mutate(|up| {
+				up.push(UMP_SEPARATOR);
+				up.push(UMPSignal::SelectCore(core_selector.0, core_selector.1).encode());
+			});
+		}
 	}
 
 	/// Open HRMP channel for using it in benchmarks or tests.

cumulus/pallets/parachain-system/src/mock.rs

@@ -22,7 +22,7 @@ use super::*;
 
 use alloc::collections::vec_deque::VecDeque;
 use codec::Encode;
-use core::num::NonZeroU32;
+use core::{cell::Cell, num::NonZeroU32};
 use cumulus_primitives_core::{
 	relay_chain::BlockNumber as RelayBlockNumber, AggregateMessageOrigin, InboundDownwardMessage,
 	InboundHrmpMessage, PersistedValidationData,
@@ -94,7 +94,31 @@ impl Config for Test {
 	type CheckAssociatedRelayNumber = AnyRelayNumber;
 	type ConsensusHook = TestConsensusHook;
 	type WeightInfo = ();
-	type SelectCore = DefaultCoreSelector<Test>;
+	type SelectCore = TestCoreSelector<DefaultCoreSelector<Test>>;
+}
+
+std::thread_local! {
+	pub static USE_CORE_SELECTOR: Cell<bool> = Cell::new(true);
+}
+
+pub struct TestCoreSelector<Selector>(PhantomData<Selector>);
+
+impl<Selector: SelectCore> SelectCore for TestCoreSelector<Selector> {
+	fn selected_core() -> Option<(CoreSelector, ClaimQueueOffset)> {
+		if USE_CORE_SELECTOR.get() {
+			Selector::selected_core()
+		} else {
+			None
+		}
+	}
+
+	fn select_next_core() -> Option<(CoreSelector, ClaimQueueOffset)> {
+		if USE_CORE_SELECTOR.get() {
+			Selector::select_next_core()
+		} else {
+			None
+		}
+	}
 }
 
 std::thread_local! {