Removes duty manager service in favour of tokio timer

validator_client/src/duties/mod.rs

@@ -1,21 +1,19 @@
 mod epoch_duties;
 mod grpc;
-mod service;
 #[cfg(test)]
 mod test_node;
 mod traits;
 
 pub use self::epoch_duties::EpochDutiesMap;
 use self::epoch_duties::{EpochDuties, EpochDutiesMapError};
-pub use self::service::DutiesManagerService;
 use self::traits::{BeaconNode, BeaconNodeError};
 use bls::PublicKey;
 use slot_clock::SlotClock;
 use std::sync::Arc;
-use types::{ChainSpec, Epoch};
+use types::{ChainSpec, Epoch, Slot};
 
 #[derive(Debug, PartialEq, Clone, Copy)]
-pub enum PollOutcome {
+pub enum UpdateOutcome {
     /// The `EpochDuties` were not updated during this poll.
     NoChange(Epoch),
     /// The `EpochDuties` for the `epoch` were previously unknown, but obtained in the poll.
@@ -50,19 +48,11 @@ pub struct DutiesManager<T: SlotClock, U: BeaconNode> {
 }
 
 impl<T: SlotClock, U: BeaconNode> DutiesManager<T, U> {
-    /// Poll the Beacon Node for `EpochDuties`.
+    /// Check the Beacon Node for `EpochDuties`.
     ///
     /// The present `epoch` will be learned from the supplied `SlotClock`. In production this will
     /// be a wall-clock (e.g., system time, remote server time, etc.).
-    //TODO: Remove the poll and trust the tokio system-clock timer. Leave for now to ensure the
+    pub fn update(&self, slot: Slot) -> Result<UpdateOutcome, Error> {
-    //timer is accurate.
-    pub fn poll(&self) -> Result<PollOutcome, Error> {
-        let slot = self
-            .slot_clock
-            .present_slot()
-            .map_err(|_| Error::SlotClockError)?
-            .ok_or(Error::SlotUnknowable)?;
-
         let epoch = slot.epoch(self.spec.slots_per_epoch);
 
         if let Some(duties) = self
@@ -72,17 +62,17 @@ impl<T: SlotClock, U: BeaconNode> DutiesManager<T, U> {
             // If these duties were known, check to see if they're updates or identical.
             let result = if let Some(known_duties) = self.duties_map.get(epoch)? {
                 if known_duties == duties {
-                    Ok(PollOutcome::NoChange(epoch))
+                    Ok(UpdateOutcome::NoChange(epoch))
                 } else {
-                    Ok(PollOutcome::DutiesChanged(epoch, duties))
+                    Ok(UpdateOutcome::DutiesChanged(epoch, duties))
                 }
             } else {
-                Ok(PollOutcome::NewDuties(epoch, duties))
+                Ok(UpdateOutcome::NewDuties(epoch, duties))
             };
             self.duties_map.insert(epoch, duties)?;
             result
         } else {
-            Ok(PollOutcome::UnknownValidatorOrEpoch(epoch))
+            Ok(UpdateOutcome::UnknownValidatorOrEpoch(epoch))
         }
     }
 }

validator_client/src/duties/service.rs

@@ -1,40 +0,0 @@
-use super::traits::BeaconNode;
-use super::{DutiesManager, PollOutcome};
-use slog::{debug, error, info, Logger};
-use slot_clock::SlotClock;
-use std::time::Duration;
-
-pub struct DutiesManagerService<T: SlotClock, U: BeaconNode> {
-    pub manager: DutiesManager<T, U>,
-    pub poll_interval_millis: u64,
-    pub log: Logger,
-}
-
-impl<T: SlotClock, U: BeaconNode> DutiesManagerService<T, U> {
-    /// Run a loop which polls the manager each `poll_interval_millis` milliseconds.
-    ///
-    /// Logs the results of the polls.
-    pub fn run(&mut self) {
-        loop {
-            match self.manager.poll() {
-                Err(error) => {
-                    error!(self.log, "Epoch duties poll error"; "error" => format!("{:?}", error))
-                }
-                Ok(PollOutcome::NoChange(epoch)) => {
-                    debug!(self.log, "No change in duties"; "epoch" => epoch)
-                }
-                Ok(PollOutcome::DutiesChanged(epoch, duties)) => {
-                    info!(self.log, "Duties changed (potential re-org)"; "epoch" => epoch, "duties" => format!("{:?}", duties))
-                }
-                Ok(PollOutcome::NewDuties(epoch, duties)) => {
-                    info!(self.log, "New duties obtained"; "epoch" => epoch, "duties" => format!("{:?}", duties))
-                }
-                Ok(PollOutcome::UnknownValidatorOrEpoch(epoch)) => {
-                    error!(self.log, "Epoch or validator unknown"; "epoch" => epoch)
-                }
-            };
-
-            std::thread::sleep(Duration::from_millis(self.poll_interval_millis));
-        }
-    }
-}

validator_client/src/service.rs

@@ -2,8 +2,8 @@
 use crate::attester_service::{AttestationGrpcClient, AttesterService};
 use crate::block_producer_service::{BeaconBlockGrpcClient, BlockProducerService};
 use crate::config::Config as ValidatorConfig;
-use crate::duties::PollOutcome;
+use crate::duties::UpdateOutcome;
-use crate::duties::{DutiesManager, DutiesManagerService, EpochDutiesMap};
+use crate::duties::{DutiesManager, EpochDutiesMap};
 use crate::error as error_chain;
 use crate::error::ErrorKind;
 use attester::test_utils::EpochMap;
@@ -230,9 +230,10 @@ impl Service {
             beacon_node: service.validator_client.clone(),
         };
 
+        // run the core thread
         runtime
             .block_on(interval.for_each(move |_| {
-                // update duties
+                // get the current slot
                 let current_slot = match service.slot_clock.present_slot() {
                     Err(e) => {
                         error!(service.log, "SystemTimeError {:?}", e);
@@ -246,23 +247,24 @@ impl Service {
                     "The Timer should poll a new slot"
                 );
 
-                debug!(service.log, "Processing slot: {}", current_slot.as_u64());
+                info!(service.log, "Processing slot: {}", current_slot.as_u64());
 
                 // check for new duties
-                match manager.poll() {
+                // TODO: Convert to its own thread
+                match manager.update(current_slot) {
                     Err(error) => {
                         error!(service.log, "Epoch duties poll error"; "error" => format!("{:?}", error))
                     }
-                    Ok(PollOutcome::NoChange(epoch)) => {
+                    Ok(UpdateOutcome::NoChange(epoch)) => {
                         debug!(service.log, "No change in duties"; "epoch" => epoch)
                     }
-                    Ok(PollOutcome::DutiesChanged(epoch, duties)) => {
+                    Ok(UpdateOutcome::DutiesChanged(epoch, duties)) => {
                         info!(service.log, "Duties changed (potential re-org)"; "epoch" => epoch, "duties" => format!("{:?}", duties))
                     }
-                    Ok(PollOutcome::NewDuties(epoch, duties)) => {
+                    Ok(UpdateOutcome::NewDuties(epoch, duties)) => {
                         info!(service.log, "New duties obtained"; "epoch" => epoch, "duties" => format!("{:?}", duties))
                     }
-                    Ok(PollOutcome::UnknownValidatorOrEpoch(epoch)) => {
+                    Ok(UpdateOutcome::UnknownValidatorOrEpoch(epoch)) => {
                         error!(service.log, "Epoch or validator unknown"; "epoch" => epoch)
                     }
                 };