Split the telemetry net status report in two (#3887)

* Split the telemetry net status report in two

* Update core/service/src/lib.rs

Co-Authored-By: Bastian Köcher <bkchr@users.noreply.github.com>

* Remove clone()

* Move code to status_sinks.rs instead

* Add basic usage for status_sinks

* Update core/service/src/status_sinks.rs

Co-Authored-By: Bastian Köcher <bkchr@users.noreply.github.com>

substrate/Cargo.lock

@@ -5539,6 +5539,7 @@ dependencies = [
  "substrate-transaction-pool 2.0.0",
  "sysinfo 0.9.5 (registry+https://github.com/rust-lang/crates.io-index)",
  "target_info 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)",
+ "tokio 0.1.22 (registry+https://github.com/rust-lang/crates.io-index)",
  "tokio-executor 0.1.8 (registry+https://github.com/rust-lang/crates.io-index)",
  "tokio-timer 0.2.11 (registry+https://github.com/rust-lang/crates.io-index)",
 ]

substrate/core/cli/src/informant.rs

@@ -22,6 +22,7 @@ use futures03::{StreamExt as _, TryStreamExt as _};
 use log::{info, warn};
 use sr_primitives::traits::Header;
 use service::AbstractService;
+use std::time::Duration;
 
 mod display;
 
@@ -31,11 +32,13 @@ pub fn build(service: &impl AbstractService) -> impl Future<Item = (), Error = (
 
 	let mut display = display::InformantDisplay::new();
 
-	let display_notifications = service.network_status().for_each(move |(net_status, _)| {
-		let info = client.info();
-		display.display(&info, net_status);
-		Ok(())
-	});
+	let display_notifications = service
+		.network_status(Duration::from_millis(5000))
+		.for_each(move |(net_status, _)| {
+			let info = client.info();
+			display.display(&info, net_status);
+			Ok(())
+		});
 
 	let client = service.client();
 	let mut last_best = {

substrate/core/service/Cargo.toml

@@ -49,3 +49,4 @@ node-runtime = { path = "../../node/runtime" }
 babe-primitives = { package = "substrate-consensus-babe-primitives", path = "../../core/consensus/babe/primitives" }
 grandpa = { package = "substrate-finality-grandpa", path = "../../core/finality-grandpa" }
 grandpa-primitives = { package = "substrate-finality-grandpa-primitives", path = "../../core/finality-grandpa/primitives" }
+tokio = "0.1"

substrate/core/service/src/builder.rs

@@ -17,6 +17,7 @@
 use crate::{NewService, NetworkStatus, NetworkState, error::{self, Error}, DEFAULT_PROTOCOL_ID};
 use crate::{SpawnTaskHandle, start_rpc_servers, build_network_future, TransactionPoolAdapter};
 use crate::TaskExecutor;
+use crate::status_sinks;
 use crate::config::Configuration;
 use client::{
 	BlockchainEvents, Client, runtime_api,

substrate/core/service/src/lib.rs

@@ -24,6 +24,8 @@ pub mod config;
 pub mod chain_ops;
 pub mod error;
 
+mod status_sinks;
+
 use std::io;
 use std::marker::PhantomData;
 use std::net::SocketAddr;
@@ -74,9 +76,8 @@ pub struct NewService<TBl, TCl, TSc, TNetStatus, TNet, TTxPool, TOc> {
 	select_chain: Option<TSc>,
 	network: Arc<TNet>,
 	/// Sinks to propagate network status updates.
-	network_status_sinks: Arc<Mutex<Vec<mpsc::UnboundedSender<(
-		TNetStatus, NetworkState
-	)>>>>,
+	/// For each element, every time the `Interval` fires we push an element on the sender.
+	network_status_sinks: Arc<Mutex<status_sinks::StatusSinks<(TNetStatus, NetworkState)>>>,
 	transaction_pool: Arc<TTxPool>,
 	/// A future that resolves when the service has exited, this is useful to
 	/// make sure any internally spawned futures stop when the service does.
@@ -210,7 +211,7 @@ macro_rules! new_impl {
 		let has_bootnodes = !network_params.network_config.boot_nodes.is_empty();
 		let network_mut = network::NetworkWorker::new(network_params)?;
 		let network = network_mut.service().clone();
-		let network_status_sinks = Arc::new(Mutex::new(Vec::new()));
+		let network_status_sinks = Arc::new(Mutex::new(status_sinks::StatusSinks::new()));
 
 		let offchain_storage = backend.offchain_storage();
 		let offchain_workers = match ($config.offchain_worker, offchain_storage) {
@@ -296,9 +297,9 @@ macro_rules! new_impl {
 		let client_ = client.clone();
 		let mut sys = System::new();
 		let self_pid = get_current_pid().ok();
-		let (netstat_tx, netstat_rx) = mpsc::unbounded::<(NetworkStatus<_>, NetworkState)>();
-		network_status_sinks.lock().push(netstat_tx);
-		let tel_task = netstat_rx.for_each(move |(net_status, network_state)| {
+		let (state_tx, state_rx) = mpsc::unbounded::<(NetworkStatus<_>, NetworkState)>();
+		network_status_sinks.lock().push(std::time::Duration::from_millis(5000), state_tx);
+		let tel_task = state_rx.for_each(move |(net_status, _)| {
 			let info = client_.info();
 			let best_number = info.chain.best_number.saturated_into::<u64>();
 			let best_hash = info.chain.best_hash;
@@ -325,7 +326,6 @@ macro_rules! new_impl {
 			telemetry!(
 				SUBSTRATE_INFO;
 				"system.interval";
-				"network_state" => network_state,
 				"peers" => num_peers,
 				"height" => best_number,
 				"best" => ?best_hash,
@@ -343,6 +343,19 @@ macro_rules! new_impl {
 		}).select(exit.clone()).then(|_| Ok(()));
 		let _ = to_spawn_tx.unbounded_send(Box::new(tel_task));
 
+		// Periodically send the network state to the telemetry.
+		let (netstat_tx, netstat_rx) = mpsc::unbounded::<(NetworkStatus<_>, NetworkState)>();
+		network_status_sinks.lock().push(std::time::Duration::from_secs(30), netstat_tx);
+		let tel_task_2 = netstat_rx.for_each(move |(_, network_state)| {
+			telemetry!(
+				SUBSTRATE_INFO;
+				"system.network_state";
+				"state" => network_state,
+			);
+			Ok(())
+		}).select(exit.clone()).then(|_| Ok(()));
+		let _ = to_spawn_tx.unbounded_send(Box::new(tel_task_2));
+
 		// RPC
 		let (system_rpc_tx, system_rpc_rx) = futures03::channel::mpsc::unbounded();
 		let gen_handler = || {
@@ -507,7 +520,7 @@ pub trait AbstractService: 'static + Future<Item = (), Error = Error> +
 	fn network(&self) -> Arc<NetworkService<Self::Block, Self::NetworkSpecialization, H256>>;
 
 	/// Returns a receiver that periodically receives a status of the network.
-	fn network_status(&self) -> mpsc::UnboundedReceiver<(NetworkStatus<Self::Block>, NetworkState)>;
+	fn network_status(&self, interval: Duration) -> mpsc::UnboundedReceiver<(NetworkStatus<Self::Block>, NetworkState)>;
 
 	/// Get shared transaction pool instance.
 	fn transaction_pool(&self) -> Arc<TransactionPool<Self::TransactionPoolApi>>;
@@ -590,9 +603,9 @@ where
 		self.network.clone()
 	}
 
-	fn network_status(&self) -> mpsc::UnboundedReceiver<(NetworkStatus<Self::Block>, NetworkState)> {
+	fn network_status(&self, interval: Duration) -> mpsc::UnboundedReceiver<(NetworkStatus<Self::Block>, NetworkState)> {
 		let (sink, stream) = mpsc::unbounded();
-		self.network_status_sinks.lock().push(sink);
+		self.network_status_sinks.lock().push(interval, sink);
 		stream
 	}
 
@@ -666,7 +679,7 @@ fn build_network_future<
 	roles: Roles,
 	mut network: network::NetworkWorker<B, S, H>,
 	client: Arc<C>,
-	status_sinks: Arc<Mutex<Vec<mpsc::UnboundedSender<(NetworkStatus<B>, NetworkState)>>>>,
+	status_sinks: Arc<Mutex<status_sinks::StatusSinks<(NetworkStatus<B>, NetworkState)>>>,
 	rpc_rx: futures03::channel::mpsc::UnboundedReceiver<rpc::system::Request<B>>,
 	should_have_peers: bool,
 	dht_event_tx: Option<mpsc::Sender<DhtEvent>>,
@@ -675,10 +688,6 @@ fn build_network_future<
 	// See https://github.com/paritytech/substrate/issues/3099
 	let mut rpc_rx = futures03::compat::Compat::new(rpc_rx.map(|v| Ok::<_, ()>(v)));
 
-	// Interval at which we send status updates on the status stream.
-	const STATUS_INTERVAL: Duration = Duration::from_millis(5000);
-	let mut status_interval = tokio_timer::Interval::new_interval(STATUS_INTERVAL);
-
 	let mut imported_blocks_stream = client.import_notification_stream().fuse()
 		.map(|v| Ok::<_, ()>(v)).compat();
 	let mut finality_notification_stream = client.finality_notification_stream().fuse()
@@ -746,7 +755,7 @@ fn build_network_future<
 		}
 
 		// Interval report for the external API.
-		while let Ok(Async::Ready(_)) = status_interval.poll() {
+		status_sinks.lock().poll(|| {
 			let status = NetworkStatus {
 				sync_state: network.sync_state(),
 				best_seen_block: network.best_seen_block(),
@@ -757,9 +766,8 @@ fn build_network_future<
 				average_upload_per_sec: network.average_upload_per_sec(),
 			};
 			let state = network.network_state();
-
-			status_sinks.lock().retain(|sink| sink.unbounded_send((status.clone(), state.clone())).is_ok());
-		}
+			(status, state)
+		});
 
 		// Main network polling.
 		while let Ok(Async::Ready(Some(Event::Dht(event)))) = network.poll().map_err(|err| {

substrate/core/service/src/status_sinks.rs

@@ -0,0 +1,149 @@
+// Copyright 2019 Parity Technologies (UK) Ltd.
+// This file is part of Substrate.
+
+// Substrate is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// Substrate is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+
+// You should have received a copy of the GNU General Public License
+// along with Substrate.  If not, see <http://www.gnu.org/licenses/>.
+
+use futures::prelude::*;
+use futures::sync::mpsc;
+use futures::stream::futures_unordered::FuturesUnordered;
+use std::time::{Duration, Instant};
+use tokio_timer::Delay;
+
+/// Holds a list of `UnboundedSender`s, each associated with a certain time period. Every time the
+/// period elapses, we push an element on the sender.
+///
+/// Senders are removed only when they are closed.
+pub struct StatusSinks<T> {
+	entries: FuturesUnordered<YieldAfter<T>>,
+}
+
+struct YieldAfter<T> {
+	delay: tokio_timer::Delay,
+	interval: Duration,
+	sender: Option<mpsc::UnboundedSender<T>>,
+}
+
+impl<T> StatusSinks<T> {
+	/// Builds a new empty collection.
+	pub fn new() -> StatusSinks<T> {
+		StatusSinks {
+			entries: FuturesUnordered::new(),
+		}
+	}
+
+	/// Adds a sender to the collection.
+	///
+	/// The `interval` is the time period between two pushes on the sender.
+	pub fn push(&mut self, interval: Duration, sender: mpsc::UnboundedSender<T>) {
+		self.entries.push(YieldAfter {
+			delay: Delay::new(Instant::now() + interval),
+			interval,
+			sender: Some(sender),
+		})
+	}
+
+	/// Processes all the senders. If any sender is ready, calls the `status_grab` function and
+	/// pushes what it returns to the sender.
+	///
+	/// This function doesn't return anything, but it should be treated as if it implicitly
+	/// returns `Ok(Async::NotReady)`. In particular, it should be called again when the task
+	/// is waken up.
+	///
+	/// # Panic
+	///
+	/// Panics if not called within the context of a task.
+	pub fn poll(&mut self, mut status_grab: impl FnMut() -> T) {
+		loop {
+			match self.entries.poll() {
+				Ok(Async::Ready(Some((sender, interval)))) => {
+					let status = status_grab();
+					if sender.unbounded_send(status).is_ok() {
+						self.entries.push(YieldAfter {
+							// Note that since there's a small delay between the moment a task is
+							// waken up and the moment it is polled, the period is actually not
+							// `interval` but `interval + <delay>`. We ignore this problem in
+							// practice.
+							delay: Delay::new(Instant::now() + interval),
+							interval,
+							sender: Some(sender),
+						});
+					}
+				}
+				Err(()) |
+				Ok(Async::Ready(None)) |
+				Ok(Async::NotReady) => break,
+			}
+		}
+	}
+}
+
+impl<T> Future for YieldAfter<T> {
+	type Item = (mpsc::UnboundedSender<T>, Duration);
+	type Error = ();
+
+	fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+		match self.delay.poll() {
+			Ok(Async::NotReady) => Ok(Async::NotReady),
+			Ok(Async::Ready(())) => {
+				let sender = self.sender.take()
+					.expect("sender is always Some unless the future is finished; qed");
+				Ok(Async::Ready((sender, self.interval)))
+			},
+			Err(_) => Err(()),
+		}
+	}
+}
+
+#[cfg(test)]
+mod tests {
+	use super::StatusSinks;
+	use futures::prelude::*;
+	use futures::sync::mpsc;
+	use std::time::Duration;
+
+	#[test]
+	fn basic_usage() {
+		let mut status_sinks = StatusSinks::new();
+
+		let (tx1, rx1) = mpsc::unbounded();
+		status_sinks.push(Duration::from_millis(200), tx1);
+
+		let (tx2, rx2) = mpsc::unbounded();
+		status_sinks.push(Duration::from_millis(500), tx2);
+
+		let mut runtime = tokio::runtime::Runtime::new().unwrap();
+
+		let mut val_order = 5;
+		runtime.spawn(futures::future::poll_fn(move || {
+			status_sinks.poll(|| { val_order += 1; val_order });
+			Ok(Async::NotReady)
+		}));
+
+		let done = rx1
+			.into_future()
+			.and_then(|(item, rest)| {
+				assert_eq!(item, Some(6));
+				rest.into_future()
+			})
+			.and_then(|(item, _)| {
+				assert_eq!(item, Some(7));
+				rx2.into_future()
+			})
+			.map(|(item, _)| {
+				assert_eq!(item, Some(8));
+			});
+
+		runtime.block_on(done).unwrap();
+	}
+}