use ThreadPool to execute spawn_worker(fn)

core/offchain/Cargo.toml

@@ -16,6 +16,8 @@ futures-timer = "0.4.0"
 hyper = "0.12.35"
 hyper-tls = "0.3.2"
 log = "0.4.8"
+threadpool = "1.0"
+num_cpus = "1.6"
 offchain-primitives = { package = "substrate-offchain-primitives", path = "./primitives" }
 codec = { package = "parity-scale-codec", version = "1.0.0", features = ["derive"] }
 parking_lot = "0.9.0"

core/offchain/src/lib.rs

@@ -33,19 +33,20 @@
 
 #![warn(missing_docs)]
 
-use std::{
-	fmt,
-	marker::PhantomData,
-	sync::Arc,
-};
+extern crate num_cpus;
+use std::{fmt, marker::PhantomData, sync::Arc};
 
 use client::runtime_api::ApiExt;
 use futures::future::Future;
 use log::{debug, warn};
 use network::NetworkStateInfo;
 use primitives::{offchain, ExecutionContext};
-use sr_primitives::{generic::BlockId, traits::{self, ProvideRuntimeApi}};
-use transaction_pool::txpool::{Pool, ChainApi};
+use sr_primitives::{
+    generic::BlockId,
+    traits::{self, ProvideRuntimeApi},
+};
+use threadpool::ThreadPool;
+use transaction_pool::txpool::{ChainApi, Pool};
 
 mod api;
 
@@ -55,132 +56,132 @@ pub use offchain_primitives::OffchainWorkerApi;
 
 /// An offchain workers manager.
 pub struct OffchainWorkers<Client, Storage, Block: traits::Block> {
-	client: Arc<Client>,
-	db: Storage,
-	_block: PhantomData<Block>,
+    client: Arc<Client>,
+    db: Storage,
+    _block: PhantomData<Block>,
+    thread_pool: ThreadPool,
 }
 
 impl<Client, Storage, Block: traits::Block> OffchainWorkers<Client, Storage, Block> {
-	/// Creates new `OffchainWorkers`.
-	pub fn new(client: Arc<Client>, db: Storage) -> Self {
-		Self {
-			client,
-			db,
-			_block: PhantomData,
-		}
-	}
+    /// Creates new `OffchainWorkers`.
+    pub fn new(client: Arc<Client>, db: Storage) -> Self {
+        Self {
+            client,
+            db,
+            _block: PhantomData,
+            thread_pool: ThreadPool::new(num_cpus::get()),
+        }
+    }
 }
 
-impl<Client, Storage, Block: traits::Block> fmt::Debug for OffchainWorkers<
-	Client,
-	Storage,
-	Block,
-> {
-	fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
-		f.debug_tuple("OffchainWorkers").finish()
-	}
+impl<Client, Storage, Block: traits::Block> fmt::Debug for OffchainWorkers<Client, Storage, Block> {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        f.debug_tuple("OffchainWorkers").finish()
+    }
 }
 
-impl<Client, Storage, Block> OffchainWorkers<
-	Client,
-	Storage,
-	Block,
-> where
-	Block: traits::Block,
-	Client: ProvideRuntimeApi + Send + Sync + 'static,
-	Client::Api: OffchainWorkerApi<Block>,
-	Storage: client::backend::OffchainStorage + 'static,
+impl<Client, Storage, Block> OffchainWorkers<Client, Storage, Block>
+where
+    Block: traits::Block,
+    Client: ProvideRuntimeApi + Send + Sync + 'static,
+    Client::Api: OffchainWorkerApi<Block>,
+    Storage: client::backend::OffchainStorage + 'static,
 {
-	/// Start the offchain workers after given block.
-	#[must_use]
-	pub fn on_block_imported<A>(
-		&self,
-		number: &<Block::Header as traits::Header>::Number,
-		pool: &Arc<Pool<A>>,
-		network_state: Arc<dyn NetworkStateInfo + Send + Sync>,
-		is_validator: bool,
-	) -> impl Future<Output = ()> where A: ChainApi<Block=Block> + 'static {
-		let runtime = self.client.runtime_api();
-		let at = BlockId::number(*number);
-		let has_api = runtime.has_api::<dyn OffchainWorkerApi<Block>>(&at);
-		debug!("Checking offchain workers at {:?}: {:?}", at, has_api);
-
-		if has_api.unwrap_or(false) {
-			let (api, runner) = api::AsyncApi::new(
-				pool.clone(),
-				self.db.clone(),
-				at.clone(),
-				network_state.clone(),
-				is_validator,
-			);
-			debug!("Spawning offchain workers at {:?}", at);
-			let number = *number;
-			let client = self.client.clone();
-			spawn_worker(move || {
-				let runtime = client.runtime_api();
-				let api = Box::new(api);
-				debug!("Running offchain workers at {:?}", at);
-				let run = runtime.offchain_worker_with_context(
-					&at,
-					ExecutionContext::OffchainCall(Some((api, offchain::Capabilities::all()))),
-					number,
-				);
-				if let Err(e) =	run {
-					log::error!("Error running offchain workers at {:?}: {:?}", at, e);
-				}
-			});
-			futures::future::Either::Left(runner.process())
-		} else {
-			futures::future::Either::Right(futures::future::ready(()))
-		}
-	}
-}
-
-/// Spawns a new offchain worker.
-///
-/// We spawn offchain workers for each block in a separate thread,
-/// since they can run for a significant amount of time
-/// in a blocking fashion and we don't want to block the runtime.
-///
-/// Note that we should avoid that if we switch to future-based runtime in the future,
-/// alternatively:
-/// TODO [ToDr] (#1458) we can consider using a thread pool instead.
-fn spawn_worker(f: impl FnOnce() -> () + Send + 'static) {
-	std::thread::spawn(f);
+    /// Start the offchain workers after given block.
+    #[must_use]
+    pub fn on_block_imported<A>(
+        &self,
+        number: &<Block::Header as traits::Header>::Number,
+        pool: &Arc<Pool<A>>,
+        network_state: Arc<dyn NetworkStateInfo + Send + Sync>,
+        is_validator: bool,
+    ) -> impl Future<Output = ()>
+    where
+        A: ChainApi<Block = Block> + 'static,
+    {
+        let runtime = self.client.runtime_api();
+        let at = BlockId::number(*number);
+        let has_api = runtime.has_api::<dyn OffchainWorkerApi<Block>>(&at);
+        debug!("Checking offchain workers at {:?}: {:?}", at, has_api);
+
+        if has_api.unwrap_or(false) {
+            let (api, runner) = api::AsyncApi::new(
+                pool.clone(),
+                self.db.clone(),
+                at.clone(),
+                network_state.clone(),
+                is_validator,
+            );
+            debug!("Spawning offchain workers at {:?}", at);
+            let number = *number;
+            let client = self.client.clone();
+            self.spawn_worker(move || {
+                let runtime = client.runtime_api();
+                let api = Box::new(api);
+                debug!("Running offchain workers at {:?}", at);
+                let run = runtime.offchain_worker_with_context(
+                    &at,
+                    ExecutionContext::OffchainCall(Some((api, offchain::Capabilities::all()))),
+                    number,
+                );
+                if let Err(e) = run {
+                    log::error!("Error running offchain workers at {:?}: {:?}", at, e);
+                }
+            });
+            futures::future::Either::Left(runner.process())
+        } else {
+            futures::future::Either::Right(futures::future::ready(()))
+        }
+    }
+
+    /// Spawns a new offchain worker.
+    ///
+    /// We spawn offchain workers for each block in a separate thread,
+    /// since they can run for a significant amount of time
+    /// in a blocking fashion and we don't want to block the runtime.
+    ///
+    /// Note that we should avoid that if we switch to future-based runtime in the future,
+    /// alternatively:
+    fn spawn_worker(&self, f: impl FnOnce() -> () + Send + 'static) {
+        self.thread_pool.execute(f);
+    }
 }
 
 #[cfg(test)]
 mod tests {
-	use super::*;
-	use network::{Multiaddr, PeerId};
-
-	struct MockNetworkStateInfo();
-
-	impl NetworkStateInfo for MockNetworkStateInfo {
-		fn external_addresses(&self) -> Vec<Multiaddr> {
-			Vec::new()
-		}
-
-		fn peer_id(&self) -> PeerId {
-			PeerId::random()
-		}
-	}
-
-	#[test]
-	fn should_call_into_runtime_and_produce_extrinsic() {
-		// given
-		let _ = env_logger::try_init();
-		let client = Arc::new(test_client::new());
-		let pool = Arc::new(Pool::new(Default::default(), transaction_pool::FullChainApi::new(client.clone())));
-		let db = client_db::offchain::LocalStorage::new_test();
-		let network_state = Arc::new(MockNetworkStateInfo());
-
-		// when
-		let offchain = OffchainWorkers::new(client, db);
-		futures::executor::block_on(offchain.on_block_imported(&0u64, &pool, network_state, false));
-
-		// then
-		assert_eq!(pool.status().ready, 1);
-		assert_eq!(pool.ready().next().unwrap().is_propagateable(), false);
-	}
+    use super::*;
+    use network::{Multiaddr, PeerId};
+
+    struct MockNetworkStateInfo();
+
+    impl NetworkStateInfo for MockNetworkStateInfo {
+        fn external_addresses(&self) -> Vec<Multiaddr> {
+            Vec::new()
+        }
+
+        fn peer_id(&self) -> PeerId {
+            PeerId::random()
+        }
+    }
+
+    #[test]
+    fn should_call_into_runtime_and_produce_extrinsic() {
+        // given
+        let _ = env_logger::try_init();
+        let client = Arc::new(test_client::new());
+        let pool = Arc::new(Pool::new(
+            Default::default(),
+            transaction_pool::FullChainApi::new(client.clone()),
+        ));
+        let db = client_db::offchain::LocalStorage::new_test();
+        let network_state = Arc::new(MockNetworkStateInfo());
+
+        // when
+        let offchain = OffchainWorkers::new(client, db);
+        futures::executor::block_on(offchain.on_block_imported(&0u64, &pool, network_state, false));
+
+        // then
+        assert_eq!(pool.status().ready, 1);
+        assert_eq!(pool.ready().next().unwrap().is_propagateable(), false);
+    }
 }