wip

polkadot/node/subsystem-bench/Cargo.toml

@@ -28,6 +28,7 @@ polkadot-primitives = { path = "../../primitives" }
 polkadot-node-network-protocol = { path = "../network/protocol" }
 polkadot-availability-recovery = { path = "../network/availability-recovery", features = ["subsystem-benchmarks"] }
 polkadot-availability-distribution = { path = "../network/availability-distribution" }
+polkadot-statement-distribution = { path = "../network/statement-distribution" }
 polkadot-node-core-av-store = { path = "../core/av-store" }
 polkadot-node-core-chain-api = { path = "../core/chain-api" }
 polkadot-availability-bitfield-distribution = { path = "../network/bitfield-distribution" }

polkadot/node/subsystem-bench/src/cli/subsystem-bench.rs

@@ -175,7 +175,7 @@ impl BenchCli {
 				},
 				TestObjective::StatementDistribution => {
 					let state = statement::TestState::new(&test_config);
-					let mut env = statement::prepare_test(&state, true);
+					let (mut env, _protocol_config) = statement::prepare_test(&state, true);
 					env.runtime().block_on(statement::benchmark_statement_distribution(
 						&benchmark_name,
 						&mut env,

polkadot/node/subsystem-bench/src/lib/mock/mod.rs

@@ -22,6 +22,7 @@ pub mod av_store;
 pub mod chain_api;
 pub mod dummy;
 pub mod network_bridge;
+pub mod prospective_parachains;
 pub mod runtime_api;
 
 pub struct AlwaysSupportsParachains {}

polkadot/node/subsystem-bench/src/lib/mock/prospective_parachains.rs

@@ -0,0 +1,72 @@
+// Copyright (C) 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/>.
+
+//! A generic runtime api subsystem mockup suitable to be used in benchmarks.
+
+use futures::FutureExt;
+use polkadot_node_subsystem::{
+	messages::ProspectiveParachainsMessage, overseer, SpawnedSubsystem, SubsystemError,
+};
+use polkadot_node_subsystem_types::OverseerSignal;
+
+const LOG_TARGET: &str = "subsystem-bench::prospective-parachains-mock";
+
+pub struct MockProspectiveParachains {}
+
+impl MockProspectiveParachains {
+	pub fn new() -> Self {
+		Self {}
+	}
+}
+
+#[overseer::subsystem(ProspectiveParachains, error=SubsystemError, prefix=self::overseer)]
+impl<Context> MockProspectiveParachains {
+	fn start(self, ctx: Context) -> SpawnedSubsystem {
+		let future = self.run(ctx).map(|_| Ok(())).boxed();
+
+		SpawnedSubsystem { name: "test-environment", future }
+	}
+}
+
+#[overseer::contextbounds(ProspectiveParachains, prefix = self::overseer)]
+impl MockProspectiveParachains {
+	async fn run<Context>(self, mut ctx: Context) {
+		loop {
+			let msg = ctx.recv().await.expect("Overseer never fails us");
+			match msg {
+				orchestra::FromOrchestra::Signal(signal) =>
+					if signal == OverseerSignal::Conclude {
+						return
+					},
+				orchestra::FromOrchestra::Communication { msg } => {
+					gum::debug!(target: LOG_TARGET, msg=?msg, "recv message");
+
+					match msg {
+						ProspectiveParachainsMessage::GetMinimumRelayParents(_relay_parent, tx) => {
+							tx.send(vec![]).unwrap();
+						},
+						ProspectiveParachainsMessage::GetHypotheticalFrontier(_req, tx) => {
+							tx.send(vec![]).unwrap();
+						},
+						_ => {
+							unimplemented!("Unexpected chain-api message")
+						},
+					}
+				},
+			}
+		}
+	}
+}

polkadot/node/subsystem-bench/src/lib/mock/runtime_api.rs

@@ -26,8 +26,8 @@ use polkadot_node_subsystem::{
 };
 use polkadot_node_subsystem_types::OverseerSignal;
 use polkadot_primitives::{
-	CandidateEvent, CandidateReceipt, CoreState, GroupIndex, IndexedVec, NodeFeatures,
-	OccupiedCore, SessionIndex, SessionInfo, ValidatorIndex,
+	AsyncBackingParams, CandidateEvent, CandidateReceipt, CoreState, GroupIndex, GroupRotationInfo,
+	IndexedVec, NodeFeatures, OccupiedCore, SessionIndex, SessionInfo, ValidatorIndex,
 };
 use sp_consensus_babe::Epoch as BabeEpoch;
 use sp_core::H256;
@@ -223,6 +223,43 @@ impl MockRuntimeApi {
 								.clone()
 								.expect("Babe epoch unpopulated")));
 						},
+						RuntimeApiMessage::Request(
+							_block_hash,
+							RuntimeApiRequest::AsyncBackingParams(sender),
+						) => {
+							let _ = sender.send(Ok(AsyncBackingParams {
+								max_candidate_depth: 1,
+								allowed_ancestry_len: 1,
+							}));
+						},
+						RuntimeApiMessage::Request(_parent, RuntimeApiRequest::Version(tx)) => {
+							tx.send(Ok(RuntimeApiRequest::DISABLED_VALIDATORS_RUNTIME_REQUIREMENT))
+								.unwrap();
+						},
+						RuntimeApiMessage::Request(
+							_parent,
+							RuntimeApiRequest::DisabledValidators(tx),
+						) => {
+							tx.send(Ok(vec![])).unwrap();
+						},
+						RuntimeApiMessage::Request(
+							_parent,
+							RuntimeApiRequest::MinimumBackingVotes(_session_index, tx),
+						) => {
+							tx.send(Ok(2)).unwrap();
+						},
+						RuntimeApiMessage::Request(
+							_parent,
+							RuntimeApiRequest::ValidatorGroups(tx),
+						) => {
+							let groups = self.session_info().validator_groups.to_vec();
+							let group_rotation_info = GroupRotationInfo {
+								session_start_block: 1,
+								group_rotation_frequency: 12,
+								now: 1,
+							};
+							tx.send(Ok((groups, group_rotation_info))).unwrap();
+						},
 						// Long term TODO: implement more as needed.
 						message => {
 							unimplemented!("Unexpected runtime-api message: {:?}", message)

polkadot/node/subsystem-bench/src/lib/statement/mod.rs

@@ -14,56 +14,120 @@
 // You should have received a copy of the GNU General Public License
 // along with Polkadot.  If not, see <http://www.gnu.org/licenses/>.
 
-use colored::Colorize;
-use polkadot_overseer::{
-	Handle as OverseerHandle, MetricsTrait, Overseer, OverseerConnector, OverseerMetrics, SpawnGlue,
-};
-use sc_service::SpawnTaskHandle;
-use std::time::Instant;
-pub use test_state::TestState;
-
-const LOG_TARGET: &str = "subsystem-bench::availability";
-
 use crate::{
 	dummy_builder,
-	environment::{TestEnvironment, TestEnvironmentDependencies},
-	mock::AlwaysSupportsParachains,
+	environment::{TestEnvironment, TestEnvironmentDependencies, GENESIS_HASH},
+	mock::{
+		chain_api::{ChainApiState, MockChainApi},
+		network_bridge::{MockNetworkBridgeRx, MockNetworkBridgeTx},
+		prospective_parachains::MockProspectiveParachains,
+		runtime_api::MockRuntimeApi,
+		AlwaysSupportsParachains,
+	},
 	network::new_network,
 	usage::BenchmarkUsage,
 };
+use colored::Colorize;
+use polkadot_node_metrics::metrics::Metrics;
+use polkadot_node_network_protocol::request_response::{IncomingRequest, ReqProtocolNames};
+use polkadot_node_primitives::{SignedFullStatementWithPVD, Statement};
+use polkadot_node_subsystem::messages::{AllMessages, StatementDistributionMessage};
+use polkadot_overseer::{
+	Handle as OverseerHandle, Overseer, OverseerConnector, OverseerMetrics, SpawnGlue,
+};
+use polkadot_primitives::{Block, Hash, SigningContext, ValidatorIndex, ValidatorPair};
+use polkadot_statement_distribution::StatementDistributionSubsystem;
+use rand::SeedableRng;
+use sc_keystore::LocalKeystore;
+use sc_network::request_responses::ProtocolConfig;
+use sc_service::SpawnTaskHandle;
+use sp_core::Pair;
+use std::{sync::Arc, time::Instant};
+pub use test_state::TestState;
 
 mod test_state;
 
-/// Helper function to build an overseer with the real implementation for `ApprovalDistribution` and
-/// `ApprovalVoting` subsystems and mock subsystems for all others.
+const LOG_TARGET: &str = "subsystem-bench::availability";
+
 fn build_overseer(
+	state: &TestState,
+	network_bridge: (MockNetworkBridgeTx, MockNetworkBridgeRx),
 	dependencies: &TestEnvironmentDependencies,
-) -> (Overseer<SpawnGlue<SpawnTaskHandle>, AlwaysSupportsParachains>, OverseerHandle) {
+) -> (
+	Overseer<SpawnGlue<SpawnTaskHandle>, AlwaysSupportsParachains>,
+	OverseerHandle,
+	Vec<ProtocolConfig>,
+) {
 	let overseer_connector = OverseerConnector::with_event_capacity(64000);
 	let overseer_metrics = OverseerMetrics::try_register(&dependencies.registry).unwrap();
 	let spawn_task_handle = dependencies.task_manager.spawn_handle();
-	let dummy = dummy_builder!(spawn_task_handle, overseer_metrics);
+	let mock_runtime_api = MockRuntimeApi::new(
+		state.config.clone(),
+		state.test_authorities.clone(),
+		state.candidate_receipts.clone(),
+		Default::default(),
+		Default::default(),
+		0,
+	);
+	let chain_api_state = ChainApiState { block_headers: state.block_headers.clone() };
+	let mock_chain_api = MockChainApi::new(chain_api_state);
+	let mock_prospective_parachains = MockProspectiveParachains::new();
+	let (statement_req_receiver, statement_req_cfg) = IncomingRequest::get_config_receiver::<
+		Block,
+		sc_network::NetworkWorker<Block, Hash>,
+	>(&ReqProtocolNames::new(GENESIS_HASH, None));
+	let (candidate_req_receiver, candidate_req_cfg) = IncomingRequest::get_config_receiver::<
+		Block,
+		sc_network::NetworkWorker<Block, Hash>,
+	>(&ReqProtocolNames::new(GENESIS_HASH, None));
+	let subsystem = StatementDistributionSubsystem::new(
+		Arc::new(LocalKeystore::in_memory()),
+		statement_req_receiver,
+		candidate_req_receiver,
+		Metrics::try_register(&dependencies.registry).unwrap(),
+		rand::rngs::StdRng::from_entropy(),
+	);
+	let dummy = dummy_builder!(spawn_task_handle, overseer_metrics)
+		.replace_runtime_api(|_| mock_runtime_api)
+		.replace_chain_api(|_| mock_chain_api)
+		.replace_prospective_parachains(|_| mock_prospective_parachains)
+		.replace_statement_distribution(|_| subsystem)
+		.replace_network_bridge_tx(|_| network_bridge.0)
+		.replace_network_bridge_rx(|_| network_bridge.1);
 	let (overseer, raw_handle) =
 		dummy.build_with_connector(overseer_connector).expect("Should not fail");
 	let overseer_handle = OverseerHandle::new(raw_handle);
 
-	(overseer, overseer_handle)
+	(overseer, overseer_handle, vec![statement_req_cfg, candidate_req_cfg])
 }
 
-pub fn prepare_test(state: &TestState, with_prometheus_endpoint: bool) -> TestEnvironment {
+pub fn prepare_test(
+	state: &TestState,
+	with_prometheus_endpoint: bool,
+) -> (TestEnvironment, Vec<ProtocolConfig>) {
 	let dependencies = TestEnvironmentDependencies::default();
-	let (network, _network_interface, _network_receiver) =
+	let (network, network_interface, network_receiver) =
 		new_network(&state.config, &dependencies, &state.test_authorities, vec![]);
-	let (overseer, overseer_handle) = build_overseer(&dependencies);
-
-	TestEnvironment::new(
-		dependencies,
-		state.config.clone(),
-		network,
-		overseer,
-		overseer_handle,
+	let network_bridge_tx = MockNetworkBridgeTx::new(
+		network.clone(),
+		network_interface.subsystem_sender(),
 		state.test_authorities.clone(),
-		with_prometheus_endpoint,
+	);
+	let network_bridge_rx = MockNetworkBridgeRx::new(network_receiver, None);
+	let (overseer, overseer_handle, cfg) =
+		build_overseer(state, (network_bridge_tx, network_bridge_rx), &dependencies);
+
+	(
+		TestEnvironment::new(
+			dependencies,
+			state.config.clone(),
+			network,
+			overseer,
+			overseer_handle,
+			state.test_authorities.clone(),
+			with_prometheus_endpoint,
+		),
+		cfg,
 	)
 }
 
@@ -77,12 +141,39 @@ pub async fn benchmark_statement_distribution(
 	env.metrics().set_n_validators(config.n_validators);
 	env.metrics().set_n_cores(config.n_cores);
 
+	let pair = ValidatorPair::generate().0;
+
 	let test_start = Instant::now();
 	for block_info in state.block_infos.iter() {
 		let block_num = block_info.number as usize;
 		gum::info!(target: LOG_TARGET, "Current block {}/{}", block_num, config.num_blocks);
 		env.metrics().set_current_block(block_num);
 		env.import_block(block_info.clone()).await;
+
+		let receipts = state
+			.commited_candidate_receipts
+			.get(&block_info.hash)
+			.expect("Pregenerated")
+			.clone();
+
+		for receipt in receipts {
+			let statement = Statement::Seconded(receipt);
+			let context = SigningContext { parent_hash: block_info.parent_hash, session_index: 0 };
+			let payload = statement.to_compact().signing_payload(&context);
+			let signature = pair.sign(&payload[..]);
+			let message = AllMessages::StatementDistribution(StatementDistributionMessage::Share(
+				block_info.hash,
+				SignedFullStatementWithPVD::new(
+					statement.supply_pvd(state.persisted_validation_data.clone()),
+					ValidatorIndex(0),
+					signature,
+					&context,
+					&pair.public(),
+				)
+				.unwrap(),
+			));
+			env.send_message(message).await;
+		}
 	}
 
 	let duration: u128 = test_start.elapsed().as_millis();