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

Use fewer fixed vecs in lookups #39

Open
wants to merge 2 commits into
base: pr-5583-review
Choose a base branch
from
Open

Use fewer fixed vecs in lookups #39

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
23129c4
add back lookup verify error
realbigsean Apr 17, 2024
53020a3
remove fixed vecs
realbigsean Apr 18, 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
13 changes: 4 additions & 9 deletions beacon_node/beacon_chain/src/beacon_chain.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -121,7 +121,6 @@ use store::{
use task_executor::{ShutdownReason, TaskExecutor};
use tokio_stream::Stream;
use tree_hash::TreeHash;
use types::blob_sidecar::FixedBlobSidecarList;
use types::payload::BlockProductionVersion;
use types::*;

Expand Down Expand Up @@ -2913,7 +2912,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
self: &Arc<Self>,
slot: Slot,
block_root: Hash256,
blobs: FixedBlobSidecarList<T::EthSpec>,
blobs: Vec<Arc<BlobSidecar<T::EthSpec>>>,
) -> Result<AvailabilityProcessingStatus, BlockError<T::EthSpec>> {
// If this block has already been imported to forkchoice it must have been available, so
// we don't need to process its blobs again.
Expand All @@ -2927,7 +2926,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {

if let Some(event_handler) = self.event_handler.as_ref() {
if event_handler.has_blob_sidecar_subscribers() {
for blob in blobs.iter().filter_map(|maybe_blob| maybe_blob.as_ref()) {
for blob in blobs.iter() {
event_handler.register(EventKind::BlobSidecar(
SseBlobSidecar::from_blob_sidecar(blob),
));
Expand Down Expand Up @@ -3185,17 +3184,13 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
self: &Arc<Self>,
slot: Slot,
block_root: Hash256,
blobs: FixedBlobSidecarList<T::EthSpec>,
blobs: Vec<Arc<BlobSidecar<T::EthSpec>>>,
) -> Result<AvailabilityProcessingStatus, BlockError<T::EthSpec>> {
// Need to scope this to ensure the lock is dropped before calling `process_availability`
// Even an explicit drop is not enough to convince the borrow checker.
{
let mut slashable_cache = self.observed_slashable.write();
for header in blobs
.into_iter()
.filter_map(|b| b.as_ref().map(|b| b.signed_block_header.clone()))
.unique()
{
for header in blobs.iter().map(|b| b.signed_block_header.clone()).unique() {
if verify_header_signature::<T, BlockError<T::EthSpec>>(self, &header).is_ok() {
slashable_cache
.observe_slashable(
Expand Down
6 changes: 3 additions & 3 deletions beacon_node/beacon_chain/src/data_availability_checker.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@ use std::fmt::Debug;
use std::num::NonZeroUsize;
use std::sync::Arc;
use task_executor::TaskExecutor;
use types::blob_sidecar::{BlobIdentifier, BlobSidecar, FixedBlobSidecarList};
use types::blob_sidecar::{BlobIdentifier, BlobSidecar};
use types::{BlobSidecarList, ChainSpec, Epoch, EthSpec, Hash256, SignedBeaconBlock};

mod child_components;
Expand Down Expand Up @@ -170,13 +170,13 @@ impl<T: BeaconChainTypes> DataAvailabilityChecker<T> {
pub fn put_rpc_blobs(
&self,
block_root: Hash256,
blobs: FixedBlobSidecarList<T::EthSpec>,
blobs: Vec<Arc<BlobSidecar<T::EthSpec>>>,
) -> Result<Availability<T::EthSpec>, AvailabilityCheckError> {
let Some(kzg) = self.kzg.as_ref() else {
return Err(AvailabilityCheckError::KzgNotInitialized);
};

let verified_blobs = KzgVerifiedBlobList::new(Vec::from(blobs).into_iter().flatten(), kzg)
let verified_blobs = KzgVerifiedBlobList::new(blobs.into_iter(), kzg)
.map_err(AvailabilityCheckError::Kzg)?;

self.availability_cache
Expand Down
48 changes: 30 additions & 18 deletions beacon_node/beacon_chain/src/data_availability_checker/child_components.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,16 +14,6 @@ pub struct ChildComponents<E: EthSpec> {
pub downloaded_blobs: FixedBlobSidecarList<E>,
}

impl<E: EthSpec> From<RpcBlock<E>> for ChildComponents<E> {
fn from(value: RpcBlock<E>) -> Self {
let (block_root, block, blobs) = value.deconstruct();
let fixed_blobs = blobs.map(|blobs| {
FixedBlobSidecarList::from(blobs.into_iter().map(Some).collect::<Vec<_>>())
});
Self::new(block_root, Some(block), fixed_blobs)
}
}

impl<E: EthSpec> ChildComponents<E> {
pub fn empty(block_root: Hash256) -> Self {
Self {
Expand All @@ -32,34 +22,56 @@ impl<E: EthSpec> ChildComponents<E> {
downloaded_blobs: <_>::default(),
}
}

pub fn new(
block_root: Hash256,
block: Option<Arc<SignedBeaconBlock<E>>>,
blobs: Option<FixedBlobSidecarList<E>>,
) -> Self {
blobs: Option<Vec<Arc<BlobSidecar<E>>>>,
) -> Result<Self, String> {
let mut cache = Self::empty(block_root);
if let Some(block) = block {
cache.merge_block(block);
}
if let Some(blobs) = blobs {
cache.merge_blobs(blobs);
cache.merge_blobs(blobs)?;
}
cache
Ok(cache)
}

pub fn new_from_rpc_block(value: RpcBlock<E>) -> Result<Self, String> {
let (block_root, block, blobs) = value.deconstruct();
// Safe to unwrap because we are constructing the struct from a valid RpcBlock
Self::new(block_root, Some(block), blobs.map(Into::into))
}

pub fn merge_block(&mut self, block: Arc<SignedBeaconBlock<E>>) {
self.downloaded_block = Some(block);
}

pub fn merge_blob(&mut self, blob: Arc<BlobSidecar<E>>) {
pub fn merge_blob(&mut self, blob: Arc<BlobSidecar<E>>) -> Result<(), String> {
if blob.index >= E::max_blobs_per_block() as u64 {
return Err(format!("Blob index {} is out of range", blob.index));
}
self.merge_blob_unchecked(blob);
Ok(())
}

pub fn merge_blobs(&mut self, blobs: Vec<Arc<BlobSidecar<E>>>) -> Result<(), String> {
for blob in blobs.into_iter() {
self.merge_blob(blob)?;
}
Ok(())
}

pub fn merge_blob_unchecked(&mut self, blob: Arc<BlobSidecar<E>>) {
if let Some(blob_ref) = self.downloaded_blobs.get_mut(blob.index as usize) {
*blob_ref = Some(blob);
}
}

pub fn merge_blobs(&mut self, blobs: FixedBlobSidecarList<E>) {
for blob in blobs.iter().flatten() {
self.merge_blob(blob.clone());
pub fn merge_blobs_unchecked(&mut self, blobs: Vec<Arc<BlobSidecar<E>>>) {
for blob in blobs.into_iter() {
self.merge_blob_unchecked(blob);
}
}

Expand Down
5 changes: 2 additions & 3 deletions beacon_node/network/src/network_beacon_processor/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,6 @@ use tokio::sync::mpsc::{self, error::TrySendError};
use types::*;

pub use sync_methods::ChainSegmentProcessId;
use types::blob_sidecar::FixedBlobSidecarList;

pub type Error<T> = TrySendError<BeaconWorkEvent<T>>;

Expand Down Expand Up @@ -426,11 +425,11 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
pub fn send_rpc_blobs(
self: &Arc<Self>,
block_root: Hash256,
blobs: FixedBlobSidecarList<T::EthSpec>,
blobs: Vec<Arc<BlobSidecar<T::EthSpec>>>,
seen_timestamp: Duration,
process_type: BlockProcessType,
) -> Result<(), Error<T::EthSpec>> {
let blob_count = blobs.iter().filter(|b| b.is_some()).count();
let blob_count = blobs.len();
if blob_count == 0 {
return Ok(());
}
Expand Down
18 changes: 5 additions & 13 deletions beacon_node/network/src/network_beacon_processor/sync_methods.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,8 +23,7 @@ use std::time::Duration;
use store::KzgCommitment;
use tokio::sync::mpsc;
use types::beacon_block_body::format_kzg_commitments;
use types::blob_sidecar::FixedBlobSidecarList;
use types::{Epoch, Hash256};
use types::{BlobSidecar, Epoch, Hash256};

/// Id associated to a batch processing request, either a sync batch or a parent lookup.
#[derive(Clone, Debug, PartialEq)]
Expand Down Expand Up @@ -201,7 +200,7 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
pub fn generate_rpc_blobs_process_fn(
self: Arc<Self>,
block_root: Hash256,
blobs: FixedBlobSidecarList<T::EthSpec>,
blobs: Vec<Arc<BlobSidecar<T::EthSpec>>>,
seen_timestamp: Duration,
process_type: BlockProcessType,
) -> AsyncFn {
Expand All @@ -217,24 +216,17 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
pub async fn process_rpc_blobs(
self: Arc<NetworkBeaconProcessor<T>>,
block_root: Hash256,
blobs: FixedBlobSidecarList<T::EthSpec>,
blobs: Vec<Arc<BlobSidecar<T::EthSpec>>>,
seen_timestamp: Duration,
process_type: BlockProcessType,
) {
let Some(slot) = blobs
.iter()
.find_map(|blob| blob.as_ref().map(|blob| blob.slot()))
else {
let Some(slot) = blobs.iter().map(|blob| blob.slot()).next() else {
return;
};

let (indices, commitments): (Vec<u64>, Vec<KzgCommitment>) = blobs
.iter()
.filter_map(|blob_opt| {
blob_opt
.as_ref()
.map(|blob| (blob.index, blob.kzg_commitment))
})
.map(|blob| (blob.index, blob.kzg_commitment))
.unzip();
let commitments = format_kzg_commitments(&commitments);

Expand Down
20 changes: 9 additions & 11 deletions beacon_node/network/src/sync/block_lookups/common.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,8 +14,7 @@ use beacon_chain::data_availability_checker::ChildComponents;
use beacon_chain::BeaconChainTypes;
use std::sync::Arc;
use std::time::Duration;
use types::blob_sidecar::FixedBlobSidecarList;
use types::{Hash256, SignedBeaconBlock};
use types::{BlobSidecar, Hash256, SignedBeaconBlock};

#[derive(Debug, Copy, Clone)]
pub enum ResponseType {
Expand Down Expand Up @@ -266,8 +265,8 @@ impl<L: Lookup, T: BeaconChainTypes> RequestState<L, T> for BlockRequestState<L>

impl<L: Lookup, T: BeaconChainTypes> RequestState<L, T> for BlobRequestState<L, T::EthSpec> {
type RequestType = BlobsByRootSingleBlockRequest;
type VerifiedResponseType = FixedBlobSidecarList<T::EthSpec>;
type ReconstructedResponseType = FixedBlobSidecarList<T::EthSpec>;
type VerifiedResponseType = Vec<Arc<BlobSidecar<T::EthSpec>>>;
type ReconstructedResponseType = Vec<Arc<BlobSidecar<T::EthSpec>>>;

fn new_request(&self) -> Self::RequestType {
BlobsByRootSingleBlockRequest {
Expand All @@ -286,33 +285,32 @@ impl<L: Lookup, T: BeaconChainTypes> RequestState<L, T> for BlobRequestState<L,
.map_err(LookupRequestError::SendFailed)
}

fn get_parent_root(verified_response: &FixedBlobSidecarList<T::EthSpec>) -> Option<Hash256> {
fn get_parent_root(verified_response: &Vec<Arc<BlobSidecar<T::EthSpec>>>) -> Option<Hash256> {
verified_response
.into_iter()
.filter_map(|blob| blob.as_ref())
.map(|blob| blob.block_parent_root())
.next()
}

fn add_to_child_components(
verified_response: FixedBlobSidecarList<T::EthSpec>,
verified_response: Vec<Arc<BlobSidecar<T::EthSpec>>>,
components: &mut ChildComponents<T::EthSpec>,
) {
components.merge_blobs(verified_response);
components.merge_blobs(verified_response).unwrap();
}

fn verified_to_reconstructed(
_block_root: Hash256,
blobs: FixedBlobSidecarList<T::EthSpec>,
) -> FixedBlobSidecarList<T::EthSpec> {
blobs: Vec<Arc<BlobSidecar<T::EthSpec>>>,
) -> Vec<Arc<BlobSidecar<T::EthSpec>>> {
blobs
}

fn send_reconstructed_for_processing(
id: Id,
bl: &BlockLookups<T>,
block_root: Hash256,
verified: FixedBlobSidecarList<T::EthSpec>,
verified: Vec<Arc<BlobSidecar<T::EthSpec>>>,
duration: Duration,
cx: &SyncNetworkContext<T>,
) -> Result<(), LookupRequestError> {
Expand Down
Loading