Add subsystem benchmarks for availability-distribution and biftield-distribution (availability write) (#2970)

Introduce a new test objective : `DataAvailabilityWrite`.

The new benchmark measures the network and cpu usage of
`availability-distribution`, `biftield-distribution` and
`availability-store` subsystems from the perspective of a validator node
during the process when candidates are made available.

Additionally I refactored the networking emulation to support bandwidth
acounting and limits of incoming and outgoing requests.

Screenshot of succesful run


<img width="1293" alt="Screenshot 2024-01-17 at 19 17 44"
src="https://github.com/paritytech/polkadot-sdk/assets/54316454/fde11280-e25b-4dc3-9dc9-d4b9752f9b7a">

---------

Signed-off-by: Andrei Sandu <andrei-mihail@parity.io>

polkadot/node/subsystem-bench/src/availability/av_store_helpers.rs

@@ -0,0 +1,57 @@
+// 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/>.
+
+use super::*;
+
+use polkadot_node_metrics::metrics::Metrics;
+
+use polkadot_node_core_av_store::Config;
+use polkadot_node_subsystem_util::database::Database;
+
+use polkadot_node_core_av_store::AvailabilityStoreSubsystem;
+
+mod columns {
+	pub const DATA: u32 = 0;
+	pub const META: u32 = 1;
+	pub const NUM_COLUMNS: u32 = 2;
+}
+
+const TEST_CONFIG: Config = Config { col_data: columns::DATA, col_meta: columns::META };
+
+struct DumbOracle;
+
+impl sp_consensus::SyncOracle for DumbOracle {
+	fn is_major_syncing(&self) -> bool {
+		false
+	}
+
+	fn is_offline(&self) -> bool {
+		unimplemented!("oh no!")
+	}
+}
+
+pub fn new_av_store(dependencies: &TestEnvironmentDependencies) -> AvailabilityStoreSubsystem {
+	let metrics = Metrics::try_register(&dependencies.registry).unwrap();
+
+	AvailabilityStoreSubsystem::new(test_store(), TEST_CONFIG, Box::new(DumbOracle), metrics)
+}
+
+fn test_store() -> Arc<dyn Database> {
+	let db = kvdb_memorydb::create(columns::NUM_COLUMNS);
+	let db =
+		polkadot_node_subsystem_util::database::kvdb_impl::DbAdapter::new(db, &[columns::META]);
+	Arc::new(db)
+}

polkadot/node/subsystem-bench/src/availability/mod.rs

@@ -13,25 +13,41 @@
 
 // You should have received a copy of the GNU General Public License
 // along with Polkadot.  If not, see <http://www.gnu.org/licenses/>.
+use crate::{core::mock::ChainApiState, TestEnvironment};
+use av_store::NetworkAvailabilityState;
+use bitvec::bitvec;
+use colored::Colorize;
 use itertools::Itertools;
+use polkadot_availability_bitfield_distribution::BitfieldDistribution;
+use polkadot_node_core_av_store::AvailabilityStoreSubsystem;
+use polkadot_node_subsystem::{Overseer, OverseerConnector, SpawnGlue};
+use polkadot_node_subsystem_types::{
+	messages::{AvailabilityStoreMessage, NetworkBridgeEvent},
+	Span,
+};
+use polkadot_overseer::Handle as OverseerHandle;
+use sc_network::{request_responses::ProtocolConfig, PeerId};
+use sp_core::H256;
 use std::{collections::HashMap, iter::Cycle, ops::Sub, sync::Arc, time::Instant};
 
-use crate::TestEnvironment;
-use polkadot_node_subsystem::{Overseer, OverseerConnector, SpawnGlue};
-use polkadot_node_subsystem_test_helpers::derive_erasure_chunks_with_proofs_and_root;
-use polkadot_overseer::Handle as OverseerHandle;
-use sc_network::request_responses::ProtocolConfig;
-
-use colored::Colorize;
-
+use av_store_helpers::new_av_store;
 use futures::{channel::oneshot, stream::FuturesUnordered, StreamExt};
+use polkadot_availability_distribution::{
+	AvailabilityDistributionSubsystem, IncomingRequestReceivers,
+};
 use polkadot_node_metrics::metrics::Metrics;
 
 use polkadot_availability_recovery::AvailabilityRecoverySubsystem;
+use polkadot_node_primitives::{AvailableData, ErasureChunk};
 
 use crate::GENESIS_HASH;
 use parity_scale_codec::Encode;
-use polkadot_node_network_protocol::request_response::{IncomingRequest, ReqProtocolNames};
+use polkadot_node_network_protocol::{
+	request_response::{v1::ChunkFetchingRequest, IncomingRequest, ReqProtocolNames},
+	OurView, Versioned, VersionedValidationProtocol,
+};
+use sc_network::request_responses::IncomingRequest as RawIncomingRequest;
+
 use polkadot_node_primitives::{BlockData, PoV};
 use polkadot_node_subsystem::messages::{AllMessages, AvailabilityRecoveryMessage};
 
@@ -39,8 +55,8 @@ use crate::core::{
 	environment::TestEnvironmentDependencies,
 	mock::{
 		av_store,
-		network_bridge::{self, MockNetworkBridgeTx, NetworkAvailabilityState},
-		runtime_api, MockAvailabilityStore, MockRuntimeApi,
+		network_bridge::{self, MockNetworkBridgeRx, MockNetworkBridgeTx},
+		runtime_api, MockAvailabilityStore, MockChainApi, MockRuntimeApi,
 	},
 };
 
@@ -48,24 +64,26 @@ use super::core::{configuration::TestConfiguration, mock::dummy_builder, network
 
 const LOG_TARGET: &str = "subsystem-bench::availability";
 
-use polkadot_node_primitives::{AvailableData, ErasureChunk};
-
 use super::{cli::TestObjective, core::mock::AlwaysSupportsParachains};
-use polkadot_node_subsystem_test_helpers::mock::new_block_import_info;
+use polkadot_node_subsystem_test_helpers::{
+	derive_erasure_chunks_with_proofs_and_root, mock::new_block_import_info,
+};
 use polkadot_primitives::{
-	CandidateHash, CandidateReceipt, GroupIndex, Hash, HeadData, PersistedValidationData,
+	AvailabilityBitfield, BlockNumber, CandidateHash, CandidateReceipt, GroupIndex, Hash, HeadData,
+	Header, PersistedValidationData, Signed, SigningContext, ValidatorIndex,
 };
 use polkadot_primitives_test_helpers::{dummy_candidate_receipt, dummy_hash};
 use sc_service::SpawnTaskHandle;
 
+mod av_store_helpers;
 mod cli;
 pub use cli::{DataAvailabilityReadOptions, NetworkEmulation};
 
-fn build_overseer(
+fn build_overseer_for_availability_read(
 	spawn_task_handle: SpawnTaskHandle,
 	runtime_api: MockRuntimeApi,
 	av_store: MockAvailabilityStore,
-	network_bridge: MockNetworkBridgeTx,
+	network_bridge: (MockNetworkBridgeTx, MockNetworkBridgeRx),
 	availability_recovery: AvailabilityRecoverySubsystem,
 ) -> (Overseer<SpawnGlue<SpawnTaskHandle>, AlwaysSupportsParachains>, OverseerHandle) {
 	let overseer_connector = OverseerConnector::with_event_capacity(64000);
@@ -73,7 +91,8 @@ fn build_overseer(
 	let builder = dummy
 		.replace_runtime_api(|_| runtime_api)
 		.replace_availability_store(|_| av_store)
-		.replace_network_bridge_tx(|_| network_bridge)
+		.replace_network_bridge_tx(|_| network_bridge.0)
+		.replace_network_bridge_rx(|_| network_bridge.1)
 		.replace_availability_recovery(|_| availability_recovery);
 
 	let (overseer, raw_handle) =
@@ -82,11 +101,38 @@ fn build_overseer(
 	(overseer, OverseerHandle::new(raw_handle))
 }
 
-/// Takes a test configuration and uses it to creates the `TestEnvironment`.
+fn build_overseer_for_availability_write(
+	spawn_task_handle: SpawnTaskHandle,
+	runtime_api: MockRuntimeApi,
+	network_bridge: (MockNetworkBridgeTx, MockNetworkBridgeRx),
+	availability_distribution: AvailabilityDistributionSubsystem,
+	chain_api: MockChainApi,
+	availability_store: AvailabilityStoreSubsystem,
+	bitfield_distribution: BitfieldDistribution,
+) -> (Overseer<SpawnGlue<SpawnTaskHandle>, AlwaysSupportsParachains>, OverseerHandle) {
+	let overseer_connector = OverseerConnector::with_event_capacity(64000);
+	let dummy = dummy_builder!(spawn_task_handle);
+	let builder = dummy
+		.replace_runtime_api(|_| runtime_api)
+		.replace_availability_store(|_| availability_store)
+		.replace_network_bridge_tx(|_| network_bridge.0)
+		.replace_network_bridge_rx(|_| network_bridge.1)
+		.replace_chain_api(|_| chain_api)
+		.replace_bitfield_distribution(|_| bitfield_distribution)
+		// This is needed to test own chunk recovery for `n_cores`.
+		.replace_availability_distribution(|_| availability_distribution);
+
+	let (overseer, raw_handle) =
+		builder.build_with_connector(overseer_connector).expect("Should not fail");
+
+	(overseer, OverseerHandle::new(raw_handle))
+}
+
+/// Takes a test configuration and uses it to create the `TestEnvironment`.
 pub fn prepare_test(
 	config: TestConfiguration,
 	state: &mut TestState,
-) -> (TestEnvironment, ProtocolConfig) {
+) -> (TestEnvironment, Vec<ProtocolConfig>) {
 	prepare_test_inner(config, state, TestEnvironmentDependencies::default())
 }
 
@@ -94,14 +140,38 @@ fn prepare_test_inner(
 	config: TestConfiguration,
 	state: &mut TestState,
 	dependencies: TestEnvironmentDependencies,
-) -> (TestEnvironment, ProtocolConfig) {
+) -> (TestEnvironment, Vec<ProtocolConfig>) {
 	// Generate test authorities.
 	let test_authorities = config.generate_authorities();
 
-	let runtime_api = runtime_api::MockRuntimeApi::new(config.clone(), test_authorities.clone());
+	let mut candidate_hashes: HashMap<H256, Vec<CandidateReceipt>> = HashMap::new();
 
-	let av_store =
-		av_store::MockAvailabilityStore::new(state.chunks.clone(), state.candidate_hashes.clone());
+	// Prepare per block candidates.
+	// Genesis block is always finalized, so we start at 1.
+	for block_num in 1..=config.num_blocks {
+		for _ in 0..config.n_cores {
+			candidate_hashes
+				.entry(Hash::repeat_byte(block_num as u8))
+				.or_default()
+				.push(state.next_candidate().expect("Cycle iterator"))
+		}
+
+		// First candidate is our backed candidate.
+		state.backed_candidates.push(
+			candidate_hashes
+				.get(&Hash::repeat_byte(block_num as u8))
+				.expect("just inserted above")
+				.get(0)
+				.expect("just inserted above")
+				.clone(),
+		);
+	}
+
+	let runtime_api = runtime_api::MockRuntimeApi::new(
+		config.clone(),
+		test_authorities.clone(),
+		candidate_hashes,
+	);
 
 	let availability_state = NetworkAvailabilityState {
 		candidate_hashes: state.candidate_hashes.clone(),
@@ -109,45 +179,112 @@ fn prepare_test_inner(
 		chunks: state.chunks.clone(),
 	};
 
-	let req_protocol_names = ReqProtocolNames::new(GENESIS_HASH, None);
-	let (collation_req_receiver, req_cfg) =
-		IncomingRequest::get_config_receiver(&req_protocol_names);
+	let mut req_cfgs = Vec::new();
 
-	let network =
-		NetworkEmulator::new(&config, &dependencies, &test_authorities, req_protocol_names);
+	let (collation_req_receiver, collation_req_cfg) =
+		IncomingRequest::get_config_receiver(&ReqProtocolNames::new(GENESIS_HASH, None));
+	req_cfgs.push(collation_req_cfg);
+
+	let (pov_req_receiver, pov_req_cfg) =
+		IncomingRequest::get_config_receiver(&ReqProtocolNames::new(GENESIS_HASH, None));
+
+	let (chunk_req_receiver, chunk_req_cfg) =
+		IncomingRequest::get_config_receiver(&ReqProtocolNames::new(GENESIS_HASH, None));
+	req_cfgs.push(pov_req_cfg);
+
+	let (network, network_interface, network_receiver) =
+		new_network(&config, &dependencies, &test_authorities, vec![Arc::new(availability_state)]);
 
 	let network_bridge_tx = network_bridge::MockNetworkBridgeTx::new(
-		config.clone(),
-		availability_state,
 		network.clone(),
+		network_interface.subsystem_sender(),
 	);
 
-	let use_fast_path = match &state.config().objective {
-		TestObjective::DataAvailabilityRead(options) => options.fetch_from_backers,
-		_ => panic!("Unexpected objective"),
+	let network_bridge_rx =
+		network_bridge::MockNetworkBridgeRx::new(network_receiver, Some(chunk_req_cfg.clone()));
+
+	let (overseer, overseer_handle) = match &state.config().objective {
+		TestObjective::DataAvailabilityRead(options) => {
+			let use_fast_path = options.fetch_from_backers;
+
+			let subsystem = if use_fast_path {
+				AvailabilityRecoverySubsystem::with_fast_path(
+					collation_req_receiver,
+					Metrics::try_register(&dependencies.registry).unwrap(),
+				)
+			} else {
+				AvailabilityRecoverySubsystem::with_chunks_only(
+					collation_req_receiver,
+					Metrics::try_register(&dependencies.registry).unwrap(),
+				)
+			};
+
+			// Use a mocked av-store.
+			let av_store = av_store::MockAvailabilityStore::new(
+				state.chunks.clone(),
+				state.candidate_hashes.clone(),
+			);
+
+			build_overseer_for_availability_read(
+				dependencies.task_manager.spawn_handle(),
+				runtime_api,
+				av_store,
+				(network_bridge_tx, network_bridge_rx),
+				subsystem,
+			)
+		},
+		TestObjective::DataAvailabilityWrite => {
+			let availability_distribution = AvailabilityDistributionSubsystem::new(
+				test_authorities.keyring.keystore(),
+				IncomingRequestReceivers { pov_req_receiver, chunk_req_receiver },
+				Metrics::try_register(&dependencies.registry).unwrap(),
+			);
+
+			let block_headers = (0..=config.num_blocks)
+				.map(|block_number| {
+					(
+						Hash::repeat_byte(block_number as u8),
+						Header {
+							digest: Default::default(),
+							number: block_number as BlockNumber,
+							parent_hash: Default::default(),
+							extrinsics_root: Default::default(),
+							state_root: Default::default(),
+						},
+					)
+				})
+				.collect::<HashMap<_, _>>();
+
+			let chain_api_state = ChainApiState { block_headers };
+			let chain_api = MockChainApi::new(chain_api_state);
+			let bitfield_distribution =
+				BitfieldDistribution::new(Metrics::try_register(&dependencies.registry).unwrap());
+			build_overseer_for_availability_write(
+				dependencies.task_manager.spawn_handle(),
+				runtime_api,
+				(network_bridge_tx, network_bridge_rx),
+				availability_distribution,
+				chain_api,
+				new_av_store(&dependencies),
+				bitfield_distribution,
+			)
+		},
+		_ => {
+			unimplemented!("Invalid test objective")
+		},
 	};
 
-	let subsystem = if use_fast_path {
-		AvailabilityRecoverySubsystem::with_fast_path(
-			collation_req_receiver,
-			Metrics::try_register(&dependencies.registry).unwrap(),
-		)
-	} else {
-		AvailabilityRecoverySubsystem::with_chunks_only(
-			collation_req_receiver,
-			Metrics::try_register(&dependencies.registry).unwrap(),
-		)
-	};
-
-	let (overseer, overseer_handle) = build_overseer(
-		dependencies.task_manager.spawn_handle(),
-		runtime_api,
-		av_store,
-		network_bridge_tx,
-		subsystem,
-	);
-
-	(TestEnvironment::new(dependencies, config, network, overseer, overseer_handle), req_cfg)
+	(
+		TestEnvironment::new(
+			dependencies,
+			config,
+			network,
+			overseer,
+			overseer_handle,
+			test_authorities,
+		),
+		req_cfgs,
+	)
 }
 
 #[derive(Clone)]
@@ -169,6 +306,8 @@ pub struct TestState {
 	available_data: Vec<AvailableData>,
 	// Per candiadte index chunks
 	chunks: Vec<Vec<ErasureChunk>>,
+	// Per relay chain block - candidate backed by our backing group
+	backed_candidates: Vec<CandidateReceipt>,
 }
 
 impl TestState {
@@ -255,24 +394,27 @@ impl TestState {
 			candidate_receipt_templates.push(candidate_receipt);
 		}
 
-		let pov_sizes = config.pov_sizes().to_owned();
-		let pov_sizes = pov_sizes.into_iter().cycle();
 		gum::info!(target: LOG_TARGET, "{}","Created test environment.".bright_blue());
 
 		let mut _self = Self {
-			config,
 			available_data,
 			candidate_receipt_templates,
 			chunks,
 			pov_size_to_candidate,
-			pov_sizes,
+			pov_sizes: Vec::from(config.pov_sizes()).into_iter().cycle(),
 			candidate_hashes: HashMap::new(),
 			candidates: Vec::new().into_iter().cycle(),
+			backed_candidates: Vec::new(),
+			config,
 		};
 
 		_self.generate_candidates();
 		_self
 	}
+
+	pub fn backed_candidates(&mut self) -> &mut Vec<CandidateReceipt> {
+		&mut self.backed_candidates
+	}
 }
 
 pub async fn benchmark_availability_read(env: &mut TestEnvironment, mut state: TestState) {
@@ -280,15 +422,15 @@ pub async fn benchmark_availability_read(env: &mut TestEnvironment, mut state: T
 
 	env.import_block(new_block_import_info(Hash::repeat_byte(1), 1)).await;
 
-	let start_marker = Instant::now();
+	let test_start = Instant::now();
 	let mut batch = FuturesUnordered::new();
 	let mut availability_bytes = 0u128;
 
 	env.metrics().set_n_validators(config.n_validators);
 	env.metrics().set_n_cores(config.n_cores);
 
-	for block_num in 0..env.config().num_blocks {
-		gum::info!(target: LOG_TARGET, "Current block {}/{}", block_num + 1, env.config().num_blocks);
+	for block_num in 1..=env.config().num_blocks {
+		gum::info!(target: LOG_TARGET, "Current block {}/{}", block_num, env.config().num_blocks);
 		env.metrics().set_current_block(block_num);
 
 		let block_start_ts = Instant::now();
@@ -311,7 +453,7 @@ pub async fn benchmark_availability_read(env: &mut TestEnvironment, mut state: T
 			env.send_message(message).await;
 		}
 
-		gum::info!("{}", format!("{} recoveries pending", batch.len()).bright_black());
+		gum::info!(target: LOG_TARGET, "{}", format!("{} recoveries pending", batch.len()).bright_black());
 		while let Some(completed) = batch.next().await {
 			let available_data = completed.unwrap().unwrap();
 			env.metrics().on_pov_size(available_data.encoded_size());
@@ -320,22 +462,199 @@ pub async fn benchmark_availability_read(env: &mut TestEnvironment, mut state: T
 
 		let block_time = Instant::now().sub(block_start_ts).as_millis() as u64;
 		env.metrics().set_block_time(block_time);
-		gum::info!("All work for block completed in {}", format!("{:?}ms", block_time).cyan());
+		gum::info!(target: LOG_TARGET, "All work for block completed in {}", format!("{:?}ms", block_time).cyan());
 	}
 
-	let duration: u128 = start_marker.elapsed().as_millis();
+	let duration: u128 = test_start.elapsed().as_millis();
 	let availability_bytes = availability_bytes / 1024;
-	gum::info!("All blocks processed in {}", format!("{:?}ms", duration).cyan());
-	gum::info!(
+	gum::info!(target: LOG_TARGET, "All blocks processed in {}", format!("{:?}ms", duration).cyan());
+	gum::info!(target: LOG_TARGET,
 		"Throughput: {}",
 		format!("{} KiB/block", availability_bytes / env.config().num_blocks as u128).bright_red()
 	);
-	gum::info!(
-		"Block time: {}",
-		format!("{} ms", start_marker.elapsed().as_millis() / env.config().num_blocks as u128)
-			.red()
+	gum::info!(target: LOG_TARGET,
+		"Avg block time: {}",
+		format!("{} ms", test_start.elapsed().as_millis() / env.config().num_blocks as u128).red()
 	);
 
-	gum::info!("{}", &env);
+	env.display_network_usage();
+	env.display_cpu_usage(&["availability-recovery"]);
 	env.stop().await;
 }
+
+pub async fn benchmark_availability_write(env: &mut TestEnvironment, mut state: TestState) {
+	let config = env.config().clone();
+
+	env.metrics().set_n_validators(config.n_validators);
+	env.metrics().set_n_cores(config.n_cores);
+
+	gum::info!(target: LOG_TARGET, "Seeding availability store with candidates ...");
+	for backed_candidate in state.backed_candidates().clone() {
+		let candidate_index = *state.candidate_hashes.get(&backed_candidate.hash()).unwrap();
+		let available_data = state.available_data[candidate_index].clone();
+		let (tx, rx) = oneshot::channel();
+		env.send_message(AllMessages::AvailabilityStore(
+			AvailabilityStoreMessage::StoreAvailableData {
+				candidate_hash: backed_candidate.hash(),
+				n_validators: config.n_validators as u32,
+				available_data,
+				expected_erasure_root: backed_candidate.descriptor().erasure_root,
+				tx,
+			},
+		))
+		.await;
+
+		rx.await
+			.unwrap()
+			.expect("Test candidates are stored nicely in availability store");
+	}
+
+	gum::info!(target: LOG_TARGET, "Done");
+
+	let test_start = Instant::now();
+
+	for block_num in 1..=env.config().num_blocks {
+		gum::info!(target: LOG_TARGET, "Current block #{}", block_num);
+		env.metrics().set_current_block(block_num);
+
+		let block_start_ts = Instant::now();
+		let relay_block_hash = Hash::repeat_byte(block_num as u8);
+		env.import_block(new_block_import_info(relay_block_hash, block_num as BlockNumber))
+			.await;
+
+		// Inform bitfield distribution about our view of current test block
+		let message = polkadot_node_subsystem_types::messages::BitfieldDistributionMessage::NetworkBridgeUpdate(
+			NetworkBridgeEvent::OurViewChange(OurView::new(vec![(relay_block_hash, Arc::new(Span::Disabled))], 0))
+		);
+		env.send_message(AllMessages::BitfieldDistribution(message)).await;
+
+		let chunk_fetch_start_ts = Instant::now();
+
+		// Request chunks of our own backed candidate from all other validators.
+		let mut receivers = Vec::new();
+		for index in 1..config.n_validators {
+			let (pending_response, pending_response_receiver) = oneshot::channel();
+
+			let request = RawIncomingRequest {
+				peer: PeerId::random(),
+				payload: ChunkFetchingRequest {
+					candidate_hash: state.backed_candidates()[block_num - 1].hash(),
+					index: ValidatorIndex(index as u32),
+				}
+				.encode(),
+				pending_response,
+			};
+
+			let peer = env
+				.authorities()
+				.validator_authority_id
+				.get(index)
+				.expect("all validators have keys");
+
+			if env.network().is_peer_connected(peer) &&
+				env.network().send_request_from_peer(peer, request).is_ok()
+			{
+				receivers.push(pending_response_receiver);
+			}
+		}
+
+		gum::info!(target: LOG_TARGET, "Waiting for all emulated peers to receive their chunk from us ...");
+		for receiver in receivers.into_iter() {
+			let response = receiver.await.expect("Chunk is always served succesfully");
+			// TODO: check if chunk is the one the peer expects to receive.
+			assert!(response.result.is_ok());
+		}
+
+		let chunk_fetch_duration = Instant::now().sub(chunk_fetch_start_ts).as_millis();
+
+		gum::info!(target: LOG_TARGET, "All chunks received in {}ms", chunk_fetch_duration);
+
+		let signing_context = SigningContext { session_index: 0, parent_hash: relay_block_hash };
+		let network = env.network().clone();
+		let authorities = env.authorities().clone();
+		let n_validators = config.n_validators;
+
+		// Spawn a task that will generate `n_validator` - 1 signed bitfiends and
+		// send them from the emulated peers to the subsystem.
+		// TODO: Implement topology.
+		env.spawn_blocking("send-bitfields", async move {
+			for index in 1..n_validators {
+				let validator_public =
+					authorities.validator_public.get(index).expect("All validator keys are known");
+
+				// Node has all the chunks in the world.
+				let payload: AvailabilityBitfield =
+					AvailabilityBitfield(bitvec![u8, bitvec::order::Lsb0; 1u8; 32]);
+				// TODO(soon): Use pre-signed messages. This is quite intensive on the CPU.
+				let signed_bitfield = Signed::<AvailabilityBitfield>::sign(
+					&authorities.keyring.keystore(),
+					payload,
+					&signing_context,
+					ValidatorIndex(index as u32),
+					validator_public,
+				)
+				.ok()
+				.flatten()
+				.expect("should be signed");
+
+				let from_peer = &authorities.validator_authority_id[index];
+
+				let message = peer_bitfield_message_v2(relay_block_hash, signed_bitfield);
+
+				// Send the action from peer only if it is connected to our node.
+				if network.is_peer_connected(from_peer) {
+					let _ = network.send_message_from_peer(from_peer, message);
+				}
+			}
+		});
+
+		gum::info!(
+			"Waiting for {} bitfields to be received and processed",
+			config.connected_count()
+		);
+
+		// Wait for all bitfields to be processed.
+		env.wait_until_metric_eq(
+			"polkadot_parachain_received_availabilty_bitfields_total",
+			config.connected_count() * block_num,
+		)
+		.await;
+
+		gum::info!(target: LOG_TARGET, "All bitfields processed");
+
+		let block_time = Instant::now().sub(block_start_ts).as_millis() as u64;
+		env.metrics().set_block_time(block_time);
+		gum::info!(target: LOG_TARGET, "All work for block completed in {}", format!("{:?}ms", block_time).cyan());
+	}
+
+	let duration: u128 = test_start.elapsed().as_millis();
+	gum::info!(target: LOG_TARGET, "All blocks processed in {}", format!("{:?}ms", duration).cyan());
+	gum::info!(target: LOG_TARGET,
+		"Avg block time: {}",
+		format!("{} ms", test_start.elapsed().as_millis() / env.config().num_blocks as u128).red()
+	);
+
+	env.display_network_usage();
+
+	env.display_cpu_usage(&[
+		"availability-distribution",
+		"bitfield-distribution",
+		"availability-store",
+	]);
+
+	env.stop().await;
+}
+
+pub fn peer_bitfield_message_v2(
+	relay_hash: H256,
+	signed_bitfield: Signed<AvailabilityBitfield>,
+) -> VersionedValidationProtocol {
+	let bitfield = polkadot_node_network_protocol::v2::BitfieldDistributionMessage::Bitfield(
+		relay_hash,
+		signed_bitfield.into(),
+	);
+
+	Versioned::V2(polkadot_node_network_protocol::v2::ValidationProtocol::BitfieldDistribution(
+		bitfield,
+	))
+}