offline tracker

polkadot/consensus/src/lib.rs

@@ -80,20 +80,26 @@ use futures::prelude::*;
 use futures::future;
 use collation::CollationFetch;
 use dynamic_inclusion::DynamicInclusion;
+use parking_lot::RwLock;
 
 pub use self::collation::{validate_collation, Collators};
 pub use self::error::{ErrorKind, Error};
+pub use self::offline_tracker::OfflineTracker;
 pub use self::shared_table::{SharedTable, StatementProducer, ProducedStatements, Statement, SignedStatement, GenericStatement};
 pub use service::Service;
 
 mod dynamic_inclusion;
 mod evaluation;
 mod error;
+mod offline_tracker;
 mod service;
 mod shared_table;
 
 pub mod collation;
 
+/// Shared offline validator tracker.
+pub type SharedOfflineTracker = Arc<RwLock<OfflineTracker>>;
+
 // block size limit.
 const MAX_TRANSACTIONS_SIZE: usize = 4 * 1024 * 1024;
 
@@ -240,6 +246,8 @@ pub struct ProposerFactory<C, N, P> {
 	pub parachain_empty_duration: Duration,
 	/// Store for extrinsic data.
 	pub extrinsic_store: ExtrinsicStore,
+	/// Offline-tracker.
+	pub offline: SharedOfflineTracker,
 }
 
 impl<C, N, P> bft::Environment<Block> for ProposerFactory<C, N, P>
@@ -255,10 +263,11 @@ impl<C, N, P> bft::Environment<Block> for ProposerFactory<C, N, P>
 	type Output = N::Output;
 	type Error = Error;
 
-	fn init(&self,
+	fn init(
+		&self,
 		parent_header: &Header,
 		authorities: &[AuthorityId],
-		sign_with: Arc<ed25519::Pair>
+		sign_with: Arc<ed25519::Pair>,
 	) -> Result<(Self::Proposer, Self::Input, Self::Output), Error> {
 		use runtime_primitives::traits::{Hash as HashT, BlakeTwo256};
 
@@ -269,6 +278,9 @@ impl<C, N, P> bft::Environment<Block> for ProposerFactory<C, N, P>
 		let random_seed = self.client.random_seed(&id)?;
 		let random_seed = BlakeTwo256::hash(&*random_seed);
 
+		let validators = self.client.validators(&id)?;
+		self.offline.write().note_new_block(&validators[..]);
+
 		let (group_info, local_duty) = make_group_info(
 			duty_roster,
 			authorities,
@@ -326,6 +338,7 @@ impl<C, N, P> bft::Environment<Block> for ProposerFactory<C, N, P>
 			random_seed,
 			table,
 			transaction_pool: self.transaction_pool.clone(),
+			offline: self.offline.clone(),
 			_drop_signal: drop_signal,
 		};
 
@@ -403,6 +416,7 @@ pub struct Proposer<C: PolkadotApi> {
 	random_seed: Hash,
 	table: Arc<SharedTable>,
 	transaction_pool: Arc<TransactionPool<C>>,
+	offline: SharedOfflineTracker,
 	_drop_signal: exit_future::Signal,
 }
 
@@ -452,6 +466,7 @@ impl<C> bft::Proposer<Block> for Proposer<C>
 			client: self.client.clone(),
 			transaction_pool: self.transaction_pool.clone(),
 			table: self.table.clone(),
+			offline: self.offline.clone(),
 			timing,
 		})
 	}
@@ -692,6 +707,7 @@ pub struct CreateProposal<C: PolkadotApi>  {
 	transaction_pool: Arc<TransactionPool<C>>,
 	table: Arc<SharedTable>,
 	timing: ProposalTiming,
+	offline: SharedOfflineTracker,
 }
 
 impl<C> CreateProposal<C> where C: PolkadotApi {
@@ -701,6 +717,11 @@ impl<C> CreateProposal<C> where C: PolkadotApi {
 
 		// TODO: handle case when current timestamp behind that in state.
 		let timestamp = current_timestamp();
+
+		// TODO: include reports into block.
+		let validators = self.client.validators(&self.parent_id)?;
+		let _reports = self.offline.read().reports(&validators[..]);
+
 		let mut block_builder = self.client.build_block(&self.parent_id, timestamp, candidates)?;
 
 		{

polkadot/consensus/src/offline_tracker.rs

@@ -0,0 +1,92 @@
+// Copyright 2018 Parity Technologies (UK) Ltd.
+// This file is part of Polkadot.
+
+// Polkadot 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.
+
+// Polkadot 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 Polkadot.  If not, see <http://www.gnu.org/licenses/>.
+
+//! Tracks offline validators.
+
+use polkadot_primitives::AccountId;
+
+use std::collections::HashMap;
+use std::time::{Instant, Duration};
+
+// time before we report a validator.
+const REPORT_TIME: Duration = Duration::from_secs(60 * 5);
+
+struct Observed {
+	last_round_end: Instant,
+	offline_since: Instant,
+}
+
+impl Observed {
+	fn new() -> Observed {
+		let now = Instant::now();
+		Observed {
+			last_round_end: now,
+			offline_since: now,
+		}
+	}
+
+	fn note_round_end(&mut self, was_online: bool) {
+		let now = Instant::now();
+
+		self.last_round_end = now;
+		if was_online {
+			self.offline_since = now;
+		}
+	}
+
+	fn is_active(&self) -> bool {
+		// can happen if clocks are not monotonic
+		if self.offline_since > self.last_round_end { return true }
+		self.last_round_end.duration_since(self.offline_since) < REPORT_TIME
+	}
+}
+
+/// Tracks offline validators and can issue a report for those offline.
+pub struct OfflineTracker {
+	observed: HashMap<AccountId, Observed>,
+}
+
+impl OfflineTracker {
+	/// Create a new tracker.
+	pub fn new() -> Self {
+		OfflineTracker { observed: HashMap::new() }
+	}
+
+	/// Note new consensus is starting with the given set of validators.
+	pub fn note_new_block(&mut self, validators: &[AccountId]) {
+		use std::collections::HashSet;
+
+		let set: HashSet<_> = validators.iter().cloned().collect();
+		self.observed.retain(|k, _| set.contains(k));
+	}
+
+	/// Note that a round has ended.
+	pub fn note_round_end(&mut self, validator: AccountId, was_online: bool) {
+		self.observed.entry(validator)
+			.or_insert_with(Observed::new)
+			.note_round_end(was_online);
+	}
+
+	/// Generate a vector of reports for the given account IDs.
+	/// The length of the vector will be the same as the length of the validator
+	/// list.
+	/// `None` is "not enough data", `true` is "online", and `false` is "offline".
+	pub fn reports(&self, validators: &[AccountId]) -> Vec<Option<bool>> {
+		validators.iter()
+			.map(|v| self.observed.get(v).map(Observed::is_active))
+			.collect()
+	}
+}

polkadot/consensus/src/service.rs

@@ -166,6 +166,9 @@ impl Service {
 			N::TableRouter: Send + 'static,
 			<N::Collation as IntoFuture>::Future: Send + 'static,
 	{
+		use parking_lot::RwLock;
+		use super::OfflineTracker;
+
 		let (signal, exit) = ::exit_future::signal();
 		let thread = thread::spawn(move || {
 			let mut runtime = LocalRuntime::new().expect("Could not create local runtime");
@@ -179,6 +182,7 @@ impl Service {
 				parachain_empty_duration,
 				handle: thread_pool.clone(),
 				extrinsic_store: extrinsic_store.clone(),
+				offline: Arc::new(RwLock::new(OfflineTracker::new())),
 			};
 			let bft_service = Arc::new(BftService::new(client.clone(), key, factory));