From b006586d19d44d40882747d10f5b04346b9f0895 Mon Sep 17 00:00:00 2001 From: Paul Hauner Date: Wed, 27 Mar 2019 10:36:20 +1100 Subject: [PATCH] Add slot timer to beacon node --- beacon_node/Cargo.toml | 1 + beacon_node/beacon_chain/src/beacon_chain.rs | 14 ++++ beacon_node/client/Cargo.toml | 1 + beacon_node/client/src/lib.rs | 76 ++++++++++++++++++- beacon_node/client/src/notifier.rs | 4 +- beacon_node/src/run.rs | 2 + eth2/utils/slot_clock/src/lib.rs | 3 + .../slot_clock/src/system_time_slot_clock.rs | 28 +++++++ .../slot_clock/src/testing_slot_clock.rs | 6 ++ 9 files changed, 132 insertions(+), 3 deletions(-) diff --git a/beacon_node/Cargo.toml b/beacon_node/Cargo.toml index e7aaf938de..a090c1cc54 100644 --- a/beacon_node/Cargo.toml +++ b/beacon_node/Cargo.toml @@ -14,6 +14,7 @@ slog-term = "^2.4.0" slog-async = "^2.3.0" ctrlc = { version = "3.1.1", features = ["termination"] } tokio = "0.1.15" +tokio-timer = "0.2.10" futures = "0.1.25" exit-future = "0.1.3" state_processing = { path = "../eth2/state_processing" } diff --git a/beacon_node/beacon_chain/src/beacon_chain.rs b/beacon_node/beacon_chain/src/beacon_chain.rs index 2caee2f988..6fcb3fde40 100644 --- a/beacon_node/beacon_chain/src/beacon_chain.rs +++ b/beacon_node/beacon_chain/src/beacon_chain.rs @@ -409,6 +409,20 @@ where } } + /// Reads the slot clock (see `self.read_slot_clock()` and returns the number of slots since + /// genesis. + pub fn slots_since_genesis(&self) -> Option { + let now = self.read_slot_clock()?; + + if now < self.spec.genesis_slot { + None + } else { + Some(SlotHeight::from( + now.as_u64() - self.spec.genesis_slot.as_u64(), + )) + } + } + /// Returns slot of the present state. /// /// This is distinct to `read_slot_clock`, which reads from the actual system clock. If diff --git a/beacon_node/client/Cargo.toml b/beacon_node/client/Cargo.toml index 12c1b5c802..8956dbb07d 100644 --- a/beacon_node/client/Cargo.toml +++ b/beacon_node/client/Cargo.toml @@ -14,6 +14,7 @@ types = { path = "../../eth2/types" } slot_clock = { path = "../../eth2/utils/slot_clock" } error-chain = "0.12.0" slog = "^2.2.3" +ssz = { path = "../../eth2/utils/ssz" } tokio = "0.1.15" clap = "2.32.0" dirs = "1.0.3" diff --git a/beacon_node/client/src/lib.rs b/beacon_node/client/src/lib.rs index b24d2cb7f2..2ce181aa19 100644 --- a/beacon_node/client/src/lib.rs +++ b/beacon_node/client/src/lib.rs @@ -9,11 +9,17 @@ use beacon_chain::BeaconChain; pub use client_config::ClientConfig; pub use client_types::ClientTypes; use exit_future::Signal; +use futures::{future::Future, Stream}; use network::Service as NetworkService; -use slog::o; +use slog::{error, info, o}; +use slot_clock::SlotClock; +use ssz::TreeHash; use std::marker::PhantomData; use std::sync::Arc; +use std::time::{Duration, Instant}; use tokio::runtime::TaskExecutor; +use tokio::timer::Interval; +use types::Hash256; /// Main beacon node client service. This provides the connection and initialisation of the clients /// sub-services in multiple threads. @@ -26,6 +32,8 @@ pub struct Client { pub network: Arc, /// Signal to terminate the RPC server. pub rpc_exit_signal: Option, + /// Signal to terminate the slot timer. + pub slot_timer_exit_signal: Option, /// The clients logger. log: slog::Logger, /// Marker to pin the beacon chain generics. @@ -42,6 +50,17 @@ impl Client { // generate a beacon chain let beacon_chain = TClientType::initialise_beacon_chain(&config); + { + let state = beacon_chain.state.read(); + let state_root = Hash256::from_slice(&state.hash_tree_root()); + info!( + log, + "ChainInitialized"; + "state_root" => format!("{}", state_root), + "genesis_time" => format!("{}", state.genesis_time), + ); + } + // Start the network service, libp2p and syncing threads // TODO: Add beacon_chain reference to network parameters let network_config = &config.net_conf; @@ -65,10 +84,65 @@ impl Client { )); } + let (slot_timer_exit_signal, exit) = exit_future::signal(); + if let Ok(Some(duration_to_next_slot)) = beacon_chain.slot_clock.duration_to_next_slot() { + // set up the validator work interval - start at next slot and proceed every slot + let interval = { + // Set the interval to start at the next slot, and every slot after + let slot_duration = Duration::from_secs(config.spec.seconds_per_slot); + //TODO: Handle checked add correctly + Interval::new(Instant::now() + duration_to_next_slot, slot_duration) + }; + + let chain = beacon_chain.clone(); + let log = log.new(o!("Service" => "SlotTimer")); + + let state_slot = chain.state.read().slot; + let wall_clock_slot = chain.read_slot_clock().unwrap(); + let slots_since_genesis = chain.slots_since_genesis().unwrap(); + info!( + log, + "Starting SlotTimer"; + "state_slot" => state_slot, + "wall_clock_slot" => wall_clock_slot, + "slots_since_genesis" => slots_since_genesis, + "catchup_distance" => wall_clock_slot - state_slot, + ); + executor.spawn( + exit.until( + interval + .for_each(move |_| { + if let Some(genesis_height) = chain.slots_since_genesis() { + match chain.catchup_state() { + Ok(_) => info!( + log, + "NewSlot"; + "slot" => chain.state.read().slot, + "slots_since_genesis" => genesis_height, + ), + Err(e) => error!( + log, + "StateCatchupFailed"; + "state_slot" => chain.state.read().slot, + "slots_since_genesis" => genesis_height, + "error" => format!("{:?}", e), + ), + }; + } + + Ok(()) + }) + .map_err(|_| ()), + ) + .map(|_| ()), + ); + } + Ok(Client { config, beacon_chain, rpc_exit_signal, + slot_timer_exit_signal: Some(slot_timer_exit_signal), log, network, phantom: PhantomData, diff --git a/beacon_node/client/src/notifier.rs b/beacon_node/client/src/notifier.rs index 335183c7de..91a9f3a261 100644 --- a/beacon_node/client/src/notifier.rs +++ b/beacon_node/client/src/notifier.rs @@ -2,7 +2,7 @@ use crate::Client; use crate::ClientTypes; use exit_future::Exit; use futures::{Future, Stream}; -use slog::{debug, info, o}; +use slog::{debug, o}; use std::sync::{Arc, Mutex}; use std::time::{Duration, Instant}; use tokio::runtime::TaskExecutor; @@ -22,7 +22,7 @@ pub fn run(client: &Client, executor: TaskExecutor, exit: Exi // build heartbeat logic here let heartbeat = move |_| { - info!(log, "Temp heartbeat output"); + debug!(log, "Temp heartbeat output"); //TODO: Remove this logic. Testing only let mut count = counter.lock().unwrap(); *count += 1; diff --git a/beacon_node/src/run.rs b/beacon_node/src/run.rs index b3b2844526..1d9156124f 100644 --- a/beacon_node/src/run.rs +++ b/beacon_node/src/run.rs @@ -6,10 +6,12 @@ use futures::Future; use slog::info; use std::cell::RefCell; use tokio::runtime::Builder; +use tokio_timer::clock::Clock; pub fn run_beacon_node(config: ClientConfig, log: &slog::Logger) -> error::Result<()> { let mut runtime = Builder::new() .name_prefix("main-") + .clock(Clock::system()) .build() .map_err(|e| format!("{:?}", e))?; diff --git a/eth2/utils/slot_clock/src/lib.rs b/eth2/utils/slot_clock/src/lib.rs index 0379d50d9a..fd5a2d1d7d 100644 --- a/eth2/utils/slot_clock/src/lib.rs +++ b/eth2/utils/slot_clock/src/lib.rs @@ -3,10 +3,13 @@ mod testing_slot_clock; pub use crate::system_time_slot_clock::{Error as SystemTimeSlotClockError, SystemTimeSlotClock}; pub use crate::testing_slot_clock::{Error as TestingSlotClockError, TestingSlotClock}; +use std::time::Duration; pub use types::Slot; pub trait SlotClock: Send + Sync { type Error; fn present_slot(&self) -> Result, Self::Error>; + + fn duration_to_next_slot(&self) -> Result, Self::Error>; } diff --git a/eth2/utils/slot_clock/src/system_time_slot_clock.rs b/eth2/utils/slot_clock/src/system_time_slot_clock.rs index 816180d6e6..4dfc6b37da 100644 --- a/eth2/utils/slot_clock/src/system_time_slot_clock.rs +++ b/eth2/utils/slot_clock/src/system_time_slot_clock.rs @@ -54,6 +54,10 @@ impl SlotClock for SystemTimeSlotClock { .and_then(|s| Some(s + self.genesis_slot))), } } + + fn duration_to_next_slot(&self) -> Result, Error> { + duration_to_next_slot(self.genesis_seconds, self.slot_duration_seconds) + } } impl From for Error { @@ -67,6 +71,30 @@ fn slot_from_duration(slot_duration_seconds: u64, duration: Duration) -> Option< duration.as_secs().checked_div(slot_duration_seconds)?, )) } +// calculate the duration to the next slot +fn duration_to_next_slot( + genesis_time: u64, + seconds_per_slot: u64, +) -> Result, Error> { + let now = SystemTime::now().duration_since(SystemTime::UNIX_EPOCH)?; + let genesis_time = Duration::from_secs(genesis_time); + + if now < genesis_time { + return Ok(None); + } + + let since_genesis = now - genesis_time; + + let elapsed_slots = since_genesis.as_secs() / seconds_per_slot; + + let next_slot_start_seconds = (elapsed_slots + 1) + .checked_mul(seconds_per_slot) + .expect("Next slot time should not overflow u64"); + + let time_to_next_slot = Duration::from_secs(next_slot_start_seconds) - since_genesis; + + Ok(Some(time_to_next_slot)) +} #[cfg(test)] mod tests { diff --git a/eth2/utils/slot_clock/src/testing_slot_clock.rs b/eth2/utils/slot_clock/src/testing_slot_clock.rs index 80ee405397..b5c36dfa0a 100644 --- a/eth2/utils/slot_clock/src/testing_slot_clock.rs +++ b/eth2/utils/slot_clock/src/testing_slot_clock.rs @@ -1,5 +1,6 @@ use super::SlotClock; use std::sync::RwLock; +use std::time::Duration; use types::Slot; #[derive(Debug, PartialEq)] @@ -32,6 +33,11 @@ impl SlotClock for TestingSlotClock { let slot = *self.slot.read().expect("TestingSlotClock poisoned."); Ok(Some(Slot::new(slot))) } + + /// Always returns a duration of 1 second. + fn duration_to_next_slot(&self) -> Result, Error> { + Ok(Some(Duration::from_secs(1))) + } } #[cfg(test)]