WIP Replace NetworkStatusSinks with Arch<NetworkService>

Author: kpp

bin/node/cli/src/service.rs

@@ -204,7 +204,6 @@ pub struct NewFullBase {
 	pub task_manager: TaskManager,
 	pub client: Arc<FullClient>,
 	pub network: Arc<NetworkService<Block, <Block as BlockT>::Hash>>,
-	pub network_status_sinks: sc_service::NetworkStatusSinks<Block>,
 	pub transaction_pool: Arc<sc_transaction_pool::FullPool<Block, FullClient>>,
 }
 
@@ -242,7 +241,7 @@ pub fn new_full_base(
 		)
 	);
 
-	let (network, network_status_sinks, system_rpc_tx, network_starter) =
+	let (network, system_rpc_tx, network_starter) =
 		sc_service::build_network(sc_service::BuildNetworkParams {
 			config: &config,
 			client: client.clone(),
@@ -279,7 +278,6 @@ pub fn new_full_base(
 			task_manager: &mut task_manager,
 			on_demand: None,
 			remote_blockchain: None,
-			network_status_sinks: network_status_sinks.clone(),
 			system_rpc_tx,
 			telemetry: telemetry.as_mut(),
 		},
@@ -415,7 +413,6 @@ pub fn new_full_base(
 		task_manager,
 		client,
 		network,
-		network_status_sinks,
 		transaction_pool,
 	})
 }
@@ -519,7 +516,7 @@ pub fn new_light_base(
 		telemetry.as_ref().map(|x| x.handle()),
 	)?;
 
-	let (network, network_status_sinks, system_rpc_tx, network_starter) =
+	let (network, system_rpc_tx, network_starter) =
 		sc_service::build_network(sc_service::BuildNetworkParams {
 			config: &config,
 			client: client.clone(),
@@ -554,7 +551,7 @@ pub fn new_light_base(
 			client: client.clone(),
 			transaction_pool: transaction_pool.clone(),
 			keystore: keystore_container.sync_keystore(),
-			config, backend, network_status_sinks, system_rpc_tx,
+			config, backend, system_rpc_tx,
 			network: network.clone(),
 			task_manager: &mut task_manager,
 			telemetry: telemetry.as_mut(),

client/informant/src/lib.rs

@@ -23,11 +23,11 @@ use futures::prelude::*;
 use log::{info, trace, warn};
 use parity_util_mem::MallocSizeOf;
 use sc_client_api::{BlockchainEvents, UsageProvider};
-use sc_network::NetworkStatus;
+use sc_network::NetworkService;
 use sp_blockchain::HeaderMetadata;
 use sp_runtime::traits::{Block as BlockT, Header};
 use sp_transaction_pool::TransactionPool;
-use sp_utils::{status_sinks, mpsc::tracing_unbounded};
+use sp_utils::mpsc::tracing_unbounded;
 use std::{fmt::Display, sync::Arc, time::Duration, collections::VecDeque};
 
 mod display;
@@ -66,7 +66,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<status_sinks::StatusSinks<NetworkStatus<B>>>,
+	network: Arc<NetworkService<B, <B as BlockT>::Hash>>,
 	pool: Arc<impl TransactionPoolAndMaybeMallogSizeOf>,
 	format: OutputFormat,
 ) -> impl futures::Future<Output = ()>
@@ -78,7 +78,7 @@ where
 
 	let client_1 = client.clone();
 	let (network_status_sink, network_status_stream) = tracing_unbounded("mpsc_network_status");
-	network_status_sinks.push(Duration::from_millis(5000), network_status_sink);
+	network.status_sinks.status.push(Duration::from_millis(5000), network_status_sink);
 
 	let display_notifications = network_status_stream
 		.for_each(move |net_status| {

client/network/src/lib.rs

@@ -276,6 +276,10 @@ pub use service::{
 
 pub use sc_peerset::ReputationChange;
 use sp_runtime::traits::{Block as BlockT, NumberFor};
+use sp_utils::{status_sinks, mpsc::{tracing_unbounded, TracingUnboundedReceiver}};
+use crate::network_state::NetworkState;
+use std::sync::Arc;
+use std::time::Duration;
 
 /// The maximum allowed number of established connections per peer.
 ///
@@ -322,3 +326,41 @@ pub struct NetworkStatus<B: BlockT> {
 	/// The total number of bytes sent.
 	pub total_bytes_outbound: u64,
 }
+
+/// Sinks to propagate network status updates.
+/// For each element, every time the `Interval` fires we push an element on the sender.
+#[derive(Clone)]
+pub struct NetworkStatusSinks<Block: BlockT> {
+	/// TODO
+	pub status: Arc<status_sinks::StatusSinks<NetworkStatus<Block>>>,
+	/// TODO
+	pub state: Arc<status_sinks::StatusSinks<NetworkState>>,
+}
+
+impl<Block: BlockT> NetworkStatusSinks<Block> {
+	fn new() -> Self {
+		Self {
+			status: Arc::new(status_sinks::StatusSinks::new()),
+			state: Arc::new(status_sinks::StatusSinks::new()),
+		}
+	}
+
+	/// Returns a receiver that periodically yields a [`NetworkStatus`].
+	pub fn status_stream(&self, interval: Duration)
+		-> TracingUnboundedReceiver<NetworkStatus<Block>>
+	{
+		let (sink, stream) = tracing_unbounded("mpsc_network_status");
+		self.status.push(interval, sink);
+		stream
+	}
+
+	/// Returns a receiver that periodically yields a [`NetworkState`].
+	pub fn state_stream(&self, interval: Duration)
+		-> TracingUnboundedReceiver<NetworkState>
+	{
+		let (sink, stream) = tracing_unbounded("mpsc_network_state");
+		self.state.push(interval, sink);
+		stream
+	}
+
+}

client/network/src/service.rs

@@ -28,7 +28,7 @@
 //! which is then processed by [`NetworkWorker::poll`].
 
 use crate::{
-	ExHashT, NetworkStateInfo, NetworkStatus,
+	ExHashT, NetworkStateInfo, NetworkStatus, NetworkStatusSinks,
 	behaviour::{self, Behaviour, BehaviourOut},
 	config::{parse_str_addr, Params, TransportConfig},
 	DhtEvent,
@@ -123,6 +123,8 @@ pub struct NetworkService<B: BlockT + 'static, H: ExHashT> {
 	local_peer_id: PeerId,
 	/// Bandwidth logging system. Can be queried to know the average bandwidth consumed.
 	bandwidth: Arc<transport::BandwidthSinks>,
+	/// Sinks to propagate network status updates.
+	pub status_sinks: NetworkStatusSinks<B>,
 	/// Peerset manager (PSM); manages the reputation of nodes and indicates the network which
 	/// nodes it should be connected to or not.
 	peerset: PeersetHandle,
@@ -400,9 +402,11 @@ impl<B: BlockT + 'static, H: ExHashT> NetworkWorker<B, H> {
 
 		let external_addresses = Arc::new(Mutex::new(Vec::new()));
 		let peers_notifications_sinks = Arc::new(Mutex::new(HashMap::new()));
+		let status_sinks = NetworkStatusSinks::new();
 
 		let service = Arc::new(NetworkService {
 			bandwidth,
+			status_sinks,
 			external_addresses: external_addresses.clone(),
 			num_connected: num_connected.clone(),
 			is_major_syncing: is_major_syncing.clone(),

client/service/src/builder.rs

@@ -17,7 +17,7 @@
 // along with this program. If not, see <https://www.gnu.org/licenses/>.
 
 use crate::{
-	error::Error, MallocSizeOfWasm, RpcHandlers, NetworkStatusSinks,
+	error::Error, MallocSizeOfWasm, RpcHandlers,
 	start_rpc_servers, build_network_future, TransactionPoolAdapter, TaskManager, SpawnTaskHandle,
 	metrics::MetricsService,
 	client::{light, Client, ClientConfig},
@@ -504,8 +504,6 @@ pub struct SpawnTasksParams<'a, TBl: BlockT, TCl, TExPool, TRpc, Backend> {
 	pub remote_blockchain: Option<Arc<dyn RemoteBlockchain<TBl>>>,
 	/// A shared network instance.
 	pub network: Arc<NetworkService<TBl, <TBl as BlockT>::Hash>>,
-	/// Sinks to propagate network status updates.
-	pub network_status_sinks: NetworkStatusSinks<TBl>,
 	/// A Sender for RPC requests.
 	pub system_rpc_tx: TracingUnboundedSender<sc_rpc::system::Request<TBl>>,
 	/// Telemetry instance for this node.
@@ -575,7 +573,6 @@ pub fn spawn_tasks<TBl, TBackend, TExPool, TRpc, TCl>(
 		rpc_extensions_builder,
 		remote_blockchain,
 		network,
-		network_status_sinks,
 		system_rpc_tx,
 		telemetry,
 	} = params;
@@ -639,7 +636,7 @@ pub fn spawn_tasks<TBl, TBackend, TExPool, TRpc, TCl>(
 		metrics_service.run(
 			client.clone(),
 			transaction_pool.clone(),
-			network_status_sinks.clone()
+			network.clone()
 		)
 	);
 
@@ -664,7 +661,7 @@ pub fn spawn_tasks<TBl, TBackend, TExPool, TRpc, TCl>(
 	// Spawn informant task
 	spawn_handle.spawn("informant", sc_informant::build(
 		client.clone(),
-		network_status_sinks.status.clone(),
+		network.clone(),
 		transaction_pool.clone(),
 		config.informant_output_format,
 	));
@@ -850,7 +847,6 @@ pub fn build_network<TBl, TExPool, TImpQu, TCl>(
 ) -> Result<
 	(
 		Arc<NetworkService<TBl, <TBl as BlockT>::Hash>>,
-		NetworkStatusSinks<TBl>,
 		TracingUnboundedSender<sc_rpc::system::Request<TBl>>,
 		NetworkStarter,
 	),
@@ -944,15 +940,13 @@ pub fn build_network<TBl, TExPool, TImpQu, TCl>(
 	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 = NetworkStatusSinks::new();
 
 	let (system_rpc_tx, system_rpc_rx) = tracing_unbounded("mpsc_system_rpc");
 
 	let future = build_network_future(
 		config.role.clone(),
 		network_mut,
 		client,
-		network_status_sinks.clone(),
 		system_rpc_rx,
 		has_bootnodes,
 		config.announce_block,
@@ -995,7 +989,7 @@ pub fn build_network<TBl, TExPool, TImpQu, TCl>(
 		future.await
 	});
 
-	Ok((network, network_status_sinks, system_rpc_tx, NetworkStarter(network_start_tx)))
+	Ok((network, system_rpc_tx, NetworkStarter(network_start_tx)))
 }
 
 /// Object used to start the network.

client/service/src/lib.rs

@@ -37,7 +37,6 @@ mod task_manager;
 use std::{io, pin::Pin};
 use std::net::SocketAddr;
 use std::collections::HashMap;
-use std::time::Duration;
 use std::task::Poll;
 
 use futures::{Future, FutureExt, Stream, StreamExt, stream, compat::*};
@@ -47,7 +46,7 @@ use codec::{Encode, Decode};
 use sp_runtime::generic::BlockId;
 use sp_runtime::traits::{Block as BlockT, Header as HeaderT};
 use parity_util_mem::MallocSizeOf;
-use sp_utils::{status_sinks, mpsc::{tracing_unbounded, TracingUnboundedReceiver}};
+use sp_utils::mpsc::TracingUnboundedReceiver;
 
 pub use self::error::Error;
 pub use self::builder::{
@@ -124,42 +123,6 @@ impl RpcHandlers {
 	}
 }
 
-/// Sinks to propagate network status updates.
-/// For each element, every time the `Interval` fires we push an element on the sender.
-#[derive(Clone)]
-pub struct NetworkStatusSinks<Block: BlockT> {
-	status: Arc<status_sinks::StatusSinks<NetworkStatus<Block>>>,
-	state: Arc<status_sinks::StatusSinks<NetworkState>>,
-}
-
-impl<Block: BlockT> NetworkStatusSinks<Block> {
-	fn new() -> Self {
-		Self {
-			status: Arc::new(status_sinks::StatusSinks::new()),
-			state: Arc::new(status_sinks::StatusSinks::new()),
-		}
-	}
-
-	/// Returns a receiver that periodically yields a [`NetworkStatus`].
-	pub fn status_stream(&self, interval: Duration)
-		-> TracingUnboundedReceiver<NetworkStatus<Block>>
-	{
-		let (sink, stream) = tracing_unbounded("mpsc_network_status");
-		self.status.push(interval, sink);
-		stream
-	}
-
-	/// Returns a receiver that periodically yields a [`NetworkState`].
-	pub fn state_stream(&self, interval: Duration)
-		-> TracingUnboundedReceiver<NetworkState>
-	{
-		let (sink, stream) = tracing_unbounded("mpsc_network_state");
-		self.state.push(interval, sink);
-		stream
-	}
-
-}
-
 /// An incomplete set of chain components, but enough to run the chain ops subcommands.
 pub struct PartialComponents<Client, Backend, SelectChain, ImportQueue, TransactionPool, Other> {
 	/// A shared client instance.
@@ -191,7 +154,6 @@ async fn build_network_future<
 	role: Role,
 	mut network: sc_network::NetworkWorker<B, H>,
 	client: Arc<C>,
-	status_sinks: NetworkStatusSinks<B>,
 	mut rpc_rx: TracingUnboundedReceiver<sc_rpc::system::Request<B>>,
 	should_have_peers: bool,
 	announce_imported_blocks: bool,
@@ -340,13 +302,13 @@ async fn build_network_future<
 			// detailed state information of the network on what are called
 			// "status sinks".
 
-			status_sink = status_sinks.status.next().fuse() => {
-				status_sink.send(network.status());
-			}
+			// FIXME status_sink = network.service().status_sinks.status.next().fuse() => {
+			//	status_sink.send(network.status());
+			//}
 
-			state_sink = status_sinks.state.next().fuse() => {
-				state_sink.send(network.network_state());
-			}
+			// FIXME state_sink = network.service().status_sinks.state.next().fuse() => {
+			//	state_sink.send(network.network_state());
+			//}
 		}
 	}
 }

client/service/src/metrics.rs

@@ -18,7 +18,7 @@
 
 use std::{convert::TryFrom, time::SystemTime};
 
-use crate::{NetworkStatus, NetworkState, NetworkStatusSinks, config::Configuration};
+use crate::{NetworkStatus, NetworkState, config::Configuration};
 use futures_timer::Delay;
 use prometheus_endpoint::{register, Gauge, U64, Registry, PrometheusError, Opts, GaugeVec};
 use sc_telemetry::{telemetry, TelemetryHandle, SUBSTRATE_INFO};
@@ -29,6 +29,7 @@ use sp_utils::metrics::register_globals;
 use sp_utils::mpsc::TracingUnboundedReceiver;
 use sc_client_api::{ClientInfo, UsageProvider};
 use sc_network::config::Role;
+use sc_network::NetworkService;
 use std::sync::Arc;
 use std::time::Duration;
 use wasm_timer::Instant;
@@ -163,7 +164,7 @@ impl MetricsService {
 		mut self,
 		client: Arc<TCl>,
 		transactions: Arc<TExPool>,
-		network: NetworkStatusSinks<TBl>,
+		network: Arc<NetworkService<TBl, <TBl as Block>::Hash>>,
 	) where
 		TBl: Block,
 		TCl: ProvideRuntimeApi<TBl> + UsageProvider<TBl>,
@@ -177,8 +178,8 @@ impl MetricsService {
 		let net_state_interval = Duration::from_secs(30);
 
 		// Source of network information.
-		let mut net_status_rx = Some(network.status_stream(net_status_interval));
-		let mut net_state_rx = Some(network.state_stream(net_state_interval));
+		let mut net_status_rx = Some(network.status_sinks.status_stream(net_status_interval));
+		let mut net_state_rx = Some(network.status_sinks.state_stream(net_state_interval));
 
 		loop {
 			// Wait for the next tick of the timer.