Skip to content

Partial processing #3

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

Merged
realbigsean merged 3 commits into from
Mar 21, 2023
Merged

Partial processing #3

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
3a40094
Revert "merge cleanuo"
realbigsean Mar 21, 2023
525448a
Revert "Revert "merge cleanuo""
realbigsean Mar 21, 2023
589c21f
fix rpc methods
realbigsean Mar 21, 2023
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
108 changes: 40 additions & 68 deletions beacon_node/network/src/beacon_processor/worker/rpc_methods.rs
View file
Open in desktop
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_blob_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_blobs_checking_early_attester_cache(&root)
.await
{
Ok(Some(_blob_sidecar_list)) => {
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 @@ -833,40 +825,20 @@ impl<T: BeaconChainTypes> Worker<T> {
// remove all skip slots
let block_roots = block_roots.into_iter().flatten().collect::<Vec<_>>();

let blobs_sent = 0;
let mut blobs_sent = 0;
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,
// });
// }
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