XCM Revamp Continued

Cargo.toml

@@ -45,6 +45,7 @@ members = [
 	"xcm",
 	"xcm/xcm-builder",
 	"xcm/xcm-executor",
+	"xcm/pallet-xcm",
 	"node/collation-generation",
 	"node/core/approval-voting",
 	"node/core/av-store",

runtime/common/src/xcm_sender.rs

@@ -22,21 +22,22 @@ use xcm::opaque::{VersionedXcm, v0::{SendXcm, MultiLocation, Junction, Xcm, Resu
 use runtime_parachains::{configuration, dmp};
 
 /// Xcm sender for relay chain. It only sends downward message.
-pub struct RelayChainXcmSender<T>(PhantomData<T>);
+pub struct ChildParachainRouter<T>(PhantomData<T>);
 
-impl<T: configuration::Config + dmp::Config> SendXcm for RelayChainXcmSender<T> {
+impl<T: configuration::Config + dmp::Config> SendXcm for ChildParachainRouter<T> {
 	fn send_xcm(dest: MultiLocation, msg: Xcm) -> Result {
-		if let MultiLocation::X1(Junction::Parachain { id }) = dest {
-			// Downward message passing.
-			let config = <configuration::Module<T>>::config();
-			<dmp::Module<T>>::queue_downward_message(
-				&config,
-				id.into(),
-				VersionedXcm::from(msg).encode(),
-			).map_err(Into::<Error>::into)?;
-			Ok(())
-		} else {
-			Err(Error::CannotReachDestination("UnsupportedDestination"))
+		match dest {
+			MultiLocation::X1(Junction::Parachain { id }) => {
+				// Downward message passing.
+				let config = <configuration::Module<T>>::config();
+				<dmp::Module<T>>::queue_downward_message(
+					&config,
+					id.into(),
+					VersionedXcm::from(msg).encode(),
+				).map_err(Into::<Error>::into)?;
+				Ok(())
+			}
+			d => Err(Error::CannotReachDestination(d, msg)),
 		}
 	}
 }

runtime/parachains/src/mock.rs

@@ -117,8 +117,13 @@ impl crate::paras::Config for Test {
 
 impl crate::dmp::Config for Test { }
 
+parameter_types! {
+	pub const FirstMessageFactorPercent: u64 = 100;
+}
+
 impl crate::ump::Config for Test {
 	type UmpSink = crate::ump::mock_sink::MockUmpSink;
+	type FirstMessageFactorPercent = FirstMessageFactorPercent;
 }
 
 impl crate::hrmp::Config for Test {

runtime/parachains/src/ump.rs

@@ -18,9 +18,8 @@ use crate::{
 	configuration::{self, HostConfiguration},
 	initializer,
 };
-use sp_std::{fmt, prelude::*};
+use sp_std::{prelude::*, fmt, marker::PhantomData};
 use sp_std::collections::{btree_map::BTreeMap, vec_deque::VecDeque};
-use sp_runtime::traits::Zero;
 use frame_support::{decl_module, decl_storage, StorageMap, StorageValue, weights::Weight, traits::Get};
 use primitives::v1::{Id as ParaId, UpwardMessage};
 
@@ -42,53 +41,50 @@ const LOG_TARGET: &str = "runtime::ump-sink";
 /// It is possible that by the time the message is sank the origin parachain was offboarded. It is
 /// up to the implementer to check that if it cares.
 pub trait UmpSink {
-	/// Process an incoming upward message and return the amount of weight it consumed.
+	/// Process an incoming upward message and return the amount of weight it consumed, or `None` if
+	/// it did not begin processing a message since it would otherwise exceed `max_weight`.
 	///
 	/// See the trait docs for more details.
-	fn process_upward_message(origin: ParaId, msg: Vec<u8>) -> Weight;
+	fn process_upward_message(origin: ParaId, msg: &[u8], max_weight: Weight) -> Option<Weight>;
 }
 
-/// An implementation of a sink that just swallows the message without consuming any weight.
+/// An implementation of a sink that just swallows the message without consuming any weight. Returns
+/// `Some(0)` indicating that no messages existed for it to process.
 impl UmpSink for () {
-	fn process_upward_message(_: ParaId, _: Vec<u8>) -> Weight {
-		0
+	fn process_upward_message(_: ParaId, _: &[u8], _: Weight) -> Option<Weight> {
+		Some(0)
 	}
 }
 
 /// A specific implementation of a UmpSink where messages are in the XCM format
 /// and will be forwarded to the XCM Executor.
-pub struct XcmSink<Config>(sp_std::marker::PhantomData<Config>);
+pub struct XcmSink<XcmExecutor, Call>(PhantomData<(XcmExecutor, Call)>);
 
-impl<Config: xcm_executor::Config> UmpSink for XcmSink<Config> {
-	fn process_upward_message(origin: ParaId, msg: Vec<u8>) -> Weight {
+impl<XcmExecutor: xcm::v0::ExecuteXcm<Call>, Call> UmpSink for XcmSink<XcmExecutor, Call> {
+	fn process_upward_message(origin: ParaId, mut msg: &[u8], max_weight: Weight) -> Option<Weight> {
 		use parity_scale_codec::Decode;
 		use xcm::VersionedXcm;
-		use xcm::v0::{Junction, MultiLocation, ExecuteXcm};
-		use xcm_executor::XcmExecutor;
+		use xcm::v0::{Junction, MultiLocation, Outcome, Error as XcmError};
 
-		// TODO: #2841 #UMPQUEUE Get a proper weight limit here. Probably from Relay Chain Config
-		let weight_limit = Weight::max_value();
-		let weight = if let Ok(versioned_xcm_message) = VersionedXcm::decode(&mut &msg[..]) {
+		if let Ok(versioned_xcm_message) = VersionedXcm::decode(&mut msg) {
 			match versioned_xcm_message {
 				VersionedXcm::V0(xcm_message) => {
 					let xcm_junction: Junction = Junction::Parachain { id: origin.into() };
 					let xcm_location: MultiLocation = xcm_junction.into();
-					let result = XcmExecutor::<Config>::execute_xcm(xcm_location, xcm_message, weight_limit);
-					result.weight_used()
+					match XcmExecutor::execute_xcm(xcm_location, xcm_message, max_weight) {
+						Outcome::Complete(w) | Outcome::Incomplete(w, _) => Some(w),
+						Outcome::Error(XcmError::WeightLimitReached) => None,
+						Outcome::Error(_) => Some(0),
+					}
 				}
 			}
 		} else {
 			log::error!(
 				target: LOG_TARGET,
 				"Failed to decode versioned XCM from upward message.",
 			);
-			Weight::zero()
-		};
-
-		// TODO: #2841 #UMPQUEUE to be sound, this implementation must ensure that returned (and thus consumed)
-		//  weight is limited to some small portion of the total block weight (as a ballpark, 1/4, 1/8
-		//  or lower).
-		weight
+			None
+		}
 	}
 }
 
@@ -148,6 +144,9 @@ impl fmt::Debug for AcceptanceCheckErr {
 pub trait Config: frame_system::Config + configuration::Config {
 	/// A place where all received upward messages are funneled.
 	type UmpSink: UmpSink;
+
+	/// The factor by which the weight limit it multiplied for the first UMP message to execute with.
+	type FirstMessageFactorPercent: Get<Weight>;
 }
 
 decl_storage! {
@@ -337,12 +336,21 @@ impl<T: Config> Module<T> {
 				// if so - bail.
 				break;
 			}
+			let max_weight = if used_weight_so_far == 0 {
+				// we increase the amount of weight that we're allowed to use on the first message to try to prevent
+				// the possibility of blockage of the queue.
+				config.preferred_dispatchable_upward_messages_step_weight * T::FirstMessageFactorPercent::get() / 100
+			} else {
+				config.preferred_dispatchable_upward_messages_step_weight - used_weight_so_far
+			};
 
 			// dequeue the next message from the queue of the dispatchee
 			let (upward_message, became_empty) = queue_cache.dequeue::<T>(dispatchee);
 			if let Some(upward_message) = upward_message {
-				used_weight_so_far +=
-					T::UmpSink::process_upward_message(dispatchee, upward_message);
+				match T::UmpSink::process_upward_message(dispatchee, &upward_message[..], max_weight) {
+					None => break,
+					Some(used) => used_weight_so_far += used,
+				}
 			}
 
 			if became_empty {
@@ -555,28 +563,25 @@ pub(crate) mod mock_sink {
 
 	pub struct MockUmpSink;
 	impl UmpSink for MockUmpSink {
-		fn process_upward_message(actual_origin: ParaId, actual_msg: Vec<u8>) -> Weight {
-			HOOK.with(|opt_hook| match &mut *opt_hook.borrow_mut() {
-				Some(hook) => {
-					let UmpExpectation {
-						expected_origin,
-						expected_msg,
-						mock_weight,
-					} = match hook.pop_front() {
-						Some(expectation) => expectation,
-						None => {
-							panic!(
-								"The probe is active but didn't expect the message:\n\n\t{:?}.",
-								actual_msg,
-							);
-						}
-					};
-					assert_eq!(expected_origin, actual_origin);
-					assert_eq!(expected_msg, actual_msg);
-					mock_weight
-				}
-				None => 0,
-			})
+		fn process_upward_message(actual_origin: ParaId, actual_msg: &[u8], _max_weight: Weight) -> Option<Weight> {
+			HOOK.with(|opt_hook| opt_hook.borrow_mut().as_mut().map(|hook| {
+				let UmpExpectation {
+					expected_origin,
+					expected_msg,
+					mock_weight,
+				} = match hook.pop_front() {
+					Some(expectation) => expectation,
+					None => {
+						panic!(
+							"The probe is active but didn't expect the message:\n\n\t{:?}.",
+							actual_msg,
+						);
+					}
+				};
+				assert_eq!(expected_origin, actual_origin);
+				assert_eq!(expected_msg, &actual_msg[..]);
+				mock_weight
+			}))
 		}
 	}
 

runtime/rococo/Cargo.toml

@@ -36,6 +36,7 @@ pallet-authorship = { git = "https://github.com/paritytech/substrate", branch =
 pallet-babe = { git = "https://github.com/paritytech/substrate", branch = "master", default-features = false }
 pallet-beefy = { git = "https://github.com/paritytech/grandpa-bridge-gadget", branch = "master", default-features = false }
 pallet-balances = { git = "https://github.com/paritytech/substrate", branch = "master", default-features = false }
+pallet-collective = { git = "https://github.com/paritytech/substrate", branch = "master", default-features = false }
 pallet-grandpa = { git = "https://github.com/paritytech/substrate", branch = "master", default-features = false }
 pallet-im-online = { git = "https://github.com/paritytech/substrate", branch = "master", default-features = false }
 pallet-indices = { git = "https://github.com/paritytech/substrate", branch = "master", default-features = false }
@@ -64,6 +65,7 @@ runtime-parachains = { package = "polkadot-runtime-parachains", path = "../parac
 xcm = { package = "xcm", path = "../../xcm", default-features = false }
 xcm-executor = { package = "xcm-executor", path = "../../xcm/xcm-executor", default-features = false }
 xcm-builder = { package = "xcm-builder", path = "../../xcm/xcm-builder", default-features = false }
+pallet-xcm = { path = "../../xcm/pallet-xcm", default-features = false }
 
 [build-dependencies]
 substrate-wasm-builder = "3.0.0"
@@ -81,6 +83,7 @@ std = [
 	"pallet-babe/std",
 	"beefy-primitives/std",
 	"pallet-balances/std",
+	"pallet-collective/std",
 	"pallet-beefy/std",
 	"pallet-grandpa/std",
 	"pallet-sudo/std",
@@ -118,6 +121,7 @@ std = [
 	"xcm/std",
 	"xcm-executor/std",
 	"xcm-builder/std",
+	"pallet-xcm/std",
 	"log/std",
 ]
 # When enabled, the runtime api will not be build.
@@ -133,11 +137,14 @@ runtime-benchmarks = [
 	"sp-runtime/runtime-benchmarks",
 	"pallet-babe/runtime-benchmarks",
 	"pallet-balances/runtime-benchmarks",
+	"pallet-collective/runtime-benchmarks",
 	"pallet-grandpa/runtime-benchmarks",
 	"pallet-im-online/runtime-benchmarks",
 	"pallet-indices/runtime-benchmarks",
 	"pallet-staking/runtime-benchmarks",
 	"pallet-timestamp/runtime-benchmarks",
+	"pallet-xcm/runtime-benchmarks",
+	"xcm-builder/runtime-benchmarks",
 ]
 try-runtime = [
 	"frame-executive/try-runtime",
@@ -147,6 +154,7 @@ try-runtime = [
 	"pallet-authorship/try-runtime",
 	"pallet-babe/try-runtime",
 	"pallet-balances/try-runtime",
+	"pallet-collective/try-runtime",
 	"pallet-grandpa/try-runtime",
 	"pallet-sudo/try-runtime",
 	"pallet-indices/try-runtime",

runtime/rococo/src/lib.rs

@@ -86,6 +86,7 @@ pub use pallet_balances::Call as BalancesCall;
 
 use polkadot_parachain::primitives::Id as ParaId;
 use xcm::v0::{MultiLocation, NetworkId};
+use xcm_executor::XcmExecutor;
 use xcm_builder::{
 	AccountId32Aliases, ChildParachainConvertsVia, SovereignSignedViaLocation,
 	CurrencyAdapter as XcmCurrencyAdapter, ChildParachainAsNative,
@@ -274,6 +275,9 @@ construct_runtime! {
 
 		Utility: pallet_utility::{Pallet, Call, Event} = 90,
 		Proxy: pallet_proxy::{Pallet, Call, Storage, Event<T>} = 91,
+
+		// Pallet for sending XCM.
+		XcmPallet: pallet_xcm::{Pallet, Call, Storage, Event<T>} = 99,
 	}
 }
 
@@ -619,10 +623,17 @@ parameter_types! {
 	pub const RocFee: (MultiLocation, u128) = (RocLocation::get(), 1 * CENTS);
 }
 
+/// The XCM router. When we want to send an XCM message, we use this type. It amalgamates all of our
+/// individual routers.
+pub type XcmRouter = (
+	// Only one router so far - use DMP to communicate with child parachains.
+	xcm_sender::ChildParachainRouter<Runtime>,
+);
+
 pub struct XcmConfig;
 impl xcm_executor::Config for XcmConfig {
 	type Call = Call;
-	type XcmSender = xcm_sender::RelayChainXcmSender<Runtime>;
+	type XcmSender = XcmRouter;
 	type AssetTransactor = LocalAssetTransactor;
 	type OriginConverter = LocalOriginConverter;
 	type IsReserve = ();
@@ -634,10 +645,29 @@ impl xcm_executor::Config for XcmConfig {
 	type ResponseHandler = ();
 }
 
+/// Type to convert an `Origin` type value into a `MultiLocation` value which represents an interior location
+/// of this chain.
+pub type LocalOriginToLocation = (
+);
+
+impl pallet_xcm::Config for Runtime {
+	type Event = Event;
+	type SendXcmOrigin = xcm_builder::EnsureXcmOrigin<Origin, LocalOriginToLocation>;
+	type XcmRouter = XcmRouter;
+	// Right now nobody but root is allowed to dispatch local XCM messages.
+	type ExecuteXcmOrigin = xcm_builder::EnsureXcmOrigin<Origin, ()>;
+	type XcmExecutor = XcmExecutor<XcmConfig>;
+}
+
 impl parachains_session_info::Config for Runtime {}
 
+parameter_types! {
+	pub const FirstMessageFactorPercent: u64 = 100;
+}
+
 impl parachains_ump::Config for Runtime {
-	type UmpSink = crate::parachains_ump::XcmSink<XcmConfig>;
+	type UmpSink = crate::parachains_ump::XcmSink<XcmExecutor<XcmConfig>, Call>;
+	type FirstMessageFactorPercent = FirstMessageFactorPercent;
 }
 
 impl parachains_dmp::Config for Runtime {}

runtime/test-runtime/src/lib.rs

@@ -474,8 +474,13 @@ impl parachains_paras::Config for Runtime {
 
 impl parachains_dmp::Config for Runtime {}
 
+parameter_types! {
+	pub const FirstMessageFactorPercent: u64 = 100;
+}
+
 impl parachains_ump::Config for Runtime {
 	type UmpSink = ();
+	type FirstMessageFactorPercent = FirstMessageFactorPercent;
 }
 
 impl parachains_hrmp::Config for Runtime {

xcm/Cargo.toml

@@ -6,6 +6,7 @@ description = "The basic XCM datastructures."
 edition = "2018"
 
 [dependencies]
+impl-trait-for-tuples = "0.2.0"
 parity-scale-codec = { version = "2.0.0", default-features = false, features = [ "derive" ] }
 derivative = {version = "2.2.0", default-features = false, features = [ "use_core" ] }
 

xcm/pallet-xcm/Cargo.toml

@@ -0,0 +1,29 @@
+[package]
+authors = ["Parity Technologies <admin@parity.io>"]
+edition = "2018"
+name = "pallet-xcm"
+version = "0.1.0"
+
+[dependencies]
+codec = { package = "parity-scale-codec", version = "2.0.0", default-features = false, features = ["derive"] }
+serde = { version = "1.0.101", optional = true, features = ["derive"] }
+
+sp-std = { git = "https://github.com/paritytech/substrate", default-features = false, branch = "master" }
+sp-runtime = { git = "https://github.com/paritytech/substrate", default-features = false, branch = "master" }
+frame-support = { git = "https://github.com/paritytech/substrate", default-features = false, branch = "master" }
+frame-system = { git = "https://github.com/paritytech/substrate", default-features = false, branch = "master" }
+
+xcm = { path = "..", default-features = false }
+
+[features]
+default = ["std"]
+std = [
+	"codec/std",
+	"serde",
+	"sp-std/std",
+	"sp-runtime/std",
+	"frame-support/std",
+	"frame-system/std",
+	"xcm/std",
+]
+runtime-benchmarks = []