From a06a9c0525c99397a01f4043a5307f8f5479d55c Mon Sep 17 00:00:00 2001 From: Oliver Tale-Yazdi Date: Thu, 2 Nov 2023 15:31:38 +0100 Subject: [PATCH] Use `Message Queue` as DMP and XCMP dispatch queue (#1246) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit (imported from https://github.com/paritytech/cumulus/pull/2157) ## Changes This MR refactores the XCMP, Parachains System and DMP pallets to use the [MessageQueue](https://github.com/paritytech/substrate/pull/12485) for delayed execution of incoming messages. The DMP pallet is entirely replaced by the MQ and thereby removed. This allows for PoV-bounded execution and resolves a number of issues that stem from the current work-around. All System Parachains adopt this change. The most important changes are in `primitives/core/src/lib.rs`, `parachains/common/src/process_xcm_message.rs`, `pallets/parachain-system/src/lib.rs`, `pallets/xcmp-queue/src/lib.rs` and the runtime configs. ### DMP Queue Pallet The pallet got removed and its logic refactored into parachain-system. Overweight message management can be done directly through the MQ pallet. Final undeployment migrations are provided by `cumulus_pallet_dmp_queue::UndeployDmpQueue` and `DeleteDmpQueue` that can be configured with an aux config trait like: ```rust parameter_types! { pub const DmpQueuePalletName: &'static str = \"DmpQueue\" < CHANGE ME; pub const RelayOrigin: AggregateMessageOrigin = AggregateMessageOrigin::Parent; } impl cumulus_pallet_dmp_queue::MigrationConfig for Runtime { type PalletName = DmpQueuePalletName; type DmpHandler = frame_support::traits::EnqueueWithOrigin; type DbWeight = ::DbWeight; } // And adding them to your Migrations tuple: pub type Migrations = ( ... cumulus_pallet_dmp_queue::UndeployDmpQueue, cumulus_pallet_dmp_queue::DeleteDmpQueue, ); ``` ### XCMP Queue pallet Removed all dispatch queue functionality. Incoming XCMP messages are now either: Immediately handled if they are Signals, enqueued into the MQ pallet otherwise. New config items for the XCMP queue pallet: ```rust /// The actual queue implementation that retains the messages for later processing. type XcmpQueue: EnqueueMessage; /// How a XCM over HRMP from a sibling parachain should be processed. type XcmpProcessor: ProcessMessage; /// The maximal number of suspended XCMP channels at the same time. #[pallet::constant] type MaxInboundSuspended: Get; ``` How to configure those: ```rust // Use the MessageQueue pallet to store messages for later processing. The `TransformOrigin` is needed since // the MQ pallet itself operators on `AggregateMessageOrigin` but we want to enqueue `ParaId`s. type XcmpQueue = TransformOrigin; // Process XCMP messages from siblings. This is type-safe to only accept `ParaId`s. They will be dispatched // with origin `Junction::Sibling(…)`. type XcmpProcessor = ProcessFromSibling< ProcessXcmMessage< AggregateMessageOrigin, xcm_executor::XcmExecutor, RuntimeCall, >, >; // Not really important what to choose here. Just something larger than the maximal number of channels. type MaxInboundSuspended = sp_core::ConstU32<1_000>; ``` The `InboundXcmpStatus` storage item was replaced by `InboundXcmpSuspended` since it now only tracks inbound queue suspension and no message indices anymore. Now only sends the most recent channel `Signals`, as all prio ones are out-dated anyway. ### Parachain System pallet For `DMP` messages instead of forwarding them to the `DMP` pallet, it now pushes them to the configured `DmpQueue`. The message processing which was triggered in `set_validation_data` is now being done by the MQ pallet `on_initialize`. XCMP messages are still handed off to the `XcmpMessageHandler` (XCMP-Queue pallet) - no change here. New config items for the parachain system pallet: ```rust /// Queues inbound downward messages for delayed processing. /// /// Analogous to the `XcmpQueue` of the XCMP queue pallet. type DmpQueue: EnqueueMessage; ``` How to configure: ```rust /// Use the MQ pallet to store DMP messages for delayed processing. type DmpQueue = MessageQueue; ``` ## Message Flow The flow of messages on the parachain side. Messages come in from the left via the `Validation Data` and finally end up at the `Xcm Executor` on the right. ![Untitled (1)](https://github.com/paritytech/cumulus/assets/10380170/6cf8b377-88c9-4aed-96df-baace266e04d) ## Further changes - Bumped the default suspension, drop and resume thresholds in `QueueConfigData::default()`. - `XcmpQueue::{suspend_xcm_execution, resume_xcm_execution}` errors when they would be a noop. - Properly validate the `QueueConfigData` before setting it. - Marked weight files as auto-generated so they wont auto-expand in the MR files view. - Move the `hypothetical` asserts to `frame_support` under the name `experimental_hypothetically` Questions: - [ ] What about the ugly `#[cfg(feature = \"runtime-benchmarks\")]` in the runtimes? Not sure how to best fix. Just having them like this makes tests fail that rely on the real message processor when the feature is enabled. - [ ] Need a good weight for `MessageQueueServiceWeight`. The scheduler already takes 80% so I put it to 10% but that is quite low. TODO: - [x] Remove c&p code after https://github.com/paritytech/polkadot/pull/6271 - [x] Use `HandleMessage` once it is public in Substrate - [x] fix `runtime-benchmarks` feature https://github.com/paritytech/polkadot/pull/6966 - [x] Benchmarks - [x] Tests - [ ] Migrate `InboundXcmpStatus` to `InboundXcmpSuspended` - [x] Possibly cleanup Migrations (DMP+XCMP) - [x] optional: create `TransformProcessMessageOrigin` in Substrate and replace `ProcessFromSibling` - [ ] Rerun weights on ref HW --------- Signed-off-by: Oliver Tale-Yazdi Co-authored-by: Liam Aharon Co-authored-by: joe petrowski <25483142+joepetrowski@users.noreply.github.com> Co-authored-by: Kian Paimani <5588131+kianenigma@users.noreply.github.com> Co-authored-by: command-bot <> --- substrate/frame/message-queue/src/lib.rs | 42 +++++++-------- substrate/frame/message-queue/src/mock.rs | 8 ++- substrate/frame/message-queue/src/tests.rs | 51 +++++++++++++++---- substrate/frame/safe-mode/Cargo.toml | 1 + substrate/frame/safe-mode/src/tests.rs | 28 +--------- substrate/frame/support/src/lib.rs | 29 ++++++++++- substrate/frame/support/src/traits.rs | 3 +- .../frame/support/src/traits/messages.rs | 21 +++++--- substrate/primitives/core/Cargo.toml | 2 +- 9 files changed, 116 insertions(+), 69 deletions(-) diff --git a/substrate/frame/message-queue/src/lib.rs b/substrate/frame/message-queue/src/lib.rs index 04bbea121ddc..12d289478b37 100644 --- a/substrate/frame/message-queue/src/lib.rs +++ b/substrate/frame/message-queue/src/lib.rs @@ -195,7 +195,7 @@ use frame_support::{ pallet_prelude::*, traits::{ DefensiveTruncateFrom, EnqueueMessage, ExecuteOverweightError, Footprint, ProcessMessage, - ProcessMessageError, QueuePausedQuery, ServiceQueues, + ProcessMessageError, QueueFootprint, QueuePausedQuery, ServiceQueues, }, BoundedSlice, CloneNoBound, DefaultNoBound, }; @@ -423,14 +423,23 @@ impl Default for BookState { } } +impl From> for QueueFootprint { + fn from(book: BookState) -> Self { + QueueFootprint { + pages: book.count, + storage: Footprint { count: book.message_count, size: book.size }, + } + } +} + /// Handler code for when the items in a queue change. pub trait OnQueueChanged { /// Note that the queue `id` now has `item_count` items in it, taking up `items_size` bytes. - fn on_queue_changed(id: Id, items_count: u64, items_size: u64); + fn on_queue_changed(id: Id, fp: QueueFootprint); } impl OnQueueChanged for () { - fn on_queue_changed(_: Id, _: u64, _: u64) {} + fn on_queue_changed(_: Id, _: QueueFootprint) {} } #[frame_support::pallet] @@ -907,11 +916,7 @@ impl Pallet { T::WeightInfo::execute_overweight_page_updated() }; BookStateFor::::insert(&origin, &book_state); - T::QueueChangeHandler::on_queue_changed( - origin, - book_state.message_count, - book_state.size, - ); + T::QueueChangeHandler::on_queue_changed(origin, book_state.into()); Ok(weight_counter.consumed().saturating_add(page_weight)) }, } @@ -976,11 +981,7 @@ impl Pallet { book_state.message_count.saturating_reduce(page.remaining.into() as u64); book_state.size.saturating_reduce(page.remaining_size.into() as u64); BookStateFor::::insert(origin, &book_state); - T::QueueChangeHandler::on_queue_changed( - origin.clone(), - book_state.message_count, - book_state.size, - ); + T::QueueChangeHandler::on_queue_changed(origin.clone(), book_state.into()); Self::deposit_event(Event::PageReaped { origin: origin.clone(), index: page_index }); Ok(()) @@ -1035,11 +1036,7 @@ impl Pallet { } BookStateFor::::insert(&origin, &book_state); if total_processed > 0 { - T::QueueChangeHandler::on_queue_changed( - origin, - book_state.message_count, - book_state.size, - ); + T::QueueChangeHandler::on_queue_changed(origin, book_state.into()); } (total_processed > 0, next_ready) } @@ -1482,7 +1479,7 @@ impl EnqueueMessage> for Pallet { ) { Self::do_enqueue_message(&origin, message); let book_state = BookStateFor::::get(&origin); - T::QueueChangeHandler::on_queue_changed(origin, book_state.message_count, book_state.size); + T::QueueChangeHandler::on_queue_changed(origin, book_state.into()); } fn enqueue_messages<'a>( @@ -1493,7 +1490,7 @@ impl EnqueueMessage> for Pallet { Self::do_enqueue_message(&origin, message); } let book_state = BookStateFor::::get(&origin); - T::QueueChangeHandler::on_queue_changed(origin, book_state.message_count, book_state.size); + T::QueueChangeHandler::on_queue_changed(origin, book_state.into()); } fn sweep_queue(origin: MessageOriginOf) { @@ -1508,8 +1505,7 @@ impl EnqueueMessage> for Pallet { BookStateFor::::insert(&origin, &book_state); } - fn footprint(origin: MessageOriginOf) -> Footprint { - let book_state = BookStateFor::::get(&origin); - Footprint { count: book_state.message_count, size: book_state.size } + fn footprint(origin: MessageOriginOf) -> QueueFootprint { + BookStateFor::::get(&origin).into() } } diff --git a/substrate/frame/message-queue/src/mock.rs b/substrate/frame/message-queue/src/mock.rs index e6af0d9f1ee7..55a645743542 100644 --- a/substrate/frame/message-queue/src/mock.rs +++ b/substrate/frame/message-queue/src/mock.rs @@ -278,8 +278,8 @@ parameter_types! { /// Records all queue changes into [`QueueChanges`]. pub struct RecordingQueueChangeHandler; impl OnQueueChanged for RecordingQueueChangeHandler { - fn on_queue_changed(id: MessageOrigin, items_count: u64, items_size: u64) { - QueueChanges::mutate(|cs| cs.push((id, items_count, items_size))); + fn on_queue_changed(id: MessageOrigin, fp: QueueFootprint) { + QueueChanges::mutate(|cs| cs.push((id, fp.storage.count, fp.storage.size))); } } @@ -366,3 +366,7 @@ pub fn num_overweight_enqueued_events() -> u32 { }) .count() as u32 } + +pub fn fp(pages: u32, count: u64, size: u64) -> QueueFootprint { + QueueFootprint { storage: Footprint { count, size }, pages } +} diff --git a/substrate/frame/message-queue/src/tests.rs b/substrate/frame/message-queue/src/tests.rs index 5a235a8750e1..d94ad581ea0d 100644 --- a/substrate/frame/message-queue/src/tests.rs +++ b/substrate/frame/message-queue/src/tests.rs @@ -49,6 +49,7 @@ fn enqueue_within_one_page_works() { MessageQueue::enqueue_message(msg("c"), Here); assert_eq!(MessageQueue::service_queues(2.into_weight()), 2.into_weight()); assert_eq!(MessagesProcessed::take(), vec![(b"a".to_vec(), Here), (b"b".to_vec(), Here)]); + assert_eq!(MessageQueue::footprint(Here).pages, 1); assert_eq!(MessageQueue::service_queues(2.into_weight()), 1.into_weight()); assert_eq!(MessagesProcessed::take(), vec![(b"c".to_vec(), Here)]); @@ -314,6 +315,7 @@ fn reap_page_permanent_overweight_works() { MessageQueue::enqueue_message(msg("weight=2"), Here); } assert_eq!(Pages::::iter().count(), n); + assert_eq!(MessageQueue::footprint(Here).pages, n as u32); assert_eq!(QueueChanges::take().len(), n); // Mark all pages as stale since their message is permanently overweight. MessageQueue::service_queues(1.into_weight()); @@ -339,6 +341,7 @@ fn reap_page_permanent_overweight_works() { assert_noop!(MessageQueue::do_reap_page(&o, i), Error::::NotReapable); assert!(QueueChanges::take().is_empty()); } + assert_eq!(MessageQueue::footprint(Here).pages, 3); }); } @@ -1022,8 +1025,9 @@ fn footprint_works() { BookStateFor::::insert(origin, book); let info = MessageQueue::footprint(origin); - assert_eq!(info.count as usize, msgs); - assert_eq!(info.size, page.remaining_size as u64); + assert_eq!(info.storage.count as usize, msgs); + assert_eq!(info.storage.size, page.remaining_size as u64); + assert_eq!(info.pages, 1); // Sweeping a queue never calls OnQueueChanged. assert!(QueueChanges::take().is_empty()); @@ -1044,16 +1048,44 @@ fn footprint_invalid_works() { fn footprint_on_swept_works() { use MessageOrigin::*; build_and_execute::(|| { - let mut book = empty_book::(); - book.message_count = 3; - book.size = 10; - BookStateFor::::insert(Here, &book); - knit(&Here); + build_ring::(&[Here]); MessageQueue::sweep_queue(Here); let fp = MessageQueue::footprint(Here); - assert_eq!(fp.count, 3); - assert_eq!(fp.size, 10); + assert_eq!((1, 1, 1), (fp.storage.count, fp.storage.size, fp.pages)); + }) +} + +/// The number of reported pages takes overweight pages into account. +#[test] +fn footprint_num_pages_works() { + use MessageOrigin::*; + build_and_execute::(|| { + MessageQueue::enqueue_message(msg("weight=2"), Here); + MessageQueue::enqueue_message(msg("weight=3"), Here); + + assert_eq!(MessageQueue::footprint(Here), fp(2, 2, 16)); + + // Mark the messages as overweight. + assert_eq!(MessageQueue::service_queues(1.into_weight()), 0.into_weight()); + assert_eq!(System::events().len(), 2); + // Overweight does not change the footprint. + assert_eq!(MessageQueue::footprint(Here), fp(2, 2, 16)); + + // Now execute the second message. + assert_eq!( + ::execute_overweight(3.into_weight(), (Here, 1, 0)) + .unwrap(), + 3.into_weight() + ); + assert_eq!(MessageQueue::footprint(Here), fp(1, 1, 8)); + // And the first one: + assert_eq!( + ::execute_overweight(2.into_weight(), (Here, 0, 0)) + .unwrap(), + 2.into_weight() + ); + assert_eq!(MessageQueue::footprint(Here), Default::default()); }) } @@ -1143,6 +1175,7 @@ fn permanently_overweight_book_unknits() { assert_ring(&[]); assert_eq!(MessagesProcessed::take().len(), 0); assert_eq!(BookStateFor::::get(Here).message_count, 1); + assert_eq!(MessageQueue::footprint(Here).pages, 1); // Now if we enqueue another message, it will become ready again. MessageQueue::enqueue_messages([msg("weight=1")].into_iter(), Here); assert_ring(&[Here]); diff --git a/substrate/frame/safe-mode/Cargo.toml b/substrate/frame/safe-mode/Cargo.toml index 1b8054c51a8b..ac469bb385c9 100644 --- a/substrate/frame/safe-mode/Cargo.toml +++ b/substrate/frame/safe-mode/Cargo.toml @@ -30,6 +30,7 @@ sp-io = { path = "../../primitives/io" } pallet-balances = { path = "../balances" } pallet-utility = { path = "../utility" } pallet-proxy = { path = "../proxy" } +frame-support = { path = "../support", features = ["experimental"] } [features] default = [ "std" ] diff --git a/substrate/frame/safe-mode/src/tests.rs b/substrate/frame/safe-mode/src/tests.rs index ca1d7eb1d934..b92c5b87a530 100644 --- a/substrate/frame/safe-mode/src/tests.rs +++ b/substrate/frame/safe-mode/src/tests.rs @@ -22,32 +22,8 @@ use super::*; use crate::mock::{RuntimeCall, *}; -use frame_support::{assert_err, assert_noop, assert_ok, traits::Currency}; -use sp_runtime::{traits::Dispatchable, TransactionOutcome}; - -/// Do something hypothetically by rolling back any changes afterwards. -/// -/// Returns the original result of the closure. -macro_rules! hypothetically { - ( $e:expr ) => { - frame_support::storage::transactional::with_transaction( - || -> TransactionOutcome> { - sp_runtime::TransactionOutcome::Rollback(Ok($e)) - }, - ) - .expect("Always returning Ok; qed") - }; -} - -/// Assert something to be [*hypothetically*] `Ok` without actually committing it. -/// -/// Reverts any storage changes made by the closure. -macro_rules! hypothetically_ok { - ($e:expr $(, $args:expr)* $(,)?) => { - let result = hypothetically!($e); - assert_ok!(result $(, $args)*); - }; -} +use frame_support::{assert_err, assert_noop, assert_ok, hypothetically_ok, traits::Currency}; +use sp_runtime::traits::Dispatchable; #[test] fn fails_to_filter_calls_to_safe_mode_pallet() { diff --git a/substrate/frame/support/src/lib.rs b/substrate/frame/support/src/lib.rs index 4888b8996d1a..d49588686851 100644 --- a/substrate/frame/support/src/lib.rs +++ b/substrate/frame/support/src/lib.rs @@ -51,7 +51,9 @@ pub mod __private { pub use sp_metadata_ir as metadata_ir; #[cfg(feature = "std")] pub use sp_runtime::{bounded_btree_map, bounded_vec}; - pub use sp_runtime::{traits::Dispatchable, RuntimeDebug, StateVersion}; + pub use sp_runtime::{ + traits::Dispatchable, DispatchError, RuntimeDebug, StateVersion, TransactionOutcome, + }; #[cfg(feature = "std")] pub use sp_state_machine::BasicExternalities; pub use sp_std; @@ -781,6 +783,31 @@ macro_rules! assert_error_encoded_size { } => {}; } +/// Do something hypothetically by rolling back any changes afterwards. +/// +/// Returns the original result of the closure. +#[macro_export] +#[cfg(feature = "experimental")] +macro_rules! hypothetically { + ( $e:expr ) => { + $crate::storage::transactional::with_transaction(|| -> $crate::__private::TransactionOutcome> { + $crate::__private::TransactionOutcome::Rollback(Ok($e)) + }, + ).expect("Always returning Ok; qed") + }; +} + +/// Assert something to be *hypothetically* `Ok`, without actually committing it. +/// +/// Reverts any storage changes made by the closure. +#[macro_export] +#[cfg(feature = "experimental")] +macro_rules! hypothetically_ok { + ($e:expr $(, $args:expr)* $(,)?) => { + $crate::assert_ok!($crate::hypothetically!($e) $(, $args)*); + }; +} + #[doc(hidden)] pub use serde::{Deserialize, Serialize}; diff --git a/substrate/frame/support/src/traits.rs b/substrate/frame/support/src/traits.rs index 75c43d8cc508..4c692f848fde 100644 --- a/substrate/frame/support/src/traits.rs +++ b/substrate/frame/support/src/traits.rs @@ -113,7 +113,8 @@ pub use preimages::{Bounded, BoundedInline, FetchResult, QueryPreimage, StorePre mod messages; pub use messages::{ EnqueueMessage, EnqueueWithOrigin, ExecuteOverweightError, HandleMessage, NoopServiceQueues, - ProcessMessage, ProcessMessageError, QueuePausedQuery, ServiceQueues, TransformOrigin, + ProcessMessage, ProcessMessageError, QueueFootprint, QueuePausedQuery, ServiceQueues, + TransformOrigin, }; mod safe_mode; diff --git a/substrate/frame/support/src/traits/messages.rs b/substrate/frame/support/src/traits/messages.rs index 0db163e072b1..654380dbb0a8 100644 --- a/substrate/frame/support/src/traits/messages.rs +++ b/substrate/frame/support/src/traits/messages.rs @@ -116,6 +116,15 @@ impl ServiceQueues for NoopServiceQueues { } } +/// The resource footprint of a queue. +#[derive(Default, Copy, Clone, Eq, PartialEq, RuntimeDebug)] +pub struct QueueFootprint { + /// The number of pages in the queue (including overweight pages). + pub pages: u32, + /// The storage footprint of the queue (including overweight messages). + pub storage: Footprint, +} + /// Can enqueue messages for multiple origins. pub trait EnqueueMessage { /// The maximal length any enqueued message may have. @@ -134,7 +143,7 @@ pub trait EnqueueMessage { fn sweep_queue(origin: Origin); /// Return the state footprint of the given queue. - fn footprint(origin: Origin) -> Footprint; + fn footprint(origin: Origin) -> QueueFootprint; } impl EnqueueMessage for () { @@ -146,8 +155,8 @@ impl EnqueueMessage for () { ) { } fn sweep_queue(_: Origin) {} - fn footprint(_: Origin) -> Footprint { - Footprint::default() + fn footprint(_: Origin) -> QueueFootprint { + QueueFootprint::default() } } @@ -173,7 +182,7 @@ impl, O: MaxEncodedLen, N: MaxEncodedLen, C: Convert> E::sweep_queue(C::convert(origin)); } - fn footprint(origin: N) -> Footprint { + fn footprint(origin: N) -> QueueFootprint { E::footprint(C::convert(origin)) } } @@ -195,7 +204,7 @@ pub trait HandleMessage { fn sweep_queue(); /// Return the state footprint of the queue. - fn footprint() -> Footprint; + fn footprint() -> QueueFootprint; } /// Adapter type to transform an [`EnqueueMessage`] with an origin into a [`HandleMessage`] impl. @@ -220,7 +229,7 @@ where E::sweep_queue(O::get()); } - fn footprint() -> Footprint { + fn footprint() -> QueueFootprint { E::footprint(O::get()) } } diff --git a/substrate/primitives/core/Cargo.toml b/substrate/primitives/core/Cargo.toml index 9098b6e48ed9..fe0905c3bed4 100644 --- a/substrate/primitives/core/Cargo.toml +++ b/substrate/primitives/core/Cargo.toml @@ -14,7 +14,7 @@ targets = ["x86_64-unknown-linux-gnu"] [dependencies] codec = { package = "parity-scale-codec", version = "3.6.1", default-features = false, features = ["derive","max-encoded-len"] } -scale-info = { version = "2.10.0", default-features = false, features = ["derive"] } +scale-info = { version = "2.5.0", default-features = false, features = ["derive"] } log = { version = "0.4.17", default-features = false } serde = { version = "1.0.188", optional = true, default-features = false, features = ["derive", "alloc"] } bounded-collections = { version = "0.1.8", default-features = false }