Use async/await in build_network_future

client/informant/src/lib.rs

@@ -29,7 +29,6 @@ use sp_runtime::traits::{Block as BlockT, Header};
 use sp_transaction_pool::TransactionPool;
 use sp_utils::{status_sinks, mpsc::tracing_unbounded};
 use std::{fmt::Display, sync::Arc, time::Duration, collections::VecDeque};
-use parking_lot::Mutex;
 
 mod display;
 
@@ -82,7 +81,7 @@ impl<T: TransactionPool + MallocSizeOf> TransactionPoolAndMaybeMallogSizeOf for
 /// Builds the informant and returns a `Future` that drives the informant.
 pub fn build<B: BlockT, C>(
 	client: Arc<C>,
-	network_status_sinks: Arc<Mutex<status_sinks::StatusSinks<(NetworkStatus<B>, NetworkState)>>>,
+	network_status_sinks: Arc<status_sinks::StatusSinks<(NetworkStatus<B>, NetworkState)>>,
 	pool: Arc<impl TransactionPoolAndMaybeMallogSizeOf>,
 	format: OutputFormat,
 ) -> impl futures::Future<Output = ()>
@@ -94,7 +93,7 @@ where
 
 	let client_1 = client.clone();
 	let (network_status_sink, network_status_stream) = tracing_unbounded("mpsc_network_status");
-	network_status_sinks.lock().push(Duration::from_millis(5000), network_status_sink);
+	network_status_sinks.push(Duration::from_millis(5000), network_status_sink);
 
 	let display_notifications = network_status_stream
 		.for_each(move |(net_status, _)| {

client/service/src/builder.rs

@@ -1237,7 +1237,7 @@ async fn telemetry_periodic_send<TBl, TBackend, TExec, TRtApi, TExPool>(
 	client: Arc<Client<TBackend, TExec, TBl, TRtApi>>,
 	transaction_pool: Arc<TExPool>,
 	mut metrics_service: MetricsService,
-	network_status_sinks: Arc<Mutex<status_sinks::StatusSinks<(NetworkStatus<TBl>, NetworkState)>>>
+	network_status_sinks: Arc<status_sinks::StatusSinks<(NetworkStatus<TBl>, NetworkState)>>
 )
 	where
 		TBl: BlockT,
@@ -1247,7 +1247,7 @@ async fn telemetry_periodic_send<TBl, TBackend, TExec, TRtApi, TExPool>(
 		TBackend: sc_client_api::backend::Backend<TBl>,
 {
 	let (state_tx, state_rx) = tracing_unbounded::<(NetworkStatus<_>, NetworkState)>("mpsc_netstat1");
-	network_status_sinks.lock().push(std::time::Duration::from_millis(5000), state_tx);
+	network_status_sinks.push(std::time::Duration::from_millis(5000), state_tx);
 	state_rx.for_each(move |(net_status, _)| {
 		let info = client.usage_info();
 		metrics_service.tick(
@@ -1260,11 +1260,11 @@ async fn telemetry_periodic_send<TBl, TBackend, TExec, TRtApi, TExPool>(
 }
 
 async fn telemetry_periodic_network_state<TBl: BlockT>(
-	network_status_sinks: Arc<Mutex<status_sinks::StatusSinks<(NetworkStatus<TBl>, NetworkState)>>>
+	network_status_sinks: Arc<status_sinks::StatusSinks<(NetworkStatus<TBl>, NetworkState)>>
 ) {
 	// Periodically send the network state to the telemetry.
 	let (netstat_tx, netstat_rx) = tracing_unbounded::<(NetworkStatus<_>, NetworkState)>("mpsc_netstat2");
-	network_status_sinks.lock().push(std::time::Duration::from_secs(30), netstat_tx);
+	network_status_sinks.push(std::time::Duration::from_secs(30), netstat_tx);
 	netstat_rx.for_each(move |(_, network_state)| {
 		telemetry!(
 			SUBSTRATE_INFO;
@@ -1429,7 +1429,7 @@ fn build_network<TBl, TBackend, TExec, TRtApi, TExPool, TImpQu>(
 ) -> Result<
 	(
 		Arc<NetworkService<TBl, <TBl as BlockT>::Hash>>,
-		Arc<Mutex<status_sinks::StatusSinks<(NetworkStatus<TBl>, NetworkState)>>>,
+		Arc<status_sinks::StatusSinks<(NetworkStatus<TBl>, NetworkState)>>,
 		Pin<Box<dyn Future<Output = ()> + Send>>
 	),
 	Error
@@ -1490,7 +1490,7 @@ fn build_network<TBl, TBackend, TExec, TRtApi, TExPool, TImpQu>(
 	let has_bootnodes = !network_params.network_config.boot_nodes.is_empty();
 	let network_mut = sc_network::NetworkWorker::new(network_params)?;
 	let network = network_mut.service().clone();
-	let network_status_sinks = Arc::new(Mutex::new(status_sinks::StatusSinks::new()));
+	let network_status_sinks = Arc::new(status_sinks::StatusSinks::new());
 
 	let future = build_network_future(
 		config.role.clone(),