Skip to content

Commit

Permalink
Merge of #5700
Browse files Browse the repository at this point in the history
  • Loading branch information
@mergify
mergify[bot] authored May 6, 2024
2 parents da9d386 + b1f2ddd commit 0b49f0d
Show file tree
Hide file tree
Showing 3 changed files with 51 additions and 37 deletions.
49 changes: 30 additions & 19 deletions beacon_node/network/src/sync/block_lookups/mod.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@ use fnv::FnvHashMap;
use lighthouse_network::{PeerAction, PeerId};
use lru_cache::LRUTimeCache;
pub use single_block_lookup::{BlobRequestState, BlockRequestState};
use slog::{debug, error, trace, warn, Logger};
use slog::{debug, error, warn, Logger};
use std::collections::hash_map::Entry;
use std::sync::Arc;
use std::time::Duration;
Expand Down Expand Up @@ -233,13 +233,17 @@ impl<T: BeaconChainTypes> BlockLookups<T> {
.iter_mut()
.find(|(_id, lookup)| lookup.is_for_block(block_root))
{
trace!(self.log, "Adding peer to existing single block lookup"; "block_root" => %block_root);
lookup.add_peers(peers);
for peer in peers {
if lookup.add_peer(*peer) {
debug!(self.log, "Adding peer to existing single block lookup"; "block_root" => ?block_root, "peer" => ?peer);
}
}

if let Some(block_component) = block_component {
let component_type = block_component.get_type();
let imported = lookup.add_child_components(block_component);
if !imported {
debug!(self.log, "Lookup child component ignored"; "block_root" => %block_root, "type" => component_type);
debug!(self.log, "Lookup child component ignored"; "block_root" => ?block_root, "type" => component_type);
}
}
return true;
Expand All @@ -252,10 +256,15 @@ impl<T: BeaconChainTypes> BlockLookups<T> {
.iter()
.any(|(_, lookup)| lookup.is_for_block(awaiting_parent))
{
warn!(self.log, "Ignoring child lookup parent lookup not found"; "block_root" => ?awaiting_parent);
return false;
}
}

// If we know that this lookup has unknown parent (is awaiting a parent lookup to resolve),
// signal here to hold processing downloaded data.
let mut lookup = SingleBlockLookup::new(block_root, peers, cx.next_id(), awaiting_parent);

let msg = if block_component.is_some() {
"Searching for components of a block with unknown parent"
} else {
Expand All @@ -265,14 +274,11 @@ impl<T: BeaconChainTypes> BlockLookups<T> {
self.log,
"{}", msg;
"peer_ids" => ?peers,
"block" => ?block_root,
"block_root" => ?block_root,
"id" => lookup.id,
);
metrics::inc_counter(&metrics::SYNC_LOOKUP_CREATED);

// If we know that this lookup has unknown parent (is awaiting a parent lookup to resolve),
// signal here to hold processing downloaded data.
let mut lookup = SingleBlockLookup::new(block_root, peers, cx.next_id(), awaiting_parent);

// Add block components to the new request
if let Some(block_component) = block_component {
lookup.add_child_components(block_component);
Expand Down Expand Up @@ -338,7 +344,8 @@ impl<T: BeaconChainTypes> BlockLookups<T> {
Ok((response, seen_timestamp)) => {
debug!(self.log,
"Received lookup download success";
"block_root" => %block_root,
"block_root" => ?block_root,
"id" => id,
"peer_id" => %peer_id,
"response_type" => ?response_type,
);
Expand All @@ -357,7 +364,8 @@ impl<T: BeaconChainTypes> BlockLookups<T> {
Err(e) => {
debug!(self.log,
"Received lookup download failure";
"block_root" => %block_root,
"block_root" => ?block_root,
"id" => id,
"peer_id" => %peer_id,
"response_type" => ?response_type,
"error" => %e,
Expand All @@ -380,7 +388,7 @@ impl<T: BeaconChainTypes> BlockLookups<T> {
req.should_drop_lookup_on_disconnected_peer(peer_id );

if should_drop_lookup {
debug!(self.log, "Dropping single lookup after peer disconnection"; "block_root" => %req.block_root());
debug!(self.log, "Dropping single lookup after peer disconnection"; "block_root" => ?req.block_root());
}

!should_drop_lookup
Expand Down Expand Up @@ -425,6 +433,7 @@ impl<T: BeaconChainTypes> BlockLookups<T> {
"Received lookup processing result";
"component" => ?R::response_type(),
"block_root" => ?block_root,
"id" => lookup_id,
"result" => ?result,
);

Expand Down Expand Up @@ -496,7 +505,7 @@ impl<T: BeaconChainTypes> BlockLookups<T> {
debug!(
self.log,
"Single block lookup failed. Execution layer is offline / unsynced / misconfigured";
"block_root" => %block_root,
"block_root" => ?block_root,
"error" => ?e
);
Action::Drop
Expand All @@ -505,7 +514,7 @@ impl<T: BeaconChainTypes> BlockLookups<T> {
if e.category() == AvailabilityCheckErrorCategory::Internal =>
{
// There errors indicate internal problems and should not downscore the peer
warn!(self.log, "Internal availability check failure"; "block_root" => %block_root, "error" => ?e);
warn!(self.log, "Internal availability check failure"; "block_root" => ?block_root, "error" => ?e);

// Here we choose *not* to call `on_processing_failure` because this could result in a bad
// lookup state transition. This error invalidates both blob and block requests, and we don't know the
Expand All @@ -514,7 +523,7 @@ impl<T: BeaconChainTypes> BlockLookups<T> {
Action::Drop
}
other => {
debug!(self.log, "Invalid lookup component"; "block_root" => %block_root, "component" => ?R::response_type(), "error" => ?other);
debug!(self.log, "Invalid lookup component"; "block_root" => ?block_root, "component" => ?R::response_type(), "error" => ?other);
let peer_id = request_state.on_processing_failure()?;
cx.report_peer(
peer_id,
Expand All @@ -540,7 +549,7 @@ impl<T: BeaconChainTypes> BlockLookups<T> {
Action::ParentUnknown { parent_root } => {
let peers = lookup.all_available_peers().cloned().collect::<Vec<_>>();
lookup.set_awaiting_parent(parent_root);
debug!(self.log, "Marking lookup as awaiting parent"; "lookup" => %block_root, "parent_root" => %parent_root);
debug!(self.log, "Marking lookup as awaiting parent"; "id" => lookup.id, "block_root" => ?block_root, "parent_root" => ?parent_root);
self.search_parent_of_child(parent_root, block_root, &peers, cx);
Ok(LookupResult::Pending)
}
Expand All @@ -562,7 +571,7 @@ impl<T: BeaconChainTypes> BlockLookups<T> {
for (id, lookup) in self.single_block_lookups.iter_mut() {
if lookup.awaiting_parent() == Some(block_root) {
lookup.resolve_awaiting_parent();
debug!(self.log, "Continuing child lookup"; "parent_root" => %block_root, "block_root" => %lookup.block_root());
debug!(self.log, "Continuing child lookup"; "parent_root" => ?block_root, "id" => id, "block_root" => ?lookup.block_root());
let result = lookup.continue_requests(cx);
lookup_results.push((*id, result));
}
Expand All @@ -578,7 +587,7 @@ impl<T: BeaconChainTypes> BlockLookups<T> {
/// dropped.
pub fn drop_lookup_and_children(&mut self, dropped_id: SingleLookupId) {
if let Some(dropped_lookup) = self.single_block_lookups.remove(&dropped_id) {
debug!(self.log, "Dropping child lookup"; "id" => ?dropped_id, "block_root" => %dropped_lookup.block_root());
debug!(self.log, "Dropping child lookup"; "id" => ?dropped_id, "block_root" => ?dropped_lookup.block_root());

let child_lookups = self
.single_block_lookups
Expand All @@ -605,11 +614,13 @@ impl<T: BeaconChainTypes> BlockLookups<T> {
Ok(LookupResult::Pending) => {} // no action
Ok(LookupResult::Completed) => {
if let Some(lookup) = self.single_block_lookups.remove(&id) {
debug!(self.log, "Dropping completed lookup"; "block" => %lookup.block_root());
debug!(self.log, "Dropping completed lookup"; "block" => ?lookup.block_root(), "id" => id);
metrics::inc_counter(&metrics::SYNC_LOOKUP_COMPLETED);
// Block imported, continue the requests of pending child blocks
self.continue_child_lookups(lookup.block_root(), cx);
self.update_metrics();
} else {
debug!(self.log, "Attempting to drop non-existent lookup"; "id" => id);
}
}
Err(error) => {
Expand Down
24 changes: 10 additions & 14 deletions beacon_node/network/src/sync/block_lookups/single_block_lookup.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -172,17 +172,12 @@ impl<T: BeaconChainTypes> SingleBlockLookup<T> {
)
}

/// Add all given peers to both block and blob request states.
pub fn add_peer(&mut self, peer_id: PeerId) {
self.block_request_state.state.add_peer(&peer_id);
self.blob_request_state.state.add_peer(&peer_id);
}

/// Add all given peers to both block and blob request states.
pub fn add_peers(&mut self, peers: &[PeerId]) {
for peer in peers {
self.add_peer(*peer);
}
/// Add peer to all request states. The peer must be able to serve this request.
/// Returns true if the peer was newly inserted into some request state.
pub fn add_peer(&mut self, peer_id: PeerId) -> bool {
let inserted_block = self.block_request_state.state.add_peer(&peer_id);
let inserted_blob = self.blob_request_state.state.add_peer(&peer_id);
inserted_block || inserted_blob
}

/// Returns true if the block has already been downloaded.
Expand Down Expand Up @@ -464,9 +459,10 @@ impl<T: Clone> SingleLookupRequestState<T> {
self.failed_processing >= self.failed_downloading
}

/// This method should be used for peers wrapped in `PeerId::BlockAndBlobs`.
pub fn add_peer(&mut self, peer_id: &PeerId) {
self.available_peers.insert(*peer_id);
/// Add peer to this request states. The peer must be able to serve this request.
/// Returns true if the peer is newly inserted.
pub fn add_peer(&mut self, peer_id: &PeerId) -> bool {
self.available_peers.insert(*peer_id)
}

/// If a peer disconnects, this request could be failed. If so, an error is returned
Expand Down
15 changes: 11 additions & 4 deletions beacon_node/network/src/sync/manager.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -56,7 +56,7 @@ use lighthouse_network::rpc::RPCError;
use lighthouse_network::types::{NetworkGlobals, SyncState};
use lighthouse_network::SyncInfo;
use lighthouse_network::{PeerAction, PeerId};
use slog::{crit, debug, error, info, trace, warn, Logger};
use slog::{crit, debug, error, info, o, trace, warn, Logger};
use std::ops::Sub;
use std::sync::Arc;
use std::time::Duration;
Expand Down Expand Up @@ -257,9 +257,16 @@ impl<T: BeaconChainTypes> SyncManager<T> {
beacon_chain.clone(),
log.clone(),
),
range_sync: RangeSync::new(beacon_chain.clone(), log.clone()),
backfill_sync: BackFillSync::new(beacon_chain.clone(), network_globals, log.clone()),
block_lookups: BlockLookups::new(log.clone()),
range_sync: RangeSync::new(
beacon_chain.clone(),
log.new(o!("service" => "range_sync")),
),
backfill_sync: BackFillSync::new(
beacon_chain.clone(),
network_globals,
log.new(o!("service" => "backfill_sync")),
),
block_lookups: BlockLookups::new(log.new(o!("service"=> "lookup_sync"))),
log: log.clone(),
}
}
Expand Down

0 comments on commit 0b49f0d

Please sign in to comment.