Replace task_executor fn's input by proper TaskExecutor type (cleaner)

client/cli/src/config.rs

@@ -28,15 +28,12 @@ use names::{Generator, Name};
 use sc_client_api::execution_extensions::ExecutionStrategies;
 use sc_service::config::{
 	BasePath, Configuration, DatabaseConfig, ExtTransport, KeystoreConfig, NetworkConfiguration,
-	NodeKeyConfig, OffchainWorkerConfig, PrometheusConfig, PruningMode, Role, RpcMethods, TaskType,
-	TelemetryEndpoints, TransactionPoolOptions, WasmExecutionMethod,
+	NodeKeyConfig, OffchainWorkerConfig, PrometheusConfig, PruningMode, Role, RpcMethods,
+	TaskExecutor, TelemetryEndpoints, TransactionPoolOptions, WasmExecutionMethod,
 };
 use sc_service::{ChainSpec, TracingReceiver};
-use std::future::Future;
 use std::net::SocketAddr;
 use std::path::PathBuf;
-use std::pin::Pin;
-use std::sync::Arc;
 
 /// The maximum number of characters for a node name.
 pub(crate) const NODE_NAME_MAX_LENGTH: usize = 32;
@@ -402,7 +399,7 @@ pub trait CliConfiguration: Sized {
 	fn create_configuration<C: SubstrateCli>(
 		&self,
 		cli: &C,
-		task_executor: Arc<dyn Fn(Pin<Box<dyn Future<Output = ()> + Send>>, TaskType) + Send + Sync>,
+		task_executor: TaskExecutor,
 	) -> Result<Configuration> {
 		let is_dev = self.is_dev()?;
 		let chain_id = self.chain_id(is_dev)?;
@@ -432,7 +429,7 @@ pub trait CliConfiguration: Sized {
 		Ok(Configuration {
 			impl_name: C::impl_name(),
 			impl_version: C::impl_version(),
-			task_executor: task_executor.into(),
+			task_executor,
 			transaction_pool: self.transaction_pool()?,
 			network: self.network_config(
 				&chain_spec,

client/cli/src/lib.rs

@@ -37,11 +37,8 @@ use log::info;
 pub use params::*;
 use regex::Regex;
 pub use runner::*;
-use sc_service::{ChainSpec, Configuration, TaskType};
-use std::future::Future;
+use sc_service::{ChainSpec, Configuration, TaskExecutor};
 use std::io::Write;
-use std::pin::Pin;
-use std::sync::Arc;
 pub use structopt;
 use structopt::{
 	clap::{self, AppSettings},
@@ -199,7 +196,7 @@ pub trait SubstrateCli: Sized {
 	fn create_configuration<T: CliConfiguration>(
 		&self,
 		command: &T,
-		task_executor: Arc<dyn Fn(Pin<Box<dyn Future<Output = ()> + Send>>, TaskType) + Send + Sync>,
+		task_executor: TaskExecutor,
 	) -> error::Result<Configuration> {
 		command.create_configuration(self, task_executor)
 	}

client/cli/src/runner.rs

@@ -25,11 +25,13 @@ use futures::pin_mut;
 use futures::select;
 use futures::{future, future::FutureExt, Future};
 use log::info;
-use sc_service::{AbstractService, Configuration, Role, ServiceBuilderCommand, TaskType};
+use sc_service::{
+	AbstractService, Configuration, Role, ServiceBuilderCommand, TaskType, config::TaskExecutor,
+};
 use sp_runtime::traits::{Block as BlockT, Header as HeaderT};
 use sp_utils::metrics::{TOKIO_THREADS_ALIVE, TOKIO_THREADS_TOTAL};
 use sp_version::RuntimeVersion;
-use std::{fmt::Debug, marker::PhantomData, str::FromStr, sync::Arc};
+use std::{fmt::Debug, marker::PhantomData, str::FromStr};
 
 #[cfg(target_family = "unix")]
 async fn main<F, E>(func: F) -> std::result::Result<(), Box<dyn std::error::Error>>
@@ -119,7 +121,7 @@ impl<C: SubstrateCli> Runner<C> {
 		let tokio_runtime = build_runtime()?;
 		let runtime_handle = tokio_runtime.handle().clone();
 
-		let task_executor = Arc::new(
+		let task_executor = TaskExecutor::from_fn(
 			move |fut, task_type| {
 				match task_type {
 					TaskType::Async => { runtime_handle.spawn(fut); }

client/service/src/config.rs

@@ -254,26 +254,29 @@ impl std::convert::From<PathBuf> for BasePath {
 	}
 }
 
+type TaskExecutorInner = Arc<dyn Fn(Pin<Box<dyn Future<Output = ()> + Send>>, TaskType) + Send + Sync>;
+
 /// Callable object that execute tasks.
 #[derive(Clone)]
-pub struct TaskExecutor(Arc<dyn Fn(Pin<Box<dyn Future<Output = ()> + Send>>, TaskType) + Send + Sync>);
+pub struct TaskExecutor(TaskExecutorInner);
 
 impl std::fmt::Debug for TaskExecutor {
 	fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
 		write!(f, "TaskExecutor")
 	}
 }
 
-impl std::convert::From<Arc<dyn Fn(Pin<Box<dyn Future<Output = ()> + Send>>, TaskType) + Send + Sync>>
-for TaskExecutor {
-	fn from(x: Arc<dyn Fn(Pin<Box<dyn Future<Output = ()> + Send>>, TaskType) + Send + Sync>)
+/*
+impl std::convert::From<TaskExecutorInner> for TaskExecutor {
+	fn from(x: TaskExecutorInner)
 	-> Self {
 		Self(x)
 	}
 }
+*/
 
 impl std::ops::Deref for TaskExecutor {
-	type Target = Arc<dyn Fn(Pin<Box<dyn Future<Output = ()> + Send>>, TaskType) + Send + Sync>;
+	type Target = TaskExecutorInner;
 
 	fn deref(&self) -> &Self::Target {
 		&self.0
@@ -282,7 +285,9 @@ impl std::ops::Deref for TaskExecutor {
 
 impl TaskExecutor {
 	/// Create a `TaskExecutor` from a function
-	pub fn from_fn(f: impl Fn(Pin<Box<dyn Future<Output = ()> + Send>>, TaskType) + Send + Sync + 'static) -> Self {
+	pub fn from_fn(
+		f: impl Fn(Pin<Box<dyn Future<Output = ()> + Send>>, TaskType) + Send + Sync + 'static,
+	) -> Self {
 		Self(Arc::new(f))
 	}
 }

client/service/src/lib.rs

@@ -66,7 +66,9 @@ pub use self::builder::{
 	ServiceBuilder, ServiceBuilderCommand, TFullClient, TLightClient, TFullBackend, TLightBackend,
 	TFullCallExecutor, TLightCallExecutor, RpcExtensionBuilder,
 };
-pub use config::{BasePath, Configuration, DatabaseConfig, PruningMode, Role, RpcMethods, TaskType};
+pub use config::{
+	BasePath, Configuration, DatabaseConfig, PruningMode, Role, RpcMethods, TaskExecutor, TaskType,
+};
 pub use sc_chain_spec::{
 	ChainSpec, GenericChainSpec, Properties, RuntimeGenesis, Extension as ChainSpecExtension,
 	NoExtension, ChainType,

client/service/test/src/lib.rs

@@ -38,7 +38,7 @@ use sc_service::{
 	RuntimeGenesis,
 	Role,
 	Error,
-	TaskType,
+	TaskExecutor,
 };
 use sp_blockchain::HeaderBackend;
 use sc_network::{multiaddr, Multiaddr};
@@ -142,7 +142,7 @@ fn node_config<G: RuntimeGenesis + 'static, E: ChainSpecExtension + Clone + 'sta
 	index: usize,
 	spec: &GenericChainSpec<G, E>,
 	role: Role,
-	task_executor: Arc<dyn Fn(Pin<Box<dyn futures::Future<Output = ()> + Send>>, TaskType) + Send + Sync>,
+	task_executor: TaskExecutor,
 	key_seed: Option<String>,
 	base_port: u16,
 	root: &TempDir,
@@ -176,7 +176,7 @@ fn node_config<G: RuntimeGenesis + 'static, E: ChainSpecExtension + Clone + 'sta
 		impl_name: "network-test-impl",
 		impl_version: "0.1",
 		role,
-		task_executor: task_executor.into(),
+		task_executor,
 		transaction_pool: Default::default(),
 		network: network_config,
 		keystore: KeystoreConfig::Path {
@@ -255,17 +255,21 @@ impl<G, E, F, L, U> TestNet<G, E, F, L, U> where
 		authorities: impl Iterator<Item = (String, impl FnOnce(Configuration) -> Result<(F, U), Error>)>
 	) {
 		let executor = self.runtime.executor();
+		let task_executor = {
+			let executor = executor.clone();
+			TaskExecutor::from_fn(
+				move |fut: Pin<Box<dyn futures::Future<Output = ()> + Send>>, _| {
+					executor.spawn(fut.unit_error().compat());
+				},
+			)
+		};
 
 		for (key, authority) in authorities {
-			let task_executor = {
-				let executor = executor.clone();
-				Arc::new(move |fut: Pin<Box<dyn futures::Future<Output = ()> + Send>>, _| executor.spawn(fut.unit_error().compat()))
-			};
 			let node_config = node_config(
 				self.nodes,
 				&self.chain_spec,
 				Role::Authority { sentry_nodes: Vec::new() },
-				task_executor,
+				task_executor.clone(),
 				Some(key),
 				self.base_port,
 				&temp,
@@ -281,11 +285,15 @@ impl<G, E, F, L, U> TestNet<G, E, F, L, U> where
 		}
 
 		for full in full {
-			let task_executor = {
-				let executor = executor.clone();
-				Arc::new(move |fut: Pin<Box<dyn futures::Future<Output = ()> + Send>>, _| executor.spawn(fut.unit_error().compat()))
-			};
-			let node_config = node_config(self.nodes, &self.chain_spec, Role::Full, task_executor, None, self.base_port, &temp);
+			let node_config = node_config(
+				self.nodes,
+				&self.chain_spec,
+				Role::Full,
+				task_executor.clone(),
+				None,
+				self.base_port,
+				&temp,
+			);
 			let addr = node_config.network.listen_addresses.iter().next().unwrap().clone();
 			let (service, user_data) = full(node_config).expect("Error creating test node service");
 			let service = SyncService::from(service);
@@ -297,11 +305,15 @@ impl<G, E, F, L, U> TestNet<G, E, F, L, U> where
 		}
 
 		for light in light {
-			let task_executor = {
-				let executor = executor.clone();
-				Arc::new(move |fut: Pin<Box<dyn futures::Future<Output = ()> + Send>>, _| executor.spawn(fut.unit_error().compat()))
-			};
-			let node_config = node_config(self.nodes, &self.chain_spec, Role::Light, task_executor, None, self.base_port, &temp);
+			let node_config = node_config(
+				self.nodes,
+				&self.chain_spec,
+				Role::Light,
+				task_executor.clone(),
+				None,
+				self.base_port,
+				&temp,
+			);
 			let addr = node_config.network.listen_addresses.iter().next().unwrap().clone();
 			let service = SyncService::from(light(node_config).expect("Error creating test node service"));