Revert "Use SpawnTaskHandles for spawning tasks in the tx pool (#8958)" (#8983)

This reverts commit bfef07c.

Author: expenses

Cargo.lock

@@ -16,6 +16,7 @@ targets = ["x86_64-unknown-linux-gnu"]
 codec = { package = "parity-scale-codec", version = "2.0.0" }
 thiserror = "1.0.21"
 futures = { version = "0.3.1", features = ["compat"] }
+futures-diagnose = "1.0"
 intervalier = "0.4.0"
 log = "0.4.8"
 parity-util-mem = { version = "0.9.0", default-features = false, features = ["primitive-types"] }

client/transaction-pool/Cargo.toml

@@ -21,7 +21,7 @@
 use std::{marker::PhantomData, pin::Pin, sync::Arc};
 use codec::{Decode, Encode};
 use futures::{
-	channel::oneshot, future::{Future, FutureExt, ready, Ready},
+	channel::oneshot, executor::{ThreadPool, ThreadPoolBuilder}, future::{Future, FutureExt, ready, Ready},
 };
 
 use sc_client_api::{
@@ -31,7 +31,6 @@ use sp_runtime::{
 	generic::BlockId, traits::{self, Block as BlockT, BlockIdTo, Header as HeaderT, Hash as HashT},
 	transaction_validity::{TransactionValidity, TransactionSource},
 };
-use sp_core::traits::SpawnNamed;
 use sp_transaction_pool::runtime_api::TaggedTransactionQueue;
 use sp_api::{ProvideRuntimeApi, ApiExt};
 use prometheus_endpoint::Registry as PrometheusRegistry;
@@ -41,7 +40,7 @@ use crate::{metrics::{ApiMetrics, ApiMetricsExt}, error::{self, Error}};
 /// The transaction pool logic for full client.
 pub struct FullChainApi<Client, Block> {
 	client: Arc<Client>,
-	spawner: Box<dyn SpawnNamed>,
+	pool: ThreadPool,
 	_marker: PhantomData<Block>,
 	metrics: Option<Arc<ApiMetrics>>,
 }
@@ -51,7 +50,6 @@ impl<Client, Block> FullChainApi<Client, Block> {
 	pub fn new(
 		client: Arc<Client>,
 		prometheus: Option<&PrometheusRegistry>,
-		spawner: impl SpawnNamed + 'static,
 	) -> Self {
 		let metrics = prometheus.map(ApiMetrics::register).and_then(|r| {
 			match r {
@@ -69,9 +67,13 @@ impl<Client, Block> FullChainApi<Client, Block> {
 
 		FullChainApi {
 			client,
+			pool: ThreadPoolBuilder::new()
+				.pool_size(2)
+				.name_prefix("txpool-verifier")
+				.create()
+				.expect("Failed to spawn verifier threads, that are critical for node operation."),
 			_marker: Default::default(),
 			metrics,
-			spawner: Box::new(spawner) ,
 		}
 	}
 }
@@ -107,9 +109,9 @@ where
 		let metrics = self.metrics.clone();
 		metrics.report(|m| m.validations_scheduled.inc());
 
-		self.spawner.spawn_blocking(
+		self.pool.spawn_ok(futures_diagnose::diagnose(
 			"validate-transaction",
-			Box::pin(async move {
+			async move {
 				let res = validate_transaction_blocking(&*client, &at, source, uxt);
 				if let Err(e) = tx.send(res) {
 					log::warn!("Unable to send a validate transaction result: {:?}", e);

client/transaction-pool/src/api.rs

@@ -366,10 +366,10 @@ where
 		options: sc_transaction_graph::Options,
 		is_validator: txpool::IsValidator,
 		prometheus: Option<&PrometheusRegistry>,
-		spawner: impl SpawnNamed + Clone + 'static,
+		spawner: impl SpawnNamed,
 		client: Arc<Client>,
 	) -> Arc<Self> {
-		let pool_api = Arc::new(FullChainApi::new(client.clone(), prometheus, spawner.clone()));
+		let pool_api = Arc::new(FullChainApi::new(client.clone(), prometheus));
 		let pool = Arc::new(Self::with_revalidation_type(
 			options, is_validator, pool_api, prometheus, RevalidationType::Full, spawner
 		));

client/transaction-pool/src/lib.rs

@@ -35,7 +35,6 @@ use std::collections::BTreeSet;
 use sc_client_api::client::BlockchainEvents;
 use sc_block_builder::BlockBuilderProvider;
 use sp_consensus::BlockOrigin;
-use sp_core::testing::TaskExecutor;
 
 fn pool() -> Pool<TestApi> {
 	Pool::new(Default::default(), true.into(), TestApi::with_alice_nonce(209).into())
@@ -936,7 +935,7 @@ fn should_not_accept_old_signatures() {
 	let client = Arc::new(substrate_test_runtime_client::new());
 
 	let pool = Arc::new(
-		BasicPool::new_test(Arc::new(FullChainApi::new(client, None, TaskExecutor::new()))).0
+		BasicPool::new_test(Arc::new(FullChainApi::new(client, None))).0
 	);
 
 	let transfer = Transfer {
@@ -972,7 +971,7 @@ fn import_notification_to_pool_maintain_works() {
 	let mut client = Arc::new(substrate_test_runtime_client::new());
 
 	let pool = Arc::new(
-		BasicPool::new_test(Arc::new(FullChainApi::new(client.clone(), None, TaskExecutor::new()))).0
+		BasicPool::new_test(Arc::new(FullChainApi::new(client.clone(), None))).0
 	);
 
 	// Prepare the extrisic, push it to the pool and check that it was added.