mirror of
https://github.com/sigp/lighthouse.git
synced 2026-03-02 16:21:42 +00:00
Use RpcSend on RPC::self_limiter::ready_requests (#6634)
We don't need to store `BehaviourAction` for `ready_requests` and therefore avoid having an `unreachable!` on #6625. Therefore this PR should be merged before it
This commit is contained in:
@@ -232,14 +232,14 @@ impl<Id: ReqId, E: EthSpec> RPC<Id, E> {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
} else {
|
} else {
|
||||||
ToSwarm::NotifyHandler {
|
RPCSend::Request(request_id, req)
|
||||||
peer_id,
|
|
||||||
handler: NotifyHandler::Any,
|
|
||||||
event: 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
|
/// Lighthouse wishes to disconnect from this peer by sending a Goodbye message. This
|
||||||
|
|||||||
@@ -35,7 +35,7 @@ pub(crate) struct SelfRateLimiter<Id: ReqId, E: EthSpec> {
|
|||||||
/// Rate limiter for our own requests.
|
/// Rate limiter for our own requests.
|
||||||
limiter: RateLimiter,
|
limiter: RateLimiter,
|
||||||
/// Requests that are ready to be sent.
|
/// Requests that are ready to be sent.
|
||||||
ready_requests: SmallVec<[BehaviourAction<Id, E>; 3]>,
|
ready_requests: SmallVec<[(PeerId, RPCSend<Id, E>); 3]>,
|
||||||
/// Slog logger.
|
/// Slog logger.
|
||||||
log: Logger,
|
log: Logger,
|
||||||
}
|
}
|
||||||
@@ -76,7 +76,7 @@ impl<Id: ReqId, E: EthSpec> SelfRateLimiter<Id, E> {
|
|||||||
peer_id: PeerId,
|
peer_id: PeerId,
|
||||||
request_id: Id,
|
request_id: Id,
|
||||||
req: RequestType<E>,
|
req: RequestType<E>,
|
||||||
) -> Result<BehaviourAction<Id, E>, Error> {
|
) -> Result<RPCSend<Id, E>, Error> {
|
||||||
let protocol = req.versioned_protocol().protocol();
|
let protocol = req.versioned_protocol().protocol();
|
||||||
// First check that there are not already other requests waiting to be sent.
|
// 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)) {
|
if let Some(queued_requests) = self.delayed_requests.get_mut(&(peer_id, protocol)) {
|
||||||
@@ -108,13 +108,9 @@ impl<Id: ReqId, E: EthSpec> SelfRateLimiter<Id, E> {
|
|||||||
request_id: Id,
|
request_id: Id,
|
||||||
req: RequestType<E>,
|
req: RequestType<E>,
|
||||||
log: &Logger,
|
log: &Logger,
|
||||||
) -> Result<BehaviourAction<Id, E>, (QueuedRequest<Id, E>, Duration)> {
|
) -> Result<RPCSend<Id, E>, (QueuedRequest<Id, E>, Duration)> {
|
||||||
match limiter.allows(&peer_id, &req) {
|
match limiter.allows(&peer_id, &req) {
|
||||||
Ok(()) => Ok(BehaviourAction::NotifyHandler {
|
Ok(()) => Ok(RPCSend::Request(request_id, req)),
|
||||||
peer_id,
|
|
||||||
handler: NotifyHandler::Any,
|
|
||||||
event: RPCSend::Request(request_id, req),
|
|
||||||
}),
|
|
||||||
Err(e) => {
|
Err(e) => {
|
||||||
let protocol = req.versioned_protocol();
|
let protocol = req.versioned_protocol();
|
||||||
match e {
|
match e {
|
||||||
@@ -126,11 +122,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.";
|
"Self rate limiting error for a batch that will never fit. Sending request anyway. Check configuration parameters.";
|
||||||
"protocol" => %req.versioned_protocol().protocol()
|
"protocol" => %req.versioned_protocol().protocol()
|
||||||
);
|
);
|
||||||
Ok(BehaviourAction::NotifyHandler {
|
Ok(RPCSend::Request(request_id, req))
|
||||||
peer_id,
|
|
||||||
handler: NotifyHandler::Any,
|
|
||||||
event: RPCSend::Request(request_id, req),
|
|
||||||
})
|
|
||||||
}
|
}
|
||||||
RateLimitedErr::TooSoon(wait_time) => {
|
RateLimitedErr::TooSoon(wait_time) => {
|
||||||
debug!(log, "Self rate limiting"; "protocol" => %protocol.protocol(), "wait_time_ms" => wait_time.as_millis(), "peer_id" => %peer_id);
|
debug!(log, "Self rate limiting"; "protocol" => %protocol.protocol(), "wait_time_ms" => wait_time.as_millis(), "peer_id" => %peer_id);
|
||||||
@@ -156,7 +148,7 @@ impl<Id: ReqId, E: EthSpec> SelfRateLimiter<Id, E> {
|
|||||||
// If one fails just wait for the next window that allows sending requests.
|
// If one fails just wait for the next window that allows sending requests.
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
Ok(event) => self.ready_requests.push(event),
|
Ok(event) => self.ready_requests.push((peer_id, event)),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
if queued_requests.is_empty() {
|
if queued_requests.is_empty() {
|
||||||
@@ -203,8 +195,12 @@ impl<Id: ReqId, E: EthSpec> SelfRateLimiter<Id, E> {
|
|||||||
let _ = self.limiter.poll_unpin(cx);
|
let _ = self.limiter.poll_unpin(cx);
|
||||||
|
|
||||||
// Finally return any queued events.
|
// Finally return any queued events.
|
||||||
if !self.ready_requests.is_empty() {
|
if let Some((peer_id, event)) = self.ready_requests.pop() {
|
||||||
return Poll::Ready(self.ready_requests.remove(0));
|
return Poll::Ready(BehaviourAction::NotifyHandler {
|
||||||
|
peer_id,
|
||||||
|
handler: NotifyHandler::Any,
|
||||||
|
event,
|
||||||
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
Poll::Pending
|
Poll::Pending
|
||||||
|
|||||||
Reference in New Issue
Block a user