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

Misc custody lookup improvements #5821

Merged
merged 5 commits into from
May 24, 2024
Merged

Misc custody lookup improvements #5821

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
538d56f
Improve custody requests
dapplion May 22, 2024
3ef3253
Type DataColumnsByRootRequestId
dapplion May 22, 2024
75af5ec
Prioritize peers and load balance
dapplion May 22, 2024
1b6dca7
Update tests
dapplion May 22, 2024
0d8b606
Address PR review
dapplion May 23, 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
137 changes: 79 additions & 58 deletions beacon_node/network/src/sync/block_lookups/tests.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,9 +1,7 @@
use crate::network_beacon_processor::NetworkBeaconProcessor;

use crate::service::RequestId;
use crate::sync::manager::{
DataColumnsByRootRequestId, DataColumnsByRootRequester, RequestId as SyncRequestId,
SingleLookupReqId, SyncManager,
DataColumnsByRootRequester, RequestId as SyncRequestId, SingleLookupReqId, SyncManager,
};
use crate::sync::sampling::{SamplingConfig, SamplingRequester};
use crate::sync::{SamplingId, SyncMessage};
Expand Down Expand Up @@ -94,7 +92,8 @@ const D: Duration = Duration::new(0, 0);
const PARENT_FAIL_TOLERANCE: u8 = SINGLE_BLOCK_LOOKUP_MAX_ATTEMPTS;
const SAMPLING_REQUIRED_SUCCESSES: usize = 2;

type SamplingIds = Vec<(DataColumnsByRootRequestId, ColumnIndex)>;
type DCByRootIds = Vec<DCByRootId>;
type DCByRootId = (SyncRequestId, Vec<ColumnIndex>);

struct TestRigConfig {
peer_das_enabled: bool,
Expand Down Expand Up @@ -609,18 +608,18 @@ impl TestRig {
})
}

fn return_empty_sampling_requests(&mut self, sampling_ids: SamplingIds) {
for (id, column_index) in sampling_ids {
self.log(&format!("return empty data column for {column_index}"));
fn return_empty_sampling_requests(&mut self, ids: DCByRootIds) {
for id in ids {
self.log(&format!("return empty data column for {id:?}"));
self.return_empty_sampling_request(id)
}
}

fn return_empty_sampling_request(&mut self, id: DataColumnsByRootRequestId) {
fn return_empty_sampling_request(&mut self, (request_id, _): DCByRootId) {
let peer_id = PeerId::random();
// Send stream termination
self.send_sync_message(SyncMessage::RpcDataColumn {
request_id: SyncRequestId::DataColumnsByRoot(id),
request_id,
peer_id,
data_column: None,
seen_timestamp: timestamp_now(),
Expand Down Expand Up @@ -655,26 +654,29 @@ impl TestRig {

fn complete_valid_sampling_column_requests(
&mut self,
sampling_ids: SamplingIds,
ids: DCByRootIds,
data_columns: Vec<Arc<DataColumnSidecar<E>>>,
) {
for (id, column_index) in sampling_ids {
self.log(&format!("return valid data column for {column_index}"));
self.complete_valid_sampling_column_request(
id,
data_columns[column_index as usize].clone(),
);
for id in ids {
self.log(&format!("return valid data column for {id:?}"));
let indices = &id.1;
let columns_to_send = indices
.iter()
.map(|&i| data_columns[i as usize].clone())
.collect::<Vec<_>>();
self.complete_valid_sampling_column_request(id, &columns_to_send);
}
}

fn complete_valid_sampling_column_request(
&mut self,
id: DataColumnsByRootRequestId,
data_column: Arc<DataColumnSidecar<E>>,
id: DCByRootId,
data_columns: &[Arc<DataColumnSidecar<E>>],
) {
let block_root = data_column.block_root();
let column_index = data_column.index;
self.complete_data_columns_by_root_request(id, data_column);
let first_dc = data_columns.first().unwrap();
let block_root = first_dc.block_root();
let column_index = first_dc.index;
self.complete_data_columns_by_root_request(id, data_columns);

// Expect work event
// TODO(das): worth it to append sender id to the work event for stricter assertion?
Expand All @@ -692,25 +694,29 @@ impl TestRig {

fn complete_valid_custody_request(
&mut self,
sampling_ids: SamplingIds,
ids: DCByRootIds,
data_columns: Vec<Arc<DataColumnSidecar<E>>>,
missing_components: bool,
) {
let lookup_id = if let DataColumnsByRootRequester::Custody(id) =
sampling_ids.first().unwrap().0.requester
{
id.id.0.lookup_id
} else {
panic!("not a custody requester")
};
let lookup_id =
if let SyncRequestId::DataColumnsByRoot(_, DataColumnsByRootRequester::Custody(id)) =
ids.first().unwrap().0
{
id.requester.0.lookup_id
} else {
panic!("not a custody requester")
};

let first_column = data_columns.first().cloned().unwrap();

for (id, column_index) in sampling_ids {
self.log(&format!("return valid data column for {column_index}"));

let data_column = data_columns[column_index as usize].clone();
self.complete_data_columns_by_root_request(id, data_column);
for id in ids {
self.log(&format!("return valid data column for {id:?}"));
let indices = &id.1;
let columns_to_send = indices
.iter()
.map(|&i| data_columns[i as usize].clone())
.collect::<Vec<_>>();
self.complete_data_columns_by_root_request(id, &columns_to_send);
}

// Expect work event
Expand All @@ -735,20 +741,22 @@ impl TestRig {

fn complete_data_columns_by_root_request(
&mut self,
id: DataColumnsByRootRequestId,
data_column: Arc<DataColumnSidecar<E>>,
(request_id, _): DCByRootId,
data_columns: &[Arc<DataColumnSidecar<E>>],
) {
let peer_id = PeerId::random();
// Send chunk
self.send_sync_message(SyncMessage::RpcDataColumn {
request_id: SyncRequestId::DataColumnsByRoot(id),
peer_id,
data_column: Some(data_column),
seen_timestamp: timestamp_now(),
});
for data_column in data_columns {
// Send chunks
self.send_sync_message(SyncMessage::RpcDataColumn {
request_id,
peer_id,
data_column: Some(data_column.clone()),
seen_timestamp: timestamp_now(),
});
}
// Send stream termination
self.send_sync_message(SyncMessage::RpcDataColumn {
request_id: SyncRequestId::DataColumnsByRoot(id),
request_id,
peer_id,
data_column: None,
seen_timestamp: timestamp_now(),
Expand Down Expand Up @@ -921,41 +929,54 @@ impl TestRig {
.unwrap_or_else(|e| panic!("Expected blob parent request for {for_block:?}: {e}"))
}

/// Retrieves an unknown number of requests for data columns of `block_root`. Because peer enrs
/// are random, and peer selection is random the total number of batches requests in unknown.
fn expect_data_columns_by_root_requests(
&mut self,
for_block: Hash256,
block_root: Hash256,
count: usize,
) -> SamplingIds {
(0..count)
.map(|i| {
self.pop_received_network_event(|ev| match ev {
) -> DCByRootIds {
let mut requests: DCByRootIds = vec![];
loop {
let req = self
.pop_received_network_event(|ev| match ev {
NetworkMessage::SendRequest {
peer_id: _,
request: Request::DataColumnsByRoot(request),
request_id: RequestId::Sync(SyncRequestId::DataColumnsByRoot(id)),
request_id: RequestId::Sync(id @ SyncRequestId::DataColumnsByRoot { .. }),
} if request
.data_column_ids
.to_vec()
.iter()
.any(|r| r.block_root == for_block) =>
.any(|r| r.block_root == block_root) =>
{
let index = request.data_column_ids.to_vec().first().unwrap().index;
Some((*id, index))
let indices = request
.data_column_ids
.to_vec()
.iter()
.map(|cid| cid.index)
.collect::<Vec<_>>();
Some((*id, indices))
}
_ => None,
})
.unwrap_or_else(|e| {
panic!("Expected DataColumnsByRoot request {i}/{count} for {for_block:?}: {e}")
})
})
.collect()
panic!("Expected more DataColumnsByRoot requests for {block_root:?}: {e}")
});
requests.push(req);

// Should never infinite loop because sync does not send requests for 0 columns
if requests.iter().map(|r| r.1.len()).sum::<usize>() >= count {
return requests;
}
}
}

fn expect_only_data_columns_by_root_requests(
&mut self,
for_block: Hash256,
count: usize,
) -> SamplingIds {
) -> DCByRootIds {
let ids = self.expect_data_columns_by_root_requests(for_block, count);
self.expect_empty_network();
ids
Expand Down
41 changes: 21 additions & 20 deletions beacon_node/network/src/sync/manager.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,7 +36,8 @@
use super::backfill_sync::{BackFillSync, ProcessResult, SyncStart};
use super::block_lookups::BlockLookups;
use super::network_context::{
BlockOrBlob, CustodyId, RangeRequestId, RpcEvent, SyncNetworkContext,
BlockOrBlob, CustodyId, DataColumnsByRootRequestId, RangeRequestId, RpcEvent,
SyncNetworkContext,
};
use super::peer_sync_info::{remote_sync_type, PeerSyncType};
use super::range_sync::{RangeSync, RangeSyncType, EPOCHS_PER_BATCH};
Expand Down Expand Up @@ -92,17 +93,11 @@ pub enum RequestId {
/// Request searching for a set of blobs given a hash.
SingleBlob { id: SingleLookupReqId },
/// Request searching for a set of data columns given a hash and list of column indices.
DataColumnsByRoot(DataColumnsByRootRequestId),
DataColumnsByRoot(DataColumnsByRootRequestId, DataColumnsByRootRequester),
/// Range request that is composed by both a block range request and a blob range request.
RangeBlockComponents(Id),
}

#[derive(Debug, Hash, PartialEq, Eq, Clone, Copy)]
pub struct DataColumnsByRootRequestId {
pub requester: DataColumnsByRootRequester,
pub req_id: Id,
}

#[derive(Debug, Hash, PartialEq, Eq, Clone, Copy)]
pub enum DataColumnsByRootRequester {
Sampling(SamplingId),
Expand Down Expand Up @@ -398,9 +393,12 @@ impl<T: BeaconChainTypes> SyncManager<T> {
RequestId::SingleBlob { id } => {
self.on_single_blob_response(id, peer_id, RpcEvent::RPCError(error))
}
RequestId::DataColumnsByRoot(id) => {
self.on_single_data_column_response(id, peer_id, RpcEvent::RPCError(error))
}
RequestId::DataColumnsByRoot(req_id, requester) => self.on_single_data_column_response(
req_id,
requester,
peer_id,
RpcEvent::RPCError(error),
),
RequestId::RangeBlockComponents(id) => {
if let Some(sender_id) = self.network.range_request_failed(id) {
match sender_id {
Expand Down Expand Up @@ -975,9 +973,10 @@ impl<T: BeaconChainTypes> SyncManager<T> {
RequestId::SingleBlock { .. } | RequestId::SingleBlob { .. } => {
crit!(self.log, "bad request id for data_column"; "peer_id" => %peer_id );
}
RequestId::DataColumnsByRoot(id) => {
RequestId::DataColumnsByRoot(req_id, requester) => {
self.on_single_data_column_response(
id,
req_id,
requester,
peer_id,
match data_column {
Some(data_column) => RpcEvent::Response(data_column, seen_timestamp),
Expand Down Expand Up @@ -1015,13 +1014,14 @@ impl<T: BeaconChainTypes> SyncManager<T> {

fn on_single_data_column_response(
&mut self,
id: DataColumnsByRootRequestId,
req_id: DataColumnsByRootRequestId,
requester: DataColumnsByRootRequester,
peer_id: PeerId,
data_column: RpcEvent<Arc<DataColumnSidecar<T::EthSpec>>>,
) {
if let Some((requester, resp)) =
if let Some(resp) =
self.network
.on_data_columns_by_root_response(id, peer_id, data_column)
.on_data_columns_by_root_response(req_id, peer_id, data_column)
{
match requester {
DataColumnsByRootRequester::Sampling(id) => {
Expand All @@ -1032,15 +1032,16 @@ impl<T: BeaconChainTypes> SyncManager<T> {
self.on_sampling_result(requester, result)
}
}
DataColumnsByRootRequester::Custody(id) => {
if let Some((requester, custody_columns)) =
self.network.on_custody_by_root_response(id, peer_id, resp)
DataColumnsByRootRequester::Custody(custody_id) => {
if let Some(custody_columns) = self
.network
.on_custody_by_root_response(custody_id, req_id, peer_id, resp)
{
// TODO(das): get proper timestamp
let seen_timestamp = timestamp_now();
self.block_lookups
.on_download_response::<CustodyRequestState<T::EthSpec>>(
requester.0,
custody_id.requester.0,
custody_columns.map(|(columns, peer_group)| {
(columns, peer_group, seen_timestamp)
}),
Expand Down
Loading
Loading