feat: better blocksize heuristics (#3748)

crates/net/downloaders/src/bodies/bodies.rs

@@ -19,6 +19,7 @@ use reth_tasks::{TaskSpawner, TokioTaskExecutor};
 use std::{
     cmp::Ordering,
     collections::BinaryHeap,
+    mem,
     ops::RangeInclusive,
     pin::Pin,
     sync::Arc,
@@ -225,13 +226,16 @@ where
         self.metrics.buffered_responses.decrement(1.);
         self.buffered_blocks_size_bytes -= resp.size();
         self.metrics.buffered_blocks.decrement(resp.len() as f64);
-        self.metrics.buffered_blocks_size_bytes.set(resp.size() as f64);
+        self.metrics.buffered_blocks_size_bytes.set(self.buffered_blocks_size_bytes as f64);
         Some(resp)
     }
 
     /// Adds a new response to the internal buffer
     fn buffer_bodies_response(&mut self, response: Vec<BlockResponse>) {
-        let size = response.iter().map(|b| b.size()).sum::<usize>();
+        // take into account capacity
+        let size = response.iter().map(BlockResponse::size).sum::<usize>() +
+            response.capacity() * mem::size_of::<BlockResponse>();
+
         let response = OrderedBodiesResponse { resp: response, size };
         let response_len = response.len();
 
@@ -516,7 +520,7 @@ impl Default for BodiesDownloaderBuilder {
         Self {
             request_limit: 200,
             stream_batch_size: 10_000,
-            max_buffered_blocks_size_bytes: 4 * 1024 * 1024 * 1024, // ~4GB
+            max_buffered_blocks_size_bytes: 2 * 1024 * 1024 * 1024, // ~2GB
             concurrent_requests_range: 5..=100,
         }
     }

crates/net/downloaders/src/bodies/request.rs

@@ -1,4 +1,4 @@
-use crate::metrics::BodyDownloaderMetrics;
+use crate::metrics::{BodyDownloaderMetrics, ResponseMetrics};
 use futures::{Future, FutureExt};
 use reth_interfaces::{
     consensus::{Consensus as ConsensusTrait, Consensus},
@@ -11,6 +11,7 @@ use reth_interfaces::{
 use reth_primitives::{BlockBody, PeerId, SealedBlock, SealedHeader, WithPeerId, H256};
 use std::{
     collections::VecDeque,
+    mem,
     pin::Pin,
     sync::Arc,
     task::{ready, Context, Poll},
@@ -39,6 +40,9 @@ pub(crate) struct BodiesRequestFuture<B: BodiesClient> {
     client: Arc<B>,
     consensus: Arc<dyn Consensus>,
     metrics: BodyDownloaderMetrics,
+    /// Metrics for individual responses. This can be used to observe how the size (in bytes) of
+    /// responses change while bodies are being downloaded.
+    response_metrics: ResponseMetrics,
     // Headers to download. The collection is shrunk as responses are buffered.
     pending_headers: VecDeque<SealedHeader>,
     /// Internal buffer for all blocks
@@ -62,6 +66,7 @@ where
             client,
             consensus,
             metrics,
+            response_metrics: Default::default(),
             pending_headers: Default::default(),
             buffer: Default::default(),
             last_request_len: None,
@@ -153,8 +158,11 @@ where
     /// This method removes headers from the internal collection.
     /// If the response fails validation, then the header will be put back.
     fn try_buffer_blocks(&mut self, bodies: Vec<BlockBody>) -> DownloadResult<()> {
+        let bodies_capacity = bodies.capacity();
+        let bodies_len = bodies.len();
         let mut bodies = bodies.into_iter().peekable();
 
+        let mut total_size = bodies_capacity * mem::size_of::<BlockBody>();
         while bodies.peek().is_some() {
             let next_header = match self.pending_headers.pop_front() {
                 Some(header) => header,
@@ -162,15 +170,16 @@ where
             };
 
             if next_header.is_empty() {
+                // increment empty block body metric
+                total_size += mem::size_of::<BlockBody>();
                 self.buffer.push(BlockResponse::Empty(next_header));
             } else {
                 let next_body = bodies.next().unwrap();
-                let block = SealedBlock {
-                    header: next_header,
-                    body: next_body.transactions,
-                    ommers: next_body.ommers,
-                    withdrawals: next_body.withdrawals,
-                };
+
+                // increment full block body metric
+                total_size += next_body.size();
+
+                let block = SealedBlock::new(next_header, next_body);
 
                 if let Err(error) = self.consensus.validate_block(&block) {
                     // Body is invalid, put the header back and return an error
@@ -183,6 +192,10 @@ where
             }
         }
 
+        // Increment per-response metric
+        self.response_metrics.response_size_bytes.set(total_size as f64);
+        self.response_metrics.response_length.set(bodies_len as f64);
+
         Ok(())
     }
 }

crates/net/downloaders/src/metrics.rs

@@ -62,6 +62,19 @@ impl BodyDownloaderMetrics {
     }
 }
 
+/// Metrics for an individual response, i.e. the size in bytes, and length (number of bodies) in the
+/// response.
+///
+/// These metrics will be initialized with the `downloaders.bodies.response` scope.
+#[derive(Clone, Metrics)]
+#[metrics(scope = "downloaders.bodies.response")]
+pub struct ResponseMetrics {
+    /// The size (in bytes) of an individual bodies response received by the downloader.
+    pub response_size_bytes: Gauge,
+    /// The number of bodies in an individual bodies response received by the downloader.
+    pub response_length: Gauge,
+}
+
 /// Common header downloader metrics.
 ///
 /// These metrics will be initialized with the `downloaders.headers` scope.