diff --git a/substrate/Cargo.lock b/substrate/Cargo.lock index 96c7bfa1e6..a87da6586c 100644 --- a/substrate/Cargo.lock +++ b/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)", ] diff --git a/substrate/core/cli/src/informant.rs b/substrate/core/cli/src/informant.rs index 9d7f04b8f6..f7c23ba479 100644 --- a/substrate/core/cli/src/informant.rs +++ b/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 { select_chain: Option, network: Arc, /// Sinks to propagate network status updates. - network_status_sinks: Arc>>>, + /// For each element, every time the `Interval` fires we push an element on the sender. + network_status_sinks: Arc>>, transaction_pool: Arc, /// 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::(); 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 + fn network(&self) -> Arc>; /// Returns a receiver that periodically receives a status of the network. - fn network_status(&self) -> mpsc::UnboundedReceiver<(NetworkStatus, NetworkState)>; + fn network_status(&self, interval: Duration) -> mpsc::UnboundedReceiver<(NetworkStatus, NetworkState)>; /// Get shared transaction pool instance. fn transaction_pool(&self) -> Arc>; @@ -590,9 +603,9 @@ where self.network.clone() } - fn network_status(&self) -> mpsc::UnboundedReceiver<(NetworkStatus, NetworkState)> { + fn network_status(&self, interval: Duration) -> mpsc::UnboundedReceiver<(NetworkStatus, 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, client: Arc, - status_sinks: Arc, NetworkState)>>>>, + status_sinks: Arc, NetworkState)>>>, rpc_rx: futures03::channel::mpsc::UnboundedReceiver>, should_have_peers: bool, dht_event_tx: Option>, @@ -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| { diff --git a/substrate/core/service/src/status_sinks.rs b/substrate/core/service/src/status_sinks.rs new file mode 100644 index 0000000000..079ecf6353 --- /dev/null +++ b/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 . + +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 { + entries: FuturesUnordered>, +} + +struct YieldAfter { + delay: tokio_timer::Delay, + interval: Duration, + sender: Option>, +} + +impl StatusSinks { + /// Builds a new empty collection. + pub fn new() -> StatusSinks { + 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) { + 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 + `. 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 Future for YieldAfter { + type Item = (mpsc::UnboundedSender, Duration); + type Error = (); + + fn poll(&mut self) -> Poll { + 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(); + } +}