Transaction pool: Adds benchmark and improves performance (#9958)

* Yep

* Make it compile

* Make the benchmark work

* Some stuff

* Optimize transaction pool `BestIterator`

* Some docs

* Fix more warnings

* Fix compilation

* FMT

substrate/bin/node/cli/Cargo.toml

@@ -46,6 +46,7 @@ structopt = { version = "0.3.8", optional = true }
 sp-authority-discovery = { version = "4.0.0-dev", path = "../../../primitives/authority-discovery" }
 sp-consensus-babe = { version = "0.10.0-dev", path = "../../../primitives/consensus/babe" }
 grandpa-primitives = { version = "4.0.0-dev", package = "sp-finality-grandpa", path = "../../../primitives/finality-grandpa" }
+sp-api = { version = "4.0.0-dev", path = "../../../primitives/api" }
 sp-core = { version = "4.0.0-dev", path = "../../../primitives/core" }
 sp-runtime = { version = "4.0.0-dev", path = "../../../primitives/runtime" }
 sp-timestamp = { version = "4.0.0-dev", path = "../../../primitives/timestamp" }
@@ -78,6 +79,7 @@ sc-sync-state-rpc = { version = "0.10.0-dev", path = "../../../client/sync-state
 
 # frame dependencies
 frame-system = { version = "4.0.0-dev", path = "../../../frame/system" }
+frame-system-rpc-runtime-api = { version = "4.0.0-dev", path = "../../../frame/system/rpc/runtime-api" }
 pallet-transaction-payment = { version = "4.0.0-dev", path = "../../../frame/transaction-payment" }
 pallet-im-online = { version = "4.0.0-dev", default-features = false, path = "../../../frame/im-online" }
 
@@ -124,8 +126,9 @@ regex = "1"
 platforms = "1.1"
 async-std = { version = "1.10.0", features = ["attributes"] }
 soketto = "0.4.2"
-jsonrpsee-ws-client = { version = "0.3.0", default-features = false, features = ["tokio1"] }
+criterion = { version = "0.3.5", features = [ "async_tokio" ] }
 tokio = { version = "1.10", features = ["macros", "time"] }
+jsonrpsee-ws-client = { version = "0.3.0", default-features = false, features = ["tokio1"] }
 wait-timeout = "0.2"
 remote-externalities = { path = "../../../utils/frame/remote-externalities" }
 
@@ -137,6 +140,7 @@ substrate-build-script-utils = { version = "3.0.0", optional = true, path = "../
 substrate-frame-cli = { version = "4.0.0-dev", optional = true, path = "../../../utils/frame/frame-utilities-cli" }
 try-runtime-cli = { version = "0.10.0-dev", optional = true, path = "../../../utils/frame/try-runtime/cli" }
 sc-cli = { version = "0.10.0-dev", path = "../../../client/cli", optional = true }
+pallet-balances = { version = "4.0.0-dev", path = "../../../frame/balances" }
 
 [features]
 default = ["cli"]
@@ -158,3 +162,7 @@ runtime-benchmarks = [
 # Enable features that allow the runtime to be tried and debugged. Name might be subject to change
 # in the near future.
 try-runtime = ["node-runtime/try-runtime", "try-runtime-cli"]
+
+[[bench]]
+name = "transaction_pool"
+harness = false

substrate/bin/node/cli/benches/transaction_pool.rs

@@ -0,0 +1,274 @@
+// This file is part of Substrate.
+
+// Copyright (C) 2021 Parity Technologies (UK) Ltd.
+// SPDX-License-Identifier: GPL-3.0-or-later WITH Classpath-exception-2.0
+
+// This program 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.
+
+// This program 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 this program. If not, see <https://www.gnu.org/licenses/>.
+
+use criterion::{criterion_group, criterion_main, BatchSize, Criterion, Throughput};
+use futures::{future, StreamExt};
+use node_cli::service::{create_extrinsic, fetch_nonce, FullClient, TransactionPool};
+use node_primitives::AccountId;
+use node_runtime::{constants::currency::*, BalancesCall, SudoCall};
+use sc_client_api::execution_extensions::ExecutionStrategies;
+use sc_service::{
+	config::{
+		DatabaseSource, KeepBlocks, KeystoreConfig, NetworkConfiguration, OffchainWorkerConfig,
+		PruningMode, TransactionPoolOptions, TransactionStorageMode, WasmExecutionMethod,
+	},
+	BasePath, Configuration, Role,
+};
+use sc_transaction_pool::PoolLimit;
+use sc_transaction_pool_api::{TransactionPool as _, TransactionSource, TransactionStatus};
+use sp_core::{crypto::Pair, sr25519};
+use sp_keyring::Sr25519Keyring;
+use sp_runtime::{generic::BlockId, OpaqueExtrinsic};
+use tokio::runtime::Handle;
+
+fn new_node(tokio_handle: Handle) -> node_cli::service::NewFullBase {
+	let base_path = BasePath::new_temp_dir().expect("Creates base path");
+	let root = base_path.path().to_path_buf();
+
+	let network_config = NetworkConfiguration::new(
+		Sr25519Keyring::Alice.to_seed(),
+		"network/test/0.1",
+		Default::default(),
+		None,
+	);
+
+	let spec = Box::new(node_cli::chain_spec::development_config());
+
+	let config = Configuration {
+		impl_name: "BenchmarkImpl".into(),
+		impl_version: "1.0".into(),
+		role: Role::Authority,
+		tokio_handle,
+		transaction_pool: TransactionPoolOptions {
+			ready: PoolLimit { count: 100_000, total_bytes: 100 * 1024 * 1024 },
+			future: PoolLimit { count: 100_000, total_bytes: 100 * 1024 * 1024 },
+			reject_future_transactions: false,
+		},
+		network: network_config,
+		keystore: KeystoreConfig::InMemory,
+		keystore_remote: Default::default(),
+		database: DatabaseSource::RocksDb { path: root.join("db"), cache_size: 128 },
+		state_cache_size: 67108864,
+		state_cache_child_ratio: None,
+		state_pruning: PruningMode::ArchiveAll,
+		keep_blocks: KeepBlocks::All,
+		transaction_storage: TransactionStorageMode::BlockBody,
+		chain_spec: spec,
+		wasm_method: WasmExecutionMethod::Interpreted,
+		// NOTE: we enforce the use of the native runtime to make the errors more debuggable
+		execution_strategies: ExecutionStrategies {
+			syncing: sc_client_api::ExecutionStrategy::NativeWhenPossible,
+			importing: sc_client_api::ExecutionStrategy::NativeWhenPossible,
+			block_construction: sc_client_api::ExecutionStrategy::NativeWhenPossible,
+			offchain_worker: sc_client_api::ExecutionStrategy::NativeWhenPossible,
+			other: sc_client_api::ExecutionStrategy::NativeWhenPossible,
+		},
+		rpc_http: None,
+		rpc_ws: None,
+		rpc_ipc: None,
+		rpc_ws_max_connections: None,
+		rpc_cors: None,
+		rpc_methods: Default::default(),
+		rpc_max_payload: None,
+		prometheus_config: None,
+		telemetry_endpoints: None,
+		default_heap_pages: None,
+		offchain_worker: OffchainWorkerConfig { enabled: true, indexing_enabled: false },
+		force_authoring: false,
+		disable_grandpa: false,
+		dev_key_seed: Some(Sr25519Keyring::Alice.to_seed()),
+		tracing_targets: None,
+		tracing_receiver: Default::default(),
+		max_runtime_instances: 8,
+		announce_block: true,
+		base_path: Some(base_path),
+		informant_output_format: Default::default(),
+		wasm_runtime_overrides: None,
+		disable_log_reloading: false,
+	};
+
+	node_cli::service::new_full_base(config, |_, _| ()).expect("Creates node")
+}
+
+fn create_accounts(num: usize) -> Vec<sr25519::Pair> {
+	(0..num)
+		.map(|i| {
+			Pair::from_string(&format!("{}/{}", Sr25519Keyring::Alice.to_seed(), i), None)
+				.expect("Creates account pair")
+		})
+		.collect()
+}
+
+/// Create the extrinsics that will initialize the accounts from the sudo account (Alice).
+///
+/// `start_nonce` is the current nonce of Alice.
+fn create_account_extrinsics(
+	client: &FullClient,
+	accounts: &[sr25519::Pair],
+) -> Vec<OpaqueExtrinsic> {
+	let start_nonce = fetch_nonce(client, Sr25519Keyring::Alice.pair());
+
+	accounts
+		.iter()
+		.enumerate()
+		.map(|(i, a)| {
+			vec![
+				// Reset the nonce by removing any funds
+				create_extrinsic(
+					client,
+					Sr25519Keyring::Alice.pair(),
+					SudoCall::sudo {
+						call: Box::new(
+							BalancesCall::set_balance {
+								who: AccountId::from(a.public()).into(),
+								new_free: 0,
+								new_reserved: 0,
+							}
+							.into(),
+						),
+					},
+					Some(start_nonce + (i as u32) * 2),
+				),
+				// Give back funds
+				create_extrinsic(
+					client,
+					Sr25519Keyring::Alice.pair(),
+					SudoCall::sudo {
+						call: Box::new(
+							BalancesCall::set_balance {
+								who: AccountId::from(a.public()).into(),
+								new_free: 1_000_000 * DOLLARS,
+								new_reserved: 0,
+							}
+							.into(),
+						),
+					},
+					Some(start_nonce + (i as u32) * 2 + 1),
+				),
+			]
+		})
+		.flatten()
+		.map(OpaqueExtrinsic::from)
+		.collect()
+}
+
+fn create_benchmark_extrinsics(
+	client: &FullClient,
+	accounts: &[sr25519::Pair],
+	extrinsics_per_account: usize,
+) -> Vec<OpaqueExtrinsic> {
+	accounts
+		.iter()
+		.map(|account| {
+			(0..extrinsics_per_account).map(move |nonce| {
+				create_extrinsic(
+					client,
+					account.clone(),
+					BalancesCall::transfer {
+						dest: Sr25519Keyring::Bob.to_account_id().into(),
+						value: 1 * DOLLARS,
+					},
+					Some(nonce as u32),
+				)
+			})
+		})
+		.flatten()
+		.map(OpaqueExtrinsic::from)
+		.collect()
+}
+
+async fn submit_tx_and_wait_for_inclusion(
+	tx_pool: &TransactionPool,
+	tx: OpaqueExtrinsic,
+	client: &FullClient,
+	wait_for_finalized: bool,
+) {
+	let best_hash = client.chain_info().best_hash;
+
+	let mut watch = tx_pool
+		.submit_and_watch(&BlockId::Hash(best_hash), TransactionSource::External, tx.clone())
+		.await
+		.expect("Submits tx to pool")
+		.fuse();
+
+	loop {
+		match watch.select_next_some().await {
+			TransactionStatus::Finalized(_) => break,
+			TransactionStatus::InBlock(_) if !wait_for_finalized => break,
+			_ => {},
+		}
+	}
+}
+
+fn transaction_pool_benchmarks(c: &mut Criterion) {
+	sp_tracing::try_init_simple();
+
+	let runtime = tokio::runtime::Runtime::new().expect("Creates tokio runtime");
+	let tokio_handle = runtime.handle().clone();
+
+	let node = new_node(tokio_handle.clone());
+
+	let account_num = 10;
+	let extrinsics_per_account = 2000;
+	let accounts = create_accounts(account_num);
+
+	let mut group = c.benchmark_group("Transaction pool");
+
+	group.sample_size(10);
+	group.throughput(Throughput::Elements(account_num as u64 * extrinsics_per_account as u64));
+
+	let mut counter = 1;
+	group.bench_function(
+		format!("{} transfers from {} accounts", account_num * extrinsics_per_account, account_num),
+		move |b| {
+			b.iter_batched(
+				|| {
+					let prepare_extrinsics = create_account_extrinsics(&*node.client, &accounts);
+
+					runtime.block_on(future::join_all(prepare_extrinsics.into_iter().map(|tx| {
+						submit_tx_and_wait_for_inclusion(
+							&node.transaction_pool,
+							tx,
+							&*node.client,
+							true,
+						)
+					})));
+
+					create_benchmark_extrinsics(&*node.client, &accounts, extrinsics_per_account)
+				},
+				|extrinsics| {
+					runtime.block_on(future::join_all(extrinsics.into_iter().map(|tx| {
+						submit_tx_and_wait_for_inclusion(
+							&node.transaction_pool,
+							tx,
+							&*node.client,
+							false,
+						)
+					})));
+
+					println!("Finished {}", counter);
+					counter += 1;
+				},
+				BatchSize::SmallInput,
+			)
+		},
+	);
+}
+
+criterion_group!(benches, transaction_pool_benchmarks);
+criterion_main!(benches);

substrate/bin/node/cli/src/lib.rs

@@ -33,7 +33,7 @@
 pub mod chain_spec;
 
 #[macro_use]
-mod service;
+pub mod service;
 #[cfg(feature = "cli")]
 mod cli;
 #[cfg(feature = "cli")]

substrate/bin/node/cli/src/service.rs

@@ -20,20 +20,25 @@
 
 //! Service implementation. Specialized wrapper over substrate service.
 
+use codec::Encode;
+use frame_system_rpc_runtime_api::AccountNonceApi;
 use futures::prelude::*;
 use node_executor::ExecutorDispatch;
 use node_primitives::Block;
 use node_runtime::RuntimeApi;
-use sc_client_api::{ExecutorProvider, RemoteBackend};
+use sc_client_api::{BlockBackend, ExecutorProvider, RemoteBackend};
 use sc_consensus_babe::{self, SlotProportion};
 use sc_executor::NativeElseWasmExecutor;
 use sc_network::{Event, NetworkService};
 use sc_service::{config::Configuration, error::Error as ServiceError, RpcHandlers, TaskManager};
 use sc_telemetry::{Telemetry, TelemetryWorker};
-use sp_runtime::traits::Block as BlockT;
+use sp_api::ProvideRuntimeApi;
+use sp_core::crypto::Pair;
+use sp_runtime::{generic, traits::Block as BlockT, SaturatedConversion};
 use std::sync::Arc;
 
-type FullClient =
+/// The full client type definition.
+pub type FullClient =
 	sc_service::TFullClient<Block, RuntimeApi, NativeElseWasmExecutor<ExecutorDispatch>>;
 type FullBackend = sc_service::TFullBackend<Block>;
 type FullSelectChain = sc_consensus::LongestChain<FullBackend, Block>;
@@ -41,7 +46,80 @@ type FullGrandpaBlockImport =
 	grandpa::GrandpaBlockImport<FullBackend, Block, FullClient, FullSelectChain>;
 type LightClient =
 	sc_service::TLightClient<Block, RuntimeApi, NativeElseWasmExecutor<ExecutorDispatch>>;
+/// The transaction pool type defintion.
+pub type TransactionPool = sc_transaction_pool::FullPool<Block, FullClient>;
 
+/// Fetch the nonce of the given `account` from the chain state.
+///
+/// Note: Should only be used for tests.
+pub fn fetch_nonce(client: &FullClient, account: sp_core::sr25519::Pair) -> u32 {
+	let best_hash = client.chain_info().best_hash;
+	client
+		.runtime_api()
+		.account_nonce(&generic::BlockId::Hash(best_hash), account.public().into())
+		.expect("Fetching account nonce works; qed")
+}
+
+/// Create a transaction using the given `call`.
+///
+/// The transaction will be signed by `sender`. If `nonce` is `None` it will be fetched from the
+/// state of the best block.
+///
+/// Note: Should only be used for tests.
+pub fn create_extrinsic(
+	client: &FullClient,
+	sender: sp_core::sr25519::Pair,
+	function: impl Into<node_runtime::Call>,
+	nonce: Option<u32>,
+) -> node_runtime::UncheckedExtrinsic {
+	let function = function.into();
+	let genesis_hash = client.block_hash(0).ok().flatten().expect("Genesis block exists; qed");
+	let best_hash = client.chain_info().best_hash;
+	let best_block = client.chain_info().best_number;
+	let nonce = nonce.unwrap_or_else(|| fetch_nonce(client, sender.clone()));
+
+	let period = node_runtime::BlockHashCount::get()
+		.checked_next_power_of_two()
+		.map(|c| c / 2)
+		.unwrap_or(2) as u64;
+	let tip = 0;
+	let extra: node_runtime::SignedExtra = (
+		frame_system::CheckSpecVersion::<node_runtime::Runtime>::new(),
+		frame_system::CheckTxVersion::<node_runtime::Runtime>::new(),
+		frame_system::CheckGenesis::<node_runtime::Runtime>::new(),
+		frame_system::CheckEra::<node_runtime::Runtime>::from(generic::Era::mortal(
+			period,
+			best_block.saturated_into(),
+		)),
+		frame_system::CheckNonce::<node_runtime::Runtime>::from(nonce),
+		frame_system::CheckWeight::<node_runtime::Runtime>::new(),
+		pallet_transaction_payment::ChargeTransactionPayment::<node_runtime::Runtime>::from(tip),
+	);
+
+	let raw_payload = node_runtime::SignedPayload::from_raw(
+		function.clone(),
+		extra.clone(),
+		(
+			node_runtime::VERSION.spec_version,
+			node_runtime::VERSION.transaction_version,
+			genesis_hash,
+			best_hash,
+			(),
+			(),
+			(),
+		),
+	);
+	let signature = raw_payload.using_encoded(|e| sender.sign(e));
+
+	node_runtime::UncheckedExtrinsic::new_signed(
+		function.clone(),
+		sp_runtime::AccountId32::from(sender.public()).into(),
+		node_runtime::Signature::Sr25519(signature.clone()),
+		extra.clone(),
+	)
+}
+
+/// Creates a new partial node.
 pub fn new_partial(
 	config: &Configuration,
 ) -> Result<
@@ -211,11 +289,16 @@ pub fn new_partial(
 	})
 }
 
+/// Result of [`new_full_base`].
 pub struct NewFullBase {
+	/// The task manager of the node.
 	pub task_manager: TaskManager,
+	/// The client instance of the node.
 	pub client: Arc<FullClient>,
+	/// The networking service of the node.
 	pub network: Arc<NetworkService<Block, <Block as BlockT>::Hash>>,
-	pub transaction_pool: Arc<sc_transaction_pool::FullPool<Block, FullClient>>,
+	/// The transaction pool of the node.
+	pub transaction_pool: Arc<TransactionPool>,
 }
 
 /// Creates a full service from the configuration.
@@ -433,6 +516,7 @@ pub fn new_full(config: Configuration) -> Result<TaskManager, ServiceError> {
 	new_full_base(config, |_, _| ()).map(|NewFullBase { task_manager, .. }| task_manager)
 }
 
+/// Creates a light service from the configuration.
 pub fn new_light_base(
 	mut config: Configuration,
 ) -> Result<