Spawn the network worker with spawn_blocking (#5909)

* Spawn the network worker with spawn_blocking

* Some comment adjustments

* Fix shutdown not working

substrate/client/cli/src/runner.rs

@@ -217,12 +217,16 @@ impl<C: SubstrateCli> Runner<C> {
 		// and drop the runtime first.
 		let _telemetry = service.telemetry();
 
-		let f = service.fuse();
-		pin_mut!(f);
+		{
+			let f = service.fuse();
+			self.tokio_runtime
+				.block_on(main(f))
+				.map_err(|e| e.to_string())?;
+		}
 
-		self.tokio_runtime
-			.block_on(main(f))
-			.map_err(|e| e.to_string())?;
+		// The `service` **must** have been destroyed here for the shutdown signal to propagate
+		// to all the tasks. Dropping `tokio_runtime` will block the thread until all tasks have
+		// shut down.
 		drop(self.tokio_runtime);
 
 		Ok(())

substrate/client/service/src/builder.rs

@@ -1161,7 +1161,14 @@ ServiceBuilder<
 		// This is used internally, so don't restrict access to unsafe RPC
 		let rpc_handlers = gen_handler(sc_rpc::DenyUnsafe::No);
 
-		spawn_handle.spawn(
+		// The network worker is responsible for gathering all network messages and processing
+		// them. This is quite a heavy task, and at the time of the writing of this comment it
+		// frequently happens that this future takes several seconds or in some situations
+		// even more than a minute until it has processed its entire queue. This is clearly an
+		// issue, and ideally we would like to fix the network future to take as little time as
+		// possible, but we also take the extra harm-prevention measure to execute the networking
+		// future using `spawn_blocking`.
+		spawn_handle.spawn_blocking(
 			"network-worker",
 			build_network_future(
 				config.role.clone(),