Use SpawnNamed to give tasks names (#1379)

polkadot/network/src/protocol/mod.rs

@@ -26,7 +26,7 @@ use codec::{Decode, Encode};
 use futures::channel::{mpsc, oneshot};
 use futures::future::Either;
 use futures::prelude::*;
-use futures::task::{Spawn, SpawnExt, Context, Poll};
+use futures::task::{Context, Poll};
 use futures::stream::{FuturesUnordered, StreamFuture};
 use log::{debug, trace};
 
@@ -44,6 +44,7 @@ use polkadot_validation::{
 use sc_network::{ObservedRole, Event, PeerId};
 use sp_api::ProvideRuntimeApi;
 use sp_runtime::ConsensusEngineId;
+use sp_core::traits::SpawnNamed;
 
 use std::collections::{hash_map::{Entry, HashMap}, HashSet};
 use std::pin::Pin;
@@ -126,7 +127,9 @@ enum ServiceToWorkerMsg {
 /// Messages from a background task to the main worker task.
 enum BackgroundToWorkerMsg {
 	// Spawn a given future.
-	Spawn(future::BoxFuture<'static, ()>),
+	//
+	// The name is used for the future task.
+	Spawn(&'static str, future::BoxFuture<'static, ()>),
 }
 
 /// Operations that a handle to an underlying network service should provide.
@@ -221,7 +224,7 @@ pub fn start<C, Api, SP>(
 	C: ChainContext + 'static,
 	Api: ProvideRuntimeApi<Block> + Send + Sync + 'static,
 	Api::Api: ParachainHost<Block, Error = sp_blockchain::Error>,
-	SP: Spawn + Clone + Send + 'static,
+	SP: SpawnNamed + Clone + Send + 'static,
 {
 	const SERVICE_TO_WORKER_BUF: usize = 256;
 
@@ -234,67 +237,73 @@ pub fn start<C, Api, SP>(
 		chain_context,
 		&executor,
 	);
-	executor.spawn(worker_loop(
-		config,
-		service.clone(),
-		gossip_validator,
-		api,
-		worker_receiver,
-		executor.clone(),
-	))?;
+	executor.spawn(
+		"polkadot-network-worker",
+		worker_loop(
+			config,
+			service.clone(),
+			gossip_validator,
+			api,
+			worker_receiver,
+			executor.clone(),
+		).boxed(),
+	);
 
 	let polkadot_service = Service {
 		sender: worker_sender.clone(),
 		network_service: service.clone(),
 	};
 
-	executor.spawn(async move {
-		while let Some(event) = event_stream.next().await {
-			let res = match event {
-				Event::Dht(_) => continue,
-				Event::NotificationStreamOpened {
-					remote,
-					engine_id,
-					role,
-				} => {
-					if engine_id != POLKADOT_ENGINE_ID { continue }
+	executor.spawn(
+		"polkadot-network-notifications",
+		async move {
+			while let Some(event) = event_stream.next().await {
+				let res = match event {
+					Event::Dht(_) => continue,
+					Event::NotificationStreamOpened {
+						remote,
+						engine_id,
+						role,
+					} => {
+						if engine_id != POLKADOT_ENGINE_ID { continue }
 
-					worker_sender.send(ServiceToWorkerMsg::PeerConnected(remote, role)).await
-				},
-				Event::NotificationStreamClosed {
-					remote,
-					engine_id,
-				} => {
-					if engine_id != POLKADOT_ENGINE_ID { continue }
+						worker_sender.send(ServiceToWorkerMsg::PeerConnected(remote, role)).await
+					},
+					Event::NotificationStreamClosed {
+						remote,
+						engine_id,
+					} => {
+						if engine_id != POLKADOT_ENGINE_ID { continue }
 
-					worker_sender.send(ServiceToWorkerMsg::PeerDisconnected(remote)).await
-				},
-				Event::NotificationsReceived {
-					remote,
-					messages,
-				} => {
-					let our_notifications = messages.into_iter()
-						.filter_map(|(engine, message)| if engine == POLKADOT_ENGINE_ID {
-							Some(message)
-						} else {
-							None
-						})
-						.collect();
+						worker_sender.send(ServiceToWorkerMsg::PeerDisconnected(remote)).await
+					},
+					Event::NotificationsReceived {
+						remote,
+						messages,
+					} => {
+						let our_notifications = messages.into_iter()
+							.filter_map(|(engine, message)| if engine == POLKADOT_ENGINE_ID {
+								Some(message)
+							} else {
+								None
+							})
+							.collect();
 
-					worker_sender.send(
-						ServiceToWorkerMsg::PeerMessage(remote, our_notifications)
-					).await
-				}
-			};
+						worker_sender.send(
+							ServiceToWorkerMsg::PeerMessage(remote, our_notifications)
+						).await
+					}
+				};
 
-			if let Err(e) = res {
-				// full is impossible here, as we've `await`ed the value being sent.
-				if e.is_disconnected() {
-					break
+				if let Err(e) = res {
+					// full is impossible here, as we've `await`ed the value being sent.
+					if e.is_disconnected() {
+						break
+					}
 				}
 			}
-		}
-	})?;
+		}.boxed(),
+	);
 
 	Ok(polkadot_service)
 }
@@ -845,7 +854,7 @@ struct Worker<Api, Sp, Gossip> {
 impl<Api, Sp, Gossip> Worker<Api, Sp, Gossip> where
 	Api: ProvideRuntimeApi<Block> + Send + Sync + 'static,
 	Api::Api: ParachainHost<Block, Error = sp_blockchain::Error>,
-	Sp: Spawn + Clone,
+	Sp: SpawnNamed + Clone,
 	Gossip: GossipOps,
 {
 	// spawns a background task to spawn consensus networking.
@@ -888,14 +897,17 @@ impl<Api, Sp, Gossip> Worker<Api, Sp, Gossip> where
 
 		// glue the incoming messages, shared table, and validation
 		// work together.
-		let _ = self.executor.spawn(statement_import_loop(
-			relay_parent,
-			table,
-			self.api.clone(),
-			self.gossip_handle.clone(),
-			self.background_to_main_sender.clone(),
-			exit,
-		));
+		self.executor.spawn(
+			"polkadot-statement-import-loop",
+			statement_import_loop(
+				relay_parent,
+				table,
+				self.api.clone(),
+				self.gossip_handle.clone(),
+				self.background_to_main_sender.clone(),
+				exit,
+			).boxed(),
+		);
 	}
 
 	fn handle_service_message(&mut self, message: ServiceToWorkerMsg) {
@@ -932,12 +944,15 @@ impl<Api, Sp, Gossip> Worker<Api, Sp, Gossip> where
 				// before placing in the pool, so we can safely check by candidate hash.
 				let get_msg = fetch_pov_from_gossip(&candidate, &self.gossip_handle);
 
-				let _ = self.executor.spawn(async move {
-					let res = future::select(get_msg, AwaitCanceled { inner: &mut sender }).await;
-					if let Either::Left((pov_block, _)) = res {
-						let _ = sender.send(pov_block);
-					}
-				});
+				self.executor.spawn(
+					"polkadot-fetch-pov-block",
+					async move {
+						let res = future::select(get_msg, AwaitCanceled { inner: &mut sender }).await;
+						if let Either::Left((pov_block, _)) = res {
+							let _ = sender.send(pov_block);
+						}
+					}.boxed(),
+				);
 			}
 			ServiceToWorkerMsg::FetchErasureChunk(candidate_hash, validator_index, mut sender) => {
 				let topic = crate::erasure_coding_topic(&candidate_hash);
@@ -963,12 +978,15 @@ impl<Api, Sp, Gossip> Worker<Api, Sp, Gossip> where
 						"gossip message streams do not conclude early; qed"
 					));
 
-				let _ = self.executor.spawn(async move {
-					let res = future::select(get_msg, AwaitCanceled { inner: &mut sender }).await;
-					if let Either::Left((chunk, _)) = res {
-						let _ = sender.send(chunk);
-					}
-				});
+				self.executor.spawn(
+					"polkadot-fetch-erasure-chunk",
+					async move {
+						let res = future::select(get_msg, AwaitCanceled { inner: &mut sender }).await;
+						if let Either::Left((chunk, _)) = res {
+							let _ = sender.send(chunk);
+						}
+					}.boxed(),
+				);
 			}
 			ServiceToWorkerMsg::DistributeErasureChunk(candidate_hash, erasure_chunk) => {
 				let topic = crate::erasure_coding_topic(&candidate_hash);
@@ -1017,8 +1035,8 @@ impl<Api, Sp, Gossip> Worker<Api, Sp, Gossip> where
 
 	fn handle_background_message(&mut self, message: BackgroundToWorkerMsg) {
 		match message {
-			BackgroundToWorkerMsg::Spawn(task) => {
-				let _ = self.executor.spawn(task);
+			BackgroundToWorkerMsg::Spawn(name, task) => {
+				let _ = self.executor.spawn(name, task);
 			}
 		}
 	}
@@ -1068,7 +1086,7 @@ async fn worker_loop<Api, Sp>(
 ) where
 	Api: ProvideRuntimeApi<Block> + Send + Sync + 'static,
 	Api::Api: ParachainHost<Block, Error = sp_blockchain::Error>,
-	Sp: Spawn + Clone,
+	Sp: SpawnNamed + Clone,
 {
 	const BACKGROUND_TO_MAIN_BUF: usize = 16;
 
@@ -1250,7 +1268,7 @@ async fn statement_import_loop<Api>(
 
 					let work = future::select(work.boxed(), exit.clone()).map(drop);
 					if let Err(_) = to_worker.send(
-						BackgroundToWorkerMsg::Spawn(work.boxed())
+						BackgroundToWorkerMsg::Spawn("polkadot-statement-import-loop-sub-task", work.boxed())
 					).await {
 						// can fail only if remote has hung up - worker is dead,
 						// we should die too. this is defensive, since the exit future

polkadot/network/src/protocol/tests.rs

@@ -30,11 +30,24 @@ use av_store::{Store as AvailabilityStore, ErasureNetworking};
 use sc_network_gossip::TopicNotification;
 use sp_api::{ApiRef, ProvideRuntimeApi};
 use sp_runtime::traits::Block as BlockT;
-use sp_core::crypto::Pair;
+use sp_core::{crypto::Pair, traits::SpawnNamed};
 use sp_keyring::Sr25519Keyring;
 
-use futures::executor::LocalPool;
-use futures::task::LocalSpawnExt;
+use futures::executor::{LocalPool, LocalSpawner};
+use futures::task::{LocalSpawnExt, SpawnExt};
+
+#[derive(Clone)]
+struct Executor(LocalSpawner);
+
+impl SpawnNamed for Executor {
+	fn spawn(&self, _: &'static str, future: futures::future::BoxFuture<'static, ()>) {
+		self.0.spawn_local(future).unwrap();
+	}
+
+	fn spawn_blocking(&self, name: &'static str, future: futures::future::BoxFuture<'static, ()>) {
+		self.spawn(name, future);
+	}
+}
 
 #[derive(Default)]
 pub struct MockNetworkOps {
@@ -243,7 +256,7 @@ fn test_setup(config: Config) -> (
 		mock_gossip.clone(),
 		api.clone(),
 		worker_rx,
-		pool.spawner(),
+		Executor(pool.spawner()),
 	);
 
 	let service = Service {