Fix timer panics in the wasm light client

Cargo.toml

@@ -33,6 +33,7 @@ members = [
 	"client/executor/wasmtime",
 	"client/executor/runtime-test",
 	"client/finality-grandpa",
+	"client/informant",
 	"client/tracing",
 	"client/keystore",
 	"client/network",

bin/node/cli/browser-demo/index.html

@@ -19,7 +19,7 @@
 
 	// Build our client.
 	log('Starting client');
-	let client = await start_client(ws());
+	let client = await start_client('info', ws());
 	log('Client started');
 
 	client.rpcSubscribe('{"method":"chain_subscribeNewHead","params":[],"id":1,"jsonrpc":"2.0"}',

bin/node/cli/src/browser.rs

@@ -22,19 +22,19 @@ use browser_utils::{
 	Transport, Client,
 	browser_configuration, set_console_error_panic_hook, init_console_log,
 };
+use std::str::FromStr;
 
 /// Starts the client.
 #[wasm_bindgen]
-pub async fn start_client(wasm_ext: Transport) -> Result<Client, JsValue> {
-	start_inner(wasm_ext)
+pub async fn start_client(log_level: String, wasm_ext: Transport) -> Result<Client, JsValue> {
+	start_inner(log_level, wasm_ext)
 		.await
 		.map_err(|err| JsValue::from_str(&err.to_string()))
 }
 
-async fn start_inner(wasm_ext: Transport) -> Result<Client, Box<dyn std::error::Error>> {
+async fn start_inner(log_level: String, wasm_ext: Transport) -> Result<Client, Box<dyn std::error::Error>> {
 	set_console_error_panic_hook();
-	init_console_log(log::Level::Info)?;
-
+	init_console_log(log::Level::from_str(&log_level)?)?;
 	let chain_spec = ChainSpec::FlamingFir.load()
 		.map_err(|e| format!("{:?}", e))?;
 

client/authority-discovery/Cargo.toml

@@ -14,7 +14,7 @@ bytes = "0.4.12"
 codec = { package = "parity-scale-codec", default-features = false, version = "1.0.3" }
 derive_more = "0.99.2"
 futures = "0.3.1"
-futures-timer = "2.0"
+futures-timer = "3.0.1"
 libp2p = { version = "0.15.0", default-features = false, features = ["secp256k1", "libp2p-websocket"] }
 log = "0.4.8"
 prost = "0.6.1"

client/cli/Cargo.toml

@@ -21,6 +21,7 @@ tokio = { version = "0.2.9", features = [ "signal", "rt-core", "rt-threaded" ] }
 futures = "0.3.1"
 fdlimit = "0.1.1"
 serde_json = "1.0.41"
+sc-informant = { version = "0.8", path = "../informant" }
 sp-panic-handler = { version = "2.0.0", path = "../../primitives/panic-handler" }
 sc-client-api = { version = "2.0.0", path = "../api" }
 sp-blockchain = { version = "2.0.0", path = "../../primitives/blockchain" }

client/cli/src/lib.rs

@@ -24,7 +24,6 @@ mod traits;
 mod params;
 mod execution_strategy;
 pub mod error;
-pub mod informant;
 mod runtime;
 mod node_key;
 

client/cli/src/runtime.rs

@@ -19,7 +19,6 @@ use futures::select;
 use futures::pin_mut;
 use sc_service::{AbstractService, Configuration};
 use crate::error;
-use crate::informant;
 
 #[cfg(target_family = "unix")]
 async fn main<F, E>(func: F) -> Result<(), Box<dyn std::error::Error>>
@@ -122,7 +121,7 @@ where
 
 	let service = service_builder(config)?;
 
-	let informant_future = informant::build(&service);
+	let informant_future = sc_informant::build(&service);
 	let _informant_handle = runtime.spawn(informant_future);
 
 	// we eagerly drop the service so that the internal exit future is fired,

client/consensus/aura/Cargo.toml

@@ -16,7 +16,7 @@ codec = { package = "parity-scale-codec", version = "1.0.0" }
 sp-consensus = { version = "0.8", path = "../../../primitives/consensus/common" }
 derive_more = "0.99.2"
 futures = "0.3.1"
-futures-timer = "0.4.0"
+futures-timer = "3.0.1"
 sp-inherents = { version = "2.0.0", path = "../../../primitives/inherents" }
 sc-keystore = { version = "2.0.0", path = "../../keystore" }
 log = "0.4.8"

client/consensus/babe/Cargo.toml

@@ -31,7 +31,7 @@ sc-consensus-slots = { version = "0.8", path = "../slots" }
 sp-runtime = { version = "2.0.0", path = "../../../primitives/runtime" }
 fork-tree = { version = "2.0.0", path = "../../../utils/fork-tree" }
 futures = "0.3.1"
-futures-timer = "0.4.0"
+futures-timer = "3.0.1"
 parking_lot = "0.9.0"
 log = "0.4.8"
 schnorrkel = { version = "0.8.5", features = ["preaudit_deprecated"] }

client/consensus/slots/Cargo.toml

@@ -19,7 +19,7 @@ sc-telemetry = { version = "2.0.0", path = "../../telemetry" }
 sp-consensus = { version = "0.8", path = "../../../primitives/consensus/common" }
 sp-inherents = { version = "2.0.0", path = "../../../primitives/inherents" }
 futures = "0.3.1"
-futures-timer = "2.0"
+futures-timer = "3.0.1"
 parking_lot = "0.9.0"
 log = "0.4.8"
 

client/finality-grandpa/Cargo.toml

@@ -8,7 +8,7 @@ license = "GPL-3.0"
 [dependencies]
 fork-tree = { version = "2.0.0", path = "../../utils/fork-tree" }
 futures = "0.3.1"
-futures-timer = "2.0.2"
+futures-timer = "3.0.1"
 log = "0.4.8"
 parking_lot = "0.9.0"
 rand = "0.7.2"

client/finality-grandpa/src/communication/periodic.rs

@@ -19,7 +19,7 @@
 use futures_timer::Delay;
 use futures::{channel::mpsc, future::{FutureExt as _}, prelude::*, ready, stream::Stream};
 use log::debug;
-use std::{pin::Pin, task::{Context, Poll}, time::{Instant, Duration}};
+use std::{pin::Pin, task::{Context, Poll}, time::Duration};
 
 use sc_network::PeerId;
 use sp_runtime::traits::{NumberFor, Block as BlockT};
@@ -28,10 +28,6 @@ use super::gossip::{NeighborPacket, GossipMessage};
 // How often to rebroadcast, in cases where no new packets are created.
 const REBROADCAST_AFTER: Duration = Duration::from_secs(2 * 60);
 
-fn rebroadcast_instant() -> Instant {
-	Instant::now() + REBROADCAST_AFTER
-}
-
 /// A sender used to send neighbor packets to a background job.
 #[derive(Clone)]
 pub(super) struct NeighborPacketSender<B: BlockT>(
@@ -85,7 +81,7 @@ impl <B: BlockT> Stream for NeighborPacketWorker<B> {
 		match this.rx.poll_next_unpin(cx) {
 			Poll::Ready(None) => return Poll::Ready(None),
 			Poll::Ready(Some((to, packet))) => {
-				this.delay.reset(rebroadcast_instant());
+				this.delay.reset(REBROADCAST_AFTER);
 				this.last = Some((to.clone(), packet.clone()));
 
 				return Poll::Ready(Some((to, GossipMessage::<B>::from(packet.clone()))));
@@ -98,7 +94,7 @@ impl <B: BlockT> Stream for NeighborPacketWorker<B> {
 
 		// Getting this far here implies that the timer fired.
 
-		this.delay.reset(rebroadcast_instant());
+		this.delay.reset(REBROADCAST_AFTER);
 
 		// Make sure the underlying task is scheduled for wake-up.
 		//

client/informant/Cargo.toml

@@ -0,0 +1,18 @@
+[package]
+name = "sc-informant"
+version = "0.8.0"
+authors = ["Parity Technologies <admin@parity.io>"]
+description = "Substrate informant."
+edition = "2018"
+license = "GPL-3.0"
+
+[dependencies]
+ansi_term = "0.12.1"
+wasm-timer = "0.2"
+futures = "0.3.1"
+log = "0.4.8"
+sc-client-api = { version = "2.0.0", path = "../api" }
+sc-network = { version = "0.8", path = "../network" }
+sc-service = { version = "0.8", default-features = false, path = "../service" }
+sp-blockchain = { version = "2.0.0", path = "../../primitives/blockchain" }
+sp-runtime = { version = "2.0.0", path = "../../primitives/runtime" }

client/cli/src/informant/display.rs → client/informant/src/display.rs

@@ -20,7 +20,8 @@ use log::info;
 use sc_network::SyncState;
 use sp_runtime::traits::{Block as BlockT, CheckedDiv, NumberFor, Zero, Saturating};
 use sc_service::NetworkStatus;
-use std::{convert::{TryFrom, TryInto}, fmt, time};
+use std::{convert::{TryFrom, TryInto}, fmt};
+use wasm_timer::Instant;
 
 /// State of the informant display system.
 ///
@@ -40,24 +41,26 @@ pub struct InformantDisplay<B: BlockT> {
 	/// `None` if `display` has never been called.
 	last_number: Option<NumberFor<B>>,
 	/// The last time `display` or `new` has been called.
-	last_update: time::Instant,
+	last_update: Instant,
 }
 
 impl<B: BlockT> InformantDisplay<B> {
 	/// Builds a new informant display system.
 	pub fn new() -> InformantDisplay<B> {
 		InformantDisplay {
 			last_number: None,
-			last_update: time::Instant::now(),
+			last_update: Instant::now(),
 		}
 	}
 
 	/// Displays the informant by calling `info!`.
 	pub fn display(&mut self, info: &ClientInfo<B>, net_status: NetworkStatus<B>) {
 		let best_number = info.chain.best_number;
 		let best_hash = info.chain.best_hash;
+		let finalized_number = info.chain.finalized_number;
+		let num_connected_peers = net_status.num_connected_peers;
 		let speed = speed::<B>(best_number, self.last_number, self.last_update);
-		self.last_update = time::Instant::now();
+		self.last_update = Instant::now();
 		self.last_number = Some(best_number);
 
 		let (status, target) = match (net_status.sync_state, net_status.best_seen_block) {
@@ -66,15 +69,29 @@ impl<B: BlockT> InformantDisplay<B> {
 			(SyncState::Downloading, Some(n)) => (format!("Syncing{}", speed), format!(", target=#{}", n)),
 		};
 
+		// Colour informant output if we're running natively.
+
+		#[cfg(not(target_os = "unknown"))]
+		let status = Colour::White.bold().paint(&status);
+
+		#[cfg(not(target_os = "unknown"))]
+		let num_connected_peers = Colour::White.bold().paint(format!("{}", num_connected_peers));
+
+		#[cfg(not(target_os = "unknown"))]
+		let best_number = Colour::White.paint(format!("{}", best_number));
+
+		#[cfg(not(target_os = "unknown"))]
+		let finalized_number = Colour::White.paint(format!("{}", finalized_number));
+
 		info!(
 			target: "substrate",
 			"{}{} ({} peers), best: #{} ({}), finalized #{} ({}), ⬇ {} ⬆ {}",
-			Colour::White.bold().paint(&status),
+			status,
 			target,
-			Colour::White.bold().paint(format!("{}", net_status.num_connected_peers)),
-			Colour::White.paint(format!("{}", best_number)),
+			num_connected_peers,
+			best_number,
 			best_hash,
-			Colour::White.paint(format!("{}", info.chain.finalized_number)),
+			finalized_number,
 			info.chain.finalized_hash,
 			TransferRateFormat(net_status.average_download_per_sec),
 			TransferRateFormat(net_status.average_upload_per_sec),
@@ -87,7 +104,7 @@ impl<B: BlockT> InformantDisplay<B> {
 fn speed<B: BlockT>(
 	best_number: NumberFor<B>,
 	last_number: Option<NumberFor<B>>,
-	last_update: time::Instant
+	last_update: Instant
 ) -> String {
 	// Number of milliseconds elapsed since last time.
 	let elapsed_ms = {

client/network-gossip/Cargo.toml

@@ -9,8 +9,9 @@ edition = "2018"
 [dependencies]
 log = "0.4.8"
 futures = { version = "0.3.1", features = ["compat"] }
+wasm-timer = "0.2"
+futures-timer = "3.0.1"
 futures01 = { package = "futures", version = "0.1.29" }
-futures-timer = "0.4.0"
 libp2p = { version = "0.15.0", default-features = false, features = ["libp2p-websocket"] }
 lru = "0.1.2"
 parking_lot = "0.9.0"

client/network-gossip/src/state_machine.rs

@@ -28,6 +28,7 @@ use sp_runtime::traits::{Block as BlockT, Hash, HashFor};
 use sp_runtime::ConsensusEngineId;
 pub use sc_network::message::generic::{Message, ConsensusMessage};
 use sc_network::config::Roles;
+use wasm_timer::Instant;
 
 // FIXME: Add additional spam/DoS attack protection: https://github.com/paritytech/substrate/issues/1115
 const KNOWN_MESSAGES_CACHE_SIZE: usize = 4096;
@@ -165,7 +166,7 @@ pub struct ConsensusGossip<B: BlockT> {
 	messages: Vec<MessageEntry<B>>,
 	known_messages: LruCache<B::Hash, ()>,
 	validators: HashMap<ConsensusEngineId, Arc<dyn Validator<B>>>,
-	next_broadcast: time::Instant,
+	next_broadcast: Instant,
 }
 
 impl<B: BlockT> ConsensusGossip<B> {
@@ -177,7 +178,7 @@ impl<B: BlockT> ConsensusGossip<B> {
 			messages: Default::default(),
 			known_messages: LruCache::new(KNOWN_MESSAGES_CACHE_SIZE),
 			validators: Default::default(),
-			next_broadcast: time::Instant::now() + REBROADCAST_INTERVAL,
+			next_broadcast: Instant::now() + REBROADCAST_INTERVAL,
 		}
 	}
 
@@ -260,9 +261,9 @@ impl<B: BlockT> ConsensusGossip<B> {
 	/// Perform periodic maintenance
 	pub fn tick(&mut self, network: &mut dyn Network<B>) {
 		self.collect_garbage();
-		if time::Instant::now() >= self.next_broadcast {
+		if Instant::now() >= self.next_broadcast {
 			self.rebroadcast(network);
-			self.next_broadcast = time::Instant::now() + REBROADCAST_INTERVAL;
+			self.next_broadcast = Instant::now() + REBROADCAST_INTERVAL;
 		}
 	}
 

client/network/Cargo.toml

@@ -17,7 +17,8 @@ fnv = "1.0.6"
 fork-tree = { version = "2.0.0", path = "../../utils/fork-tree" }
 futures = "0.3.1"
 futures_codec = "0.3.3"
-futures-timer = "0.4.0"
+futures-timer = "3.0.1"
+wasm-timer = "0.2"
 libp2p = { version = "0.15.0", default-features = false, features = ["libp2p-websocket"] }
 linked-hash-map = "0.5.2"
 linked_hash_set = "0.1.3"

client/network/src/debug_info.rs

@@ -26,7 +26,8 @@ use log::{debug, trace, error};
 use std::collections::hash_map::Entry;
 use std::pin::Pin;
 use std::task::{Context, Poll};
-use std::time::{Duration, Instant};
+use std::time::Duration;
+use wasm_timer::Instant;
 use crate::utils::interval;
 
 /// Time after we disconnect from a node before we purge its information from the cache.

client/network/src/protocol.rs

@@ -52,6 +52,7 @@ use crate::chain::{Client, FinalityProofProvider};
 use sc_client_api::{FetchChecker, ChangesProof, StorageProof};
 use crate::error;
 use util::LruHashSet;
+use wasm_timer::Instant;
 
 mod legacy_proto;
 mod util;
@@ -158,17 +159,17 @@ struct PacketStats {
 /// A peer that we are connected to
 /// and from whom we have not yet received a Status message.
 struct HandshakingPeer {
-	timestamp: time::Instant,
+	timestamp: Instant,
 }
 
 /// Peer information
 #[derive(Debug, Clone)]
 struct Peer<B: BlockT, H: ExHashT> {
 	info: PeerInfo<B>,
 	/// Current block request, if any.
-	block_request: Option<(time::Instant, message::BlockRequest<B>)>,
+	block_request: Option<(Instant, message::BlockRequest<B>)>,
 	/// Requests we are no longer insterested in.
-	obsolete_requests: HashMap<message::RequestId, time::Instant>,
+	obsolete_requests: HashMap<message::RequestId, Instant>,
 	/// Holds a set of transactions known to this peer.
 	known_extrinsics: LruHashSet<H>,
 	/// Holds a set of blocks known to this peer.
@@ -701,7 +702,7 @@ impl<B: BlockT, S: NetworkSpecialization<B>, H: ExHashT> Protocol<B, S, H> {
 	/// Called when a new peer is connected
 	pub fn on_peer_connected(&mut self, who: PeerId) {
 		trace!(target: "sync", "Connecting {}", who);
-		self.handshaking_peers.insert(who.clone(), HandshakingPeer { timestamp: time::Instant::now() });
+		self.handshaking_peers.insert(who.clone(), HandshakingPeer { timestamp: Instant::now() });
 		self.send_status(who);
 	}
 
@@ -890,7 +891,7 @@ impl<B: BlockT, S: NetworkSpecialization<B>, H: ExHashT> Protocol<B, S, H> {
 	}
 
 	fn maintain_peers(&mut self) {
-		let tick = time::Instant::now();
+		let tick = Instant::now();
 		let mut aborting = Vec::new();
 		{
 			for (who, peer) in self.context_data.peers.iter() {
@@ -1833,7 +1834,7 @@ fn send_request<B: BlockT, H: ExHashT>(
 				trace!(target: "sync", "Request {} for {} is now obsolete.", request.id, who);
 				peer.obsolete_requests.insert(request.id, timestamp);
 			}
-			peer.block_request = Some((time::Instant::now(), r.clone()));
+			peer.block_request = Some((Instant::now(), r.clone()));
 		}
 	}
 	send_message::<B>(behaviour, stats, who, message)