Use tokio runtime handle instead of TaskExecutor abstraction (#9737)

* Use tokio runtime handle instead of TaskExecutor abstraction

Before this pr we had the `TaskExecutor` abstraction which theoretically
allowed that any futures executor could have been used. However, this
was never tested and is currently not really required. Anyone running a
node currently only used tokio and nothing else (because this was hard
coded in CLI). So, this pr removes the `TaskExecutor` abstraction and
relies directly on the tokio runtime handle.

Besides this changes, this pr also makes sure that the http and ws rpc
server use the same tokio runtime. This fixes a panic that occurred when
you drop the rpc servers inside an async function (tokio doesn't like
that a tokio runtime is dropped in the async context of another tokio
runtime).

As we don't use any custom runtime in the http rpc server anymore, this
pr also removes the `rpc-http-threads` cli argument. If external parties
complain that there aren't enough threads for the rpc server, we could
bring support for increasing the thread count of the tokio runtime.

* FMT

* Fix try runtime

* Fix integration tests and some other optimizations

* Remove warnings

Author: bkchr

Cargo.lock

@@ -114,6 +114,7 @@ sc-consensus = { version = "0.10.0-dev", path = "../../../client/consensus/commo
 sc-consensus-babe = { version = "0.10.0-dev", path = "../../../client/consensus/babe" }
 sc-consensus-epochs = { version = "0.10.0-dev", path = "../../../client/consensus/epochs" }
 sc-service-test = { version = "2.0.0", path = "../../../client/service/test" }
+sp-tracing = { version = "4.0.0-dev", path = "../../../primitives/tracing" }
 futures = "0.3.16"
 tempfile = "3.1.0"
 assert_cmd = "1.0"

bin/node/cli/Cargo.toml

@@ -463,6 +463,8 @@ pub(crate) mod tests {
 	#[test]
 	#[ignore]
 	fn test_connectivity() {
+		sp_tracing::try_init_simple();
+
 		sc_service_test::connectivity(
 			integration_test_config_with_two_authorities(),
 			|config| {

bin/node/cli/src/chain_spec.rs

@@ -156,7 +156,7 @@ pub fn run() -> Result<()> {
 				// manager to do `async_run`.
 				let registry = config.prometheus_config.as_ref().map(|cfg| &cfg.registry);
 				let task_manager =
-					sc_service::TaskManager::new(config.task_executor.clone(), registry)
+					sc_service::TaskManager::new(config.tokio_handle.clone(), registry)
 						.map_err(|e| sc_cli::Error::Service(sc_service::Error::Prometheus(e)))?;
 
 				Ok((cmd.run::<Block, ExecutorDispatch>(config), task_manager))

bin/node/cli/src/command.rs

@@ -644,6 +644,8 @@ mod tests {
 	// This can be run locally with `cargo test --release -p node-cli test_sync -- --ignored`.
 	#[ignore]
 	fn test_sync() {
+		sp_tracing::try_init_simple();
+
 		let keystore_path = tempfile::tempdir().expect("Creates keystore path");
 		let keystore: SyncCryptoStorePtr =
 			Arc::new(LocalKeystore::open(keystore_path.path(), None).expect("Creates keystore"));
@@ -843,6 +845,8 @@ mod tests {
 	#[test]
 	#[ignore]
 	fn test_consensus() {
+		sp_tracing::try_init_simple();
+
 		sc_service_test::consensus(
 			crate::chain_spec::tests::integration_test_config_with_two_authorities(),
 			|config| {

bin/node/cli/src/service.rs

@@ -16,23 +16,30 @@
 // You should have received a copy of the GNU General Public License
 // along with this program. If not, see <https://www.gnu.org/licenses/>.
 
+#![cfg(unix)]
+
 use assert_cmd::cargo::cargo_bin;
-use std::{convert::TryInto, process::Command, thread, time::Duration};
+use nix::{
+	sys::signal::{
+		kill,
+		Signal::{self, SIGINT, SIGTERM},
+	},
+	unistd::Pid,
+};
+use sc_service::Deref;
+use std::{
+	convert::TryInto,
+	ops::DerefMut,
+	process::{Child, Command},
+	thread,
+	time::Duration,
+};
 use tempfile::tempdir;
 
 pub mod common;
 
 #[test]
-#[cfg(unix)]
 fn running_the_node_works_and_can_be_interrupted() {
-	use nix::{
-		sys::signal::{
-			kill,
-			Signal::{self, SIGINT, SIGTERM},
-		},
-		unistd::Pid,
-	};
-
 	fn run_command_and_kill(signal: Signal) {
 		let base_path = tempdir().expect("could not create a temp dir");
 		let mut cmd = Command::new(cargo_bin("substrate"))
@@ -55,3 +62,57 @@ fn running_the_node_works_and_can_be_interrupted() {
 	run_command_and_kill(SIGINT);
 	run_command_and_kill(SIGTERM);
 }
+
+struct KillChildOnDrop(Child);
+
+impl Drop for KillChildOnDrop {
+	fn drop(&mut self) {
+		let _ = self.0.kill();
+	}
+}
+
+impl Deref for KillChildOnDrop {
+	type Target = Child;
+
+	fn deref(&self) -> &Self::Target {
+		&self.0
+	}
+}
+
+impl DerefMut for KillChildOnDrop {
+	fn deref_mut(&mut self) -> &mut Self::Target {
+		&mut self.0
+	}
+}
+
+#[test]
+fn running_two_nodes_with_the_same_ws_port_should_work() {
+	fn start_node() -> Child {
+		Command::new(cargo_bin("substrate"))
+			.args(&["--dev", "--tmp", "--ws-port=45789"])
+			.spawn()
+			.unwrap()
+	}
+
+	let mut first_node = KillChildOnDrop(start_node());
+	let mut second_node = KillChildOnDrop(start_node());
+
+	thread::sleep(Duration::from_secs(30));
+
+	assert!(first_node.try_wait().unwrap().is_none(), "The first node should still be running");
+	assert!(second_node.try_wait().unwrap().is_none(), "The second node should still be running");
+
+	kill(Pid::from_raw(first_node.id().try_into().unwrap()), SIGINT).unwrap();
+	kill(Pid::from_raw(second_node.id().try_into().unwrap()), SIGINT).unwrap();
+
+	assert_eq!(
+		common::wait_for(&mut first_node, 30).map(|x| x.success()),
+		Some(true),
+		"The first node must exit gracefully",
+	);
+	assert_eq!(
+		common::wait_for(&mut second_node, 30).map(|x| x.success()),
+		Some(true),
+		"The second node must exit gracefully",
+	);
+}

bin/node/cli/tests/running_the_node_and_interrupt.rs

@@ -88,18 +88,17 @@ mod tests {
 	use node_cli::chain_spec::development_config;
 	use sp_keyring::sr25519::Keyring::Alice;
 	use sp_runtime::{traits::IdentifyAccount, MultiSigner};
-	use test_runner::{build_runtime, client_parts, task_executor, ConfigOrChainSpec, Node};
+	use test_runner::{build_runtime, client_parts, ConfigOrChainSpec, Node};
 
 	#[test]
 	fn test_runner() {
 		let tokio_runtime = build_runtime().unwrap();
-		let task_executor = task_executor(tokio_runtime.handle().clone());
-		let (rpc, task_manager, client, pool, command_sink, backend) = client_parts::<
-			NodeTemplateChainInfo,
-		>(
-			ConfigOrChainSpec::ChainSpec(Box::new(development_config()), task_executor),
-		)
-		.unwrap();
+		let (rpc, task_manager, client, pool, command_sink, backend) =
+			client_parts::<NodeTemplateChainInfo>(ConfigOrChainSpec::ChainSpec(
+				Box::new(development_config()),
+				tokio_runtime.handle().clone(),
+			))
+			.unwrap();
 		let node = Node::<NodeTemplateChainInfo>::new(
 			rpc,
 			task_manager,

bin/node/test-runner-example/src/lib.rs

@@ -127,10 +127,6 @@ pub struct RunCmd {
 	#[structopt(long = "ws-max-connections", value_name = "COUNT")]
 	pub ws_max_connections: Option<usize>,
 
-	/// Size of the RPC HTTP server thread pool.
-	#[structopt(long = "rpc-http-threads", value_name = "COUNT")]
-	pub rpc_http_threads: Option<usize>,
-
 	/// Specify browser Origins allowed to access the HTTP & WS RPC servers.
 	///
 	/// A comma-separated list of origins (protocol://domain or special `null`
@@ -381,10 +377,6 @@ impl CliConfiguration for RunCmd {
 		Ok(self.ws_max_connections)
 	}
 
-	fn rpc_http_threads(&self) -> Result<Option<usize>> {
-		Ok(self.rpc_http_threads)
-	}
-
 	fn rpc_cors(&self, is_dev: bool) -> Result<Option<Vec<String>>> {
 		Ok(self
 			.rpc_cors

client/cli/src/commands/run_cmd.rs

@@ -29,7 +29,7 @@ use sc_service::{
 	config::{
 		BasePath, Configuration, DatabaseSource, KeystoreConfig, NetworkConfiguration,
 		NodeKeyConfig, OffchainWorkerConfig, PrometheusConfig, PruningMode, Role, RpcMethods,
-		TaskExecutor, TelemetryEndpoints, TransactionPoolOptions, WasmExecutionMethod,
+		TelemetryEndpoints, TransactionPoolOptions, WasmExecutionMethod,
 	},
 	ChainSpec, KeepBlocks, TracingReceiver, TransactionStorageMode,
 };
@@ -348,13 +348,6 @@ pub trait CliConfiguration<DCV: DefaultConfigurationValues = ()>: Sized {
 		Ok(None)
 	}
 
-	/// Get the RPC HTTP thread pool size (`None` for a default 4-thread pool config).
-	///
-	/// By default this is `None`.
-	fn rpc_http_threads(&self) -> Result<Option<usize>> {
-		Ok(None)
-	}
-
 	/// Get the RPC cors (`None` if disabled)
 	///
 	/// By default this is `Some(Vec::new())`.
@@ -465,7 +458,7 @@ pub trait CliConfiguration<DCV: DefaultConfigurationValues = ()>: Sized {
 	fn create_configuration<C: SubstrateCli>(
 		&self,
 		cli: &C,
-		task_executor: TaskExecutor,
+		tokio_handle: tokio::runtime::Handle,
 	) -> Result<Configuration> {
 		let is_dev = self.is_dev()?;
 		let chain_id = self.chain_id(is_dev)?;
@@ -490,7 +483,7 @@ pub trait CliConfiguration<DCV: DefaultConfigurationValues = ()>: Sized {
 		Ok(Configuration {
 			impl_name: C::impl_name(),
 			impl_version: C::impl_version(),
-			task_executor,
+			tokio_handle,
 			transaction_pool: self.transaction_pool()?,
 			network: self.network_config(
 				&chain_spec,
@@ -518,7 +511,6 @@ pub trait CliConfiguration<DCV: DefaultConfigurationValues = ()>: Sized {
 			rpc_ipc: self.rpc_ipc()?,
 			rpc_methods: self.rpc_methods()?,
 			rpc_ws_max_connections: self.rpc_ws_max_connections()?,
-			rpc_http_threads: self.rpc_http_threads()?,
 			rpc_cors: self.rpc_cors(is_dev)?,
 			rpc_max_payload: self.rpc_max_payload()?,
 			prometheus_config: self.prometheus_config(DCV::prometheus_listen_port())?,

client/cli/src/config.rs

@@ -35,8 +35,8 @@ pub use config::*;
 pub use error::*;
 pub use params::*;
 pub use runner::*;
+use sc_service::Configuration;
 pub use sc_service::{ChainSpec, Role};
-use sc_service::{Configuration, TaskExecutor};
 pub use sc_tracing::logging::LoggerBuilder;
 pub use sp_version::RuntimeVersion;
 use std::io::Write;
@@ -216,9 +216,9 @@ pub trait SubstrateCli: Sized {
 	fn create_configuration<T: CliConfiguration<DVC>, DVC: DefaultConfigurationValues>(
 		&self,
 		command: &T,
-		task_executor: TaskExecutor,
+		tokio_handle: tokio::runtime::Handle,
 	) -> error::Result<Configuration> {
-		command.create_configuration(self, task_executor)
+		command.create_configuration(self, tokio_handle)
 	}
 
 	/// Create a runner for the command provided in argument. This will create a Configuration and

client/cli/src/lib.rs

@@ -20,7 +20,7 @@ use crate::{error::Error as CliError, CliConfiguration, Result, SubstrateCli};
 use chrono::prelude::*;
 use futures::{future, future::FutureExt, pin_mut, select, Future};
 use log::info;
-use sc_service::{Configuration, Error as ServiceError, TaskManager, TaskType};
+use sc_service::{Configuration, Error as ServiceError, TaskManager};
 use sc_utils::metrics::{TOKIO_THREADS_ALIVE, TOKIO_THREADS_TOTAL};
 use std::marker::PhantomData;
 
@@ -116,15 +116,8 @@ impl<C: SubstrateCli> Runner<C> {
 		let tokio_runtime = build_runtime()?;
 		let runtime_handle = tokio_runtime.handle().clone();
 
-		let task_executor = move |fut, task_type| match task_type {
-			TaskType::Async => runtime_handle.spawn(fut).map(drop),
-			TaskType::Blocking => runtime_handle
-				.spawn_blocking(move || futures::executor::block_on(fut))
-				.map(drop),
-		};
-
 		Ok(Runner {
-			config: command.create_configuration(cli, task_executor.into())?,
+			config: command.create_configuration(cli, runtime_handle)?,
 			tokio_runtime,
 			phantom: PhantomData,
 		})

client/cli/src/runner.rs

@@ -19,6 +19,7 @@ pubsub = { package = "jsonrpc-pubsub", version = "18.0.0" }
 log = "0.4.8"
 prometheus-endpoint = { package = "substrate-prometheus-endpoint", path = "../../utils/prometheus", version = "0.9.0"}
 serde_json = "1.0.41"
+tokio = "1.10"
 http = { package = "jsonrpc-http-server", version = "18.0.0" }
 ipc = { package = "jsonrpc-ipc-server", version = "18.0.0" }
 ws = { package = "jsonrpc-ws-server", version = "18.0.0" }

client/rpc-servers/Cargo.toml

@@ -36,9 +36,6 @@ pub const RPC_MAX_PAYLOAD_DEFAULT: usize = 15 * MEGABYTE;
 /// Default maximum number of connections for WS RPC servers.
 const WS_MAX_CONNECTIONS: usize = 100;
 
-/// Default thread pool size for RPC HTTP servers.
-const HTTP_THREADS: usize = 4;
-
 /// The RPC IoHandler containing all requested APIs.
 pub type RpcHandler<T> = pubsub::PubSubHandler<T, RpcMiddleware>;
 
@@ -130,17 +127,18 @@ impl ws::SessionStats for ServerMetrics {
 /// Start HTTP server listening on given address.
 pub fn start_http<M: pubsub::PubSubMetadata + Default + Unpin>(
 	addr: &std::net::SocketAddr,
-	thread_pool_size: Option<usize>,
 	cors: Option<&Vec<String>>,
 	io: RpcHandler<M>,
 	maybe_max_payload_mb: Option<usize>,
+	tokio_handle: tokio::runtime::Handle,
 ) -> io::Result<http::Server> {
 	let max_request_body_size = maybe_max_payload_mb
 		.map(|mb| mb.saturating_mul(MEGABYTE))
 		.unwrap_or(RPC_MAX_PAYLOAD_DEFAULT);
 
 	http::ServerBuilder::new(io)
-		.threads(thread_pool_size.unwrap_or(HTTP_THREADS))
+		.threads(1)
+		.event_loop_executor(tokio_handle)
 		.health_api(("/health", "system_health"))
 		.allowed_hosts(hosts_filtering(cors.is_some()))
 		.rest_api(if cors.is_some() { http::RestApi::Secure } else { http::RestApi::Unsecure })
@@ -175,13 +173,15 @@ pub fn start_ws<
 	io: RpcHandler<M>,
 	maybe_max_payload_mb: Option<usize>,
 	server_metrics: ServerMetrics,
+	tokio_handle: tokio::runtime::Handle,
 ) -> io::Result<ws::Server> {
 	let rpc_max_payload = maybe_max_payload_mb
 		.map(|mb| mb.saturating_mul(MEGABYTE))
 		.unwrap_or(RPC_MAX_PAYLOAD_DEFAULT);
 	ws::ServerBuilder::with_meta_extractor(io, |context: &ws::RequestContext| {
 		context.sender().into()
 	})
+	.event_loop_executor(tokio_handle)
 	.max_payload(rpc_max_payload)
 	.max_connections(max_connections.unwrap_or(WS_MAX_CONNECTIONS))
 	.allowed_origins(map_cors(cors))

client/rpc-servers/src/lib.rs

@@ -79,11 +79,11 @@ parity-util-mem = { version = "0.10.0", default-features = false, features = [
 	"primitive-types",
 ] }
 async-trait = "0.1.50"
+tokio = { version = "1.10", features = ["time", "rt-multi-thread"] }
 tempfile = "3.1.0"
 directories = "3.0.2"
 
 [dev-dependencies]
 substrate-test-runtime-client = { version = "2.0.0", path = "../../test-utils/runtime/client" }
 substrate-test-runtime = { version = "2.0.0", path = "../../test-utils/runtime/" }
-tokio = { version = "1.10", features = ["time"] }
 async-std = { version = "1.6.5", default-features = false }

client/service/Cargo.toml

@@ -286,7 +286,7 @@ where
 
 	let task_manager = {
 		let registry = config.prometheus_config.as_ref().map(|cfg| &cfg.registry);
-		TaskManager::new(config.task_executor.clone(), registry)?
+		TaskManager::new(config.tokio_handle.clone(), registry)?
 	};
 
 	let chain_spec = &config.chain_spec;
@@ -372,7 +372,7 @@ where
 	let keystore_container = KeystoreContainer::new(&config.keystore)?;
 	let task_manager = {
 		let registry = config.prometheus_config.as_ref().map(|cfg| &cfg.registry);
-		TaskManager::new(config.task_executor.clone(), registry)?
+		TaskManager::new(config.tokio_handle.clone(), registry)?
 	};
 
 	let db_storage = {