Organize beacon_chain typing

- Implements ClientTypes
- New network BeaconChain type for the networking service
This commit is contained in:
Age Manning 2019-03-18 17:38:23 +11:00
parent bbad4bfa19
commit 6b5debe654
No known key found for this signature in database
GPG Key ID: 05EED64B79E06A93
8 changed files with 93 additions and 45 deletions

View File

@ -106,7 +106,7 @@ pub fn initialise_test_beacon_chain(
deposit_root: Hash256::zero(), deposit_root: Hash256::zero(),
block_hash: Hash256::zero(), block_hash: Hash256::zero(),
}; };
let keypairs: Vec<Keypair> = (0..10) let keypairs: Vec<Keypair> = (0..50)
.collect::<Vec<usize>>() .collect::<Vec<usize>>()
.iter() .iter()
.map(|_| Keypair::random()) .map(|_| Keypair::random())

View File

@ -7,4 +7,7 @@ pub mod initialise;
pub use self::beacon_chain::{BeaconChain, BlockProcessingOutcome, InvalidBlock, ValidBlock}; pub use self::beacon_chain::{BeaconChain, BlockProcessingOutcome, InvalidBlock, ValidBlock};
pub use self::checkpoint::CheckPoint; pub use self::checkpoint::CheckPoint;
pub use self::errors::BeaconChainError; pub use self::errors::BeaconChainError;
pub use fork_choice::{ForkChoice, ForkChoiceAlgorithm, ForkChoiceError}; pub use db;
pub use fork_choice;
pub use parking_lot;
pub use slot_clock;

View File

@ -1,39 +1,49 @@
use db::{ClientDB, DiskDB, MemoryDB}; use crate::ClientConfig;
use fork_choice::{BitwiseLMDGhost, ForkChoice}; use beacon_chain::{
use slot_clock::{SlotClock, SystemTimeSlotClock, TestingSlotClock}; db::{ClientDB, DiskDB, MemoryDB},
use beacon_chain::initialise; fork_choice::BitwiseLMDGhost,
initialise,
slot_clock::{SlotClock, SystemTimeSlotClock, TestingSlotClock},
BeaconChain,
};
use fork_choice::ForkChoice;
use std::sync::Arc; use std::sync::Arc;
use crate::ClientConfig
pub trait ClientTypes { pub trait ClientTypes {
type ForkChoice: ForkChoice; type DB: ClientDB + 'static;
type DB: ClientDB; type SlotClock: SlotClock + 'static;
type SlotClock: SlotClock; type ForkChoice: ForkChoice + 'static;
pub fn initialise_beacon_chain(cchain_spec: &ClientConfig) -> Arc<BeaconChain<DB,SlotClock,ForkChoice>>); fn initialise_beacon_chain(
config: &ClientConfig,
) -> Arc<BeaconChain<Self::DB, Self::SlotClock, Self::ForkChoice>>;
} }
pub struct StandardClientType pub struct StandardClientType;
impl ClientTypes for StandardClientType { impl ClientTypes for StandardClientType {
type DB = DiskDB; type DB = DiskDB;
type ForkChoice = BitwiseLMDGhost<DiskDB>;
type SlotClock = SystemTimeSlotClock; type SlotClock = SystemTimeSlotClock;
type ForkChoice = BitwiseLMDGhost<DiskDB>;
pub fn initialise_beacon_chain(config: &ClientConfig) -> Arc<BeaconChain<DB,SlotClock,ForkChoice>>) { fn initialise_beacon_chain(
initialise::initialise_beacon_chain(config.chain_spec, config.db_name) config: &ClientConfig,
) -> Arc<BeaconChain<Self::DB, Self::SlotClock, Self::ForkChoice>> {
initialise::initialise_beacon_chain(&config.spec, Some(&config.db_name))
} }
} }
pub struct TestingClientType pub struct TestingClientType;
impl ClientTypes for TestingClientType { impl ClientTypes for TestingClientType {
type DB = MemoryDB; type DB = MemoryDB;
type SlotClock = TestingSlotClock; type SlotClock = SystemTimeSlotClock;
type ForkChoice = BitwiseLMDGhost<MemoryDB>; type ForkChoice = BitwiseLMDGhost<MemoryDB>;
pub fn initialise_beacon_chain(config: &ClientConfig) -> Arc<BeaconChain<DB,SlotClock,ForkChoice>>) { fn initialise_beacon_chain(
initialise::initialise_test_beacon_chain(config.chain_spec, None) config: &ClientConfig,
) -> Arc<BeaconChain<Self::DB, Self::SlotClock, Self::ForkChoice>> {
initialise::initialise_test_beacon_chain(&config.spec, None)
} }
} }

View File

@ -10,6 +10,7 @@ pub use client_config::ClientConfig;
pub use client_types::ClientTypes; pub use client_types::ClientTypes;
//use beacon_chain::BeaconChain; //use beacon_chain::BeaconChain;
use beacon_chain::BeaconChain;
use exit_future::{Exit, Signal}; use exit_future::{Exit, Signal};
use network::Service as NetworkService; use network::Service as NetworkService;
use slog::o; use slog::o;
@ -20,26 +21,35 @@ use tokio::runtime::TaskExecutor;
/// Main beacon node client service. This provides the connection and initialisation of the clients /// Main beacon node client service. This provides the connection and initialisation of the clients
/// sub-services in multiple threads. /// sub-services in multiple threads.
pub struct Client<T: ClientTypes> { pub struct Client<T: ClientTypes> {
/// Configuration for the lighthouse client.
config: ClientConfig, config: ClientConfig,
/// The beacon chain for the running client.
beacon_chain: Arc<BeaconChain<T::DB, T::SlotClock, T::ForkChoice>>, beacon_chain: Arc<BeaconChain<T::DB, T::SlotClock, T::ForkChoice>>,
/// Reference to the network service.
pub network: Arc<NetworkService>, pub network: Arc<NetworkService>,
/// Future to stop and begin shutdown of the Client.
//TODO: Decide best way to handle shutdown
pub exit: exit_future::Exit, pub exit: exit_future::Exit,
/// The sending future to call to terminate the Client.
//TODO: Decide best way to handle shutdown
pub exit_signal: Signal, pub exit_signal: Signal,
/// The clients logger.
log: slog::Logger, log: slog::Logger,
/// Marker to pin the beacon chain generics.
phantom: PhantomData<T>,
} }
impl<T: ClientTypes> Client<T> { impl<TClientType: ClientTypes> Client<TClientType> {
/// Generate an instance of the client. Spawn and link all internal subprocesses. /// Generate an instance of the client. Spawn and link all internal sub-processes.
pub fn new( pub fn new(
config: ClientConfig, config: ClientConfig,
client_type: T,
log: slog::Logger, log: slog::Logger,
executor: &TaskExecutor, executor: &TaskExecutor,
) -> error::Result<Self> { ) -> error::Result<Self> {
let (exit_signal, exit) = exit_future::signal(); let (exit_signal, exit) = exit_future::signal();
// generate a beacon chain // generate a beacon chain
let beacon_chain = client_type.initialise_beacon_chain(&config); let beacon_chain = TClientType::initialise_beacon_chain(&config);
// Start the network service, libp2p and syncing threads // Start the network service, libp2p and syncing threads
// TODO: Add beacon_chain reference to network parameters // TODO: Add beacon_chain reference to network parameters
@ -54,6 +64,7 @@ impl<T: ClientTypes> Client<T> {
Ok(Client { Ok(Client {
config, config,
beacon_chain,
exit, exit,
exit_signal: exit_signal, exit_signal: exit_signal,
log, log,

View File

@ -0,0 +1,27 @@
use beacon_chain::BeaconChain as RawBeaconChain;
use beacon_chain::{
db::ClientDB, fork_choice::ForkChoice, parking_lot::RwLockReadGuard, slot_clock::SlotClock,
CheckPoint,
};
/// The network's API to the beacon chain.
pub trait BeaconChain: Send + Sync {
fn head(&self) -> RwLockReadGuard<CheckPoint>;
fn finalized_head(&self) -> RwLockReadGuard<CheckPoint>;
}
impl<T, U, F> BeaconChain for RawBeaconChain<T, U, F>
where
T: ClientDB + Sized,
U: SlotClock,
F: ForkChoice,
{
fn head(&self) -> RwLockReadGuard<CheckPoint> {
self.head()
}
fn finalized_head(&self) -> RwLockReadGuard<CheckPoint> {
self.finalized_head()
}
}

View File

@ -1,4 +1,5 @@
/// This crate provides the network server for Lighthouse. /// This crate provides the network server for Lighthouse.
pub mod beacon_chain;
pub mod error; pub mod error;
mod message_handler; mod message_handler;
mod messages; mod messages;

View File

@ -1,14 +1,14 @@
use crate::beacon_chain::BeaconChain;
use crate::error; use crate::error;
use crate::messages::NodeMessage; use crate::messages::NodeMessage;
use beacon_chain::BeaconChain; use crossbeam_channel::{unbounded as channel, Sender};
use crossbeam_channel::{unbounded as channel, Sender, TryRecvError};
use futures::future; use futures::future;
use futures::prelude::*; use futures::prelude::*;
use libp2p::rpc; use libp2p::rpc;
use libp2p::{PeerId, RPCEvent}; use libp2p::{PeerId, RPCEvent};
use slog::debug; use slog::debug;
use std::collections::HashMap; use std::collections::HashMap;
use std::sync::{Arc, Mutex}; use std::sync::Arc;
use std::time::{Duration, Instant}; use std::time::{Duration, Instant};
use sync::SimpleSync; use sync::SimpleSync;
use types::Hash256; use types::Hash256;
@ -17,9 +17,9 @@ use types::Hash256;
const HELLO_TIMEOUT: Duration = Duration::from_secs(30); const HELLO_TIMEOUT: Duration = Duration::from_secs(30);
/// Handles messages received from the network and client and organises syncing. /// Handles messages received from the network and client and organises syncing.
pub struct MessageHandler<T: ClientTypes> { pub struct MessageHandler {
/// Currently loaded and initialised beacon chain. /// Currently loaded and initialised beacon chain.
chain: BeaconChain<T::DB, T::SlotClock, T::ForkChoice>, chain: Arc<BeaconChain>,
/// The syncing framework. /// The syncing framework.
sync: SimpleSync, sync: SimpleSync,
/// A mapping of peers we have sent a HELLO rpc request to. /// A mapping of peers we have sent a HELLO rpc request to.
@ -41,10 +41,10 @@ pub enum HandlerMessage {
RPC(RPCEvent), RPC(RPCEvent),
} }
impl<T: ClientTypes> MessageHandler<T> { impl MessageHandler {
/// Initializes and runs the MessageHandler. /// Initializes and runs the MessageHandler.
pub fn new( pub fn new(
beacon_chain: Arc<BeaconChain<T::DB, T::SlotClock, T::ForkChoice>>, beacon_chain: Arc<BeaconChain>,
executor: &tokio::runtime::TaskExecutor, executor: &tokio::runtime::TaskExecutor,
log: slog::Logger, log: slog::Logger,
) -> error::Result<Sender<HandlerMessage>> { ) -> error::Result<Sender<HandlerMessage>> {
@ -60,7 +60,7 @@ impl<T: ClientTypes> MessageHandler<T> {
// generate the Message handler // generate the Message handler
let sync = SimpleSync::new(temp_genesis); let sync = SimpleSync::new(temp_genesis);
let mut handler = MessageHandler { let mut handler = MessageHandler {
chain: beacon_chain, chain: beacon_chain.clone(),
sync, sync,
hello_requests: HashMap::new(), hello_requests: HashMap::new(),
log: log.clone(), log: log.clone(),

View File

@ -1,46 +1,42 @@
use crate::beacon_chain::BeaconChain;
use crate::error; use crate::error;
use crate::message_handler::{HandlerMessage, MessageHandler}; use crate::message_handler::{HandlerMessage, MessageHandler};
use crate::messages::{NetworkMessage, NodeMessage}; use crate::messages::{NetworkMessage, NodeMessage};
use crate::NetworkConfig; use crate::NetworkConfig;
use crossbeam_channel::{unbounded as channel, Sender, TryRecvError}; use crossbeam_channel::{unbounded as channel, Sender, TryRecvError};
use futures::future::lazy;
use futures::future::poll_fn;
use futures::prelude::*; use futures::prelude::*;
use futures::sync::oneshot; use futures::sync::oneshot;
use futures::Stream; use futures::Stream;
use libp2p::behaviour::BehaviourEvent;
use libp2p::error::Error as libp2pError;
use libp2p::Service as LibP2PService; use libp2p::Service as LibP2PService;
use libp2p::{Libp2pEvent, PeerId}; use libp2p::{Libp2pEvent, PeerId};
use slog::{debug, info, o, trace, warn, Logger}; use slog::{debug, o};
use std::sync::{Arc, Mutex}; use std::sync::Arc;
use tokio::runtime::TaskExecutor; use tokio::runtime::TaskExecutor;
use client::ClientTypes;
/// Service that handles communication between internal services and the libp2p network service. /// Service that handles communication between internal services and the libp2p network service.
pub struct Service<T: ClientTypes> { pub struct Service {
//libp2p_service: Arc<Mutex<LibP2PService>>, //libp2p_service: Arc<Mutex<LibP2PService>>,
libp2p_exit: oneshot::Sender<()>, libp2p_exit: oneshot::Sender<()>,
network_send: crossbeam_channel::Sender<NetworkMessage>, network_send: crossbeam_channel::Sender<NetworkMessage>,
//message_handler: MessageHandler, //message_handler: MessageHandler,
//message_handler_send: Sender<HandlerMessage>, //message_handler_send: Sender<HandlerMessage>,
PhantomData: T,
} }
impl<T: ClientTypes> Service<T> { impl Service {
pub fn new( pub fn new(
beacon_chain: Arc<BeaconChain<T::DB, T::SlotClock, T::ForkChoice>, beacon_chain: Arc<BeaconChain>,
config: &NetworkConfig, config: &NetworkConfig,
executor: &TaskExecutor, executor: &TaskExecutor,
log: slog::Logger, log: slog::Logger,
) -> error::Result<(Arc<Self>, Sender<NetworkMessage>)> { ) -> error::Result<(Arc<Self>, Sender<NetworkMessage>)> {
// launch message handler thread // launch message handler thread
let message_handler_log = log.new(o!("Service" => "MessageHandler")); let message_handler_log = log.new(o!("Service" => "MessageHandler"));
let message_handler_send = MessageHandler::new(beacon_chain, executor, message_handler_log)?; let message_handler_send =
MessageHandler::new(beacon_chain, executor, message_handler_log)?;
// launch libp2p service // launch libp2p service
let libp2p_log = log.new(o!("Service" => "Libp2p")); let libp2p_log = log.new(o!("Service" => "Libp2p"));
let libp2p_service = LibP2PService::new(config, libp2p_log)?; let libp2p_service = LibP2PService::new(config.clone(), libp2p_log)?;
// TODO: Spawn thread to handle libp2p messages and pass to message handler thread. // TODO: Spawn thread to handle libp2p messages and pass to message handler thread.
let (network_send, libp2p_exit) = let (network_send, libp2p_exit) =