Skip to content

Commit

Permalink
Visualisation logging for sync batch states (#6034)
Browse files Browse the repository at this point in the history
* Add visualization for batch states

* Replace icons with emojis

* Reviewers comments

* Change empty emoji and improve docs comments

* Fix lints

* Move to letters rather than emojis

* Replace 'V' with 'v'. Cargo update

* Merge latest unstable

* Improve docs around visualisation

* Merge branch 'unstable' into sync-batch-state
  • Loading branch information
AgeManning authored Jul 29, 2024
1 parent f4ddc45 commit a3b1ef3
Show file tree
Hide file tree
Showing 3 changed files with 78 additions and 11 deletions.
8 changes: 4 additions & 4 deletions Cargo.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

21 changes: 21 additions & 0 deletions beacon_node/network/src/sync/range_sync/batch.rs
Original file line number Diff line number Diff line change
Expand Up @@ -463,6 +463,11 @@ impl<E: EthSpec, B: BatchConfig> BatchInfo<E, B> {
}
}
}

// Visualizes the state of this batch using state::visualize()
pub fn visualize(&self) -> char {
self.state.visualize()
}
}

/// Represents a peer's attempt and providing the result for this batch.
Expand Down Expand Up @@ -539,3 +544,19 @@ impl<E: EthSpec> std::fmt::Debug for BatchState<E> {
}
}
}

impl<E: EthSpec> BatchState<E> {
/// Creates a character representation/visualization for the batch state to display in logs for quicker and
/// easier recognition
fn visualize(&self) -> char {
match self {
BatchState::Downloading(..) => 'D',
BatchState::Processing(_) => 'P',
BatchState::AwaitingValidation(_) => 'v',
BatchState::AwaitingDownload => 'd',
BatchState::Failed => 'F',
BatchState::AwaitingProcessing(..) => 'p',
BatchState::Poisoned => 'X',
}
}
}
60 changes: 53 additions & 7 deletions beacon_node/network/src/sync/range_sync/chain.rs
Original file line number Diff line number Diff line change
Expand Up @@ -277,7 +277,7 @@ impl<T: BeaconChainTypes> SyncingChain<T> {
let awaiting_batches = batch_id
.saturating_sub(self.optimistic_start.unwrap_or(self.processing_target))
/ EPOCHS_PER_BATCH;
debug!(self.log, "Completed batch received"; "epoch" => batch_id, "blocks" => received, "awaiting_batches" => awaiting_batches);
debug!(self.log, "Batch downloaded"; "epoch" => batch_id, "blocks" => received, "batch_state" => self.visualize_batch_state(), "awaiting_batches" => awaiting_batches);

// pre-emptively request more blocks from peers whilst we process current blocks,
self.request_batches(network)?;
Expand Down Expand Up @@ -460,6 +460,7 @@ impl<T: BeaconChainTypes> SyncingChain<T> {
) -> ProcessingResult {
// the first two cases are possible if the chain advances while waiting for a processing
// result
let batch_state = self.visualize_batch_state();
let batch = match &self.current_processing_batch {
Some(processing_id) if *processing_id != batch_id => {
debug!(self.log, "Unexpected batch result";
Expand Down Expand Up @@ -492,7 +493,7 @@ impl<T: BeaconChainTypes> SyncingChain<T> {

// Log the process result and the batch for debugging purposes.
debug!(self.log, "Batch processing result"; "result" => ?result, &batch,
"batch_epoch" => batch_id, "client" => %network.client_type(&peer));
"batch_epoch" => batch_id, "client" => %network.client_type(&peer), "batch_state" => batch_state);

// We consider three cases. Batch was successfully processed, Batch failed processing due
// to a faulty peer, or batch failed processing but the peer can't be deemed faulty.
Expand Down Expand Up @@ -859,6 +860,7 @@ impl<T: BeaconChainTypes> SyncingChain<T> {
peer_id: &PeerId,
request_id: Id,
) -> ProcessingResult {
let batch_state = self.visualize_batch_state();
if let Some(batch) = self.batches.get_mut(&batch_id) {
// A batch could be retried without the peer failing the request (disconnecting/
// sending an error /timeout) if the peer is removed from the chain for other
Expand All @@ -870,7 +872,8 @@ impl<T: BeaconChainTypes> SyncingChain<T> {
"batch_epoch" => batch_id,
"batch_state" => ?batch.state(),
"peer_id" => %peer_id,
"request_id" => %request_id
"request_id" => %request_id,
"batch_state" => batch_state
);
return Ok(KeepChain);
}
Expand All @@ -880,7 +883,8 @@ impl<T: BeaconChainTypes> SyncingChain<T> {
"batch_epoch" => batch_id,
"batch_state" => ?batch.state(),
"peer_id" => %peer_id,
"request_id" => %request_id
"request_id" => %request_id,
"batch_state" => batch_state
);
if let Some(active_requests) = self.peers.get_mut(peer_id) {
active_requests.remove(&batch_id);
Expand All @@ -898,7 +902,8 @@ impl<T: BeaconChainTypes> SyncingChain<T> {
"Batch not found";
"batch_epoch" => batch_id,
"peer_id" => %peer_id,
"request_id" => %request_id
"request_id" => %request_id,
"batch_state" => batch_state
);
// this could be an error for an old batch, removed when the chain advances
Ok(KeepChain)
Expand Down Expand Up @@ -948,6 +953,7 @@ impl<T: BeaconChainTypes> SyncingChain<T> {
batch_id: BatchId,
peer: PeerId,
) -> ProcessingResult {
let batch_state = self.visualize_batch_state();
if let Some(batch) = self.batches.get_mut(&batch_id) {
let (request, batch_type) = batch.to_blocks_by_range_request();
match network.blocks_and_blobs_by_range_request(
Expand All @@ -967,9 +973,9 @@ impl<T: BeaconChainTypes> SyncingChain<T> {
.map(|epoch| epoch == batch_id)
.unwrap_or(false)
{
debug!(self.log, "Requesting optimistic batch"; "epoch" => batch_id, &batch);
debug!(self.log, "Requesting optimistic batch"; "epoch" => batch_id, &batch, "batch_state" => batch_state);
} else {
debug!(self.log, "Requesting batch"; "epoch" => batch_id, &batch);
debug!(self.log, "Requesting batch"; "epoch" => batch_id, &batch, "batch_state" => batch_state);
}
// register the batch for this peer
return self
Expand Down Expand Up @@ -1130,6 +1136,46 @@ impl<T: BeaconChainTypes> SyncingChain<T> {
}
}
}

/// Creates a string visualization of the current state of the chain, to make it easier for debugging and understanding
/// where sync is up to from glancing at the logs.
///
/// This produces a string of the form: [D,E,E,E,E]
/// to indicate the current buffer state of the chain. The symbols are defined on each of the
/// batch states. See [BatchState::visualize] for symbol definitions.
fn visualize_batch_state(&self) -> String {
let mut visualization_string = String::with_capacity((BATCH_BUFFER_SIZE * 3) as usize);

// Start of the block
visualization_string.push('[');

for mut batch_index in 0..BATCH_BUFFER_SIZE {
if let Some(batch) = self
.batches
.get(&(self.processing_target + batch_index as u64 * EPOCHS_PER_BATCH))
{
visualization_string.push(batch.visualize());
if batch_index != BATCH_BUFFER_SIZE {
// Add a comma in between elements
visualization_string.push(',');
}
} else {
// No batch exists, it is on our list to be downloaded
// Fill in the rest of the gaps
while batch_index < BATCH_BUFFER_SIZE {
visualization_string.push('E');
// Add a comma between the empty batches
if batch_index < BATCH_BUFFER_SIZE.saturating_sub(1) {
visualization_string.push(',')
}
batch_index += 1;
}
break;
}
}
visualization_string.push(']');
visualization_string
}
}

impl<T: BeaconChainTypes> slog::KV for &mut SyncingChain<T> {
Expand Down

0 comments on commit a3b1ef3

Please sign in to comment.