use RpcSend instead of BehaviourAction on Rpc::self_limiter::ready_requests

Author: jxs

beacon_node/lighthouse_network/src/rpc/mod.rs

@@ -231,14 +231,14 @@ impl<Id: ReqId, E: EthSpec> RPC<Id, E> {
                 }
             }
         } else {
-            ToSwarm::NotifyHandler {
-                peer_id,
-                handler: NotifyHandler::Any,
-                event: RPCSend::Request(request_id, req),
-            }
+            RPCSend::Request(request_id, req)
         };
 
-        self.events.push(event);
+        self.events.push(BehaviourAction::NotifyHandler {
+            peer_id,
+            handler: NotifyHandler::Any,
+            event,
+        });
     }
 
     /// Lighthouse wishes to disconnect from this peer by sending a Goodbye message. This

beacon_node/lighthouse_network/src/rpc/self_limiter.rs

@@ -11,10 +11,12 @@ use smallvec::SmallVec;
 use tokio_util::time::DelayQueue;
 use types::EthSpec;
 
+use crate::service::BehaviourEvent;
+
 use super::{
     config::OutboundRateLimiterConfig,
     rate_limiter::{RPCRateLimiter as RateLimiter, RateLimitedErr},
-    BehaviourAction, Protocol, RPCSend, ReqId, RequestType,
+    BehaviourAction, Protocol, RPCMessage, RPCSend, ReqId, RequestType,
 };
 
 /// A request that was rate limited or waiting on rate limited requests for the same peer and
@@ -34,7 +36,7 @@ pub(crate) struct SelfRateLimiter<Id: ReqId, E: EthSpec> {
     /// Rate limiter for our own requests.
     limiter: RateLimiter,
     /// Requests that are ready to be sent.
-    ready_requests: SmallVec<[BehaviourAction<Id, E>; 3]>,
+    ready_requests: SmallVec<[(PeerId, RPCSend<Id, E>); 3]>,
     /// Slog logger.
     log: Logger,
 }
@@ -71,7 +73,7 @@ impl<Id: ReqId, E: EthSpec> SelfRateLimiter<Id, E> {
         peer_id: PeerId,
         request_id: Id,
         req: RequestType<E>,
-    ) -> Result<BehaviourAction<Id, E>, Error> {
+    ) -> Result<RPCSend<Id, E>, Error> {
         let protocol = req.versioned_protocol().protocol();
         // First check that there are not already other requests waiting to be sent.
         if let Some(queued_requests) = self.delayed_requests.get_mut(&(peer_id, protocol)) {
@@ -103,13 +105,9 @@ impl<Id: ReqId, E: EthSpec> SelfRateLimiter<Id, E> {
         request_id: Id,
         req: RequestType<E>,
         log: &Logger,
-    ) -> Result<BehaviourAction<Id, E>, (QueuedRequest<Id, E>, Duration)> {
+    ) -> Result<RPCSend<Id, E>, (QueuedRequest<Id, E>, Duration)> {
         match limiter.allows(&peer_id, &req) {
-            Ok(()) => Ok(BehaviourAction::NotifyHandler {
-                peer_id,
-                handler: NotifyHandler::Any,
-                event: RPCSend::Request(request_id, req),
-            }),
+            Ok(()) => Ok(RPCSend::Request(request_id, req)),
             Err(e) => {
                 let protocol = req.versioned_protocol();
                 match e {
@@ -121,11 +119,7 @@ impl<Id: ReqId, E: EthSpec> SelfRateLimiter<Id, E> {
                             "Self rate limiting error for a batch that will never fit. Sending request anyway. Check configuration parameters.";
                             "protocol" => %req.versioned_protocol().protocol()
                         );
-                        Ok(BehaviourAction::NotifyHandler {
-                            peer_id,
-                            handler: NotifyHandler::Any,
-                            event: RPCSend::Request(request_id, req),
-                        })
+                        Ok(RPCSend::Request(request_id, req))
                     }
                     RateLimitedErr::TooSoon(wait_time) => {
                         debug!(log, "Self rate limiting"; "protocol" => %protocol.protocol(), "wait_time_ms" => wait_time.as_millis(), "peer_id" => %peer_id);
@@ -151,7 +145,7 @@ impl<Id: ReqId, E: EthSpec> SelfRateLimiter<Id, E> {
                         // If one fails just wait for the next window that allows sending requests.
                         return;
                     }
-                    Ok(event) => self.ready_requests.push(event),
+                    Ok(event) => self.ready_requests.push((peer_id, event)),
                 }
             }
             if queued_requests.is_empty() {
@@ -198,8 +192,12 @@ impl<Id: ReqId, E: EthSpec> SelfRateLimiter<Id, E> {
         let _ = self.limiter.poll_unpin(cx);
 
         // Finally return any queued events.
-        if !self.ready_requests.is_empty() {
-            return Poll::Ready(self.ready_requests.remove(0));
+        if let Some((peer_id, event)) = self.ready_requests.pop() {
+            return Poll::Ready(BehaviourAction::NotifyHandler {
+                peer_id,
+                handler: NotifyHandler::Any,
+                event,
+            });
         }
 
         Poll::Pending