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

substrate/client/service/src/builder.rs

@@ -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 = {

substrate/client/service/src/config.rs

@@ -36,12 +36,9 @@ pub use sc_telemetry::TelemetryEndpoints;
 pub use sc_transaction_pool::Options as TransactionPoolOptions;
 use sp_core::crypto::SecretString;
 use std::{
-	future::Future,
 	io,
 	net::SocketAddr,
 	path::{Path, PathBuf},
-	pin::Pin,
-	sync::Arc,
 };
 use tempfile::TempDir;
 
@@ -54,8 +51,8 @@ pub struct Configuration {
 	pub impl_version: String,
 	/// Node role.
 	pub role: Role,
-	/// How to spawn background tasks. Mandatory, otherwise creating a `Service` will error.
-	pub task_executor: TaskExecutor,
+	/// Handle to the tokio runtime. Will be used to spawn futures by the task manager.
+	pub tokio_handle: tokio::runtime::Handle,
 	/// Extrinsic pool configuration.
 	pub transaction_pool: TransactionPoolOptions,
 	/// Network configuration.
@@ -94,8 +91,6 @@ pub struct Configuration {
 	pub rpc_ipc: Option<String>,
 	/// Maximum number of connections for WebSockets RPC server. `None` if default.
 	pub rpc_ws_max_connections: Option<usize>,
-	/// Size of the RPC HTTP server thread pool. `None` if default.
-	pub rpc_http_threads: Option<usize>,
 	/// CORS settings for HTTP & WS servers. `None` if all origins are allowed.
 	pub rpc_cors: Option<Vec<String>>,
 	/// RPC methods to expose (by default only a safe subset or all of them).
@@ -305,62 +300,3 @@ impl std::convert::From<PathBuf> for BasePath {
 		BasePath::new(path)
 	}
 }
-
-// NOTE: here for code readability.
-pub(crate) type SomeFuture = Pin<Box<dyn Future<Output = ()> + Send>>;
-pub(crate) type JoinFuture = Pin<Box<dyn Future<Output = ()> + Send>>;
-
-/// Callable object that execute tasks.
-///
-/// This struct can be created easily using `Into`.
-///
-/// # Examples
-///
-/// ## Using tokio
-///
-/// ```
-/// # use sc_service::TaskExecutor;
-/// use futures::future::FutureExt;
-/// use tokio::runtime::Runtime;
-///
-/// let runtime = Runtime::new().unwrap();
-/// let handle = runtime.handle().clone();
-/// let task_executor: TaskExecutor = (move |future, _task_type| {
-///     handle.spawn(future).map(|_| ())
-/// }).into();
-/// ```
-///
-/// ## Using async-std
-///
-/// ```
-/// # use sc_service::TaskExecutor;
-/// let task_executor: TaskExecutor = (|future, _task_type| {
-///     // NOTE: async-std's JoinHandle is not a Result so we don't need to map the result
-///     async_std::task::spawn(future)
-/// }).into();
-/// ```
-#[derive(Clone)]
-pub struct TaskExecutor(Arc<dyn Fn(SomeFuture, TaskType) -> JoinFuture + Send + Sync>);
-
-impl std::fmt::Debug for TaskExecutor {
-	fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
-		write!(f, "TaskExecutor")
-	}
-}
-
-impl<F, FUT> std::convert::From<F> for TaskExecutor
-where
-	F: Fn(SomeFuture, TaskType) -> FUT + Send + Sync + 'static,
-	FUT: Future<Output = ()> + Send + 'static,
-{
-	fn from(func: F) -> Self {
-		Self(Arc::new(move |fut, tt| Box::pin(func(fut, tt))))
-	}
-}
-
-impl TaskExecutor {
-	/// Spawns a new asynchronous task.
-	pub fn spawn(&self, future: SomeFuture, task_type: TaskType) -> JoinFuture {
-		self.0(future, task_type)
-	}
-}

substrate/client/service/src/lib.rs

@@ -58,8 +58,8 @@ pub use self::{
 	error::Error,
 };
 pub use config::{
-	BasePath, Configuration, DatabaseSource, KeepBlocks, PruningMode, Role, RpcMethods,
-	TaskExecutor, TaskType, TransactionStorageMode,
+	BasePath, Configuration, DatabaseSource, KeepBlocks, PruningMode, Role, RpcMethods, TaskType,
+	TransactionStorageMode,
 };
 pub use sc_chain_spec::{
 	ChainSpec, ChainType, Extension as ChainSpecExtension, GenericChainSpec, NoExtension,
@@ -395,7 +395,6 @@ fn start_rpc_servers<
 		maybe_start_server(config.rpc_http, |address| {
 			sc_rpc_server::start_http(
 				address,
-				config.rpc_http_threads,
 				config.rpc_cors.as_ref(),
 				gen_handler(
 					deny_unsafe(&address, &config.rpc_methods),
@@ -406,6 +405,7 @@ fn start_rpc_servers<
 					),
 				)?,
 				config.rpc_max_payload,
+				config.tokio_handle.clone(),
 			)
 			.map_err(Error::from)
 		})?
@@ -425,6 +425,7 @@ fn start_rpc_servers<
 				)?,
 				config.rpc_max_payload,
 				server_metrics.clone(),
+				config.tokio_handle.clone(),
 			)
 			.map_err(Error::from)
 		})?

substrate/client/service/src/task_manager/mod.rs

@@ -18,23 +18,20 @@
 
 //! Substrate service tasks management module.
 
-use crate::{
-	config::{JoinFuture, TaskExecutor, TaskType},
-	Error,
-};
+use crate::{config::TaskType, Error};
 use exit_future::Signal;
 use futures::{
 	future::{join_all, pending, select, try_join_all, BoxFuture, Either},
-	sink::SinkExt,
 	Future, FutureExt, StreamExt,
 };
-use log::{debug, error};
+use log::debug;
 use prometheus_endpoint::{
 	exponential_buckets, register, CounterVec, HistogramOpts, HistogramVec, Opts, PrometheusError,
 	Registry, U64,
 };
 use sc_utils::mpsc::{tracing_unbounded, TracingUnboundedReceiver, TracingUnboundedSender};
 use std::{panic, pin::Pin, result::Result};
+use tokio::{runtime::Handle, task::JoinHandle};
 use tracing_futures::Instrument;
 
 mod prometheus_future;
@@ -45,9 +42,9 @@ mod tests;
 #[derive(Clone)]
 pub struct SpawnTaskHandle {
 	on_exit: exit_future::Exit,
-	executor: TaskExecutor,
+	tokio_handle: Handle,
 	metrics: Option<Metrics>,
-	task_notifier: TracingUnboundedSender<JoinFuture>,
+	task_notifier: TracingUnboundedSender<JoinHandle<()>>,
 }
 
 impl SpawnTaskHandle {
@@ -126,19 +123,20 @@ impl SpawnTaskHandle {
 				futures::pin_mut!(task);
 				let _ = select(on_exit, task).await;
 			}
+		}
+		.in_current_span();
+
+		let join_handle = match task_type {
+			TaskType::Async => self.tokio_handle.spawn(future),
+			TaskType::Blocking => {
+				let handle = self.tokio_handle.clone();
+				self.tokio_handle.spawn_blocking(move || {
+					handle.block_on(future);
+				})
+			},
 		};
 
-		let join_handle = self.executor.spawn(future.in_current_span().boxed(), task_type);
-
-		let mut task_notifier = self.task_notifier.clone();
-		self.executor.spawn(
-			Box::pin(async move {
-				if let Err(err) = task_notifier.send(join_handle).await {
-					error!("Could not send spawned task handle to queue: {}", err);
-				}
-			}),
-			TaskType::Async,
-		);
+		let _ = self.task_notifier.unbounded_send(join_handle);
 	}
 }
 
@@ -222,8 +220,8 @@ pub struct TaskManager {
 	on_exit: exit_future::Exit,
 	/// A signal that makes the exit future above resolve, fired on service drop.
 	signal: Option<Signal>,
-	/// How to spawn background tasks.
-	executor: TaskExecutor,
+	/// Tokio runtime handle that is used to spawn futures.
+	tokio_handle: Handle,
 	/// Prometheus metric where to report the polling times.
 	metrics: Option<Metrics>,
 	/// Send a signal when a spawned essential task has concluded. The next time
@@ -234,9 +232,9 @@ pub struct TaskManager {
 	/// Things to keep alive until the task manager is dropped.
 	keep_alive: Box<dyn std::any::Any + Send>,
 	/// A sender to a stream of background tasks. This is used for the completion future.
-	task_notifier: TracingUnboundedSender<JoinFuture>,
+	task_notifier: TracingUnboundedSender<JoinHandle<()>>,
 	/// This future will complete when all the tasks are joined and the stream is closed.
-	completion_future: JoinFuture,
+	completion_future: JoinHandle<()>,
 	/// A list of other `TaskManager`'s to terminate and gracefully shutdown when the parent
 	/// terminates and gracefully shutdown. Also ends the parent `future()` if a child's essential
 	/// task fails.
@@ -247,7 +245,7 @@ impl TaskManager {
 	/// If a Prometheus registry is passed, it will be used to report statistics about the
 	/// service tasks.
 	pub fn new(
-		executor: TaskExecutor,
+		tokio_handle: Handle,
 		prometheus_registry: Option<&Registry>,
 	) -> Result<Self, PrometheusError> {
 		let (signal, on_exit) = exit_future::signal();
@@ -261,13 +259,15 @@ impl TaskManager {
 		// NOTE: for_each_concurrent will await on all the JoinHandle futures at the same time. It
 		// is possible to limit this but it's actually better for the memory foot print to await
 		// them all to not accumulate anything on that stream.
-		let completion_future = executor
-			.spawn(Box::pin(background_tasks.for_each_concurrent(None, |x| x)), TaskType::Async);
+		let completion_future =
+			tokio_handle.spawn(background_tasks.for_each_concurrent(None, |x| async move {
+				let _ = x.await;
+			}));
 
 		Ok(Self {
 			on_exit,
 			signal: Some(signal),
-			executor,
+			tokio_handle,
 			metrics,
 			essential_failed_tx,
 			essential_failed_rx,
@@ -282,7 +282,7 @@ impl TaskManager {
 	pub fn spawn_handle(&self) -> SpawnTaskHandle {
 		SpawnTaskHandle {
 			on_exit: self.on_exit.clone(),
-			executor: self.executor.clone(),
+			tokio_handle: self.tokio_handle.clone(),
 			metrics: self.metrics.clone(),
 			task_notifier: self.task_notifier.clone(),
 		}
@@ -310,14 +310,9 @@ impl TaskManager {
 
 		Box::pin(async move {
 			join_all(children_shutdowns).await;
-			completion_future.await;
+			let _ = completion_future.await;
 
-			// The keep_alive stuff is holding references to some RPC handles etc. These
-			// RPC handles spawn their own tokio stuff and that doesn't like to be closed in an
-			// async context. So, we move the deletion to some other thread.
-			std::thread::spawn(move || {
-				let _ = keep_alive;
-			});
+			let _ = keep_alive;
 		})
 	}
 

substrate/client/service/src/task_manager/tests.rs

@@ -16,7 +16,7 @@
 // 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 crate::{config::TaskExecutor, task_manager::TaskManager};
+use crate::task_manager::TaskManager;
 use futures::{future::FutureExt, pin_mut, select};
 use parking_lot::Mutex;
 use std::{any::Any, sync::Arc, time::Duration};
@@ -84,17 +84,16 @@ async fn run_background_task_blocking(duration: Duration, _keep_alive: impl Any)
 	}
 }
 
-fn new_task_manager(task_executor: TaskExecutor) -> TaskManager {
-	TaskManager::new(task_executor, None).unwrap()
+fn new_task_manager(tokio_handle: tokio::runtime::Handle) -> TaskManager {
+	TaskManager::new(tokio_handle, None).unwrap()
 }
 
 #[test]
 fn ensure_tasks_are_awaited_on_shutdown() {
 	let runtime = tokio::runtime::Runtime::new().unwrap();
 	let handle = runtime.handle().clone();
-	let task_executor: TaskExecutor = (move |future, _| handle.spawn(future).map(|_| ())).into();
 
-	let task_manager = new_task_manager(task_executor);
+	let task_manager = new_task_manager(handle);
 	let spawn_handle = task_manager.spawn_handle();
 	let drop_tester = DropTester::new();
 	spawn_handle.spawn("task1", run_background_task(drop_tester.new_ref()));
@@ -111,9 +110,8 @@ fn ensure_tasks_are_awaited_on_shutdown() {
 fn ensure_keep_alive_during_shutdown() {
 	let runtime = tokio::runtime::Runtime::new().unwrap();
 	let handle = runtime.handle().clone();
-	let task_executor: TaskExecutor = (move |future, _| handle.spawn(future).map(|_| ())).into();
 
-	let mut task_manager = new_task_manager(task_executor);
+	let mut task_manager = new_task_manager(handle);
 	let spawn_handle = task_manager.spawn_handle();
 	let drop_tester = DropTester::new();
 	task_manager.keep_alive(drop_tester.new_ref());
@@ -130,9 +128,8 @@ fn ensure_keep_alive_during_shutdown() {
 fn ensure_blocking_futures_are_awaited_on_shutdown() {
 	let runtime = tokio::runtime::Runtime::new().unwrap();
 	let handle = runtime.handle().clone();
-	let task_executor: TaskExecutor = (move |future, _| handle.spawn(future).map(|_| ())).into();
 
-	let task_manager = new_task_manager(task_executor);
+	let task_manager = new_task_manager(handle);
 	let spawn_handle = task_manager.spawn_handle();
 	let drop_tester = DropTester::new();
 	spawn_handle.spawn(
@@ -155,9 +152,8 @@ fn ensure_blocking_futures_are_awaited_on_shutdown() {
 fn ensure_no_task_can_be_spawn_after_terminate() {
 	let runtime = tokio::runtime::Runtime::new().unwrap();
 	let handle = runtime.handle().clone();
-	let task_executor: TaskExecutor = (move |future, _| handle.spawn(future).map(|_| ())).into();
 
-	let mut task_manager = new_task_manager(task_executor);
+	let mut task_manager = new_task_manager(handle);
 	let spawn_handle = task_manager.spawn_handle();
 	let drop_tester = DropTester::new();
 	spawn_handle.spawn("task1", run_background_task(drop_tester.new_ref()));
@@ -176,9 +172,8 @@ fn ensure_no_task_can_be_spawn_after_terminate() {
 fn ensure_task_manager_future_ends_when_task_manager_terminated() {
 	let runtime = tokio::runtime::Runtime::new().unwrap();
 	let handle = runtime.handle().clone();
-	let task_executor: TaskExecutor = (move |future, _| handle.spawn(future).map(|_| ())).into();
 
-	let mut task_manager = new_task_manager(task_executor);
+	let mut task_manager = new_task_manager(handle);
 	let spawn_handle = task_manager.spawn_handle();
 	let drop_tester = DropTester::new();
 	spawn_handle.spawn("task1", run_background_task(drop_tester.new_ref()));
@@ -197,9 +192,8 @@ fn ensure_task_manager_future_ends_when_task_manager_terminated() {
 fn ensure_task_manager_future_ends_with_error_when_essential_task_fails() {
 	let runtime = tokio::runtime::Runtime::new().unwrap();
 	let handle = runtime.handle().clone();
-	let task_executor: TaskExecutor = (move |future, _| handle.spawn(future).map(|_| ())).into();
 
-	let mut task_manager = new_task_manager(task_executor);
+	let mut task_manager = new_task_manager(handle);
 	let spawn_handle = task_manager.spawn_handle();
 	let spawn_essential_handle = task_manager.spawn_essential_handle();
 	let drop_tester = DropTester::new();
@@ -222,12 +216,11 @@ fn ensure_task_manager_future_ends_with_error_when_essential_task_fails() {
 fn ensure_children_tasks_ends_when_task_manager_terminated() {
 	let runtime = tokio::runtime::Runtime::new().unwrap();
 	let handle = runtime.handle().clone();
-	let task_executor: TaskExecutor = (move |future, _| handle.spawn(future).map(|_| ())).into();
 
-	let mut task_manager = new_task_manager(task_executor.clone());
-	let child_1 = new_task_manager(task_executor.clone());
+	let mut task_manager = new_task_manager(handle.clone());
+	let child_1 = new_task_manager(handle.clone());
 	let spawn_handle_child_1 = child_1.spawn_handle();
-	let child_2 = new_task_manager(task_executor.clone());
+	let child_2 = new_task_manager(handle.clone());
 	let spawn_handle_child_2 = child_2.spawn_handle();
 	task_manager.add_child(child_1);
 	task_manager.add_child(child_2);
@@ -251,13 +244,12 @@ fn ensure_children_tasks_ends_when_task_manager_terminated() {
 fn ensure_task_manager_future_ends_with_error_when_childs_essential_task_fails() {
 	let runtime = tokio::runtime::Runtime::new().unwrap();
 	let handle = runtime.handle().clone();
-	let task_executor: TaskExecutor = (move |future, _| handle.spawn(future).map(|_| ())).into();
 
-	let mut task_manager = new_task_manager(task_executor.clone());
-	let child_1 = new_task_manager(task_executor.clone());
+	let mut task_manager = new_task_manager(handle.clone());
+	let child_1 = new_task_manager(handle.clone());
 	let spawn_handle_child_1 = child_1.spawn_handle();
 	let spawn_essential_handle_child_1 = child_1.spawn_essential_handle();
-	let child_2 = new_task_manager(task_executor.clone());
+	let child_2 = new_task_manager(handle.clone());
 	let spawn_handle_child_2 = child_2.spawn_handle();
 	task_manager.add_child(child_1);
 	task_manager.add_child(child_2);
@@ -284,12 +276,11 @@ fn ensure_task_manager_future_ends_with_error_when_childs_essential_task_fails()
 fn ensure_task_manager_future_continues_when_childs_not_essential_task_fails() {
 	let runtime = tokio::runtime::Runtime::new().unwrap();
 	let handle = runtime.handle().clone();
-	let task_executor: TaskExecutor = (move |future, _| handle.spawn(future).map(|_| ())).into();
 
-	let mut task_manager = new_task_manager(task_executor.clone());
-	let child_1 = new_task_manager(task_executor.clone());
+	let mut task_manager = new_task_manager(handle.clone());
+	let child_1 = new_task_manager(handle.clone());
 	let spawn_handle_child_1 = child_1.spawn_handle();
-	let child_2 = new_task_manager(task_executor.clone());
+	let child_2 = new_task_manager(handle.clone());
 	let spawn_handle_child_2 = child_2.spawn_handle();
 	task_manager.add_child(child_1);
 	task_manager.add_child(child_2);