fluencelabs/rust-libp2p
1
0
Fork 0
mirror of https://github.com/fluencelabs/rust-libp2p synced 2025-06-21 13:51:33 +00:00
Code Issues Projects Releases Wiki Activity

*: Format with rustfmt (#2188)

Browse Source 
Co-authored-by: Thomas Eizinger <thomas@eizinger.io>
This commit is contained in:
Max Inden
2021-08-11 13:12:12 +02:00
committed by GitHub
parent 008561283e
commit f701b24ec0
171 changed files with 10051 additions and 7193 deletions
Download Patch File Download Diff File Expand all files Collapse all files

31
protocols/request-response/src/codec.rs
View File

@ -38,30 +38,43 @@ pub trait RequestResponseCodec {
/// Reads a request from the given I/O stream according to the
/// negotiated protocol.
async fn read_request<T>(&mut self, protocol: &Self::Protocol, io: &mut T)
-> io::Result<Self::Request>
async fn read_request<T>(
&mut self,
protocol: &Self::Protocol,
io: &mut T,
) -> io::Result<Self::Request>
where
T: AsyncRead + Unpin + Send;
/// Reads a response from the given I/O stream according to the
/// negotiated protocol.
async fn read_response<T>(&mut self, protocol: &Self::Protocol, io: &mut T)
-> io::Result<Self::Response>
async fn read_response<T>(
&mut self,
protocol: &Self::Protocol,
io: &mut T,
) -> io::Result<Self::Response>
where
T: AsyncRead + Unpin + Send;
/// Writes a request to the given I/O stream according to the
/// negotiated protocol.
async fn write_request<T>(&mut self, protocol: &Self::Protocol, io: &mut T, req: Self::Request)
-> io::Result<()>
async fn write_request<T>(
&mut self,
protocol: &Self::Protocol,
io: &mut T,
req: Self::Request,
) -> io::Result<()>
where
T: AsyncWrite + Unpin + Send;
/// Writes a response to the given I/O stream according to the
/// negotiated protocol.
async fn write_response<T>(&mut self, protocol: &Self::Protocol, io: &mut T, res: Self::Response)
-> io::Result<()>
async fn write_response<T>(
&mut self,
protocol: &Self::Protocol,
io: &mut T,
res: Self::Response,
) -> io::Result<()>
where
T: AsyncWrite + Unpin + Send;
}

164
protocols/request-response/src/handler.rs
View File

@ -20,37 +20,29 @@
mod protocol;
use crate::{EMPTY_QUEUE_SHRINK_THRESHOLD, RequestId};
use crate::codec::RequestResponseCodec;
use crate::{RequestId, EMPTY_QUEUE_SHRINK_THRESHOLD};
pub use protocol::{RequestProtocol, ResponseProtocol, ProtocolSupport};
pub use protocol::{ProtocolSupport, RequestProtocol, ResponseProtocol};
use futures::{
channel::oneshot,
future::BoxFuture,
prelude::*,
stream::FuturesUnordered
};
use libp2p_core::{
upgrade::{UpgradeError, NegotiationError},
};
use futures::{channel::oneshot, future::BoxFuture, prelude::*, stream::FuturesUnordered};
use libp2p_core::upgrade::{NegotiationError, UpgradeError};
use libp2p_swarm::{
SubstreamProtocol,
protocols_handler::{
KeepAlive,
ProtocolsHandler,
ProtocolsHandlerEvent,
ProtocolsHandlerUpgrErr,
}
KeepAlive, ProtocolsHandler, ProtocolsHandlerEvent, ProtocolsHandlerUpgrErr,
},
SubstreamProtocol,
};
use smallvec::SmallVec;
use std::{
collections::VecDeque,
fmt,
io,
sync::{atomic::{AtomicU64, Ordering}, Arc},
fmt, io,
sync::{
atomic::{AtomicU64, Ordering},
Arc,
},
task::{Context, Poll},
time::Duration,
task::{Context, Poll}
};
use wasm_timer::Instant;
@ -79,12 +71,19 @@ where
/// Outbound upgrades waiting to be emitted as an `OutboundSubstreamRequest`.
outbound: VecDeque<RequestProtocol<TCodec>>,
/// Inbound upgrades waiting for the incoming request.
inbound: FuturesUnordered<BoxFuture<'static,
Result<
((RequestId, TCodec::Request), oneshot::Sender<TCodec::Response>),
oneshot::Canceled
>>>,
inbound_request_id: Arc<AtomicU64>
inbound: FuturesUnordered<
BoxFuture<
'static,
Result<
(
(RequestId, TCodec::Request),
oneshot::Sender<TCodec::Response>,
),
oneshot::Canceled,
>,
>,
>,
inbound_request_id: Arc<AtomicU64>,
}
impl<TCodec> RequestResponseHandler<TCodec>
@ -96,7 +95,7 @@ where
codec: TCodec,
keep_alive_timeout: Duration,
substream_timeout: Duration,
inbound_request_id: Arc<AtomicU64>
inbound_request_id: Arc<AtomicU64>,
) -> Self {
Self {
inbound_protocols,
@ -108,7 +107,7 @@ where
inbound: FuturesUnordered::new(),
pending_events: VecDeque::new(),
pending_error: None,
inbound_request_id
inbound_request_id,
}
}
}
@ -117,18 +116,18 @@ where
#[doc(hidden)]
pub enum RequestResponseHandlerEvent<TCodec>
where
TCodec: RequestResponseCodec
TCodec: RequestResponseCodec,
{
/// A request has been received.
Request {
request_id: RequestId,
request: TCodec::Request,
sender: oneshot::Sender<TCodec::Response>
sender: oneshot::Sender<TCodec::Response>,
},
/// A response has been received.
Response {
request_id: RequestId,
response: TCodec::Response
response: TCodec::Response,
},
/// A response to an inbound request has been sent.
ResponseSent(RequestId),
@ -150,28 +149,43 @@ where
impl<TCodec: RequestResponseCodec> fmt::Debug for RequestResponseHandlerEvent<TCodec> {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
RequestResponseHandlerEvent::Request { request_id, request: _, sender: _ } => f.debug_struct("RequestResponseHandlerEvent::Request")
RequestResponseHandlerEvent::Request {
request_id,
request: _,
sender: _,
} => f
.debug_struct("RequestResponseHandlerEvent::Request")
.field("request_id", request_id)
.finish(),
RequestResponseHandlerEvent::Response { request_id, response: _ } => f.debug_struct("RequestResponseHandlerEvent::Response")
RequestResponseHandlerEvent::Response {
request_id,
response: _,
} => f
.debug_struct("RequestResponseHandlerEvent::Response")
.field("request_id", request_id)
.finish(),
RequestResponseHandlerEvent::ResponseSent(request_id) => f.debug_tuple("RequestResponseHandlerEvent::ResponseSent")
RequestResponseHandlerEvent::ResponseSent(request_id) => f
.debug_tuple("RequestResponseHandlerEvent::ResponseSent")
.field(request_id)
.finish(),
RequestResponseHandlerEvent::ResponseOmission(request_id) => f.debug_tuple("RequestResponseHandlerEvent::ResponseOmission")
RequestResponseHandlerEvent::ResponseOmission(request_id) => f
.debug_tuple("RequestResponseHandlerEvent::ResponseOmission")
.field(request_id)
.finish(),
RequestResponseHandlerEvent::OutboundTimeout(request_id) => f.debug_tuple("RequestResponseHandlerEvent::OutboundTimeout")
RequestResponseHandlerEvent::OutboundTimeout(request_id) => f
.debug_tuple("RequestResponseHandlerEvent::OutboundTimeout")
.field(request_id)
.finish(),
RequestResponseHandlerEvent::OutboundUnsupportedProtocols(request_id) => f.debug_tuple("RequestResponseHandlerEvent::OutboundUnsupportedProtocols")
RequestResponseHandlerEvent::OutboundUnsupportedProtocols(request_id) => f
.debug_tuple("RequestResponseHandlerEvent::OutboundUnsupportedProtocols")
.field(request_id)
.finish(),
RequestResponseHandlerEvent::InboundTimeout(request_id) => f.debug_tuple("RequestResponseHandlerEvent::InboundTimeout")
RequestResponseHandlerEvent::InboundTimeout(request_id) => f
.debug_tuple("RequestResponseHandlerEvent::InboundTimeout")
.field(request_id)
.finish(),
RequestResponseHandlerEvent::InboundUnsupportedProtocols(request_id) => f.debug_tuple("RequestResponseHandlerEvent::InboundUnsupportedProtocols")
RequestResponseHandlerEvent::InboundUnsupportedProtocols(request_id) => f
.debug_tuple("RequestResponseHandlerEvent::InboundUnsupportedProtocols")
.field(request_id)
.finish(),
}
@ -212,28 +226,25 @@ where
codec: self.codec.clone(),
request_sender: rq_send,
response_receiver: rs_recv,
request_id
request_id,
};
// The handler waits for the request to come in. It then emits
// `RequestResponseHandlerEvent::Request` together with a
// `ResponseChannel`.
self.inbound.push(rq_recv.map_ok(move |rq| (rq, rs_send)).boxed());
self.inbound
.push(rq_recv.map_ok(move |rq| (rq, rs_send)).boxed());
SubstreamProtocol::new(proto, request_id).with_timeout(self.substream_timeout)
}
fn inject_fully_negotiated_inbound(
&mut self,
sent: bool,
request_id: RequestId
) {
fn inject_fully_negotiated_inbound(&mut self, sent: bool, request_id: RequestId) {
if sent {
self.pending_events.push_back(
RequestResponseHandlerEvent::ResponseSent(request_id))
self.pending_events
.push_back(RequestResponseHandlerEvent::ResponseSent(request_id))
} else {
self.pending_events.push_back(
RequestResponseHandlerEvent::ResponseOmission(request_id))
self.pending_events
.push_back(RequestResponseHandlerEvent::ResponseOmission(request_id))
}
}
@ -242,9 +253,10 @@ where
response: TCodec::Response,
request_id: RequestId,
) {
self.pending_events.push_back(
RequestResponseHandlerEvent::Response {
request_id, response
self.pending_events
.push_back(RequestResponseHandlerEvent::Response {
request_id,
response,
});
}
@ -260,8 +272,8 @@ where
) {
match error {
ProtocolsHandlerUpgrErr::Timeout => {
self.pending_events.push_back(
RequestResponseHandlerEvent::OutboundTimeout(info));
self.pending_events
.push_back(RequestResponseHandlerEvent::OutboundTimeout(info));
}
ProtocolsHandlerUpgrErr::Upgrade(UpgradeError::Select(NegotiationError::Failed)) => {
// The remote merely doesn't support the protocol(s) we requested.
@ -270,7 +282,8 @@ where
// An event is reported to permit user code to react to the fact that
// the remote peer does not support the requested protocol(s).
self.pending_events.push_back(
RequestResponseHandlerEvent::OutboundUnsupportedProtocols(info));
RequestResponseHandlerEvent::OutboundUnsupportedProtocols(info),
);
}
_ => {
// Anything else is considered a fatal error or misbehaviour of
@ -283,12 +296,12 @@ where
fn inject_listen_upgrade_error(
&mut self,
info: RequestId,
error: ProtocolsHandlerUpgrErr<io::Error>
error: ProtocolsHandlerUpgrErr<io::Error>,
) {
match error {
ProtocolsHandlerUpgrErr::Timeout => {
self.pending_events.push_back(RequestResponseHandlerEvent::InboundTimeout(info))
}
ProtocolsHandlerUpgrErr::Timeout => self
.pending_events
.push_back(RequestResponseHandlerEvent::InboundTimeout(info)),
ProtocolsHandlerUpgrErr::Upgrade(UpgradeError::Select(NegotiationError::Failed)) => {
// The local peer merely doesn't support the protocol(s) requested.
// This is no reason to close the connection, which may
@ -296,7 +309,8 @@ where
// An event is reported to permit user code to react to the fact that
// the local peer does not support the requested protocol(s).
self.pending_events.push_back(
RequestResponseHandlerEvent::InboundUnsupportedProtocols(info));
RequestResponseHandlerEvent::InboundUnsupportedProtocols(info),
);
}
_ => {
// Anything else is considered a fatal error or misbehaviour of
@ -313,18 +327,17 @@ where
fn poll(
&mut self,
cx: &mut Context<'_>,
) -> Poll<
ProtocolsHandlerEvent<RequestProtocol<TCodec>, RequestId, Self::OutEvent, Self::Error>,
> {
) -> Poll<ProtocolsHandlerEvent<RequestProtocol<TCodec>, RequestId, Self::OutEvent, Self::Error>>
{
// Check for a pending (fatal) error.
if let Some(err) = self.pending_error.take() {
// The handler will not be polled again by the `Swarm`.
return Poll::Ready(ProtocolsHandlerEvent::Close(err))
return Poll::Ready(ProtocolsHandlerEvent::Close(err));
}
// Drain pending events.
if let Some(event) = self.pending_events.pop_front() {
return Poll::Ready(ProtocolsHandlerEvent::Custom(event))
return Poll::Ready(ProtocolsHandlerEvent::Custom(event));
} else if self.pending_events.capacity() > EMPTY_QUEUE_SHRINK_THRESHOLD {
self.pending_events.shrink_to_fit();
}
@ -337,8 +350,11 @@ where
self.keep_alive = KeepAlive::Yes;
return Poll::Ready(ProtocolsHandlerEvent::Custom(
RequestResponseHandlerEvent::Request {
request_id: id, request: rq, sender: rs_sender
}))
request_id: id,
request: rq,
sender: rs_sender,
},
));
}
Err(oneshot::Canceled) => {
// The inbound upgrade has errored or timed out reading
@ -351,12 +367,10 @@ where
// Emit outbound requests.
if let Some(request) = self.outbound.pop_front() {
let info = request.request_id;
return Poll::Ready(
ProtocolsHandlerEvent::OutboundSubstreamRequest {
protocol: SubstreamProtocol::new(request, info)
.with_timeout(self.substream_timeout)
},
)
return Poll::Ready(ProtocolsHandlerEvent::OutboundSubstreamRequest {
protocol: SubstreamProtocol::new(request, info)
.with_timeout(self.substream_timeout),
});
}
debug_assert!(self.outbound.is_empty());

30
protocols/request-response/src/handler/protocol.rs
View File

@ -23,8 +23,8 @@
//! receives a request and sends a response, whereas the
//! outbound upgrade send a request and receives a response.
use crate::RequestId;
use crate::codec::RequestResponseCodec;
use crate::RequestId;
use futures::{channel::oneshot, future::BoxFuture, prelude::*};
use libp2p_core::upgrade::{InboundUpgrade, OutboundUpgrade, UpgradeInfo};
@ -40,7 +40,7 @@ pub enum ProtocolSupport {
/// The protocol is only supported for outbound requests.
Outbound,
/// The protocol is supported for inbound and outbound requests.
Full
Full,
}
impl ProtocolSupport {
@ -67,19 +67,18 @@ impl ProtocolSupport {
#[derive(Debug)]
pub struct ResponseProtocol<TCodec>
where
TCodec: RequestResponseCodec
TCodec: RequestResponseCodec,
{
pub(crate) codec: TCodec,
pub(crate) protocols: SmallVec<[TCodec::Protocol; 2]>,
pub(crate) request_sender: oneshot::Sender<(RequestId, TCodec::Request)>,
pub(crate) response_receiver: oneshot::Receiver<TCodec::Response>,
pub(crate) request_id: RequestId
pub(crate) request_id: RequestId,
}
impl<TCodec> UpgradeInfo for ResponseProtocol<TCodec>
where
TCodec: RequestResponseCodec
TCodec: RequestResponseCodec,
{
type Info = TCodec::Protocol;
type InfoIter = smallvec::IntoIter<[Self::Info; 2]>;
@ -97,7 +96,11 @@ where
type Error = io::Error;
type Future = BoxFuture<'static, Result<Self::Output, Self::Error>>;
fn upgrade_inbound(mut self, mut io: NegotiatedSubstream, protocol: Self::Info) -> Self::Future {
fn upgrade_inbound(
mut self,
mut io: NegotiatedSubstream,
protocol: Self::Info,
) -> Self::Future {
async move {
let read = self.codec.read_request(&protocol, &mut io);
let request = read.await?;
@ -129,7 +132,7 @@ where
/// Sends a request and receives a response.
pub struct RequestProtocol<TCodec>
where
TCodec: RequestResponseCodec
TCodec: RequestResponseCodec,
{
pub(crate) codec: TCodec,
pub(crate) protocols: SmallVec<[TCodec::Protocol; 2]>,
@ -150,7 +153,7 @@ where
impl<TCodec> UpgradeInfo for RequestProtocol<TCodec>
where
TCodec: RequestResponseCodec
TCodec: RequestResponseCodec,
{
type Info = TCodec::Protocol;
type InfoIter = smallvec::IntoIter<[Self::Info; 2]>;
@ -168,7 +171,11 @@ where
type Error = io::Error;
type Future = BoxFuture<'static, Result<Self::Output, Self::Error>>;
fn upgrade_outbound(mut self, mut io: NegotiatedSubstream, protocol: Self::Info) -> Self::Future {
fn upgrade_outbound(
mut self,
mut io: NegotiatedSubstream,
protocol: Self::Info,
) -> Self::Future {
async move {
let write = self.codec.write_request(&protocol, &mut io, self.request);
write.await?;
@ -176,6 +183,7 @@ where
let read = self.codec.read_response(&protocol, &mut io);
let response = read.await?;
Ok(response)
}.boxed()
}
.boxed()
}
}

342
protocols/request-response/src/lib.rs
View File

@ -60,38 +60,23 @@ pub mod codec;
pub mod handler;
pub mod throttled;
pub use codec::{RequestResponseCodec, ProtocolName};
pub use codec::{ProtocolName, RequestResponseCodec};
pub use handler::ProtocolSupport;
pub use throttled::Throttled;
use futures::{
channel::oneshot,
};
use handler::{
RequestProtocol,
RequestResponseHandler,
RequestResponseHandlerEvent,
};
use libp2p_core::{
ConnectedPoint,
Multiaddr,
PeerId,
connection::ConnectionId,
};
use futures::channel::oneshot;
use handler::{RequestProtocol, RequestResponseHandler, RequestResponseHandlerEvent};
use libp2p_core::{connection::ConnectionId, ConnectedPoint, Multiaddr, PeerId};
use libp2p_swarm::{
DialPeerCondition,
NetworkBehaviour,
NetworkBehaviourAction,
NotifyHandler,
PollParameters,
DialPeerCondition, NetworkBehaviour, NetworkBehaviourAction, NotifyHandler, PollParameters,
};
use smallvec::SmallVec;
use std::{
collections::{HashMap, HashSet, VecDeque},
fmt,
time::Duration,
sync::{atomic::AtomicU64, Arc},
task::{Context, Poll}
task::{Context, Poll},
time::Duration,
};
/// An inbound request or response.
@ -117,7 +102,7 @@ pub enum RequestResponseMessage<TRequest, TResponse, TChannelResponse = TRespons
/// See [`RequestResponse::send_request`].
request_id: RequestId,
/// The response message.
response: TResponse
response: TResponse,
},
}
@ -129,7 +114,7 @@ pub enum RequestResponseEvent<TRequest, TResponse, TChannelResponse = TResponse>
/// The peer who sent the message.
peer: PeerId,
/// The incoming message.
message: RequestResponseMessage<TRequest, TResponse, TChannelResponse>
message: RequestResponseMessage<TRequest, TResponse, TChannelResponse>,
},
/// An outbound request failed.
OutboundFailure {
@ -186,8 +171,12 @@ impl fmt::Display for OutboundFailure {
match self {
OutboundFailure::DialFailure => write!(f, "Failed to dial the requested peer"),
OutboundFailure::Timeout => write!(f, "Timeout while waiting for a response"),
OutboundFailure::ConnectionClosed => write!(f, "Connection was closed before a response was received"),
OutboundFailure::UnsupportedProtocols => write!(f, "The remote supports none of the requested protocols")
OutboundFailure::ConnectionClosed => {
write!(f, "Connection was closed before a response was received")
}
OutboundFailure::UnsupportedProtocols => {
write!(f, "The remote supports none of the requested protocols")
}
}
}
}
@ -217,10 +206,20 @@ pub enum InboundFailure {
impl fmt::Display for InboundFailure {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
InboundFailure::Timeout => write!(f, "Timeout while receiving request or sending response"),
InboundFailure::ConnectionClosed => write!(f, "Connection was closed before a response could be sent"),
InboundFailure::UnsupportedProtocols => write!(f, "The local peer supports none of the protocols requested by the remote"),
InboundFailure::ResponseOmission => write!(f, "The response channel was dropped without sending a response to the remote")
InboundFailure::Timeout => {
write!(f, "Timeout while receiving request or sending response")
}
InboundFailure::ConnectionClosed => {
write!(f, "Connection was closed before a response could be sent")
}
InboundFailure::UnsupportedProtocols => write!(
f,
"The local peer supports none of the protocols requested by the remote"
),
InboundFailure::ResponseOmission => write!(
f,
"The response channel was dropped without sending a response to the remote"
),
}
}
}
@ -322,7 +321,9 @@ where
pending_events: VecDeque<
NetworkBehaviourAction<
RequestProtocol<TCodec>,
RequestResponseEvent<TCodec::Request, TCodec::Response>>>,
RequestResponseEvent<TCodec::Request, TCodec::Response>,
>,
>,
/// The currently connected peers, their pending outbound and inbound responses and their known,
/// reachable addresses, if any.
connected: HashMap<PeerId, SmallVec<[Connection; 2]>>,
@ -341,7 +342,7 @@ where
/// protocols, codec and configuration.
pub fn new<I>(codec: TCodec, protocols: I, cfg: RequestResponseConfig) -> Self
where
I: IntoIterator<Item = (TCodec::Protocol, ProtocolSupport)>
I: IntoIterator<Item = (TCodec::Protocol, ProtocolSupport)>,
{
let mut inbound_protocols = SmallVec::new();
let mut outbound_protocols = SmallVec::new();
@ -375,7 +376,7 @@ where
where
I: IntoIterator<Item = (TCodec::Protocol, ProtocolSupport)>,
TCodec: Send,
TCodec::Protocol: Sync
TCodec::Protocol: Sync,
{
Throttled::new(c, protos, cfg)
}
@ -402,11 +403,15 @@ where
};
if let Some(request) = self.try_send_request(peer, request) {
self.pending_events.push_back(NetworkBehaviourAction::DialPeer {
peer_id: *peer,
condition: DialPeerCondition::Disconnected,
});
self.pending_outbound_requests.entry(*peer).or_default().push(request);
self.pending_events
.push_back(NetworkBehaviourAction::DialPeer {
peer_id: *peer,
condition: DialPeerCondition::Disconnected,
});
self.pending_outbound_requests
.entry(*peer)
.or_default()
.push(request);
}
request_id
@ -423,9 +428,11 @@ where
///
/// The provided `ResponseChannel` is obtained from an inbound
/// [`RequestResponseMessage::Request`].
pub fn send_response(&mut self, ch: ResponseChannel<TCodec::Response>, rs: TCodec::Response)
-> Result<(), TCodec::Response>
{
pub fn send_response(
&mut self,
ch: ResponseChannel<TCodec::Response>,
rs: TCodec::Response,
) -> Result<(), TCodec::Response> {
ch.sender.send(rs)
}
@ -464,12 +471,19 @@ where
/// pending, i.e. waiting for a response.
pub fn is_pending_outbound(&self, peer: &PeerId, request_id: &RequestId) -> bool {
// Check if request is already sent on established connection.
let est_conn = self.connected.get(peer)
.map(|cs| cs.iter().any(|c| c.pending_inbound_responses.contains(request_id)))
let est_conn = self
.connected
.get(peer)
.map(|cs| {
cs.iter()
.any(|c| c.pending_inbound_responses.contains(request_id))
})
.unwrap_or(false);
// Check if request is still pending to be sent.
let pen_conn = self.pending_outbound_requests.get(peer)
.map(|rps| rps.iter().any(|rp| {rp.request_id == *request_id}))
let pen_conn = self
.pending_outbound_requests
.get(peer)
.map(|rps| rps.iter().any(|rp| rp.request_id == *request_id))
.unwrap_or(false);
est_conn || pen_conn
@ -479,8 +493,12 @@ where
/// [`PeerId`] is still pending, i.e. waiting for a response by the local
/// node through [`RequestResponse::send_response`].
pub fn is_pending_inbound(&self, peer: &PeerId, request_id: &RequestId) -> bool {
self.connected.get(peer)
.map(|cs| cs.iter().any(|c| c.pending_outbound_responses.contains(request_id)))
self.connected
.get(peer)
.map(|cs| {
cs.iter()
.any(|c| c.pending_outbound_responses.contains(request_id))
})
.unwrap_or(false)
}
@ -494,21 +512,24 @@ where
/// Tries to send a request by queueing an appropriate event to be
/// emitted to the `Swarm`. If the peer is not currently connected,
/// the given request is return unchanged.
fn try_send_request(&mut self, peer: &PeerId, request: RequestProtocol<TCodec>)
-> Option<RequestProtocol<TCodec>>
{
fn try_send_request(
&mut self,
peer: &PeerId,
request: RequestProtocol<TCodec>,
) -> Option<RequestProtocol<TCodec>> {
if let Some(connections) = self.connected.get_mut(peer) {
if connections.is_empty() {
return Some(request)
return Some(request);
}
let ix = (request.request_id.0 as usize) % connections.len();
let conn = &mut connections[ix];
conn.pending_inbound_responses.insert(request.request_id);
self.pending_events.push_back(NetworkBehaviourAction::NotifyHandler {
peer_id: *peer,
handler: NotifyHandler::One(conn.id),
event: request
});
self.pending_events
.push_back(NetworkBehaviourAction::NotifyHandler {
peer_id: *peer,
handler: NotifyHandler::One(conn.id),
event: request,
});
None
} else {
Some(request)
@ -554,9 +575,9 @@ where
peer: &PeerId,
connection: ConnectionId,
) -> Option<&mut Connection> {
self.connected.get_mut(peer).and_then(|connections| {
connections.iter_mut().find(|c| c.id == connection)
})
self.connected
.get_mut(peer)
.and_then(|connections| connections.iter_mut().find(|c| c.id == connection))
}
}
@ -573,7 +594,7 @@ where
self.codec.clone(),
self.config.connection_keep_alive,
self.config.request_timeout,
self.next_inbound_id.clone()
self.next_inbound_id.clone(),
)
}
@ -597,21 +618,35 @@ where
}
}
fn inject_connection_established(&mut self, peer: &PeerId, conn: &ConnectionId, endpoint: &ConnectedPoint) {
fn inject_connection_established(
&mut self,
peer: &PeerId,
conn: &ConnectionId,
endpoint: &ConnectedPoint,
) {
let address = match endpoint {
ConnectedPoint::Dialer { address } => Some(address.clone()),
ConnectedPoint::Listener { .. } => None
ConnectedPoint::Listener { .. } => None,
};
self.connected.entry(*peer)
self.connected
.entry(*peer)
.or_default()
.push(Connection::new(*conn, address));
}
fn inject_connection_closed(&mut self, peer_id: &PeerId, conn: &ConnectionId, _: &ConnectedPoint) {
let connections = self.connected.get_mut(peer_id)
fn inject_connection_closed(
&mut self,
peer_id: &PeerId,
conn: &ConnectionId,
_: &ConnectedPoint,
) {
let connections = self
.connected
.get_mut(peer_id)
.expect("Expected some established connection to peer before closing.");
let connection = connections.iter()
let connection = connections
.iter()
.position(|c| &c.id == conn)
.map(|p: usize| connections.remove(p))
.expect("Expected connection to be established before closing.");
@ -621,24 +656,25 @@ where
}
for request_id in connection.pending_outbound_responses {
self.pending_events.push_back(NetworkBehaviourAction::GenerateEvent(
RequestResponseEvent::InboundFailure {
peer: *peer_id,
request_id,
error: InboundFailure::ConnectionClosed
}
));
self.pending_events
.push_back(NetworkBehaviourAction::GenerateEvent(
RequestResponseEvent::InboundFailure {
peer: *peer_id,
request_id,
error: InboundFailure::ConnectionClosed,
},
));
}
for request_id in connection.pending_inbound_responses {
self.pending_events.push_back(NetworkBehaviourAction::GenerateEvent(
RequestResponseEvent::OutboundFailure {
peer: *peer_id,
request_id,
error: OutboundFailure::ConnectionClosed
}
));
self.pending_events
.push_back(NetworkBehaviourAction::GenerateEvent(
RequestResponseEvent::OutboundFailure {
peer: *peer_id,
request_id,
error: OutboundFailure::ConnectionClosed,
},
));
}
}
@ -655,13 +691,14 @@ where
// another, concurrent dialing attempt ongoing.
if let Some(pending) = self.pending_outbound_requests.remove(peer) {
for request in pending {
self.pending_events.push_back(NetworkBehaviourAction::GenerateEvent(
RequestResponseEvent::OutboundFailure {
peer: *peer,
request_id: request.request_id,
error: OutboundFailure::DialFailure
}
));
self.pending_events
.push_back(NetworkBehaviourAction::GenerateEvent(
RequestResponseEvent::OutboundFailure {
peer: *peer,
request_id: request.request_id,
error: OutboundFailure::DialFailure,
},
));
}
}
}
@ -673,49 +710,74 @@ where
event: RequestResponseHandlerEvent<TCodec>,
) {
match event {
RequestResponseHandlerEvent::Response { request_id, response } => {
RequestResponseHandlerEvent::Response {
request_id,
response,
} => {
let removed = self.remove_pending_inbound_response(&peer, connection, &request_id);
debug_assert!(
removed,
"Expect request_id to be pending before receiving response.",
);
let message = RequestResponseMessage::Response { request_id, response };
self.pending_events.push_back(
NetworkBehaviourAction::GenerateEvent(
RequestResponseEvent::Message { peer, message }));
let message = RequestResponseMessage::Response {
request_id,
response,
};
self.pending_events
.push_back(NetworkBehaviourAction::GenerateEvent(
RequestResponseEvent::Message { peer, message },
));
}
RequestResponseHandlerEvent::Request { request_id, request, sender } => {
let channel = ResponseChannel { request_id, peer, sender };
let message = RequestResponseMessage::Request { request_id, request, channel };
self.pending_events.push_back(NetworkBehaviourAction::GenerateEvent(
RequestResponseEvent::Message { peer, message }
));
RequestResponseHandlerEvent::Request {
request_id,
request,
sender,
} => {
let channel = ResponseChannel {
request_id,
peer,
sender,
};
let message = RequestResponseMessage::Request {
request_id,
request,
channel,
};
self.pending_events
.push_back(NetworkBehaviourAction::GenerateEvent(
RequestResponseEvent::Message { peer, message },
));
match self.get_connection_mut(&peer, connection) {
Some(connection) => {
let inserted = connection.pending_outbound_responses.insert(request_id);
debug_assert!(inserted, "Expect id of new request to be unknown.");
},
}
// Connection closed after `RequestResponseEvent::Request` has been emitted.
None => {
self.pending_events.push_back(NetworkBehaviourAction::GenerateEvent(
RequestResponseEvent::InboundFailure {
peer,
request_id,
error: InboundFailure::ConnectionClosed
}
));
self.pending_events
.push_back(NetworkBehaviourAction::GenerateEvent(
RequestResponseEvent::InboundFailure {
peer,
request_id,
error: InboundFailure::ConnectionClosed,
},
));
}
}
}
RequestResponseHandlerEvent::ResponseSent(request_id) => {
let removed = self.remove_pending_outbound_response(&peer, connection, request_id);
debug_assert!(removed, "Expect request_id to be pending before response is sent.");
debug_assert!(
removed,
"Expect request_id to be pending before response is sent."
);
self.pending_events.push_back(
NetworkBehaviourAction::GenerateEvent(
RequestResponseEvent::ResponseSent { peer, request_id }));
self.pending_events
.push_back(NetworkBehaviourAction::GenerateEvent(
RequestResponseEvent::ResponseSent { peer, request_id },
));
}
RequestResponseHandlerEvent::ResponseOmission(request_id) => {
let removed = self.remove_pending_outbound_response(&peer, connection, request_id);
@ -724,25 +786,30 @@ where
"Expect request_id to be pending before response is omitted.",
);
self.pending_events.push_back(
NetworkBehaviourAction::GenerateEvent(
self.pending_events
.push_back(NetworkBehaviourAction::GenerateEvent(
RequestResponseEvent::InboundFailure {
peer,
request_id,
error: InboundFailure::ResponseOmission
}));
error: InboundFailure::ResponseOmission,
},
));
}
RequestResponseHandlerEvent::OutboundTimeout(request_id) => {
let removed = self.remove_pending_inbound_response(&peer, connection, &request_id);
debug_assert!(removed, "Expect request_id to be pending before request times out.");
debug_assert!(
removed,
"Expect request_id to be pending before request times out."
);
self.pending_events.push_back(
NetworkBehaviourAction::GenerateEvent(
self.pending_events
.push_back(NetworkBehaviourAction::GenerateEvent(
RequestResponseEvent::OutboundFailure {
peer,
request_id,
error: OutboundFailure::Timeout,
}));
},
));
}
RequestResponseHandlerEvent::InboundTimeout(request_id) => {
// Note: `RequestResponseHandlerEvent::InboundTimeout` is emitted both for timing
@ -751,13 +818,14 @@ where
// not assert the request_id to be present before removing it.
self.remove_pending_outbound_response(&peer, connection, request_id);
self.pending_events.push_back(
NetworkBehaviourAction::GenerateEvent(
self.pending_events
.push_back(NetworkBehaviourAction::GenerateEvent(
RequestResponseEvent::InboundFailure {
peer,
request_id,
error: InboundFailure::Timeout,
}));
},
));
}
RequestResponseHandlerEvent::OutboundUnsupportedProtocols(request_id) => {
let removed = self.remove_pending_inbound_response(&peer, connection, &request_id);
@ -766,35 +834,41 @@ where
"Expect request_id to be pending before failing to connect.",
);
self.pending_events.push_back(
NetworkBehaviourAction::GenerateEvent(
self.pending_events
.push_back(NetworkBehaviourAction::GenerateEvent(
RequestResponseEvent::OutboundFailure {
peer,
request_id,
error: OutboundFailure::UnsupportedProtocols,
}));
},
));
}
RequestResponseHandlerEvent::InboundUnsupportedProtocols(request_id) => {
// Note: No need to call `self.remove_pending_outbound_response`,
// `RequestResponseHandlerEvent::Request` was never emitted for this request and
// thus request was never added to `pending_outbound_responses`.
self.pending_events.push_back(
NetworkBehaviourAction::GenerateEvent(
self.pending_events
.push_back(NetworkBehaviourAction::GenerateEvent(
RequestResponseEvent::InboundFailure {
peer,
request_id,
error: InboundFailure::UnsupportedProtocols,
}));
},
));
}
}
}
fn poll(&mut self, _: &mut Context<'_>, _: &mut impl PollParameters)
-> Poll<NetworkBehaviourAction<
fn poll(
&mut self,
_: &mut Context<'_>,
_: &mut impl PollParameters,
) -> Poll<
NetworkBehaviourAction<
RequestProtocol<TCodec>,
RequestResponseEvent<TCodec::Request, TCodec::Response>
>>
{
RequestResponseEvent<TCodec::Request, TCodec::Response>,
>,
> {
if let Some(ev) = self.pending_events.pop_front() {
return Poll::Ready(ev);
} else if self.pending_events.capacity() > EMPTY_QUEUE_SHRINK_THRESHOLD {
@ -821,7 +895,7 @@ struct Connection {
pending_outbound_responses: HashSet<RequestId>,
/// Pending inbound responses for previously sent requests on this
/// connection.
pending_inbound_responses: HashSet<RequestId>
pending_inbound_responses: HashSet<RequestId>,
}
impl Connection {

448
protocols/request-response/src/throttled.rs
View File

@ -36,22 +36,20 @@
mod codec;
use codec::{Codec, Message, ProtocolWrapper, Type};
use super::{
ProtocolSupport, RequestId, RequestResponse, RequestResponseCodec, RequestResponseConfig,
RequestResponseEvent, RequestResponseMessage,
};
use crate::handler::{RequestProtocol, RequestResponseHandler, RequestResponseHandlerEvent};
use codec::{Codec, Message, ProtocolWrapper, Type};
use futures::ready;
use libp2p_core::{ConnectedPoint, connection::ConnectionId, Multiaddr, PeerId};
use libp2p_core::{connection::ConnectionId, ConnectedPoint, Multiaddr, PeerId};
use libp2p_swarm::{NetworkBehaviour, NetworkBehaviourAction, PollParameters};
use lru::LruCache;
use std::{collections::{HashMap, HashSet, VecDeque}, task::{Context, Poll}};
use std::{cmp::max, num::NonZeroU16};
use super::{
ProtocolSupport,
RequestId,
RequestResponse,
RequestResponseCodec,
RequestResponseConfig,
RequestResponseEvent,
RequestResponseMessage,
use std::{
collections::{HashMap, HashSet, VecDeque},
task::{Context, Poll},
};
pub type ResponseChannel<R> = super::ResponseChannel<Message<R>>;
@ -60,7 +58,7 @@ pub type ResponseChannel<R> = super::ResponseChannel<Message<R>>;
pub struct Throttled<C>
where
C: RequestResponseCodec + Send,
C::Protocol: Sync
C::Protocol: Sync,
{
/// A random id used for logging.
id: u32,
@ -77,7 +75,7 @@ where
/// Pending events to report in `Throttled::poll`.
events: VecDeque<Event<C::Request, C::Response, Message<C::Response>>>,
/// The current credit ID.
next_grant_id: u64
next_grant_id: u64,
}
/// Information about a credit grant that is sent to remote peers.
@ -89,7 +87,7 @@ struct Grant {
request: RequestId,
/// The credit given in this grant, i.e. the number of additional
/// requests the remote is allowed to send.
credit: u16
credit: u16,
}
/// Max. number of inbound requests that can be received.
@ -99,7 +97,7 @@ struct Limit {
max_recv: NonZeroU16,
/// The next receive limit which becomes active after
/// the current limit has been reached.
next_max: NonZeroU16
next_max: NonZeroU16,
}
impl Limit {
@ -111,7 +109,7 @@ impl Limit {
// sender so we must not use `max` right away.
Limit {
max_recv: NonZeroU16::new(1).expect("1 > 0"),
next_max: max
next_max: max,
}
}
@ -191,7 +189,7 @@ impl PeerInfo {
limit: recv_limit,
remaining: 1,
sent: HashSet::new(),
}
},
}
}
@ -210,16 +208,18 @@ impl PeerInfo {
impl<C> Throttled<C>
where
C: RequestResponseCodec + Send + Clone,
C::Protocol: Sync
C::Protocol: Sync,
{
/// Create a new throttled request-response behaviour.
pub fn new<I>(c: C, protos: I, cfg: RequestResponseConfig) -> Self
where
I: IntoIterator<Item = (C::Protocol, ProtocolSupport)>,
C: Send,
C::Protocol: Sync
C::Protocol: Sync,
{
let protos = protos.into_iter().map(|(p, ps)| (ProtocolWrapper::new(b"/t/1", p), ps));
let protos = protos
.into_iter()
.map(|(p, ps)| (ProtocolWrapper::new(b"/t/1", p), ps));
Throttled::from(RequestResponse::new(Codec::new(c, 8192), protos, cfg))
}
@ -233,7 +233,7 @@ where
default_limit: Limit::new(NonZeroU16::new(1).expect("1 > 0")),
limit_overrides: HashMap::new(),
events: VecDeque::new(),
next_grant_id: 0
next_grant_id: 0,
}
}
@ -262,7 +262,10 @@ where
/// Has the limit of outbound requests been reached for the given peer?
pub fn can_send(&mut self, p: &PeerId) -> bool {
self.peer_info.get(p).map(|i| i.send_budget.remaining > 0).unwrap_or(true)
self.peer_info
.get(p)
.map(|i| i.send_budget.remaining > 0)
.unwrap_or(true)
}
/// Send a request to a peer.
@ -273,22 +276,30 @@ where
pub fn send_request(&mut self, p: &PeerId, req: C::Request) -> Result<RequestId, C::Request> {
let connected = &mut self.peer_info;
let disconnected = &mut self.offline_peer_info;
let remaining =
if let Some(info) = connected.get_mut(p).or_else(|| disconnected.get_mut(p)) {
if info.send_budget.remaining == 0 {
log::trace!("{:08x}: no more budget to send another request to {}", self.id, p);
return Err(req)
}
info.send_budget.remaining -= 1;
info.send_budget.remaining
} else {
let limit = self.limit_overrides.get(p).copied().unwrap_or(self.default_limit);
let mut info = PeerInfo::new(limit);
info.send_budget.remaining -= 1;
let remaining = info.send_budget.remaining;
self.offline_peer_info.put(*p, info);
remaining
};
let remaining = if let Some(info) = connected.get_mut(p).or_else(|| disconnected.get_mut(p))
{
if info.send_budget.remaining == 0 {
log::trace!(
"{:08x}: no more budget to send another request to {}",
self.id,
p
);
return Err(req);
}
info.send_budget.remaining -= 1;
info.send_budget.remaining
} else {
let limit = self
.limit_overrides
.get(p)
.copied()
.unwrap_or(self.default_limit);
let mut info = PeerInfo::new(limit);
info.send_budget.remaining -= 1;
let remaining = info.send_budget.remaining;
self.offline_peer_info.put(*p, info);
remaining
};
let rid = self.behaviour.send_request(p, Message::request(req));
@ -305,12 +316,20 @@ where
/// Answer an inbound request with a response.
///
/// See [`RequestResponse::send_response`] for details.
pub fn send_response(&mut self, ch: ResponseChannel<C::Response>, res: C::Response)
-> Result<(), C::Response>
{
log::trace!("{:08x}: sending response {} to peer {}", self.id, ch.request_id(), &ch.peer);
pub fn send_response(
&mut self,
ch: ResponseChannel<C::Response>,
res: C::Response,
) -> Result<(), C::Response> {
log::trace!(
"{:08x}: sending response {} to peer {}",
self.id,
ch.request_id(),
&ch.peer
);
if let Some(info) = self.peer_info.get_mut(&ch.peer) {
if info.recv_budget.remaining == 0 { // need to send more credit to the remote peer
if info.recv_budget.remaining == 0 {
// need to send more credit to the remote peer
let crd = info.recv_budget.limit.switch();
info.recv_budget.remaining = info.recv_budget.limit.max_recv.get();
self.send_credit(&ch.peer, crd);
@ -350,7 +369,6 @@ where
self.behaviour.is_pending_outbound(p, r)
}
/// Is the remote waiting for the local node to respond to the given
/// request?
///
@ -365,8 +383,18 @@ where
let cid = self.next_grant_id;
self.next_grant_id += 1;
let rid = self.behaviour.send_request(p, Message::credit(credit, cid));
log::trace!("{:08x}: sending {} credit as grant {} to {}", self.id, credit, cid, p);
let grant = Grant { id: cid, request: rid, credit };
log::trace!(
"{:08x}: sending {} credit as grant {} to {}",
self.id,
credit,
cid,
p
);
let grant = Grant {
id: cid,
request: rid,
credit,
};
info.recv_budget.grant = Some(grant);
info.recv_budget.sent.insert(rid);
}
@ -383,13 +411,13 @@ pub enum Event<Req, Res, CRes = Res> {
/// When previously reaching the send limit of a peer,
/// this event is eventually emitted when sending is
/// allowed to resume.
ResumeSending(PeerId)
ResumeSending(PeerId),
}
impl<C> NetworkBehaviour for Throttled<C>
where
C: RequestResponseCodec + Send + Clone + 'static,
C::Protocol: Sync
C::Protocol: Sync,
{
type ProtocolsHandler = RequestResponseHandler<Codec<C>>;
type OutEvent = Event<C::Request, C::Response, Message<C::Response>>;
@ -402,7 +430,12 @@ where
self.behaviour.addresses_of_peer(p)
}
fn inject_connection_established(&mut self, p: &PeerId, id: &ConnectionId, end: &ConnectedPoint) {
fn inject_connection_established(
&mut self,
p: &PeerId,
id: &ConnectionId,
end: &ConnectedPoint,
) {
self.behaviour.inject_connection_established(p, id, end)
}
@ -433,7 +466,11 @@ where
self.send_credit(p, recv_budget - 1);
}
} else {
let limit = self.limit_overrides.get(p).copied().unwrap_or(self.default_limit);
let limit = self
.limit_overrides
.get(p)
.copied()
.unwrap_or(self.default_limit);
self.peer_info.insert(*p, PeerInfo::new(limit));
}
}
@ -451,142 +488,183 @@ where
self.behaviour.inject_dial_failure(p)
}
fn inject_event(&mut self, p: PeerId, i: ConnectionId, e: RequestResponseHandlerEvent<Codec<C>>) {
fn inject_event(
&mut self,
p: PeerId,
i: ConnectionId,
e: RequestResponseHandlerEvent<Codec<C>>,
) {
self.behaviour.inject_event(p, i, e)
}
fn poll(&mut self, cx: &mut Context<'_>, params: &mut impl PollParameters)
-> Poll<NetworkBehaviourAction<RequestProtocol<Codec<C>>, Self::OutEvent>>
{
fn poll(
&mut self,
cx: &mut Context<'_>,
params: &mut impl PollParameters,
) -> Poll<NetworkBehaviourAction<RequestProtocol<Codec<C>>, Self::OutEvent>> {
loop {
if let Some(ev) = self.events.pop_front() {
return Poll::Ready(NetworkBehaviourAction::GenerateEvent(ev))
return Poll::Ready(NetworkBehaviourAction::GenerateEvent(ev));
} else if self.events.capacity() > super::EMPTY_QUEUE_SHRINK_THRESHOLD {
self.events.shrink_to_fit()
}
let event = match ready!(self.behaviour.poll(cx, params)) {
| NetworkBehaviourAction::GenerateEvent(RequestResponseEvent::Message { peer, message }) => {
NetworkBehaviourAction::GenerateEvent(RequestResponseEvent::Message {
peer,
message,
}) => {
let message = match message {
| RequestResponseMessage::Response { request_id, response } =>
match &response.header().typ {
| Some(Type::Ack) => {
if let Some(info) = self.peer_info.get_mut(&peer) {
if let Some(id) = info.recv_budget.grant.as_ref().map(|c| c.id) {
if Some(id) == response.header().ident {
log::trace!("{:08x}: received ack {} from {}", self.id, id, peer);
info.recv_budget.grant = None;
}
RequestResponseMessage::Response {
request_id,
response,
} => match &response.header().typ {
Some(Type::Ack) => {
if let Some(info) = self.peer_info.get_mut(&peer) {
if let Some(id) = info.recv_budget.grant.as_ref().map(|c| c.id)
{
if Some(id) == response.header().ident {
log::trace!(
"{:08x}: received ack {} from {}",
self.id,
id,
peer
);
info.recv_budget.grant = None;
}
info.recv_budget.sent.remove(&request_id);
}
continue
info.recv_budget.sent.remove(&request_id);
}
| Some(Type::Response) => {
log::trace!("{:08x}: received response {} from {}", self.id, request_id, peer);
if let Some(rs) = response.into_parts().1 {
RequestResponseMessage::Response { request_id, response: rs }
} else {
log::error! { "{:08x}: missing data for response {} from peer {}",
self.id,
request_id,
peer
}
continue
continue;
}
Some(Type::Response) => {
log::trace!(
"{:08x}: received response {} from {}",
self.id,
request_id,
peer
);
if let Some(rs) = response.into_parts().1 {
RequestResponseMessage::Response {
request_id,
response: rs,
}
}
| ty => {
log::trace! {
"{:08x}: unknown message type: {:?} from {}; expected response or credit",
} else {
log::error! { "{:08x}: missing data for response {} from peer {}",
self.id,
ty,
request_id,
peer
};
continue
}
continue;
}
}
| RequestResponseMessage::Request { request_id, request, channel } =>
match &request.header().typ {
| Some(Type::Credit) => {
if let Some(info) = self.peer_info.get_mut(&peer) {
let id = if let Some(n) = request.header().ident {
n
} else {
log::warn! { "{:08x}: missing credit id in message from {}",
ty => {
log::trace! {
"{:08x}: unknown message type: {:?} from {}; expected response or credit",
self.id,
ty,
peer
};
continue;
}
},
RequestResponseMessage::Request {
request_id,
request,
channel,
} => match &request.header().typ {
Some(Type::Credit) => {
if let Some(info) = self.peer_info.get_mut(&peer) {
let id = if let Some(n) = request.header().ident {
n
} else {
log::warn! { "{:08x}: missing credit id in message from {}",
self.id,
peer
}
continue;
};
let credit = request.header().credit.unwrap_or(0);
log::trace! { "{:08x}: received {} additional credit {} from {}",
self.id,
credit,
id,
peer
};
if info.send_budget.grant < Some(id) {
if info.send_budget.remaining == 0 && credit > 0 {
log::trace!(
"{:08x}: sending to peer {} can resume",
self.id,
peer
}
continue
};
let credit = request.header().credit.unwrap_or(0);
log::trace! { "{:08x}: received {} additional credit {} from {}",
self.id,
credit,
id,
peer
};
if info.send_budget.grant < Some(id) {
if info.send_budget.remaining == 0 && credit > 0 {
log::trace!("{:08x}: sending to peer {} can resume", self.id, peer);
self.events.push_back(Event::ResumeSending(peer))
}
info.send_budget.remaining += credit;
info.send_budget.grant = Some(id);
);
self.events.push_back(Event::ResumeSending(peer))
}
// Note: Failing to send a response to a credit grant is
// handled along with other inbound failures further below.
let _ = self.behaviour.send_response(channel, Message::ack(id));
info.send_budget.received.insert(request_id);
info.send_budget.remaining += credit;
info.send_budget.grant = Some(id);
}
continue
// Note: Failing to send a response to a credit grant is
// handled along with other inbound failures further below.
let _ = self.behaviour.send_response(channel, Message::ack(id));
info.send_budget.received.insert(request_id);
}
| Some(Type::Request) => {
if let Some(info) = self.peer_info.get_mut(&peer) {
log::trace! { "{:08x}: received request {} (recv. budget = {})",
self.id,
request_id,
info.recv_budget.remaining
};
if info.recv_budget.remaining == 0 {
log::debug!("{:08x}: peer {} exceeds its budget", self.id, peer);
self.events.push_back(Event::TooManyInboundRequests(peer));
continue
}
info.recv_budget.remaining -= 1;
// We consider a request as proof that our credit grant has
// reached the peer. Usually, an ACK has already been
// received.
info.recv_budget.grant = None;
}
if let Some(rq) = request.into_parts().1 {
RequestResponseMessage::Request { request_id, request: rq, channel }
} else {
log::error! { "{:08x}: missing data for request {} from peer {}",
self.id,
request_id,
peer
}
continue
}
}
| ty => {
log::trace! {
"{:08x}: unknown message type: {:?} from {}; expected request or ack",
continue;
}
Some(Type::Request) => {
if let Some(info) = self.peer_info.get_mut(&peer) {
log::trace! { "{:08x}: received request {} (recv. budget = {})",
self.id,
ty,
peer
request_id,
info.recv_budget.remaining
};
continue
if info.recv_budget.remaining == 0 {
log::debug!(
"{:08x}: peer {} exceeds its budget",
self.id,
peer
);
self.events.push_back(Event::TooManyInboundRequests(peer));
continue;
}
info.recv_budget.remaining -= 1;
// We consider a request as proof that our credit grant has
// reached the peer. Usually, an ACK has already been
// received.
info.recv_budget.grant = None;
}
if let Some(rq) = request.into_parts().1 {
RequestResponseMessage::Request {
request_id,
request: rq,
channel,
}
} else {
log::error! { "{:08x}: missing data for request {} from peer {}",
self.id,
request_id,
peer
}
continue;
}
}
ty => {
log::trace! {
"{:08x}: unknown message type: {:?} from {}; expected request or ack",
self.id,
ty,
peer
};
continue;
}
},
};
let event = RequestResponseEvent::Message { peer, message };
NetworkBehaviourAction::GenerateEvent(Event::Event(event))
}
| NetworkBehaviourAction::GenerateEvent(RequestResponseEvent::OutboundFailure {
NetworkBehaviourAction::GenerateEvent(RequestResponseEvent::OutboundFailure {
peer,
request_id,
error
error,
}) => {
if let Some(info) = self.peer_info.get_mut(&peer) {
if let Some(grant) = info.recv_budget.grant.as_mut() {
@ -606,16 +684,20 @@ where
// If the outbound failure was for a credit message, don't report it on
// the public API and retry the sending.
if info.recv_budget.sent.remove(&request_id) {
continue
continue;
}
}
let event = RequestResponseEvent::OutboundFailure { peer, request_id, error };
let event = RequestResponseEvent::OutboundFailure {
peer,
request_id,
error,
};
NetworkBehaviourAction::GenerateEvent(Event::Event(event))
}
| NetworkBehaviourAction::GenerateEvent(RequestResponseEvent::InboundFailure {
NetworkBehaviourAction::GenerateEvent(RequestResponseEvent::InboundFailure {
peer,
request_id,
error
error,
}) => {
// If the inbound failure occurred in the context of responding to a
// credit grant, don't report it on the public API.
@ -625,15 +707,19 @@ where
"{:08}: failed to acknowledge credit grant from {}: {:?}",
self.id, peer, error
};
continue
continue;
}
}
let event = RequestResponseEvent::InboundFailure { peer, request_id, error };
let event = RequestResponseEvent::InboundFailure {
peer,
request_id,
error,
};
NetworkBehaviourAction::GenerateEvent(Event::Event(event))
}
| NetworkBehaviourAction::GenerateEvent(RequestResponseEvent::ResponseSent {
NetworkBehaviourAction::GenerateEvent(RequestResponseEvent::ResponseSent {
peer,
request_id
request_id,
}) => {
// If this event is for an ACK response that was sent for
// the last received credit grant, skip it.
@ -644,25 +730,41 @@ where
self.id,
info.send_budget.grant,
}
continue
continue;
}
}
NetworkBehaviourAction::GenerateEvent(Event::Event(
RequestResponseEvent::ResponseSent { peer, request_id }))
RequestResponseEvent::ResponseSent { peer, request_id },
))
}
| NetworkBehaviourAction::DialAddress { address } =>
NetworkBehaviourAction::DialAddress { address },
| NetworkBehaviourAction::DialPeer { peer_id, condition } =>
NetworkBehaviourAction::DialPeer { peer_id, condition },
| NetworkBehaviourAction::NotifyHandler { peer_id, handler, event } =>
NetworkBehaviourAction::NotifyHandler { peer_id, handler, event },
| NetworkBehaviourAction::ReportObservedAddr { address, score } =>
NetworkBehaviourAction::ReportObservedAddr { address, score },
| NetworkBehaviourAction::CloseConnection { peer_id, connection } =>
NetworkBehaviourAction::CloseConnection { peer_id, connection }
NetworkBehaviourAction::DialAddress { address } => {
NetworkBehaviourAction::DialAddress { address }
}
NetworkBehaviourAction::DialPeer { peer_id, condition } => {
NetworkBehaviourAction::DialPeer { peer_id, condition }
}
NetworkBehaviourAction::NotifyHandler {
peer_id,
handler,
event,
} => NetworkBehaviourAction::NotifyHandler {
peer_id,
handler,
event,
},
NetworkBehaviourAction::ReportObservedAddr { address, score } => {
NetworkBehaviourAction::ReportObservedAddr { address, score }
}
NetworkBehaviourAction::CloseConnection {
peer_id,
connection,
} => NetworkBehaviourAction::CloseConnection {
peer_id,
connection,
},
};
return Poll::Ready(event)
return Poll::Ready(event);
}
}
}

123
protocols/request-response/src/throttled/codec.rs
View File

@ -18,13 +18,13 @@
// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
// DEALINGS IN THE SOFTWARE.
use super::RequestResponseCodec;
use async_trait::async_trait;
use bytes::{Bytes, BytesMut};
use futures::prelude::*;
use libp2p_core::ProtocolName;
use minicbor::{Encode, Decode};
use minicbor::{Decode, Encode};
use std::io;
use super::RequestResponseCodec;
use unsigned_varint::{aio, io::ReadError};
/// A protocol header.
@ -32,27 +32,34 @@ use unsigned_varint::{aio, io::ReadError};
#[cbor(map)]
pub struct Header {
/// The type of message.
#[n(0)] pub typ: Option<Type>,
#[n(0)]
pub typ: Option<Type>,
/// The number of additional requests the remote is willing to receive.
#[n(1)] pub credit: Option<u16>,
#[n(1)]
pub credit: Option<u16>,
/// An identifier used for sending credit grants.
#[n(2)] pub ident: Option<u64>
#[n(2)]
pub ident: Option<u64>,
}
/// A protocol message type.
#[derive(Debug, Clone, PartialEq, Eq, Encode, Decode)]
pub enum Type {
#[n(0)] Request,
#[n(1)] Response,
#[n(2)] Credit,
#[n(3)] Ack
#[n(0)]
Request,
#[n(1)]
Response,
#[n(2)]
Credit,
#[n(3)]
Ack,
}
/// A protocol message consisting of header and data.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Message<T> {
header: Header,
data: Option<T>
data: Option<T>,
}
impl<T> Message<T> {
@ -63,26 +70,40 @@ impl<T> Message<T> {
/// Create a request message.
pub fn request(data: T) -> Self {
let mut m = Message::new(Header { typ: Some(Type::Request), .. Header::default() });
let mut m = Message::new(Header {
typ: Some(Type::Request),
..Header::default()
});
m.data = Some(data);
m
}
/// Create a response message.
pub fn response(data: T) -> Self {
let mut m = Message::new(Header { typ: Some(Type::Response), .. Header::default() });
let mut m = Message::new(Header {
typ: Some(Type::Response),
..Header::default()
});
m.data = Some(data);
m
}
/// Create a credit grant.
pub fn credit(credit: u16, ident: u64) -> Self {
Message::new(Header { typ: Some(Type::Credit), credit: Some(credit), ident: Some(ident) })
Message::new(Header {
typ: Some(Type::Credit),
credit: Some(credit),
ident: Some(ident),
})
}
/// Create an acknowledge message.
pub fn ack(ident: u64) -> Self {
Message::new(Header { typ: Some(Type::Ack), credit: None, ident: Some(ident) })
Message::new(Header {
typ: Some(Type::Ack),
credit: None,
ident: Some(ident),
})
}
/// Access the message header.
@ -130,28 +151,34 @@ pub struct Codec<C> {
/// Encoding/decoding buffer.
buffer: Vec<u8>,
/// Max. header length.
max_header_len: u32
max_header_len: u32,
}
impl<C> Codec<C> {
/// Create a codec by wrapping an existing one.
pub fn new(c: C, max_header_len: u32) -> Self {
Codec { inner: c, buffer: Vec::new(), max_header_len }
Codec {
inner: c,
buffer: Vec::new(),
max_header_len,
}
}
/// Read and decode a request header.
async fn read_header<T, H>(&mut self, io: &mut T) -> io::Result<H>
where
T: AsyncRead + Unpin + Send,
H: for<'a> minicbor::Decode<'a>
H: for<'a> minicbor::Decode<'a>,
{
let header_len = aio::read_u32(&mut *io).await
.map_err(|e| match e {
ReadError::Io(e) => e,
other => io::Error::new(io::ErrorKind::Other, other)
})?;
let header_len = aio::read_u32(&mut *io).await.map_err(|e| match e {
ReadError::Io(e) => e,
other => io::Error::new(io::ErrorKind::Other, other),
})?;
if header_len > self.max_header_len {
return Err(io::Error::new(io::ErrorKind::InvalidData, "header too large to read"))
return Err(io::Error::new(
io::ErrorKind::InvalidData,
"header too large to read",
));
}
self.buffer.resize(u32_to_usize(header_len), 0u8);
io.read_exact(&mut self.buffer).await?;
@ -162,12 +189,16 @@ impl<C> Codec<C> {
async fn write_header<T, H>(&mut self, hdr: &H, io: &mut T) -> io::Result<()>
where
T: AsyncWrite + Unpin + Send,
H: minicbor::Encode
H: minicbor::Encode,
{
self.buffer.clear();
minicbor::encode(hdr, &mut self.buffer).map_err(|e| io::Error::new(io::ErrorKind::Other, e))?;
minicbor::encode(hdr, &mut self.buffer)
.map_err(|e| io::Error::new(io::ErrorKind::Other, e))?;
if self.buffer.len() > u32_to_usize(self.max_header_len) {
return Err(io::Error::new(io::ErrorKind::InvalidData, "header too large to write"))
return Err(io::Error::new(
io::ErrorKind::InvalidData,
"header too large to write",
));
}
let mut b = unsigned_varint::encode::u32_buffer();
let header_len = unsigned_varint::encode::u32(self.buffer.len() as u32, &mut b);
@ -180,7 +211,7 @@ impl<C> Codec<C> {
impl<C> RequestResponseCodec for Codec<C>
where
C: RequestResponseCodec + Send,
C::Protocol: Sync
C::Protocol: Sync,
{
type Protocol = ProtocolWrapper<C::Protocol>;
type Request = Message<C::Request>;
@ -188,7 +219,7 @@ where
async fn read_request<T>(&mut self, p: &Self::Protocol, io: &mut T) -> io::Result<Self::Request>
where
T: AsyncRead + Unpin + Send
T: AsyncRead + Unpin + Send,
{
let mut msg = Message::new(self.read_header(io).await?);
match msg.header.typ {
@ -198,15 +229,22 @@ where
}
Some(Type::Credit) => Ok(msg),
Some(Type::Response) | Some(Type::Ack) | None => {
log::debug!("unexpected {:?} when expecting request or credit grant", msg.header.typ);
log::debug!(
"unexpected {:?} when expecting request or credit grant",
msg.header.typ
);
Err(io::ErrorKind::InvalidData.into())
}
}
}
async fn read_response<T>(&mut self, p: &Self::Protocol, io: &mut T) -> io::Result<Self::Response>
async fn read_response<T>(
&mut self,
p: &Self::Protocol,
io: &mut T,
) -> io::Result<Self::Response>
where
T: AsyncRead + Unpin + Send
T: AsyncRead + Unpin + Send,
{
let mut msg = Message::new(self.read_header(io).await?);
match msg.header.typ {
@ -216,15 +254,23 @@ where
}
Some(Type::Ack) => Ok(msg),
Some(Type::Request) | Some(Type::Credit) | None => {
log::debug!("unexpected {:?} when expecting response or ack", msg.header.typ);
log::debug!(
"unexpected {:?} when expecting response or ack",
msg.header.typ
);
Err(io::ErrorKind::InvalidData.into())
}
}
}
async fn write_request<T>(&mut self, p: &Self::Protocol, io: &mut T, r: Self::Request) -> io::Result<()>
async fn write_request<T>(
&mut self,
p: &Self::Protocol,
io: &mut T,
r: Self::Request,
) -> io::Result<()>
where
T: AsyncWrite + Unpin + Send
T: AsyncWrite + Unpin + Send,
{
self.write_header(&r.header, io).await?;
if let Some(data) = r.data {
@ -233,9 +279,14 @@ where
Ok(())
}
async fn write_response<T>(&mut self, p: &Self::Protocol, io: &mut T, r: Self::Response) -> io::Result<()>
async fn write_response<T>(
&mut self,
p: &Self::Protocol,
io: &mut T,
r: Self::Response,
) -> io::Result<()>
where
T: AsyncWrite + Unpin + Send
T: AsyncWrite + Unpin + Send,
{
self.write_header(&r.header, io).await?;
if let Some(data) = r.data {