Skip to content

Commit

Permalink
Merge branch 'deneb-free-blobs' of https://github.com/sigp/lighthouse …
Browse files Browse the repository at this point in the history 
…into partial-processing
  • Loading branch information
@realbigsean
realbigsean committed Mar 21, 2023
2 parents c1ecaa9 + d1e653c commit 2bfb0bf
Show file tree
Hide file tree
Showing 10 changed files with 152 additions and 153 deletions.
43 changes: 28 additions & 15 deletions beacon_node/beacon_chain/src/beacon_chain.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -983,12 +983,23 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
Ok(self.get_block(block_root).await?.map(Arc::new))
}

pub async fn get_block_and_blobs_checking_early_attester_cache(
pub async fn get_blobs_checking_early_attester_cache(
&self,
_block_root: &Hash256,
) -> Result<Option<()>, Error> {
//TODO(sean) use the rpc blobs cache and revert this to the current block cache logic
Ok(Some(()))
block_root: &Hash256,
) -> Result<Option<BlobSidecarList<T::EthSpec>>, Error> {
// If there is no data availability boundary, the Eip4844 fork is disabled.
if let Some(finalized_data_availability_boundary) =
self.finalized_data_availability_boundary()
{
self.early_attester_cache
.get_blobs(*block_root)
.map_or_else(
|| self.get_blobs(block_root, finalized_data_availability_boundary),
|blobs| Ok(Some(blobs)),
)
} else {
Ok(None)
}
}

/// Returns the block at the given root, if any.
Expand Down Expand Up @@ -3101,12 +3112,14 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
// margin, or younger (of higher epoch number).
if block_epoch >= import_boundary {
if let Some(blobs) = blobs {
//FIXME(sean) using this for debugging for now
info!(
self.log, "Writing blobs to store";
"block_root" => ?block_root
);
ops.push(StoreOp::PutBlobs(block_root, blobs));
if !blobs.is_empty() {
//FIXME(sean) using this for debugging for now
info!(
self.log, "Writing blobs to store";
"block_root" => ?block_root
);
ops.push(StoreOp::PutBlobs(block_root, blobs));
}
}
}
}
Expand Down Expand Up @@ -4860,7 +4873,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
)
.map_err(BlockProductionError::KzgError)?;

let blob_sidecars = VariableList::from(
let blob_sidecars = BlobSidecarList::from(
blobs
.into_iter()
.enumerate()
Expand All @@ -4873,7 +4886,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
.get(blob_index)
.expect("KZG proof should exist for blob");

Ok(BlobSidecar {
Ok(Arc::new(BlobSidecar {
block_root: beacon_block_root,
index: blob_index as u64,
slot,
Expand All @@ -4882,9 +4895,9 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
blob,
kzg_commitment: *kzg_commitment,
kzg_proof: *kzg_proof,
})
}))
})
.collect::<Result<Vec<BlobSidecar<T::EthSpec>>, BlockProductionError>>()?,
.collect::<Result<Vec<_>, BlockProductionError>>()?,
);

self.proposal_blob_cache
Expand Down
15 changes: 6 additions & 9 deletions beacon_node/http_api/src/block_id.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@ use eth2::types::BlockId as CoreBlockId;
use std::fmt;
use std::str::FromStr;
use std::sync::Arc;
use types::{BlobSidecar, Hash256, SignedBeaconBlock, SignedBlindedBeaconBlock, Slot};
use types::{BlobSidecarList, Hash256, SignedBeaconBlock, SignedBlindedBeaconBlock, Slot};

/// Wraps `eth2::types::BlockId` and provides a simple way to obtain a block or root for a given
/// `BlockId`.
Expand Down Expand Up @@ -212,19 +212,16 @@ impl BlockId {
}
}

/// Return the `BlobsSidecar` identified by `self`.
pub async fn blobs_sidecar<T: BeaconChainTypes>(
/// Return the `BlobSidecarList` identified by `self`.
pub async fn blob_sidecar_list<T: BeaconChainTypes>(
&self,
chain: &BeaconChain<T>,
) -> Result<Arc<BlobSidecar<T::EthSpec>>, warp::Rejection> {
) -> Result<BlobSidecarList<T::EthSpec>, warp::Rejection> {
let root = self.root(chain)?.0;
let Some(_data_availability_boundary) = chain.data_availability_boundary() else {
return Err(warp_utils::reject::custom_not_found("Eip4844 fork disabled".into()));
};
match chain.get_blobs(&root) {
Ok(Some(_blob)) => todo!(), // Jimmy's PR will fix this,
Ok(Some(blob_sidecar_list)) => Ok(blob_sidecar_list),
Ok(None) => Err(warp_utils::reject::custom_not_found(format!(
"Blob with block root {} is not in the store",
"No blobs with block root {} found in the store",
root
))),
Err(e) => Err(warp_utils::reject::beacon_chain_error(e)),
Expand Down
76 changes: 40 additions & 36 deletions beacon_node/http_api/src/lib.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -1293,6 +1293,45 @@ pub fn serve<T: BeaconChainTypes>(
},
);

/*
* beacon/blobs
*/

// GET beacon/blobs/{block_id}
let get_blobs = eth_v1
.and(warp::path("beacon"))
.and(warp::path("blobs"))
.and(block_id_or_err)
.and(warp::path::end())
.and(chain_filter.clone())
.and(warp::header::optional::<api_types::Accept>("accept"))
.and_then(
|block_id: BlockId,
chain: Arc<BeaconChain<T>>,
accept_header: Option<api_types::Accept>| {
async move {
let blob_sidecar_list = block_id.blob_sidecar_list(&chain).await?;

match accept_header {
Some(api_types::Accept::Ssz) => Response::builder()
.status(200)
.header("Content-Type", "application/octet-stream")
.body(blob_sidecar_list.as_ssz_bytes().into())
.map_err(|e| {
warp_utils::reject::custom_server_error(format!(
"failed to create response: {}",
e
))
}),
_ => Ok(warp::reply::json(&api_types::GenericResponse::from(
blob_sidecar_list,
))
.into_response()),
}
}
},
);

/*
* beacon/pool
*/
Expand Down Expand Up @@ -3498,41 +3537,6 @@ pub fn serve<T: BeaconChainTypes>(
)
});

// GET lighthouse/beacon/blobs_sidecars/{block_id}
let get_lighthouse_blobs_sidecars = warp::path("lighthouse")
.and(warp::path("beacon"))
.and(warp::path("blobs_sidecars"))
.and(block_id_or_err)
.and(warp::path::end())
.and(chain_filter.clone())
.and(warp::header::optional::<api_types::Accept>("accept"))
.and_then(
|block_id: BlockId,
chain: Arc<BeaconChain<T>>,
accept_header: Option<api_types::Accept>| {
async move {
let blobs_sidecar = block_id.blobs_sidecar(&chain).await?;

match accept_header {
Some(api_types::Accept::Ssz) => Response::builder()
.status(200)
.header("Content-Type", "application/octet-stream")
.body(blobs_sidecar.as_ssz_bytes().into())
.map_err(|e| {
warp_utils::reject::custom_server_error(format!(
"failed to create response: {}",
e
))
}),
_ => Ok(warp::reply::json(&api_types::GenericResponse::from(
blobs_sidecar,
))
.into_response()),
}
}
},
);

let get_events = eth_v1
.and(warp::path("events"))
.and(warp::path::end())
Expand Down Expand Up @@ -3627,6 +3631,7 @@ pub fn serve<T: BeaconChainTypes>(
.uor(get_beacon_block_attestations)
.uor(get_beacon_blinded_block)
.uor(get_beacon_block_root)
.uor(get_blobs)
.uor(get_beacon_pool_attestations)
.uor(get_beacon_pool_attester_slashings)
.uor(get_beacon_pool_proposer_slashings)
Expand Down Expand Up @@ -3672,7 +3677,6 @@ pub fn serve<T: BeaconChainTypes>(
.uor(get_lighthouse_attestation_performance)
.uor(get_lighthouse_block_packing_efficiency)
.uor(get_lighthouse_merge_readiness)
.uor(get_lighthouse_blobs_sidecars.boxed())
.uor(get_events)
.recover(warp_utils::reject::handle_rejection),
)
Expand Down
110 changes: 41 additions & 69 deletions beacon_node/network/src/beacon_processor/worker/rpc_methods.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@ use lighthouse_network::rpc::*;
use lighthouse_network::{PeerId, PeerRequestId, ReportSource, Response, SyncInfo};
use slog::{debug, error, warn};
use slot_clock::SlotClock;
use std::collections::{hash_map::Entry, HashMap};
use std::sync::Arc;
use task_executor::TaskExecutor;
use types::blob_sidecar::BlobIdentifier;
Expand Down Expand Up @@ -225,45 +226,36 @@ impl<T: BeaconChainTypes> Worker<T> {
executor.spawn(
async move {
let requested_blobs = request.blob_ids.len();
let send_block_count = 0;
let mut send_blob_count = 0;
let mut send_response = true;
for BlobIdentifier{ block_root: root, index: _index } in request.blob_ids.into_iter() {
match self
.chain
.get_block_and_blobs_checking_early_attester_cache(&root)
.await
{
Ok(Some(())) => {
todo!();
// //
// // TODO: HORRIBLE NSFW CODE AHEAD
// //
// let types::SignedBeaconBlockAndBlobsSidecar {beacon_block, blobs_sidecar} = block_and_blobs;
// let types::BlobsSidecar{ beacon_block_root, beacon_block_slot, blobs: blob_bundle, kzg_aggregated_proof }: types::BlobsSidecar<_> = blobs_sidecar.as_ref().clone();
// // TODO: this should be unreachable after this is addressed seriously,
// // so for now let's be ok with a panic in the expect.
// let block = beacon_block.message_eip4844().expect("We fucked up the block blob stuff");
// // Intentionally not accessing the list directly
// for (known_index, blob) in blob_bundle.into_iter().enumerate() {
// if (known_index as u64) == index {
// let blob_sidecar = types::BlobSidecar{
// block_root: beacon_block_root,
// index,
// slot: beacon_block_slot,
// block_parent_root: block.parent_root,
// proposer_index: block.proposer_index,
// blob,
// kzg_commitment: block.body.blob_kzg_commitments[known_index], // TODO: needs to be stored in a more logical way so that this won't panic.
// kzg_proof: kzg_aggregated_proof // TODO: yeah
// };
// self.send_response(
// peer_id,
// Response::BlobsByRoot(Some(Arc::new(blob_sidecar))),
// request_id,
// );
// send_block_count += 1;
// }
// }

let mut blob_list_results = HashMap::new();
for BlobIdentifier{ block_root: root, index } in request.blob_ids.into_iter() {
let blob_list_result = match blob_list_results.entry(root) {
Entry::Vacant(entry) => {
entry.insert(self
.chain
.get_blobs_checking_early_attester_cache(&root)
.await)
}
Entry::Occupied(entry) => {
entry.into_mut()
}
};

match blob_list_result.as_ref() {
Ok(Some(blobs_sidecar_list)) => {
for blob_sidecar in blobs_sidecar_list.iter() {
if blob_sidecar.index == index {
self.send_response(
peer_id,
Response::BlobsByRoot(Some(blob_sidecar.clone())),
request_id,
);
send_blob_count += 1;
break;
}
}
}
Ok(None) => {
debug!(
Expand Down Expand Up @@ -356,7 +348,7 @@ impl<T: BeaconChainTypes> Worker<T> {
"Received BlobsByRoot Request";
"peer" => %peer_id,
"requested" => requested_blobs,
"returned" => send_block_count
"returned" => send_blob_count
);

// send stream termination
Expand Down Expand Up @@ -837,36 +829,16 @@ impl<T: BeaconChainTypes> Worker<T> {
let mut send_response = true;

for root in block_roots {
match self.chain.get_blobs(&root) {
Ok(Some(_blobs)) => {
todo!();
// // TODO: more GROSS code ahead. Reader beware
// let types::BlobsSidecar {
// beacon_block_root,
// beacon_block_slot,
// blobs: blob_bundle,
// kzg_aggregated_proof: _,
// }: types::BlobsSidecar<_> = blobs;
//
// for (blob_index, blob) in blob_bundle.into_iter().enumerate() {
// let blob_sidecar = types::BlobSidecar {
// block_root: beacon_block_root,
// index: blob_index as u64,
// slot: beacon_block_slot,
// block_parent_root: Hash256::zero(),
// proposer_index: 0,
// blob,
// kzg_commitment: types::KzgCommitment::default(),
// kzg_proof: types::KzgProof::default(),
// };
//
// blobs_sent += 1;
// self.send_network_message(NetworkMessage::SendResponse {
// peer_id,
// response: Response::BlobsByRange(Some(Arc::new(blob_sidecar))),
// id: request_id,
// });
// }
match self.chain.get_blobs(&root, data_availability_boundary) {
Ok(Some(blob_sidecar_list)) => {
for blob_sidecar in blob_sidecar_list.iter() {
blobs_sent += 1;
self.send_network_message(NetworkMessage::SendResponse {
peer_id,
response: Response::BlobsByRange(Some(blob_sidecar.clone())),
id: request_id,
});
}
}
Ok(None) => {
error!(
Expand Down
Loading

0 comments on commit 2bfb0bf

Please sign in to comment.