Skip to content

Commit

Permalink
Pass vec to range sync batch (#5710)
Browse files Browse the repository at this point in the history 
* Pass vec to range sync batch
  • Loading branch information
@dapplion
dapplion authored Jun 27, 2024
1 parent f14f21f commit 784ef5f
Show file tree
Hide file tree
Showing 5 changed files with 46 additions and 85 deletions.
12 changes: 3 additions & 9 deletions beacon_node/network/src/sync/backfill_sync/mod.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -369,7 +369,7 @@ impl<T: BeaconChainTypes> BackFillSync<T> {
batch_id: BatchId,
peer_id: &PeerId,
request_id: Id,
beacon_block: Option<RpcBlock<T::EthSpec>>,
blocks: Vec<RpcBlock<T::EthSpec>>,
) -> Result<ProcessResult, BackFillError> {
// check if we have this batch
let batch = match self.batches.get_mut(&batch_id) {
Expand All @@ -392,20 +392,14 @@ impl<T: BeaconChainTypes> BackFillSync<T> {
}
};

if let Some(block) = beacon_block {
// This is not a stream termination, simply add the block to the request
if let Err(e) = batch.add_block(block) {
self.fail_sync(BackFillError::BatchInvalidState(batch_id, e.0))?;
}
Ok(ProcessResult::Successful)
} else {
{
// A stream termination has been sent. This batch has ended. Process a completed batch.
// Remove the request from the peer's active batches
self.active_requests
.get_mut(peer_id)
.map(|active_requests| active_requests.remove(&batch_id));

match batch.download_completed() {
match batch.download_completed(blocks) {
Ok(received) => {
let awaiting_batches =
self.processing_target.saturating_sub(batch_id) / BACKFILL_EPOCHS_PER_BATCH;
Expand Down
59 changes: 26 additions & 33 deletions beacon_node/network/src/sync/manager.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -908,39 +908,32 @@ impl<T: BeaconChainTypes> SyncManager<T> {
{
match resp.responses {
Ok(blocks) => {
for block in blocks
.into_iter()
.map(Some)
// chain the stream terminator
.chain(vec![None])
{
match resp.sender_id {
RangeRequestId::RangeSync { chain_id, batch_id } => {
self.range_sync.blocks_by_range_response(
&mut self.network,
peer_id,
chain_id,
batch_id,
id,
block,
);
self.update_sync_state();
}
RangeRequestId::BackfillSync { batch_id } => {
match self.backfill_sync.on_block_response(
&mut self.network,
batch_id,
&peer_id,
id,
block,
) {
Ok(ProcessResult::SyncCompleted) => self.update_sync_state(),
Ok(ProcessResult::Successful) => {}
Err(_error) => {
// The backfill sync has failed, errors are reported
// within.
self.update_sync_state();
}
match resp.sender_id {
RangeRequestId::RangeSync { chain_id, batch_id } => {
self.range_sync.blocks_by_range_response(
&mut self.network,
peer_id,
chain_id,
batch_id,
id,
blocks,
);
self.update_sync_state();
}
RangeRequestId::BackfillSync { batch_id } => {
match self.backfill_sync.on_block_response(
&mut self.network,
batch_id,
&peer_id,
id,
blocks,
) {
Ok(ProcessResult::SyncCompleted) => self.update_sync_state(),
Ok(ProcessResult::Successful) => {}
Err(_error) => {
// The backfill sync has failed, errors are reported
// within.
self.update_sync_state();
}
}
}
Expand Down
42 changes: 10 additions & 32 deletions beacon_node/network/src/sync/range_sync/batch.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -116,7 +116,7 @@ pub enum BatchState<E: EthSpec> {
/// The batch has failed either downloading or processing, but can be requested again.
AwaitingDownload,
/// The batch is being downloaded.
Downloading(PeerId, Vec<RpcBlock<E>>, Id),
Downloading(PeerId, Id),
/// The batch has been completely downloaded and is ready for processing.
AwaitingProcessing(PeerId, Vec<RpcBlock<E>>),
/// The batch is being processed.
Expand Down Expand Up @@ -199,7 +199,7 @@ impl<E: EthSpec, B: BatchConfig> BatchInfo<E, B> {

/// Verifies if an incoming block belongs to this batch.
pub fn is_expecting_block(&self, peer_id: &PeerId, request_id: &Id) -> bool {
if let BatchState::Downloading(expected_peer, _, expected_id) = &self.state {
if let BatchState::Downloading(expected_peer, expected_id) = &self.state {
return peer_id == expected_peer && expected_id == request_id;
}
false
Expand All @@ -209,7 +209,7 @@ impl<E: EthSpec, B: BatchConfig> BatchInfo<E, B> {
pub fn current_peer(&self) -> Option<&PeerId> {
match &self.state {
BatchState::AwaitingDownload | BatchState::Failed => None,
BatchState::Downloading(peer_id, _, _)
BatchState::Downloading(peer_id, _)
| BatchState::AwaitingProcessing(peer_id, _)
| BatchState::Processing(Attempt { peer_id, .. })
| BatchState::AwaitingValidation(Attempt { peer_id, .. }) => Some(peer_id),
Expand Down Expand Up @@ -250,36 +250,18 @@ impl<E: EthSpec, B: BatchConfig> BatchInfo<E, B> {
&self.failed_processing_attempts
}

/// Adds a block to a downloading batch.
pub fn add_block(&mut self, block: RpcBlock<E>) -> Result<(), WrongState> {
match self.state.poison() {
BatchState::Downloading(peer, mut blocks, req_id) => {
blocks.push(block);
self.state = BatchState::Downloading(peer, blocks, req_id);
Ok(())
}
BatchState::Poisoned => unreachable!("Poisoned batch"),
other => {
self.state = other;
Err(WrongState(format!(
"Add block for batch in wrong state {:?}",
self.state
)))
}
}
}

/// Marks the batch as ready to be processed if the blocks are in the range. The number of
/// received blocks is returned, or the wrong batch end on failure
#[must_use = "Batch may have failed"]
pub fn download_completed(
&mut self,
blocks: Vec<RpcBlock<E>>,
) -> Result<
usize, /* Received blocks */
Result<(Slot, Slot, BatchOperationOutcome), WrongState>,
> {
match self.state.poison() {
BatchState::Downloading(peer, blocks, _request_id) => {
BatchState::Downloading(peer, _request_id) => {
// verify that blocks are in range
if let Some(last_slot) = blocks.last().map(|b| b.slot()) {
// the batch is non-empty
Expand Down Expand Up @@ -336,7 +318,7 @@ impl<E: EthSpec, B: BatchConfig> BatchInfo<E, B> {
mark_failed: bool,
) -> Result<BatchOperationOutcome, WrongState> {
match self.state.poison() {
BatchState::Downloading(peer, _, _request_id) => {
BatchState::Downloading(peer, _request_id) => {
// register the attempt and check if the batch can be tried again
if mark_failed {
self.failed_download_attempts.push(peer);
Expand Down Expand Up @@ -369,7 +351,7 @@ impl<E: EthSpec, B: BatchConfig> BatchInfo<E, B> {
) -> Result<(), WrongState> {
match self.state.poison() {
BatchState::AwaitingDownload => {
self.state = BatchState::Downloading(peer, Vec::new(), request_id);
self.state = BatchState::Downloading(peer, request_id);
Ok(())
}
BatchState::Poisoned => unreachable!("Poisoned batch"),
Expand Down Expand Up @@ -536,13 +518,9 @@ impl<E: EthSpec> std::fmt::Debug for BatchState<E> {
BatchState::AwaitingProcessing(ref peer, ref blocks) => {
write!(f, "AwaitingProcessing({}, {} blocks)", peer, blocks.len())
}
BatchState::Downloading(peer, blocks, request_id) => write!(
f,
"Downloading({}, {} blocks, {})",
peer,
blocks.len(),
request_id
),
BatchState::Downloading(peer, request_id) => {
write!(f, "Downloading({}, {})", peer, request_id)
}
BatchState::Poisoned => f.write_str("Poisoned"),
}
}
Expand Down
10 changes: 3 additions & 7 deletions beacon_node/network/src/sync/range_sync/chain.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -200,7 +200,7 @@ impl<T: BeaconChainTypes> SyncingChain<T> {
batch_id: BatchId,
peer_id: &PeerId,
request_id: Id,
beacon_block: Option<RpcBlock<T::EthSpec>>,
blocks: Vec<RpcBlock<T::EthSpec>>,
) -> ProcessingResult {
// check if we have this batch
let batch = match self.batches.get_mut(&batch_id) {
Expand All @@ -221,18 +221,14 @@ impl<T: BeaconChainTypes> SyncingChain<T> {
}
};

if let Some(block) = beacon_block {
// This is not a stream termination, simply add the block to the request
batch.add_block(block)?;
Ok(KeepChain)
} else {
{
// A stream termination has been sent. This batch has ended. Process a completed batch.
// Remove the request from the peer's active batches
self.peers
.get_mut(peer_id)
.map(|active_requests| active_requests.remove(&batch_id));

match batch.download_completed() {
match batch.download_completed(blocks) {
Ok(received) => {
let awaiting_batches = batch_id
.saturating_sub(self.optimistic_start.unwrap_or(self.processing_target))
Expand Down
8 changes: 4 additions & 4 deletions beacon_node/network/src/sync/range_sync/range.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -210,11 +210,11 @@ where
chain_id: ChainId,
batch_id: BatchId,
request_id: Id,
beacon_block: Option<RpcBlock<T::EthSpec>>,
blocks: Vec<RpcBlock<T::EthSpec>>,
) {
// check if this chunk removes the chain
match self.chains.call_by_id(chain_id, |chain| {
chain.on_block_response(network, batch_id, &peer_id, request_id, beacon_block)
chain.on_block_response(network, batch_id, &peer_id, request_id, blocks)
}) {
Ok((removed_chain, sync_type)) => {
if let Some((removed_chain, remove_reason)) = removed_chain {
Expand Down Expand Up @@ -795,7 +795,7 @@ mod tests {
rig.cx.update_execution_engine_state(EngineState::Offline);

// send the response to the request
range.blocks_by_range_response(&mut rig.cx, peer1, chain1, batch1, id1, None);
range.blocks_by_range_response(&mut rig.cx, peer1, chain1, batch1, id1, vec![]);

// the beacon processor shouldn't have received any work
rig.expect_empty_processor();
Expand All @@ -809,7 +809,7 @@ mod tests {
rig.complete_range_block_and_blobs_response(block_req, blob_req_opt);

// send the response to the request
range.blocks_by_range_response(&mut rig.cx, peer2, chain2, batch2, id2, None);
range.blocks_by_range_response(&mut rig.cx, peer2, chain2, batch2, id2, vec![]);

// the beacon processor shouldn't have received any work
rig.expect_empty_processor();
Expand Down

0 comments on commit 784ef5f

Please sign in to comment.