cerc-io/lighthouse
10
0
Fork 0
Code Issues 1 Pull Requests Packages Projects Releases Wiki Activity

Merge Engines and Engine struct in one in the execution_layer crate (#3284)

Browse Source 
## Issue Addressed

Part of https://github.com/sigp/lighthouse/issues/3118, continuation of https://github.com/sigp/lighthouse/pull/3257 and https://github.com/sigp/lighthouse/pull/3283

## Proposed Changes
- Merge the [`Engines`](9c429d0764/beacon_node/execution_layer/src/engines.rs (L161-L165)) struct and [`Engine` ](9c429d0764/beacon_node/execution_layer/src/engines.rs (L62-L67))
- Remove unnecessary generics 

## Additional Info
There is more cleanup to do that will come in subsequent PRs
This commit is contained in:
Divma 2022-07-11 01:44:41 +00:00
parent 5dbfb37d74
commit 6d42a09ff8
4 changed files with 52 additions and 77 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
beacon_node/execution_layer/src/engine_api.rs
View File

@ -71,8 +71,6 @@ impl From<builder_client::Error> for Error {
} }
} }
pub struct EngineApi;
#[derive(Clone, Copy, Debug, PartialEq)] #[derive(Clone, Copy, Debug, PartialEq)]
pub enum PayloadStatusV1Status { pub enum PayloadStatusV1Status {
Valid, Valid,

14
beacon_node/execution_layer/src/engine_api/http.rs
View File

@ -7,7 +7,6 @@ use reqwest::header::CONTENT_TYPE;
use sensitive_url::SensitiveUrl; use sensitive_url::SensitiveUrl;
use serde::de::DeserializeOwned; use serde::de::DeserializeOwned;
use serde_json::json; use serde_json::json;
use std::marker::PhantomData;
use std::time::Duration; use std::time::Duration;
use types::EthSpec; use types::EthSpec;
@ -169,7 +168,7 @@ pub mod deposit_log {
/// state of the deposit contract. /// state of the deposit contract.
pub mod deposit_methods { pub mod deposit_methods {
use super::Log; use super::Log;
use crate::{EngineApi, HttpJsonRpc}; use crate::HttpJsonRpc;
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
use serde_json::{json, Value}; use serde_json::{json, Value};
use std::fmt; use std::fmt;
@ -298,7 +297,7 @@ pub mod deposit_methods {
} }
} }
impl HttpJsonRpc<EngineApi> { impl HttpJsonRpc {
/// Get the eth1 chain id of the given endpoint. /// Get the eth1 chain id of the given endpoint.
pub async fn get_chain_id(&self, timeout: Duration) -> Result<Eth1Id, String> { pub async fn get_chain_id(&self, timeout: Duration) -> Result<Eth1Id, String> {
let chain_id: String = self let chain_id: String = self
@ -517,20 +516,18 @@ pub mod deposit_methods {
} }
} }
pub struct HttpJsonRpc<T = EngineApi> { pub struct HttpJsonRpc {
pub client: Client, pub client: Client,
pub url: SensitiveUrl, pub url: SensitiveUrl,
auth: Option<Auth>, auth: Option<Auth>,
_phantom: PhantomData<T>,
} }
impl<T> HttpJsonRpc<T> { impl HttpJsonRpc {
pub fn new(url: SensitiveUrl) -> Result<Self, Error> { pub fn new(url: SensitiveUrl) -> Result<Self, Error> {
Ok(Self { Ok(Self {
client: Client::builder().build()?, client: Client::builder().build()?,
url, url,
auth: None, auth: None,
_phantom: PhantomData,
}) })
} }
@ -539,7 +536,6 @@ impl<T> HttpJsonRpc<T> {
client: Client::builder().build()?, client: Client::builder().build()?,
url, url,
auth: Some(auth), auth: Some(auth),
_phantom: PhantomData,
}) })
} }
@ -592,7 +588,7 @@ impl std::fmt::Display for HttpJsonRpc {
} }
} }
impl HttpJsonRpc<EngineApi> { impl HttpJsonRpc {
pub async fn upcheck(&self) -> Result<(), Error> { pub async fn upcheck(&self) -> Result<(), Error> {
let result: serde_json::Value = self let result: serde_json::Value = self
.rpc_request(ETH_SYNCING, json!([]), ETH_SYNCING_TIMEOUT) .rpc_request(ETH_SYNCING, json!([]), ETH_SYNCING_TIMEOUT)

82
beacon_node/execution_layer/src/engines.rs
View File

@ -1,7 +1,7 @@
//! Provides generic behaviour for multiple execution engines, specifically fallback behaviour. //! Provides generic behaviour for multiple execution engines, specifically fallback behaviour.
use crate::engine_api::{ use crate::engine_api::{
EngineApi, Error as EngineApiError, ForkchoiceUpdatedResponse, PayloadAttributes, PayloadId, Error as EngineApiError, ForkchoiceUpdatedResponse, PayloadAttributes, PayloadId,
}; };
use crate::HttpJsonRpc; use crate::HttpJsonRpc;
use lru::LruCache; use lru::LruCache;
@ -55,20 +55,32 @@ struct PayloadIdCacheKey {
pub suggested_fee_recipient: Address, pub suggested_fee_recipient: Address,
} }
/// An execution engine. #[derive(Debug)]
pub struct Engine<T> { pub enum EngineError {
pub api: HttpJsonRpc<T>, Offline,
payload_id_cache: Mutex<LruCache<PayloadIdCacheKey, PayloadId>>, Api { error: EngineApiError },
state: RwLock<EngineState>, BuilderApi { error: EngineApiError },
Auth,
} }
impl<T> Engine<T> { /// An execution engine.
pub struct Engine {
pub api: HttpJsonRpc,
payload_id_cache: Mutex<LruCache<PayloadIdCacheKey, PayloadId>>,
state: RwLock<EngineState>,
pub latest_forkchoice_state: RwLock<Option<ForkChoiceState>>,
pub log: Logger,
}
impl Engine {
/// Creates a new, offline engine. /// Creates a new, offline engine.
pub fn new(api: HttpJsonRpc<T>) -> Self { pub fn new(api: HttpJsonRpc, log: &Logger) -> Self {
Self { Self {
api, api,
payload_id_cache: Mutex::new(LruCache::new(PAYLOAD_ID_LRU_CACHE_SIZE)), payload_id_cache: Mutex::new(LruCache::new(PAYLOAD_ID_LRU_CACHE_SIZE)),
state: RwLock::new(EngineState::Offline), state: RwLock::new(EngineState::Offline),
latest_forkchoice_state: Default::default(),
log: log.clone(),
} }
} }
@ -90,9 +102,7 @@ impl<T> Engine<T> {
}) })
.cloned() .cloned()
} }
}
impl Engine<EngineApi> {
pub async fn notify_forkchoice_updated( pub async fn notify_forkchoice_updated(
&self, &self,
forkchoice_state: ForkChoiceState, forkchoice_state: ForkChoiceState,
@ -120,26 +130,7 @@ impl Engine<EngineApi> {
Ok(response) Ok(response)
} }
}
// This structure used to hold multiple execution engines managed in a fallback manner. This
// functionality has been removed following https://github.com/sigp/lighthouse/issues/3118 and this
// struct will likely be removed in the future.
pub struct Engines {
pub engine: Engine<EngineApi>,
pub latest_forkchoice_state: RwLock<Option<ForkChoiceState>>,
pub log: Logger,
}
#[derive(Debug)]
pub enum EngineError {
Offline,
Api { error: EngineApiError },
BuilderApi { error: EngineApiError },
Auth,
}
impl Engines {
async fn get_latest_forkchoice_state(&self) -> Option<ForkChoiceState> { async fn get_latest_forkchoice_state(&self) -> Option<ForkChoiceState> {
*self.latest_forkchoice_state.read().await *self.latest_forkchoice_state.read().await
} }
@ -169,12 +160,7 @@ impl Engines {
// For simplicity, payload attributes are never included in this call. It may be // For simplicity, payload attributes are never included in this call. It may be
// reasonable to include them in the future. // reasonable to include them in the future.
if let Err(e) = self if let Err(e) = self.api.forkchoice_updated_v1(forkchoice_state, None).await {
.engine
.api
.forkchoice_updated_v1(forkchoice_state, None)
.await
{
debug!( debug!(
self.log, self.log,
"Failed to issue latest head to engine"; "Failed to issue latest head to engine";
@ -191,14 +177,14 @@ impl Engines {
/// Returns `true` if the engine has a "synced" status. /// Returns `true` if the engine has a "synced" status.
pub async fn is_synced(&self) -> bool { pub async fn is_synced(&self) -> bool {
*self.engine.state.read().await == EngineState::Synced *self.state.read().await == EngineState::Synced
} }
/// Run the `EngineApi::upcheck` function if the node's last known state is not synced. This /// Run the `EngineApi::upcheck` function if the node's last known state is not synced. This
/// might be used to recover the node if offline. /// might be used to recover the node if offline.
pub async fn upcheck_not_synced(&self, logging: Logging) { pub async fn upcheck_not_synced(&self, logging: Logging) {
let mut state_lock = self.engine.state.write().await; let mut state_lock = self.state.write().await;
if *state_lock != EngineState::Synced { if *state_lock != EngineState::Synced {
match self.engine.api.upcheck().await { match self.api.upcheck().await {
Ok(()) => { Ok(()) => {
if logging.is_enabled() { if logging.is_enabled() {
info!( info!(
@ -261,7 +247,7 @@ impl Engines {
/// upcheck it and then run the function again. /// upcheck it and then run the function again.
pub async fn first_success<'a, F, G, H>(&'a self, func: F) -> Result<H, EngineError> pub async fn first_success<'a, F, G, H>(&'a self, func: F) -> Result<H, EngineError>
where where
F: Fn(&'a Engine<EngineApi>) -> G + Copy, F: Fn(&'a Engine) -> G + Copy,
G: Future<Output = Result<H, EngineApiError>>, G: Future<Output = Result<H, EngineApiError>>,
{ {
match self.first_success_without_retry(func).await { match self.first_success_without_retry(func).await {
@ -282,18 +268,18 @@ impl Engines {
func: F, func: F,
) -> Result<H, EngineError> ) -> Result<H, EngineError>
where where
F: Fn(&'a Engine<EngineApi>) -> G, F: Fn(&'a Engine) -> G,
G: Future<Output = Result<H, EngineApiError>>, G: Future<Output = Result<H, EngineApiError>>,
{ {
let (engine_synced, engine_auth_failed) = { let (engine_synced, engine_auth_failed) = {
let state = self.engine.state.read().await; let state = self.state.read().await;
( (
*state == EngineState::Synced, *state == EngineState::Synced,
*state == EngineState::AuthFailed, *state == EngineState::AuthFailed,
) )
}; };
if engine_synced { if engine_synced {
match func(&self.engine).await { match func(self).await {
Ok(result) => Ok(result), Ok(result) => Ok(result),
Err(error) => { Err(error) => {
debug!( debug!(
@ -301,7 +287,7 @@ impl Engines {
"Execution engine call failed"; "Execution engine call failed";
"error" => ?error, "error" => ?error,
); );
*self.engine.state.write().await = EngineState::Offline; *self.state.write().await = EngineState::Offline;
Err(EngineError::Api { error }) Err(EngineError::Api { error })
} }
} }
@ -318,7 +304,7 @@ impl Engines {
/// it runs, it will try to upcheck all offline nodes and then run the function again. /// it runs, it will try to upcheck all offline nodes and then run the function again.
pub async fn broadcast<'a, F, G, H>(&'a self, func: F) -> Result<H, EngineError> pub async fn broadcast<'a, F, G, H>(&'a self, func: F) -> Result<H, EngineError>
where where
F: Fn(&'a Engine<EngineApi>) -> G + Copy, F: Fn(&'a Engine) -> G + Copy,
G: Future<Output = Result<H, EngineApiError>>, G: Future<Output = Result<H, EngineApiError>>,
{ {
match self.broadcast_without_retry(func).await { match self.broadcast_without_retry(func).await {
@ -333,14 +319,14 @@ impl Engines {
/// Runs `func` on the node if it's last state is not offline. /// Runs `func` on the node if it's last state is not offline.
pub async fn broadcast_without_retry<'a, F, G, H>(&'a self, func: F) -> Result<H, EngineError> pub async fn broadcast_without_retry<'a, F, G, H>(&'a self, func: F) -> Result<H, EngineError>
where where
F: Fn(&'a Engine<EngineApi>) -> G, F: Fn(&'a Engine) -> G,
G: Future<Output = Result<H, EngineApiError>>, G: Future<Output = Result<H, EngineApiError>>,
{ {
let func = &func; let func = &func;
if *self.engine.state.read().await == EngineState::Offline { if *self.state.read().await == EngineState::Offline {
Err(EngineError::Offline) Err(EngineError::Offline)
} else { } else {
match func(&self.engine).await { match func(self).await {
Ok(res) => Ok(res), Ok(res) => Ok(res),
Err(error) => { Err(error) => {
debug!( debug!(
@ -348,7 +334,7 @@ impl Engines {
"Execution engine call failed"; "Execution engine call failed";
"error" => ?error, "error" => ?error,
); );
*self.engine.state.write().await = EngineState::Offline; *self.state.write().await = EngineState::Offline;
Err(EngineError::Api { error }) Err(EngineError::Api { error })
} }
} }

31
beacon_node/execution_layer/src/lib.rs
View File

@ -10,7 +10,7 @@ use engine_api::Error as ApiError;
pub use engine_api::*; pub use engine_api::*;
pub use engine_api::{http, http::deposit_methods, http::HttpJsonRpc}; pub use engine_api::{http, http::deposit_methods, http::HttpJsonRpc};
pub use engines::ForkChoiceState; pub use engines::ForkChoiceState;
use engines::{Engine, EngineError, Engines, Logging}; use engines::{Engine, EngineError, Logging};
use lru::LruCache; use lru::LruCache;
use payload_status::process_payload_status; use payload_status::process_payload_status;
pub use payload_status::PayloadStatus; pub use payload_status::PayloadStatus;
@ -64,7 +64,7 @@ const CONFIG_POLL_INTERVAL: Duration = Duration::from_secs(60);
#[derive(Debug)] #[derive(Debug)]
pub enum Error { pub enum Error {
NoEngines, NoEngine,
NoPayloadBuilder, NoPayloadBuilder,
ApiError(ApiError), ApiError(ApiError),
Builder(builder_client::Error), Builder(builder_client::Error),
@ -101,7 +101,7 @@ pub struct Proposer {
} }
struct Inner<E: EthSpec> { struct Inner<E: EthSpec> {
engines: Engines, engine: Engine,
builder: Option<BuilderHttpClient>, builder: Option<BuilderHttpClient>,
execution_engine_forkchoice_lock: Mutex<()>, execution_engine_forkchoice_lock: Mutex<()>,
suggested_fee_recipient: Option<Address>, suggested_fee_recipient: Option<Address>,
@ -162,7 +162,7 @@ impl<T: EthSpec> ExecutionLayer<T> {
if urls.len() > 1 { if urls.len() > 1 {
warn!(log, "Only the first execution engine url will be used"); warn!(log, "Only the first execution engine url will be used");
} }
let execution_url = urls.into_iter().next().ok_or(Error::NoEngines)?; let execution_url = urls.into_iter().next().ok_or(Error::NoEngine)?;
// Use the default jwt secret path if not provided via cli. // Use the default jwt secret path if not provided via cli.
let secret_file = secret_files let secret_file = secret_files
@ -198,12 +198,11 @@ impl<T: EthSpec> ExecutionLayer<T> {
.map_err(Error::InvalidJWTSecret) .map_err(Error::InvalidJWTSecret)
}?; }?;
let engine: Engine<EngineApi> = { let engine: Engine = {
let auth = Auth::new(jwt_key, jwt_id, jwt_version); let auth = Auth::new(jwt_key, jwt_id, jwt_version);
debug!(log, "Loaded execution endpoint"; "endpoint" => %execution_url, "jwt_path" => ?secret_file.as_path()); debug!(log, "Loaded execution endpoint"; "endpoint" => %execution_url, "jwt_path" => ?secret_file.as_path());
let api = HttpJsonRpc::<EngineApi>::new_with_auth(execution_url, auth) let api = HttpJsonRpc::new_with_auth(execution_url, auth).map_err(Error::ApiError)?;
.map_err(Error::ApiError)?; Engine::new(api, &log)
Engine::<EngineApi>::new(api)
}; };
let builder = builder_url let builder = builder_url
@ -211,11 +210,7 @@ impl<T: EthSpec> ExecutionLayer<T> {
.transpose()?; .transpose()?;
let inner = Inner { let inner = Inner {
engines: Engines { engine,
engine,
latest_forkchoice_state: <_>::default(),
log: log.clone(),
},
builder, builder,
execution_engine_forkchoice_lock: <_>::default(), execution_engine_forkchoice_lock: <_>::default(),
suggested_fee_recipient, suggested_fee_recipient,
@ -234,8 +229,8 @@ impl<T: EthSpec> ExecutionLayer<T> {
} }
impl<T: EthSpec> ExecutionLayer<T> { impl<T: EthSpec> ExecutionLayer<T> {
fn engines(&self) -> &Engines { fn engines(&self) -> &Engine {
&self.inner.engines &self.inner.engine
} }
pub fn builder(&self) -> &Option<BuilderHttpClient> { pub fn builder(&self) -> &Option<BuilderHttpClient> {
@ -1004,7 +999,7 @@ impl<T: EthSpec> ExecutionLayer<T> {
/// https://github.com/ethereum/consensus-specs/blob/v1.1.5/specs/merge/validator.md /// https://github.com/ethereum/consensus-specs/blob/v1.1.5/specs/merge/validator.md
async fn get_pow_block_hash_at_total_difficulty( async fn get_pow_block_hash_at_total_difficulty(
&self, &self,
engine: &Engine<EngineApi>, engine: &Engine,
spec: &ChainSpec, spec: &ChainSpec,
) -> Result<Option<ExecutionBlockHash>, ApiError> { ) -> Result<Option<ExecutionBlockHash>, ApiError> {
let mut block = engine let mut block = engine
@ -1118,7 +1113,7 @@ impl<T: EthSpec> ExecutionLayer<T> {
/// https://github.com/ethereum/consensus-specs/issues/2636 /// https://github.com/ethereum/consensus-specs/issues/2636
async fn get_pow_block( async fn get_pow_block(
&self, &self,
engine: &Engine<EngineApi>, engine: &Engine,
hash: ExecutionBlockHash, hash: ExecutionBlockHash,
) -> Result<Option<ExecutionBlock>, ApiError> { ) -> Result<Option<ExecutionBlock>, ApiError> {
if let Some(cached) = self.execution_blocks().await.get(&hash).copied() { if let Some(cached) = self.execution_blocks().await.get(&hash).copied() {
@ -1153,7 +1148,7 @@ impl<T: EthSpec> ExecutionLayer<T> {
async fn get_payload_by_block_hash_from_engine( async fn get_payload_by_block_hash_from_engine(
&self, &self,
engine: &Engine<EngineApi>, engine: &Engine,
hash: ExecutionBlockHash, hash: ExecutionBlockHash,
) -> Result<Option<ExecutionPayload<T>>, ApiError> { ) -> Result<Option<ExecutionPayload<T>>, ApiError> {
let _timer = metrics::start_timer(&metrics::EXECUTION_LAYER_GET_PAYLOAD_BY_BLOCK_HASH); let _timer = metrics::start_timer(&metrics::EXECUTION_LAYER_GET_PAYLOAD_BY_BLOCK_HASH);