BFT delay adjustments (#593)

* force delay only on votes

* set proposal timestamp forward

* Adjusted timeout formula

Author: arkpar

polkadot/consensus/src/lib.rs

@@ -63,7 +63,7 @@ extern crate substrate_keyring;
 
 use std::collections::{HashMap, HashSet};
 use std::sync::Arc;
-use std::time::{Duration, Instant};
+use std::time::{self, Duration, Instant};
 
 use codec::{Decode, Encode};
 use polkadot_api::PolkadotApi;
@@ -267,6 +267,9 @@ impl<C, N, P> bft::Environment<Block> for ProposerFactory<C, N, P>
 	) -> Result<(Self::Proposer, Self::Input, Self::Output), Error> {
 		use runtime_primitives::traits::{Hash as HashT, BlakeTwo256};
 
+		// force delay in evaluation this long.
+		const FORCE_DELAY: Timestamp = 5;
+
 		let parent_hash = parent_header.hash().into();
 
 		let id = BlockId::hash(parent_hash);
@@ -330,6 +333,7 @@ impl<C, N, P> bft::Environment<Block> for ProposerFactory<C, N, P>
 			transaction_pool: self.transaction_pool.clone(),
 			offline: self.offline.clone(),
 			validators,
+			minimum_timestamp: current_timestamp() + FORCE_DELAY,
 			_drop_signal: drop_signal,
 		};
 
@@ -386,6 +390,7 @@ pub struct Proposer<C: PolkadotApi> {
 	transaction_pool: Arc<TransactionPool<C>>,
 	offline: SharedOfflineTracker,
 	validators: Vec<AccountId>,
+	minimum_timestamp: u64,
 	_drop_signal: exit_future::Signal,
 }
 
@@ -437,6 +442,7 @@ impl<C> bft::Proposer<Block> for Proposer<C>
 			table: self.table.clone(),
 			offline: self.offline.clone(),
 			validators: self.validators.clone(),
+			minimum_timestamp: self.minimum_timestamp,
 			timing,
 		})
 	}
@@ -489,9 +495,11 @@ impl<C> bft::Proposer<Block> for Proposer<C>
 			);
 
 			// the duration until the given timestamp is current
-			let proposed_timestamp = proposal.timestamp();
+			let proposed_timestamp = ::std::cmp::max(self.minimum_timestamp, proposal.timestamp());
 			let timestamp_delay = if proposed_timestamp > current_timestamp {
-				Some(now + Duration::from_secs(proposed_timestamp - current_timestamp))
+				let delay_s = proposed_timestamp - current_timestamp;
+				debug!(target: "bft", "Delaying evaluation of proposal for {} seconds", delay_s);
+				Some(now + Duration::from_secs(delay_s))
 			} else {
 				None
 			};
@@ -642,8 +650,6 @@ impl<C> bft::Proposer<Block> for Proposer<C>
 }
 
 fn current_timestamp() -> Timestamp {
-	use std::time;
-
 	time::SystemTime::now().duration_since(time::UNIX_EPOCH)
 		.expect("now always later than unix epoch; qed")
 		.as_secs()
@@ -697,6 +703,7 @@ pub struct CreateProposal<C: PolkadotApi>  {
 	timing: ProposalTiming,
 	validators: Vec<AccountId>,
 	offline: SharedOfflineTracker,
+	minimum_timestamp: Timestamp,
 }
 
 impl<C> CreateProposal<C> where C: PolkadotApi {
@@ -708,7 +715,7 @@ impl<C> CreateProposal<C> where C: PolkadotApi {
 		const MAX_VOTE_OFFLINE_SECONDS: Duration = Duration::from_secs(60);
 
 		// TODO: handle case when current timestamp behind that in state.
-		let timestamp = current_timestamp();
+		let timestamp = ::std::cmp::max(self.minimum_timestamp, current_timestamp());
 
 		let elapsed_since_start = self.timing.dynamic_inclusion.started_at().elapsed();
 		let offline_indices = if elapsed_since_start > MAX_VOTE_OFFLINE_SECONDS {

polkadot/consensus/src/service.rs

@@ -38,7 +38,7 @@ use transaction_pool::TransactionPool;
 use tokio::executor::current_thread::TaskExecutor as LocalThreadHandle;
 use tokio::runtime::TaskExecutor as ThreadPoolHandle;
 use tokio::runtime::current_thread::Runtime as LocalRuntime;
-use tokio::timer::{Delay, Interval};
+use tokio::timer::Interval;
 
 use super::{Network, Collators, ProposerFactory};
 use error;
@@ -58,25 +58,10 @@ fn start_bft<F, C>(
 	<F::Proposer as bft::Proposer<Block>>::Error: ::std::fmt::Display + Into<error::Error>,
 	<F as bft::Environment<Block>>::Error: ::std::fmt::Display
 {
-	const DELAY_UNTIL: Duration = Duration::from_millis(5000);
-
 	let mut handle = LocalThreadHandle::current();
 	match bft_service.build_upon(&header) {
-		Ok(Some(bft_work)) => {
-			// do not poll work for some amount of time.
-			let work = Delay::new(Instant::now() + DELAY_UNTIL).then(move |res| {
-				if let Err(e) = res {
-					warn!(target: "bft", "Failed to force delay of consensus: {:?}", e);
-				}
-
-				debug!(target: "bft", "Starting agreement. After forced delay for {:?}",
-					DELAY_UNTIL);
-
-				bft_work
-			});
-			if let Err(e) = handle.spawn_local(Box::new(work)) {
-		    	warn!(target: "bft", "Couldn't initialize BFT agreement: {:?}", e);
-			}
+		Ok(Some(bft_work)) => if let Err(e) = handle.spawn_local(Box::new(bft_work)) {
+		    warn!(target: "bft", "Couldn't initialize BFT agreement: {:?}", e);
 		}
 		Ok(None) => trace!(target: "bft", "Could not start agreement on top of {}", header.hash()),
 		Err(e) => warn!(target: "bft", "BFT agreement error: {}", e),

substrate/bft/src/lib.rs

@@ -237,10 +237,12 @@ impl<B: Block, P: Proposer<B>> BftInstance<B, P>
 
 {
 	fn round_timeout_duration(&self, round: usize) -> Duration {
-		const ROUND_INCREMENT_STEP: usize = 10000;
+		// 2^(min(6, x/8)) * 10
+		// Grows exponentially starting from 10 seconds, capped at 640 seconds.
+		const ROUND_INCREMENT_STEP: usize = 8;
 
 		let round = round / ROUND_INCREMENT_STEP;
-		let round = ::std::cmp::min(63, round) as u32;
+		let round = ::std::cmp::min(6, round) as u32;
 
 		let timeout = 1u64.checked_shl(round)
 			.unwrap_or_else(u64::max_value)
@@ -867,7 +869,7 @@ mod tests {
 				hash: None,
 				start_round: 0,
 			})),
-			round_timeout_multiplier: 4,
+			round_timeout_multiplier: 10,
 			key: Arc::new(Keyring::One.into()),
 			factory: DummyFactory
 		}