perf: send followup request with high priority (#2659)

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

@@ -12,7 +12,6 @@ use reth_interfaces::{
             response::BlockResponse,
         },
         error::{DownloadError, DownloadResult},
-        priority::Priority,
     },
 };
 use reth_primitives::{BlockNumber, SealedHeader};
@@ -388,7 +387,6 @@ where
                             Arc::clone(&this.client),
                             Arc::clone(&this.consensus),
                             request,
-                            Priority::Normal,
                         );
                         new_request_submitted = true;
                     }

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

@@ -7,7 +7,6 @@ use reth_interfaces::{
     p2p::{
         bodies::{client::BodiesClient, response::BlockResponse},
         error::DownloadResult,
-        priority::Priority,
     },
 };
 use reth_primitives::{BlockNumber, SealedHeader};
@@ -61,7 +60,6 @@ where
         client: Arc<B>,
         consensus: Arc<dyn Consensus>,
         request: Vec<SealedHeader>,
-        priority: Priority,
     ) {
         // Set last max requested block number
         self.last_requested_block_number = request
@@ -73,8 +71,7 @@ where
             .or(self.last_requested_block_number);
         // Create request and push into the queue.
         self.inner.push(
-            BodiesRequestFuture::new(client, consensus, priority, self.metrics.clone())
-                .with_headers(request),
+            BodiesRequestFuture::new(client, consensus, self.metrics.clone()).with_headers(request),
         )
     }
 }

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

@@ -39,7 +39,6 @@ pub(crate) struct BodiesRequestFuture<B: BodiesClient> {
     client: Arc<B>,
     consensus: Arc<dyn Consensus>,
     metrics: DownloaderMetrics,
-    priority: Priority,
     // Headers to download. The collection is shrunk as responses are buffered.
     pending_headers: VecDeque<SealedHeader>,
     /// Internal buffer for all blocks
@@ -57,14 +56,12 @@ where
     pub(crate) fn new(
         client: Arc<B>,
         consensus: Arc<dyn Consensus>,
-        priority: Priority,
         metrics: DownloaderMetrics,
     ) -> Self {
         Self {
             client,
             consensus,
             metrics,
-            priority,
             pending_headers: Default::default(),
             buffer: Default::default(),
             last_request_len: None,
@@ -78,7 +75,7 @@ where
         // Submit the request only if there are any headers to download.
         // Otherwise, the future will immediately be resolved.
         if let Some(req) = self.next_request() {
-            self.submit_request(req);
+            self.submit_request(req, Priority::Normal);
         }
         self
     }
@@ -89,7 +86,10 @@ where
         if let Some(peer_id) = peer_id {
             self.client.report_bad_message(peer_id);
         }
-        self.submit_request(self.next_request().expect("existing hashes to resubmit"));
+        self.submit_request(
+            self.next_request().expect("existing hashes to resubmit"),
+            Priority::High,
+        );
     }
 
     /// Retrieve header hashes for the next request.
@@ -99,12 +99,12 @@ where
         hashes.peek().is_some().then(|| hashes.collect())
     }
 
-    /// Submit the request.
-    fn submit_request(&mut self, req: Vec<H256>) {
+    /// Submit the request with the given priority.
+    fn submit_request(&mut self, req: Vec<H256>, priority: Priority) {
         tracing::trace!(target: "downloaders::bodies", request_len = req.len(), "Requesting bodies");
         let client = Arc::clone(&self.client);
         self.last_request_len = Some(req.len());
-        self.fut = Some(client.get_block_bodies_with_priority(req, self.priority));
+        self.fut = Some(client.get_block_bodies_with_priority(req, priority));
     }
 
     /// Process block response.
@@ -139,7 +139,7 @@ where
 
         // Submit next request if any
         if let Some(req) = self.next_request() {
-            self.submit_request(req);
+            self.submit_request(req, Priority::High);
         } else {
             self.fut = None;
         }
@@ -253,7 +253,6 @@ mod tests {
         let fut = BodiesRequestFuture::new(
             client.clone(),
             Arc::new(TestConsensus::default()),
-            Priority::Normal,
             DownloaderMetrics::new(TEST_SCOPE),
         )
         .with_headers(headers.clone());
@@ -278,7 +277,6 @@ mod tests {
         let fut = BodiesRequestFuture::new(
             client.clone(),
             Arc::new(TestConsensus::default()),
-            Priority::Normal,
             DownloaderMetrics::new(TEST_SCOPE),
         )
         .with_headers(headers.clone());