Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Add more info about why lookup is in AwaitingDownload #5838

Merged
merged 5 commits into from
May 28, 2024
Merged

Add more info about why lookup is in AwaitingDownload #5838

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
7fa6525
Add more info about why lookup is in AwaitingDownload
dapplion May 24, 2024
3ed1458
Review comments
dapplion May 24, 2024
1d305ba
Merge branch 'unstable' into awaiting-download-context
realbigsean May 24, 2024
e96fced
Merge branch 'unstable' of https://github.com/sigp/lighthouse into aw…
realbigsean May 27, 2024
63747ca
cargo fmt
realbigsean May 27, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
38 changes: 26 additions & 12 deletions beacon_node/network/src/sync/block_lookups/single_block_lookup.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -207,7 +207,12 @@ impl<T: BeaconChainTypes> SingleBlockLookup<T> {
request.get_state_mut().on_completed_request()?
}
// Sync will receive a future event to make progress on the request, do nothing now
LookupRequestResult::Pending => return Ok(()),
LookupRequestResult::Pending(reason) => {
request
.get_state_mut()
.update_awaiting_download_status(reason);
return Ok(());
}
}

// Otherwise, attempt to progress awaiting processing
Expand Down Expand Up @@ -301,7 +306,7 @@ pub struct DownloadResult<T: Clone> {

#[derive(PartialEq, Eq, IntoStaticStr)]
pub enum State<T: Clone> {
AwaitingDownload,
AwaitingDownload(&'static str),
Downloading(ReqId),
AwaitingProcess(DownloadResult<T>),
/// Request is processing, sent by lookup sync
Expand All @@ -325,15 +330,15 @@ pub struct SingleLookupRequestState<T: Clone> {
impl<T: Clone> SingleLookupRequestState<T> {
pub fn new() -> Self {
Self {
state: State::AwaitingDownload,
state: State::AwaitingDownload("not started"),
failed_processing: 0,
failed_downloading: 0,
}
}

pub fn is_awaiting_download(&self) -> bool {
match self.state {
State::AwaitingDownload => true,
State::AwaitingDownload { .. } => true,
State::Downloading { .. }
| State::AwaitingProcess { .. }
| State::Processing { .. }
Expand All @@ -343,7 +348,7 @@ impl<T: Clone> SingleLookupRequestState<T> {

pub fn is_processed(&self) -> bool {
match self.state {
State::AwaitingDownload
State::AwaitingDownload { .. }
| State::Downloading { .. }
| State::AwaitingProcess { .. }
| State::Processing { .. } => false,
Expand All @@ -353,7 +358,7 @@ impl<T: Clone> SingleLookupRequestState<T> {

pub fn peek_downloaded_data(&self) -> Option<&T> {
match &self.state {
State::AwaitingDownload => None,
State::AwaitingDownload { .. } => None,
State::Downloading { .. } => None,
State::AwaitingProcess(result) => Some(&result.value),
State::Processing(result) => Some(&result.value),
Expand All @@ -364,18 +369,27 @@ impl<T: Clone> SingleLookupRequestState<T> {
/// Switch to `AwaitingProcessing` if the request is in `AwaitingDownload` state, otherwise
/// ignore.
pub fn insert_verified_response(&mut self, result: DownloadResult<T>) -> bool {
if let State::AwaitingDownload = &self.state {
if let State::AwaitingDownload { .. } = &self.state {
self.state = State::AwaitingProcess(result);
true
} else {
false
}
}

/// Append metadata on why this request is in AwaitingDownload status. Very helpful to debug
/// stuck lookups. Not fallible as it's purely informational.
pub fn update_awaiting_download_status(&mut self, new_status: &'static str) {
match &mut self.state {
State::AwaitingDownload(status) => *status = new_status,
_ => {}
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

looks like a lint is failing here

}
}

/// Switch to `Downloading` if the request is in `AwaitingDownload` state, otherwise returns None.
pub fn on_download_start(&mut self, req_id: ReqId) -> Result<(), LookupRequestError> {
match &self.state {
State::AwaitingDownload => {
State::AwaitingDownload { .. } => {
self.state = State::Downloading(req_id);
Ok(())
}
Expand All @@ -397,7 +411,7 @@ impl<T: Clone> SingleLookupRequestState<T> {
});
}
self.failed_downloading = self.failed_downloading.saturating_add(1);
self.state = State::AwaitingDownload;
self.state = State::AwaitingDownload("retry");
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Might be good to differentiate "retry after download failure" from "retry after processing failure"

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Changed to "not started" since it will never remain in that state

Ok(())
}
other => Err(LookupRequestError::BadState(format!(
Expand Down Expand Up @@ -461,7 +475,7 @@ impl<T: Clone> SingleLookupRequestState<T> {
State::Processing(result) => {
let peer_id = result.peer_id;
self.failed_processing = self.failed_processing.saturating_add(1);
self.state = State::AwaitingDownload;
self.state = State::AwaitingDownload("retry");
Ok(peer_id)
}
other => Err(LookupRequestError::BadState(format!(
Expand All @@ -485,7 +499,7 @@ impl<T: Clone> SingleLookupRequestState<T> {
/// Mark a request as complete without any download or processing
pub fn on_completed_request(&mut self) -> Result<(), LookupRequestError> {
match &self.state {
State::AwaitingDownload => {
State::AwaitingDownload { .. } => {
self.state = State::Processed;
Ok(())
}
Expand Down Expand Up @@ -517,7 +531,7 @@ impl<T: Clone> std::fmt::Display for State<T> {
impl<T: Clone> std::fmt::Debug for State<T> {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::AwaitingDownload { .. } => write!(f, "AwaitingDownload"),
Self::AwaitingDownload(status) => write!(f, "AwaitingDownload({:?})", status),
Self::Downloading(req_id) => write!(f, "Downloading({:?})", req_id),
Self::AwaitingProcess(d) => write!(f, "AwaitingProcess({:?})", d.peer_id),
Self::Processing(d) => write!(f, "Processing({:?})", d.peer_id),
Expand Down
6 changes: 3 additions & 3 deletions beacon_node/network/src/sync/network_context.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -105,7 +105,7 @@ pub enum LookupRequestResult {
/// that makes progress on the request. For example: request is processing from a different
/// source (i.e. block received from gossip) and sync MUST receive an event with that processing
/// result.
Pending,
Pending(&'static str),
}

/// Wraps a Network channel to employ various RPC related network functionality for the Sync manager. This includes management of a global RPC request Id.
Expand Down Expand Up @@ -356,7 +356,7 @@ impl<T: BeaconChainTypes> SyncNetworkContext<T> {
// A block is on the `reqresp_pre_import_cache` but NOT in the
// `data_availability_checker` only if it is actively processing. We can expect a future
// event with the result of processing
return Ok(LookupRequestResult::Pending);
return Ok(LookupRequestResult::Pending("block in processing cache"));
}

let req_id = self.next_id();
Expand Down Expand Up @@ -410,7 +410,7 @@ impl<T: BeaconChainTypes> SyncNetworkContext<T> {
// latter handle the case where if the peer sent no blobs, penalize.
// - if `downloaded_block_expected_blobs` is Some = block is downloading or processing.
// - if `num_expected_blobs` returns Some = block is processed.
return Ok(LookupRequestResult::Pending);
return Ok(LookupRequestResult::Pending("waiting for block download"));
};

let imported_blob_indexes = self
Expand Down
Loading