More timely block import notifications (#306)

* More timely block import notifications

* Grumbles.

* More wrapping

* Fix build

* Fixes

substrate/.gitignore

@@ -11,4 +11,5 @@ substrate/pwasm-alloc/Cargo.lock
 substrate/pwasm-libc/Cargo.lock
 demo/runtime/wasm/target/
 **/._*
-.vscode
+.vscode
+polkadot.*

substrate/Cargo.lock

@@ -2145,6 +2145,7 @@ dependencies = [
  "hex-literal 0.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
  "log 0.3.9 (registry+https://github.com/rust-lang/crates.io-index)",
  "parking_lot 0.4.8 (registry+https://github.com/rust-lang/crates.io-index)",
+ "slog 2.2.3 (registry+https://github.com/rust-lang/crates.io-index)",
  "substrate-bft 0.1.0",
  "substrate-codec 0.1.0",
  "substrate-executor 0.1.0",
@@ -2154,6 +2155,7 @@ dependencies = [
  "substrate-runtime-primitives 0.1.0",
  "substrate-runtime-support 0.1.0",
  "substrate-state-machine 0.1.0",
+ "substrate-telemetry 0.2.0",
  "substrate-test-client 0.1.0",
  "triehash 0.1.2 (registry+https://github.com/rust-lang/crates.io-index)",
 ]

substrate/polkadot/cli/src/informant.rs

@@ -63,7 +63,6 @@ pub fn start<C>(service: &Service<C>, exit: ::exit_future::Exit, handle: TaskExe
 	let client = service.client();
 	let display_block_import = client.import_notification_stream().for_each(|n| {
 		info!(target: "polkadot", "Imported #{} ({})", n.header.number, n.hash);
-		telemetry!("block.import"; "height" => n.header.number, "best" => ?n.hash);
 		Ok(())
 	});
 

substrate/substrate/client/Cargo.toml

@@ -11,6 +11,7 @@ triehash = "0.1"
 hex-literal = "0.1"
 futures = "0.1.17"
 ed25519 = { path = "../ed25519" }
+slog = "^2"
 substrate-bft = { path = "../bft" }
 substrate-codec = { path = "../codec" }
 substrate-executor = { path = "../executor" }
@@ -20,6 +21,7 @@ substrate-runtime-support = { path = "../runtime-support" }
 substrate-runtime-primitives = { path = "../runtime/primitives" }
 substrate-state-machine = { path = "../state-machine" }
 substrate-keyring = { path = "../../substrate/keyring" }
+substrate-telemetry = { path = "../telemetry" }
 
 [dev-dependencies]
 substrate-test-client = { path = "../test-client" }

substrate/substrate/client/src/client.rs

@@ -21,7 +21,7 @@ use futures::sync::mpsc;
 use parking_lot::{Mutex, RwLock};
 use primitives::AuthorityId;
 use runtime_primitives::{bft::Justification, generic::{BlockId, SignedBlock, Block as RuntimeBlock}};
-use runtime_primitives::traits::{Block as BlockT, Header as HeaderT, Zero, One};
+use runtime_primitives::traits::{Block as BlockT, Header as HeaderT, Zero, One, As};
 use runtime_primitives::BuildStorage;
 use primitives::storage::{StorageKey, StorageData};
 use codec::Slicable;
@@ -98,7 +98,7 @@ pub enum BlockStatus {
 }
 
 /// Block data origin.
-#[derive(Debug, PartialEq, Eq, Clone)]
+#[derive(Debug, PartialEq, Eq, Clone, Copy)]
 pub enum BlockOrigin {
 	/// Genesis block built into the client.
 	Genesis,
@@ -297,9 +297,15 @@ impl<B, E, Block> Client<B, E, Block> where
 		}
 		let hash = header.hash();
 		let _import_lock = self.import_lock.lock();
+		let height: u64 = header.number().as_();
 		*self.importing_block.write() = Some(hash);
 		let result = self.execute_and_import_block(origin, hash, header, justification, body);
 		*self.importing_block.write() = None;
+		telemetry!("block.import";
+			"height" => height,
+			"best" => ?hash,
+			"origin" => ?origin
+		);
 		result
 	}
 

substrate/substrate/client/src/lib.rs

@@ -28,6 +28,8 @@ extern crate substrate_runtime_primitives as runtime_primitives;
 extern crate substrate_state_machine as state_machine;
 #[cfg(test)] extern crate substrate_keyring as keyring;
 #[cfg(test)] extern crate substrate_test_client as test_client;
+#[macro_use] extern crate substrate_telemetry;
+#[macro_use] extern crate slog;	// needed until we can reexport `slog_info` from `substrate_telemetry`
 
 extern crate ed25519;
 extern crate futures;

substrate/substrate/telemetry/src/lib.rs

@@ -44,6 +44,9 @@ pub struct TelemetryConfig {
 	pub on_connect: Box<Fn() + Send + 'static>,
 }
 
+/// Size of the channel for passing messages to telemetry thread.
+const CHANNEL_SIZE: usize = 262144;
+
 /// Initialise telemetry.
 pub fn init_telemetry(config: TelemetryConfig) -> slog_scope::GlobalLoggerGuard {
 	let log = slog::Logger::root(
@@ -58,7 +61,9 @@ pub fn init_telemetry(config: TelemetryConfig) -> slog_scope::GlobalLoggerGuard
 					first_time: true,	// ensures that on_connect will be called.
 				}
 			).fuse()
-		).build().fuse(), o!()
+		).chan_size(CHANNEL_SIZE)
+		.overflow_strategy(slog_async::OverflowStrategy::DropAndReport)
+		.build().fuse(), o!()
 	);
 	slog_scope::set_global_logger(log)
 }