Introduce threading to validator client

validator_client/src/block_producer/mod.rs

@@ -4,17 +4,15 @@ mod test_node;
 mod traits;
 
 use self::traits::{BeaconNode, BeaconNodeError};
+use super::EpochDutiesMap;
 use crate::duties::EpochDuties;
 use slot_clock::SlotClock;
 use spec::ChainSpec;
-use std::collections::HashMap;
 use std::sync::{Arc, RwLock};
 use types::BeaconBlock;
 
 pub use self::service::BlockProducerService;
 
-type EpochMap = HashMap<u64, EpochDuties>;
-
 #[derive(Debug, PartialEq)]
 pub enum PollOutcome {
     BlockProduced(u64),
@@ -38,7 +36,7 @@ pub enum Error {
 pub struct BlockProducer<T: SlotClock, U: BeaconNode> {
     pub last_processed_slot: u64,
     spec: Arc<ChainSpec>,
-    epoch_map: Arc<RwLock<HashMap<u64, EpochDuties>>>,
+    epoch_map: Arc<RwLock<EpochDutiesMap>>,
     slot_clock: Arc<RwLock<T>>,
     beacon_node: Arc<U>,
 }
@@ -46,7 +44,7 @@ pub struct BlockProducer<T: SlotClock, U: BeaconNode> {
 impl<T: SlotClock, U: BeaconNode> BlockProducer<T, U> {
     pub fn new(
         spec: Arc<ChainSpec>,
-        epoch_map: Arc<RwLock<EpochMap>>,
+        epoch_map: Arc<RwLock<EpochDutiesMap>>,
         slot_clock: Arc<RwLock<T>>,
         beacon_node: Arc<U>,
     ) -> Self {
@@ -151,7 +149,7 @@ mod tests {
         let mut rng = XorShiftRng::from_seed([42; 16]);
 
         let spec = Arc::new(ChainSpec::foundation());
-        let epoch_map = Arc::new(RwLock::new(EpochMap::new()));
+        let epoch_map = Arc::new(RwLock::new(EpochDutiesMap::new()));
         let slot_clock = Arc::new(RwLock::new(TestingSlotClock::new(0)));
         let beacon_node = Arc::new(TestBeaconNode::default());
 

validator_client/src/duties/grpc.rs

@@ -0,0 +1,63 @@
+use super::traits::{BeaconNode, BeaconNodeError};
+use protos::services::{
+    BeaconBlock as GrpcBeaconBlock, ProduceBeaconBlockRequest, PublishBeaconBlockRequest,
+};
+use protos::services_grpc::BeaconBlockServiceClient;
+use ssz::{ssz_encode, Decodable};
+use types::{BeaconBlock, BeaconBlockBody, Hash256, Signature};
+
+impl BeaconNode for BeaconBlockServiceClient {
+    fn produce_beacon_block(&self, slot: u64) -> Result<Option<BeaconBlock>, BeaconNodeError> {
+        let mut req = ProduceBeaconBlockRequest::new();
+        req.set_slot(slot);
+
+        let reply = self
+            .produce_beacon_block(&req)
+            .map_err(|err| BeaconNodeError::RemoteFailure(format!("{:?}", err)))?;
+
+        if reply.has_block() {
+            let block = reply.get_block();
+
+            let (signature, _) = Signature::ssz_decode(block.get_signature(), 0)
+                .map_err(|_| BeaconNodeError::DecodeFailure)?;
+
+            // TODO: this conversion is incomplete; fix it.
+            Ok(Some(BeaconBlock {
+                slot: block.get_slot(),
+                parent_root: Hash256::zero(),
+                state_root: Hash256::zero(),
+                randao_reveal: Hash256::from(block.get_randao_reveal()),
+                candidate_pow_receipt_root: Hash256::zero(),
+                signature,
+                body: BeaconBlockBody {
+                    proposer_slashings: vec![],
+                    casper_slashings: vec![],
+                    attestations: vec![],
+                    deposits: vec![],
+                    exits: vec![],
+                },
+            }))
+        } else {
+            Ok(None)
+        }
+    }
+
+    fn publish_beacon_block(&self, block: BeaconBlock) -> Result<bool, BeaconNodeError> {
+        let mut req = PublishBeaconBlockRequest::new();
+
+        // TODO: this conversion is incomplete; fix it.
+        let mut grpc_block = GrpcBeaconBlock::new();
+        grpc_block.set_slot(block.slot);
+        grpc_block.set_block_root(vec![0]);
+        grpc_block.set_randao_reveal(block.randao_reveal.to_vec());
+        grpc_block.set_signature(ssz_encode(&block.signature));
+
+        req.set_block(grpc_block);
+
+        let reply = self
+            .publish_beacon_block(&req)
+            .map_err(|err| BeaconNodeError::RemoteFailure(format!("{:?}", err)))?;
+
+        Ok(reply.get_success())
+    }
+}

validator_client/src/duties/mod.rs

@@ -9,10 +9,12 @@ mod service;
 mod test_node;
 mod traits;
 
+pub use self::service::DutiesManagerService;
+
 #[derive(Debug, PartialEq, Clone, Copy, Default)]
 pub struct EpochDuties {
     pub block_production_slot: Option<u64>,
-    pub shard: Option<u64>,
+    // Future shard info
 }
 
 impl EpochDuties {
@@ -24,7 +26,7 @@ impl EpochDuties {
     }
 }
 
-type EpochDutiesMap = HashMap<(PublicKey, u64), EpochDuties>;
+pub type EpochDutiesMap = HashMap<u64, EpochDuties>;
 
 #[derive(Debug, PartialEq, Clone, Copy)]
 pub enum PollOutcome {
@@ -73,7 +75,7 @@ impl<T: SlotClock, U: BeaconNode> DutiesManager<T, U> {
                 .map_err(|_| Error::EpochMapPoisoned)?;
 
             // If these duties were known, check to see if they're updates or identical.
-            let result = if let Some(known_duties) = map.get(&(self.pubkey.clone(), epoch)) {
+            let result = if let Some(known_duties) = map.get(&epoch) {
                 if *known_duties == duties {
                     Ok(PollOutcome::NoChange)
                 } else {
@@ -82,13 +84,12 @@ impl<T: SlotClock, U: BeaconNode> DutiesManager<T, U> {
             } else {
                 Ok(PollOutcome::NewDuties)
             };
-            map.insert((self.pubkey.clone(), epoch), duties);
+            map.insert(epoch, duties);
             result
         } else {
             Ok(PollOutcome::UnknownValidatorOrEpoch)
         }
     }
-
 }
 
 impl From<BeaconNodeError> for Error {
@@ -101,8 +102,8 @@ impl From<BeaconNodeError> for Error {
 mod tests {
     use super::test_node::TestBeaconNode;
     use super::*;
-    use slot_clock::TestingSlotClock;
     use bls::Keypair;
+    use slot_clock::TestingSlotClock;
 
     // TODO: implement more thorough testing.
     //
@@ -125,9 +126,8 @@ mod tests {
         };
 
         // Configure response from the BeaconNode.
-        beacon_node.set_next_shuffling_result(Ok(Some(EpochDuties{
+        beacon_node.set_next_shuffling_result(Ok(Some(EpochDuties {
             block_production_slot: Some(10),
-            shard: Some(12),
         })));
 
         // Get the duties for the first time...
@@ -136,9 +136,8 @@ mod tests {
         assert_eq!(manager.poll(), Ok(PollOutcome::NoChange));
 
         // Return new duties.
-        beacon_node.set_next_shuffling_result(Ok(Some(EpochDuties{
+        beacon_node.set_next_shuffling_result(Ok(Some(EpochDuties {
             block_production_slot: Some(11),
-            shard: Some(12),
         })));
         assert_eq!(manager.poll(), Ok(PollOutcome::DutiesChanged));
 

validator_client/src/duties/service.rs

@@ -1,16 +1,16 @@
-use super::traits::{BeaconNode, BeaconNodeError};
+use super::traits::BeaconNode;
 use super::{DutiesManager, PollOutcome};
-use slog::{debug, error, info, warn, Logger};
+use slog::{debug, error, info, Logger};
 use slot_clock::SlotClock;
 use std::time::Duration;
 
-pub struct DutiesService<T: SlotClock, U: BeaconNode> {
+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> DutiesService<T, U> {
+impl<T: SlotClock, U: BeaconNode> DutiesManagerService<T, U> {
     pub fn run(&mut self) {
         loop {
             match self.manager.poll() {

validator_client/src/main.rs

@@ -1,14 +1,16 @@
+use self::duties::{DutiesManager, DutiesManagerService, EpochDutiesMap};
 use crate::block_producer::{BlockProducer, BlockProducerService};
 use crate::config::ClientConfig;
+use bls::Keypair;
 use clap::{App, Arg};
 use grpcio::{ChannelBuilder, EnvBuilder};
 use protos::services_grpc::BeaconBlockServiceClient;
 use slog::{error, info, o, Drain};
 use slot_clock::SystemTimeSlotClock;
 use spec::ChainSpec;
-use std::collections::HashMap;
 use std::path::PathBuf;
 use std::sync::{Arc, RwLock};
+use std::thread;
 
 mod block_producer;
 mod config;
@@ -74,8 +76,7 @@ fn main() {
     // TODO: Permit loading a custom spec from file.
     let spec = Arc::new(ChainSpec::foundation());
 
-    // Global map of epoch -> validator duties.
-    let epoch_map = Arc::new(RwLock::new(HashMap::new()));
+    // Clock for determining the present slot.
     let slot_clock = {
         info!(log, "Genesis time"; "unix_epoch_seconds" => spec.genesis_time);
         let clock = SystemTimeSlotClock::new(spec.genesis_time, spec.slot_duration)
@@ -83,17 +84,40 @@ fn main() {
         Arc::new(RwLock::new(clock))
     };
 
-    let block_producer =
-        BlockProducer::new(spec.clone(), epoch_map.clone(), slot_clock.clone(), client);
-
     let poll_interval_millis = spec.slot_duration * 1000 / 10; // 10% epoch time precision.
     info!(log, "Starting block producer service"; "polls_per_epoch" => spec.slot_duration * 1000 / poll_interval_millis);
 
-    let mut block_producer_service = BlockProducerService {
-        block_producer,
-        poll_interval_millis,
-        log: log.clone(),
-    };
+    /*
+     * Start threads.
+     */
+    let keypairs = vec![Keypair::random()];
+    let mut threads = vec![];
 
-    block_producer_service.run();
+    for keypair in keypairs {
+        let duties_map = Arc::new(RwLock::new(EpochDutiesMap::new()));
+
+        let producer_thread = {
+            let spec = spec.clone();
+            let duties_map = duties_map.clone();
+            let slot_clock = slot_clock.clone();
+            let log = log.clone();
+            let client = client.clone();
+            thread::spawn(move || {
+                let block_producer = BlockProducer::new(spec, duties_map, slot_clock, client);
+                let mut block_producer_service = BlockProducerService {
+                    block_producer,
+                    poll_interval_millis,
+                    log,
+                };
+
+                block_producer_service.run();
+            })
+        };
+        threads.push(((), producer_thread));
+    }
+
+    for tuple in threads {
+        let (manager, producer) = tuple;
+        let _ = producer.join();
+    }
 }