Introduce subsystem benchmarking tool

Cargo.toml

@@ -135,6 +135,7 @@ members = [
 	"polkadot/node/gum/proc-macro",
 	"polkadot/node/jaeger",
 	"polkadot/node/malus",
+	"polkadot/node/subsystem-bench",
 	"polkadot/node/metrics",
 	"polkadot/node/network/approval-distribution",
 	"polkadot/node/network/availability-distribution",

polkadot/node/network/availability-recovery/Cargo.toml

@@ -8,6 +8,7 @@ license.workspace = true
 
 [dependencies]
 futures = "0.3.21"
+tokio = "1.24.2"
 schnellru = "0.2.1"
 rand = "0.8.5"
 fatality = "0.0.6"
@@ -37,3 +38,6 @@ sc-network = { path = "../../../../substrate/client/network" }
 
 polkadot-node-subsystem-test-helpers = { path = "../../subsystem-test-helpers" }
 polkadot-primitives-test-helpers = { path = "../../../primitives/test-helpers" }
+
+[features]
+subsystem-benchmarks = []

polkadot/node/network/availability-recovery/src/lib.rs

@@ -65,7 +65,7 @@ mod error;
 mod futures_undead;
 mod metrics;
 mod task;
-use metrics::Metrics;
+pub use metrics::Metrics;
 
 #[cfg(test)]
 mod tests;
@@ -603,7 +603,8 @@ impl AvailabilityRecoverySubsystem {
 		}
 	}
 
-	async fn run<Context>(self, mut ctx: Context) -> SubsystemResult<()> {
+	/// Starts the inner subsystem loop.
+	pub async fn run<Context>(self, mut ctx: Context) -> SubsystemResult<()> {
 		let mut state = State::default();
 		let Self {
 			mut req_receiver,
@@ -681,6 +682,7 @@ impl AvailabilityRecoverySubsystem {
 							&mut state,
 							signal,
 						).await? {
+							gum::debug!(target: LOG_TARGET, "subsystem concluded");
 							return Ok(());
 						}
 						FromOrchestra::Communication { msg } => {
@@ -845,12 +847,17 @@ async fn erasure_task_thread(
 				let _ = sender.send(maybe_data);
 			},
 			None => {
-				gum::debug!(
+				gum::trace!(
 					target: LOG_TARGET,
 					"Erasure task channel closed. Node shutting down ?",
 				);
 				break
 			},
 		}
+
+		// In benchmarks this is a very hot loop not yielding at all.
+		// To update CPU metrics for the task we need to yield.
+		#[cfg(feature = "subsystem-benchmarks")]
+		tokio::task::yield_now().await;
 	}
 }

polkadot/node/network/availability-recovery/src/metrics.rs

@@ -29,7 +29,10 @@ struct MetricsInner {
 	///
 	/// Gets incremented on each sent chunk requests.
 	chunk_requests_issued: Counter<U64>,
-
+	/// Total number of bytes recovered
+	///
+	/// Gets incremented on each succesful recovery
+	recovered_bytes_total: Counter<U64>,
 	/// A counter for finished chunk requests.
 	///
 	/// Split by result:
@@ -133,9 +136,10 @@ impl Metrics {
 	}
 
 	/// A full recovery succeeded.
-	pub fn on_recovery_succeeded(&self) {
+	pub fn on_recovery_succeeded(&self, bytes: usize) {
 		if let Some(metrics) = &self.0 {
-			metrics.full_recoveries_finished.with_label_values(&["success"]).inc()
+			metrics.full_recoveries_finished.with_label_values(&["success"]).inc();
+			metrics.recovered_bytes_total.inc_by(bytes as u64)
 		}
 	}
 
@@ -171,6 +175,13 @@ impl metrics::Metrics for Metrics {
 				)?,
 				registry,
 			)?,
+			recovered_bytes_total: prometheus::register(
+				Counter::new(
+					"polkadot_parachain_availability_recovery_bytes_total",
+					"Total number of bytes recovered",
+				)?,
+				registry,
+			)?,
 			chunk_requests_finished: prometheus::register(
 				CounterVec::new(
 					Opts::new(

polkadot/node/network/availability-recovery/src/task.rs

@@ -23,6 +23,7 @@ use crate::{
 	PostRecoveryCheck, LOG_TARGET,
 };
 use futures::{channel::oneshot, SinkExt};
+use parity_scale_codec::Encode;
 #[cfg(not(test))]
 use polkadot_node_network_protocol::request_response::CHUNK_REQUEST_TIMEOUT;
 use polkadot_node_network_protocol::request_response::{
@@ -432,7 +433,7 @@ where
 					return Err(err)
 				},
 				Ok(data) => {
-					self.params.metrics.on_recovery_succeeded();
+					self.params.metrics.on_recovery_succeeded(data.encoded_size());
 					return Ok(data)
 				},
 			}

polkadot/node/network/availability-recovery/src/tests.rs

@@ -24,6 +24,7 @@ use parity_scale_codec::Encode;
 use polkadot_node_network_protocol::request_response::{
 	self as req_res, IncomingRequest, Recipient, ReqProtocolNames, Requests,
 };
+use polkadot_node_subsystem_test_helpers::derive_erasure_chunks_with_proofs_and_root;
 
 use super::*;
 
@@ -456,33 +457,6 @@ fn validator_authority_id(val_ids: &[Sr25519Keyring]) -> Vec<AuthorityDiscoveryI
 	val_ids.iter().map(|v| v.public().into()).collect()
 }
 
-fn derive_erasure_chunks_with_proofs_and_root(
-	n_validators: usize,
-	available_data: &AvailableData,
-	alter_chunk: impl Fn(usize, &mut Vec<u8>),
-) -> (Vec<ErasureChunk>, Hash) {
-	let mut chunks: Vec<Vec<u8>> = obtain_chunks(n_validators, available_data).unwrap();
-
-	for (i, chunk) in chunks.iter_mut().enumerate() {
-		alter_chunk(i, chunk)
-	}
-
-	// create proofs for each erasure chunk
-	let branches = branches(chunks.as_ref());
-
-	let root = branches.root();
-	let erasure_chunks = branches
-		.enumerate()
-		.map(|(index, (proof, chunk))| ErasureChunk {
-			chunk: chunk.to_vec(),
-			index: ValidatorIndex(index as _),
-			proof: Proof::try_from(proof).unwrap(),
-		})
-		.collect::<Vec<ErasureChunk>>();
-
-	(erasure_chunks, root)
-}
-
 impl Default for TestState {
 	fn default() -> Self {
 		let validators = vec![

polkadot/node/overseer/src/lib.rs

@@ -276,6 +276,7 @@ impl From<FinalityNotification<Block>> for BlockInfo {
 
 /// An event from outside the overseer scope, such
 /// as the substrate framework or user interaction.
+#[derive(Debug)]
 pub enum Event {
 	/// A new block was imported.
 	///
@@ -300,6 +301,7 @@ pub enum Event {
 }
 
 /// Some request from outer world.
+#[derive(Debug)]
 pub enum ExternalRequest {
 	/// Wait for the activation of a particular hash
 	/// and be notified by means of the return channel.

polkadot/node/subsystem-bench/Cargo.toml

@@ -0,0 +1,61 @@
+[package]
+name = "polkadot-subsystem-bench"
+description = "Subsystem performance benchmark client"
+version = "1.0.0"
+authors.workspace = true
+edition.workspace = true
+license.workspace = true
+readme = "README.md"
+publish = false
+
+[[bin]]
+name = "subsystem-bench"
+path = "src/subsystem-bench.rs"
+
+# Prevent rustdoc error. Already documented from top-level Cargo.toml.
+doc = false
+
+[dependencies]
+polkadot-node-subsystem = { path = "../subsystem" }
+polkadot-node-subsystem-util = { path = "../subsystem-util" }
+polkadot-node-subsystem-types = { path = "../subsystem-types" }
+polkadot-node-primitives = { path = "../primitives" }
+polkadot-primitives = { path = "../../primitives" }
+polkadot-node-network-protocol = { path = "../network/protocol" }
+polkadot-availability-recovery = { path = "../network/availability-recovery", features=["subsystem-benchmarks"]}
+color-eyre = { version = "0.6.1", default-features = false }
+polkadot-overseer =  { path = "../overseer" }
+colored = "2.0.4"
+assert_matches = "1.5"
+async-trait = "0.1.57"
+sp-keystore = { path = "../../../substrate/primitives/keystore" }
+sc-keystore = { path = "../../../substrate/client/keystore" }
+sp-core = { path = "../../../substrate/primitives/core" }
+clap = { version = "4.4.6", features = ["derive"] }
+futures = "0.3.21"
+futures-timer = "3.0.2"
+gum = { package = "tracing-gum", path = "../gum" }
+polkadot-erasure-coding = { package = "polkadot-erasure-coding", path = "../../erasure-coding" }
+log = "0.4.17"
+env_logger = "0.9.0"
+rand = "0.8.5"
+parity-scale-codec = { version = "3.6.1", features = ["std", "derive"] }
+tokio = "1.24.2"
+clap-num = "1.0.2"
+polkadot-node-subsystem-test-helpers = { path = "../subsystem-test-helpers" }
+sp-keyring = { path = "../../../substrate/primitives/keyring" }
+sp-application-crypto = { path = "../../../substrate/primitives/application-crypto" }
+sc-network = { path = "../../../substrate/client/network" }
+sc-service = { path = "../../../substrate/client/service" }
+polkadot-node-metrics = { path = "../metrics" }
+itertools = "0.11.0"
+polkadot-primitives-test-helpers = { path = "../../primitives/test-helpers" }
+prometheus_endpoint = { package = "substrate-prometheus-endpoint", path = "../../../substrate/utils/prometheus" }
+prometheus = { version = "0.13.0", default-features = false }
+serde = "1.0.192"
+serde_yaml = "0.9"
+paste = "1.0.14"
+orchestra = { version = "0.3.3", default-features = false, features=["futures_channel"] }
+
+[features]
+default = []