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

Merge pull request #1294 from gakonst/fix/self-instead-of-service

Browse Source 
fix(eth1/service): use self instead of Service
This commit is contained in:
Paul Hauner 2020-06-29 10:18:36 +10:00 committed by GitHub
commit 9f6ee212ff
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 84 additions and 77 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

104
beacon_node/eth1/src/service.rs
View File

@ -280,32 +280,34 @@ impl Service {
/// ///
/// Emits logs for debugging and errors. /// Emits logs for debugging and errors.
pub async fn update( pub async fn update(
service: Self, &self,
) -> Result<(DepositCacheUpdateOutcome, BlockCacheUpdateOutcome), String> { ) -> Result<(DepositCacheUpdateOutcome, BlockCacheUpdateOutcome), String> {
let update_deposit_cache = async { let update_deposit_cache = async {
let outcome = Service::update_deposit_cache(service.clone()) let outcome = self
.update_deposit_cache()
.await .await
.map_err(|e| format!("Failed to update eth1 cache: {:?}", e))?; .map_err(|e| format!("Failed to update eth1 cache: {:?}", e))?;
trace!( trace!(
service.log, self.log,
"Updated eth1 deposit cache"; "Updated eth1 deposit cache";
"cached_deposits" => service.inner.deposit_cache.read().cache.len(), "cached_deposits" => self.inner.deposit_cache.read().cache.len(),
"logs_imported" => outcome.logs_imported, "logs_imported" => outcome.logs_imported,
"last_processed_eth1_block" => service.inner.deposit_cache.read().last_processed_block, "last_processed_eth1_block" => self.inner.deposit_cache.read().last_processed_block,
); );
Ok(outcome) Ok(outcome)
}; };
let update_block_cache = async { let update_block_cache = async {
let outcome = Service::update_block_cache(service.clone()) let outcome = self
.update_block_cache()
.await .await
.map_err(|e| format!("Failed to update eth1 cache: {:?}", e))?; .map_err(|e| format!("Failed to update eth1 cache: {:?}", e))?;
trace!( trace!(
service.log, self.log,
"Updated eth1 block cache"; "Updated eth1 block cache";
"cached_blocks" => service.inner.block_cache.read().len(), "cached_blocks" => self.inner.block_cache.read().len(),
"blocks_imported" => outcome.blocks_imported, "blocks_imported" => outcome.blocks_imported,
"head_block" => outcome.head_block_number, "head_block" => outcome.head_block_number,
); );
@ -324,33 +326,31 @@ impl Service {
/// - Err(_) if there is an error. /// - Err(_) if there is an error.
/// ///
/// Emits logs for debugging and errors. /// Emits logs for debugging and errors.
pub fn auto_update(service: Self, handle: environment::TaskExecutor) { pub fn auto_update(self, handle: environment::TaskExecutor) {
let update_interval = Duration::from_millis(service.config().auto_update_interval_millis); let update_interval = Duration::from_millis(self.config().auto_update_interval_millis);
let mut interval = interval_at(Instant::now(), update_interval); let mut interval = interval_at(Instant::now(), update_interval);
let update_future = async move { let update_future = async move {
while interval.next().await.is_some() { while interval.next().await.is_some() {
Service::do_update(service.clone(), update_interval) self.do_update(update_interval).await.ok();
.await
.ok();
} }
}; };
handle.spawn(update_future, "eth1"); handle.spawn(update_future, "eth1");
} }
async fn do_update(service: Self, update_interval: Duration) -> Result<(), ()> { async fn do_update(&self, update_interval: Duration) -> Result<(), ()> {
let update_result = Service::update(service.clone()).await; let update_result = self.update().await;
match update_result { match update_result {
Err(e) => error!( Err(e) => error!(
service.log, self.log,
"Failed to update eth1 cache"; "Failed to update eth1 cache";
"retry_millis" => update_interval.as_millis(), "retry_millis" => update_interval.as_millis(),
"error" => e, "error" => e,
), ),
Ok((deposit, block)) => debug!( Ok((deposit, block)) => debug!(
service.log, self.log,
"Updated eth1 cache"; "Updated eth1 cache";
"retry_millis" => update_interval.as_millis(), "retry_millis" => update_interval.as_millis(),
"blocks" => format!("{:?}", block), "blocks" => format!("{:?}", block),
@ -372,23 +372,23 @@ impl Service {
/// - Err(_) if there is an error. /// - Err(_) if there is an error.
/// ///
/// Emits logs for debugging and errors. /// Emits logs for debugging and errors.
pub async fn update_deposit_cache(service: Self) -> Result<DepositCacheUpdateOutcome, Error> { pub async fn update_deposit_cache(&self) -> Result<DepositCacheUpdateOutcome, Error> {
let endpoint = service.config().endpoint.clone(); let endpoint = self.config().endpoint.clone();
let follow_distance = service.config().follow_distance; let follow_distance = self.config().follow_distance;
let deposit_contract_address = service.config().deposit_contract_address.clone(); let deposit_contract_address = self.config().deposit_contract_address.clone();
let blocks_per_log_query = service.config().blocks_per_log_query; let blocks_per_log_query = self.config().blocks_per_log_query;
let max_log_requests_per_update = service let max_log_requests_per_update = self
.config() .config()
.max_log_requests_per_update .max_log_requests_per_update
.unwrap_or_else(usize::max_value); .unwrap_or_else(usize::max_value);
let next_required_block = service let next_required_block = self
.deposits() .deposits()
.read() .read()
.last_processed_block .last_processed_block
.map(|n| n + 1) .map(|n| n + 1)
.unwrap_or_else(|| service.config().deposit_contract_deploy_block); .unwrap_or_else(|| self.config().deposit_contract_deploy_block);
let range = get_new_block_numbers(&endpoint, next_required_block, follow_distance).await?; let range = get_new_block_numbers(&endpoint, next_required_block, follow_distance).await?;
@ -432,7 +432,7 @@ impl Service {
let mut logs_imported = 0; let mut logs_imported = 0;
for (block_range, log_chunk) in logs.iter() { for (block_range, log_chunk) in logs.iter() {
let mut cache = service.deposits().write(); let mut cache = self.deposits().write();
log_chunk log_chunk
.into_iter() .into_iter()
.map(|raw_log| { .map(|raw_log| {
@ -478,18 +478,18 @@ impl Service {
if logs_imported > 0 { if logs_imported > 0 {
info!( info!(
service.log, self.log,
"Imported deposit log(s)"; "Imported deposit log(s)";
"latest_block" => service.inner.deposit_cache.read().cache.latest_block_number(), "latest_block" => self.inner.deposit_cache.read().cache.latest_block_number(),
"total" => service.deposit_cache_len(), "total" => self.deposit_cache_len(),
"new" => logs_imported "new" => logs_imported
); );
} else { } else {
debug!( debug!(
service.log, self.log,
"No new deposits found"; "No new deposits found";
"latest_block" => service.inner.deposit_cache.read().cache.latest_block_number(), "latest_block" => self.inner.deposit_cache.read().cache.latest_block_number(),
"total_deposits" => service.deposit_cache_len(), "total_deposits" => self.deposit_cache_len(),
); );
} }
@ -507,23 +507,23 @@ impl Service {
/// - Err(_) if there is an error. /// - Err(_) if there is an error.
/// ///
/// Emits logs for debugging and errors. /// Emits logs for debugging and errors.
pub async fn update_block_cache(service: Self) -> Result<BlockCacheUpdateOutcome, Error> { pub async fn update_block_cache(&self) -> Result<BlockCacheUpdateOutcome, Error> {
let block_cache_truncation = service.config().block_cache_truncation; let block_cache_truncation = self.config().block_cache_truncation;
let max_blocks_per_update = service let max_blocks_per_update = self
.config() .config()
.max_blocks_per_update .max_blocks_per_update
.unwrap_or_else(usize::max_value); .unwrap_or_else(usize::max_value);
let next_required_block = service let next_required_block = self
.inner .inner
.block_cache .block_cache
.read() .read()
.highest_block_number() .highest_block_number()
.map(|n| n + 1) .map(|n| n + 1)
.unwrap_or_else(|| service.config().lowest_cached_block_number); .unwrap_or_else(|| self.config().lowest_cached_block_number);
let endpoint = service.config().endpoint.clone(); let endpoint = self.config().endpoint.clone();
let follow_distance = service.config().follow_distance; let follow_distance = self.config().follow_distance;
let range = get_new_block_numbers(&endpoint, next_required_block, follow_distance).await?; let range = get_new_block_numbers(&endpoint, next_required_block, follow_distance).await?;
// Map the range of required blocks into a Vec. // Map the range of required blocks into a Vec.
@ -545,7 +545,7 @@ impl Service {
// If the range of required blocks is larger than `max_size`, drop all // If the range of required blocks is larger than `max_size`, drop all
// existing blocks and download `max_size` count of blocks. // existing blocks and download `max_size` count of blocks.
let first_block = range.end() - max_size; let first_block = range.end() - max_size;
(*service.inner.block_cache.write()) = BlockCache::default(); (*self.inner.block_cache.write()) = BlockCache::default();
(first_block..=*range.end()).collect::<Vec<u64>>() (first_block..=*range.end()).collect::<Vec<u64>>()
} else { } else {
range.collect::<Vec<u64>>() range.collect::<Vec<u64>>()
@ -556,7 +556,7 @@ impl Service {
}; };
// Download the range of blocks and sequentially import them into the cache. // Download the range of blocks and sequentially import them into the cache.
// Last processed block in deposit cache // Last processed block in deposit cache
let latest_in_cache = service let latest_in_cache = self
.inner .inner
.deposit_cache .deposit_cache
.read() .read()
@ -576,7 +576,7 @@ impl Service {
|mut block_numbers| async { |mut block_numbers| async {
match block_numbers.next() { match block_numbers.next() {
Some(block_number) => { Some(block_number) => {
match download_eth1_block(service.inner.clone(), block_number).await { match download_eth1_block(self.inner.clone(), block_number).await {
Ok(eth1_block) => Ok(Some((eth1_block, block_numbers))), Ok(eth1_block) => Ok(Some((eth1_block, block_numbers))),
Err(e) => Err(e), Err(e) => Err(e),
} }
@ -590,8 +590,7 @@ impl Service {
let mut blocks_imported = 0; let mut blocks_imported = 0;
for eth1_block in eth1_blocks { for eth1_block in eth1_blocks {
service self.inner
.inner
.block_cache .block_cache
.write() .write()
.insert_root_or_child(eth1_block) .insert_root_or_child(eth1_block)
@ -599,12 +598,11 @@ impl Service {
metrics::set_gauge( metrics::set_gauge(
&metrics::BLOCK_CACHE_LEN, &metrics::BLOCK_CACHE_LEN,
service.inner.block_cache.read().len() as i64, self.inner.block_cache.read().len() as i64,
); );
metrics::set_gauge( metrics::set_gauge(
&metrics::LATEST_CACHED_BLOCK_TIMESTAMP, &metrics::LATEST_CACHED_BLOCK_TIMESTAMP,
service self.inner
.inner
.block_cache .block_cache
.read() .read()
.latest_block_timestamp() .latest_block_timestamp()
@ -615,14 +613,14 @@ impl Service {
} }
// Prune the block cache, preventing it from growing too large. // Prune the block cache, preventing it from growing too large.
service.inner.prune_blocks(); self.inner.prune_blocks();
metrics::set_gauge( metrics::set_gauge(
&metrics::BLOCK_CACHE_LEN, &metrics::BLOCK_CACHE_LEN,
service.inner.block_cache.read().len() as i64, self.inner.block_cache.read().len() as i64,
); );
let block_cache = service.inner.block_cache.read(); let block_cache = self.inner.block_cache.read();
let latest_block_mins = block_cache let latest_block_mins = block_cache
.latest_block_timestamp() .latest_block_timestamp()
.and_then(|timestamp| { .and_then(|timestamp| {
@ -636,7 +634,7 @@ impl Service {
if blocks_imported > 0 { if blocks_imported > 0 {
debug!( debug!(
service.log, self.log,
"Imported eth1 block(s)"; "Imported eth1 block(s)";
"latest_block_age" => latest_block_mins, "latest_block_age" => latest_block_mins,
"latest_block" => block_cache.highest_block_number(), "latest_block" => block_cache.highest_block_number(),
@ -645,7 +643,7 @@ impl Service {
); );
} else { } else {
debug!( debug!(
service.log, self.log,
"No new eth1 blocks imported"; "No new eth1 blocks imported";
"latest_block" => block_cache.highest_block_number(), "latest_block" => block_cache.highest_block_number(),
"cached_blocks" => block_cache.len(), "cached_blocks" => block_cache.len(),
@ -654,7 +652,7 @@ impl Service {
Ok(BlockCacheUpdateOutcome { Ok(BlockCacheUpdateOutcome {
blocks_imported, blocks_imported,
head_block_number: service.inner.block_cache.read().highest_block_number(), head_block_number: self.inner.block_cache.read().highest_block_number(),
}) })
} }
} }

49
beacon_node/eth1/tests/test.rs
View File

@ -145,14 +145,17 @@ mod eth1_cache {
eth1.ganache.evm_mine().await.expect("should mine block"); eth1.ganache.evm_mine().await.expect("should mine block");
} }
Service::update_deposit_cache(service.clone()) service
.update_deposit_cache()
.await .await
.expect("should update deposit cache"); .expect("should update deposit cache");
Service::update_block_cache(service.clone()) service
.update_block_cache()
.await .await
.expect("should update block cache"); .expect("should update block cache");
Service::update_block_cache(service.clone()) service
.update_block_cache()
.await .await
.expect("should update cache when nothing has changed"); .expect("should update cache when nothing has changed");
@ -205,10 +208,12 @@ mod eth1_cache {
eth1.ganache.evm_mine().await.expect("should mine block") eth1.ganache.evm_mine().await.expect("should mine block")
} }
Service::update_deposit_cache(service.clone()) service
.update_deposit_cache()
.await .await
.expect("should update deposit cache"); .expect("should update deposit cache");
Service::update_block_cache(service.clone()) service
.update_block_cache()
.await .await
.expect("should update block cache"); .expect("should update block cache");
@ -250,10 +255,12 @@ mod eth1_cache {
for _ in 0..cache_len / 2 { for _ in 0..cache_len / 2 {
eth1.ganache.evm_mine().await.expect("should mine block") eth1.ganache.evm_mine().await.expect("should mine block")
} }
Service::update_deposit_cache(service.clone()) service
.update_deposit_cache()
.await .await
.expect("should update deposit cache"); .expect("should update deposit cache");
Service::update_block_cache(service.clone()) service
.update_block_cache()
.await .await
.expect("should update block cache"); .expect("should update block cache");
} }
@ -293,14 +300,11 @@ mod eth1_cache {
eth1.ganache.evm_mine().await.expect("should mine block") eth1.ganache.evm_mine().await.expect("should mine block")
} }
futures::try_join!( futures::try_join!(
Service::update_deposit_cache(service.clone()), service.update_deposit_cache(),
Service::update_deposit_cache(service.clone()) service.update_deposit_cache()
) )
.expect("should perform two simultaneous updates of deposit cache"); .expect("should perform two simultaneous updates of deposit cache");
futures::try_join!( futures::try_join!(service.update_block_cache(), service.update_block_cache())
Service::update_block_cache(service.clone()),
Service::update_block_cache(service.clone())
)
.expect("should perform two simultaneous updates of block cache"); .expect("should perform two simultaneous updates of block cache");
assert!(service.block_cache_len() >= n, "should grow the cache"); assert!(service.block_cache_len() >= n, "should grow the cache");
@ -346,11 +350,13 @@ mod deposit_tree {
.expect("should perform a deposit"); .expect("should perform a deposit");
} }
Service::update_deposit_cache(service.clone()) service
.update_deposit_cache()
.await .await
.expect("should perform update"); .expect("should perform update");
Service::update_deposit_cache(service.clone()) service
.update_deposit_cache()
.await .await
.expect("should perform update when nothing has changed"); .expect("should perform update when nothing has changed");
@ -420,8 +426,8 @@ mod deposit_tree {
} }
futures::try_join!( futures::try_join!(
Service::update_deposit_cache(service.clone()), service.update_deposit_cache(),
Service::update_deposit_cache(service.clone()) service.update_deposit_cache()
) )
.expect("should perform two updates concurrently"); .expect("should perform two updates concurrently");
@ -661,7 +667,8 @@ mod fast {
eth1.ganache.evm_mine().await.expect("should mine block"); eth1.ganache.evm_mine().await.expect("should mine block");
} }
Service::update_deposit_cache(service.clone()) service
.update_deposit_cache()
.await .await
.expect("should perform update"); .expect("should perform update");
@ -728,7 +735,8 @@ mod persist {
.expect("should perform a deposit"); .expect("should perform a deposit");
} }
Service::update_deposit_cache(service.clone()) service
.update_deposit_cache()
.await .await
.expect("should perform update"); .expect("should perform update");
@ -739,7 +747,8 @@ mod persist {
let deposit_count = service.deposit_cache_len(); let deposit_count = service.deposit_cache_len();
Service::update_block_cache(service.clone()) service
.update_block_cache()
.await .await
.expect("should perform update"); .expect("should perform update");

6
beacon_node/genesis/src/eth1_genesis_service.rs
View File

@ -113,7 +113,8 @@ impl Eth1GenesisService {
); );
loop { loop {
let update_result = Eth1Service::update_deposit_cache(eth1_service.clone()) let update_result = eth1_service
.update_deposit_cache()
.await .await
.map_err(|e| format!("{:?}", e)); .map_err(|e| format!("{:?}", e));
@ -154,8 +155,7 @@ impl Eth1GenesisService {
} }
// Download new eth1 blocks into the cache. // Download new eth1 blocks into the cache.
let blocks_imported = match Eth1Service::update_block_cache(eth1_service.clone()).await let blocks_imported = match eth1_service.update_block_cache().await {
{
Ok(outcome) => { Ok(outcome) => {
debug!( debug!(
log, log,