General Message Queue Pallet (#12485)

* The message queue * Make fully generic * Refactor * Docs * Refactor * Use iter not slice * Per-origin queues * Multi-queue processing * Introduce MaxReady * Remove MaxReady in favour of ready ring * Cleanups * ReadyRing and tests * Stale page reaping * from_components -> from_parts Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * Move WeightCounter to sp_weights Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * Add MockedWeightInfo Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * Deploy to kitchensink Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * Use WeightCounter Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * Small fixes and logging Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * Add service_page Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * Typo Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * Move service_page below service_queue Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * Add service_message Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * Use correct weight function Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * Overweight execution * Refactor * Missing file * Fix WeightCounter usage in scheduler Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * Fix peek_index Take into account that decoding from a mutable slice modifies it. Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * Add tests and bench service_page_item Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * Add debug_info Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * Add no-progress check to service_queues Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * Add more benches Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * Bound from_message and try_append_message Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * Add PageReaped event Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * Rename BookStateOf and BookStateFor Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * Update tests and remove logging Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * Remove redundant per-message origins; add footprint() and sweep_queue() * Move testing stuff to mock.rs Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * Add integration test Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * Fix no-progress check Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * Fix debug_info Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * Fixup merge and tests Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * Fix footprint tracking * Introduce * Formatting * OverweightEnqueued event, auto-servicing config item * Update tests and benchmarks Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * Clippy Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * Add tests Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * Provide change handler * Add missing BookStateFor::insert and call QueueChangeHandler Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * Docs Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * Update benchmarks and weights Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * More tests... Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * Use weight metering functions Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * weightInfo::process_message_payload is gone Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * Add defensive_saturating_accrue Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * Rename WeightCounter to WeightMeter Ctr+Shift+H should do the trick. Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * Test on_initialize Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * Add module docs Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * Remove origin from MaxMessageLen The message origin is not encoded into the heap and does therefore not influence the max message length anymore. Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * Add BoundedVec::as_slice Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * Test Page::{from_message, try_append_message} Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * Fixup docs Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * Docs * Do nothing in sweep_queue if the queue does not exist ... otherwise it inserts default values into the storage. Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * Test ring (un)knitting Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * Upgrade stress-test Change the test to not assume that all queued messages will be processed in the next block but split it over multiple. Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * More tests... Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * Beauty fixes Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * clippy Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * Rename BoundedVec::as_slice to as_bounded_slice Conflicts with deref().as_slice() otherwise. Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * Fix imports Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * Remove ReadyRing struct Was used for testing only. Instead use 'fn assert_ring' which also check the service head and backlinks. Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * Beauty fixes Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * Fix stale page watermark Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * Cleanup Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * Fix test feature and clippy Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * QueueChanged handler is called correctly Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * Update benches Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * Abstract testing functions Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * More tests Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * Cleanup Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * Clippy Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * fmt Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * Simplify tests Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * Make stuff compile Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * Extend overweight execution benchmark Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * Remove TODOs Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * Test service queue with faulty MessageProcessor Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * fmt Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * Update pallet ui tests to 1.65 Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * More docs Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * Review doc fixes Co-authored-by: Robert Klotzner <eskimor@users.noreply.github.com> Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * Add weight_limit to extrinsic weight of execute_overweight * Correctly return unused weight * Return actual weight consumed in do_execute_overweight * Review fixes Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * Set version 7.0.0-dev Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * Make it compile Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * Switch message_size to u64 Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * Switch message_count to u64 Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * Fix benchmarks Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * Make CI green Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * Docs * Update tests Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * ".git/.scripts/bench-bot.sh" pallet dev pallet_message_queue * Dont mention README.md in the Cargo.toml Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> * Remove reference to readme Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> Co-authored-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io> Co-authored-by: parity-processbot <> Co-authored-by: Robert Klotzner <eskimor@users.noreply.github.com> Co-authored-by: Keith Yeung <kungfukeith11@gmail.com>
2026-05-01 06:37:56 +00:00 · 2022-12-09 10:38:24 +00:00
parent 487ed143df
commit 6f3d1a8143
20 changed files with 3883 additions and 15 deletions
@@ -0,0 +1,224 @@
+// Copyright 2022 Parity Technologies (UK) Ltd.
+// This file is part of Polkadot.
+
+// Polkadot is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// Polkadot is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+
+// You should have received a copy of the GNU General Public License
+// along with Polkadot.  If not, see <http://www.gnu.org/licenses/>.
+
+//! Stress tests pallet-message-queue. Defines its own runtime config to use larger constants for
+//! `HeapSize` and `MaxStale`.
+
+#![cfg(test)]
+
+use crate::{
+	mock::{
+		new_test_ext, CountingMessageProcessor, IntoWeight, MockedWeightInfo, NumMessagesProcessed,
+	},
+	*,
+};
+
+use crate as pallet_message_queue;
+use frame_support::{
+	parameter_types,
+	traits::{ConstU32, ConstU64},
+};
+use rand::{rngs::StdRng, Rng, SeedableRng};
+use rand_distr::Pareto;
+use sp_core::H256;
+use sp_runtime::{
+	testing::Header,
+	traits::{BlakeTwo256, IdentityLookup},
+};
+
+type UncheckedExtrinsic = frame_system::mocking::MockUncheckedExtrinsic<Test>;
+type Block = frame_system::mocking::MockBlock<Test>;
+
+frame_support::construct_runtime!(
+	pub enum Test where
+		Block = Block,
+		NodeBlock = Block,
+		UncheckedExtrinsic = UncheckedExtrinsic,
+	{
+		System: frame_system::{Pallet, Call, Config, Storage, Event<T>},
+		MessageQueue: pallet_message_queue::{Pallet, Call, Storage, Event<T>},
+	}
+);
+
+parameter_types! {
+	pub BlockWeights: frame_system::limits::BlockWeights =
+		frame_system::limits::BlockWeights::simple_max(frame_support::weights::Weight::from_ref_time(1024));
+}
+impl frame_system::Config for Test {
+	type BaseCallFilter = frame_support::traits::Everything;
+	type BlockWeights = ();
+	type BlockLength = ();
+	type DbWeight = ();
+	type RuntimeOrigin = RuntimeOrigin;
+	type Index = u64;
+	type BlockNumber = u64;
+	type Hash = H256;
+	type RuntimeCall = RuntimeCall;
+	type Hashing = BlakeTwo256;
+	type AccountId = u64;
+	type Lookup = IdentityLookup<Self::AccountId>;
+	type Header = Header;
+	type RuntimeEvent = RuntimeEvent;
+	type BlockHashCount = ConstU64<250>;
+	type Version = ();
+	type PalletInfo = PalletInfo;
+	type AccountData = ();
+	type OnNewAccount = ();
+	type OnKilledAccount = ();
+	type SystemWeightInfo = ();
+	type SS58Prefix = ();
+	type OnSetCode = ();
+	type MaxConsumers = ConstU32<16>;
+}
+
+parameter_types! {
+	pub const HeapSize: u32 = 32 * 1024;
+	pub const MaxStale: u32 = 32;
+	pub static ServiceWeight: Option<Weight> = Some(Weight::from_parts(100, 100));
+}
+
+impl Config for Test {
+	type RuntimeEvent = RuntimeEvent;
+	type WeightInfo = MockedWeightInfo;
+	type MessageProcessor = CountingMessageProcessor;
+	type Size = u32;
+	type QueueChangeHandler = ();
+	type HeapSize = HeapSize;
+	type MaxStale = MaxStale;
+	type ServiceWeight = ServiceWeight;
+}
+
+/// Simulates heavy usage by enqueueing and processing large amounts of messages.
+///
+/// Best to run with `-r`, `RUST_LOG=info` and `RUSTFLAGS='-Cdebug-assertions=y'`.
+///
+/// # Example output
+///
+/// ```pre
+/// Enqueued 1189 messages across 176 queues. Payload 46.97 KiB    
+/// Processing 772 of 1189 messages    
+/// Enqueued 9270 messages across 1559 queues. Payload 131.85 KiB    
+/// Processing 6262 of 9687 messages    
+/// Enqueued 5025 messages across 1225 queues. Payload 100.23 KiB    
+/// Processing 1739 of 8450 messages    
+/// Enqueued 42061 messages across 6357 queues. Payload 536.29 KiB    
+/// Processing 11675 of 48772 messages    
+/// Enqueued 20253 messages across 2420 queues. Payload 288.34 KiB    
+/// Processing 28711 of 57350 messages
+/// Processing all remaining 28639 messages
+/// ```
+#[test]
+#[ignore] // Only run in the CI.
+fn stress_test_enqueue_and_service() {
+	let blocks = 20;
+	let max_queues = 10_000;
+	let max_messages_per_queue = 10_000;
+	let max_msg_len = MaxMessageLenOf::<Test>::get();
+	let mut rng = StdRng::seed_from_u64(42);
+
+	new_test_ext::<Test>().execute_with(|| {
+		let mut msgs_remaining = 0;
+		for _ in 0..blocks {
+			// Start by enqueuing a large number of messages.
+			let (enqueued, _) =
+				enqueue_messages(max_queues, max_messages_per_queue, max_msg_len, &mut rng);
+			msgs_remaining += enqueued;
+
+			// Pick a fraction of all messages currently in queue and process them.
+			let processed = rng.gen_range(1..=msgs_remaining);
+			log::info!("Processing {} of all messages {}", processed, msgs_remaining);
+			process_messages(processed); // This also advances the block.
+			msgs_remaining -= processed;
+		}
+		log::info!("Processing all remaining {} messages", msgs_remaining);
+		process_messages(msgs_remaining);
+		post_conditions();
+	});
+}
+
+/// Enqueue a random number of random messages into a random number of queues.
+fn enqueue_messages(
+	max_queues: u32,
+	max_per_queue: u32,
+	max_msg_len: u32,
+	rng: &mut StdRng,
+) -> (u32, usize) {
+	let num_queues = rng.gen_range(1..max_queues);
+	let mut num_messages = 0;
+	let mut total_msg_len = 0;
+	for origin in 0..num_queues {
+		let num_messages_per_queue =
+			(rng.sample(Pareto::new(1.0, 1.1).unwrap()) as u32).min(max_per_queue);
+
+		for m in 0..num_messages_per_queue {
+			let mut message = format!("{}:{}", &origin, &m).into_bytes();
+			let msg_len = (rng.sample(Pareto::new(1.0, 1.0).unwrap()) as u32)
+				.clamp(message.len() as u32, max_msg_len);
+			message.resize(msg_len as usize, 0);
+			MessageQueue::enqueue_message(
+				BoundedSlice::defensive_truncate_from(&message),
+				origin.into(),
+			);
+			total_msg_len += msg_len;
+		}
+		num_messages += num_messages_per_queue;
+	}
+	log::info!(
+		"Enqueued {} messages across {} queues. Payload {:.2} KiB",
+		num_messages,
+		num_queues,
+		total_msg_len as f64 / 1024.0
+	);
+	(num_messages, total_msg_len as usize)
+}
+
+/// Process the number of messages.
+fn process_messages(num_msgs: u32) {
+	let weight = (num_msgs as u64).into_weight();
+	ServiceWeight::set(Some(weight));
+	let consumed = next_block();
+
+	assert_eq!(consumed, weight, "\n{}", MessageQueue::debug_info());
+	assert_eq!(NumMessagesProcessed::take(), num_msgs as usize);
+}
+
+/// Returns the weight consumed by `MessageQueue::on_initialize()`.
+fn next_block() -> Weight {
+	MessageQueue::on_finalize(System::block_number());
+	System::on_finalize(System::block_number());
+	System::set_block_number(System::block_number() + 1);
+	System::on_initialize(System::block_number());
+	MessageQueue::on_initialize(System::block_number())
+}
+
+/// Assert that the pallet is in the expected post state.
+fn post_conditions() {
+	// All queues are empty.
+	for (_, book) in BookStateFor::<Test>::iter() {
+		assert!(book.end >= book.begin);
+		assert_eq!(book.count, 0);
+		assert_eq!(book.size, 0);
+		assert_eq!(book.message_count, 0);
+		assert!(book.ready_neighbours.is_none());
+	}
+	// No pages remain.
+	assert_eq!(Pages::<Test>::iter().count(), 0);
+	// Service head is gone.
+	assert!(ServiceHead::<Test>::get().is_none());
+	// This still works fine.
+	assert_eq!(MessageQueue::service_queues(Weight::MAX), Weight::zero(), "Nothing left");
+	next_block();
+}