Restore RequestResponse::throttled.

protocols/request-response/Cargo.toml

@@ -11,13 +11,15 @@ categories = ["network-programming", "asynchronous"]
 
 [dependencies]
 async-trait = "0.1"
+bytes = "0.5.6"
 futures = "0.3.1"
 libp2p-core = { version = "0.22.0", path = "../../core" }
 libp2p-swarm = { version = "0.22.0", path = "../../swarm" }
 log = "0.4.11"
-lru = "0.6"
+minicbor = { version = "0.5", features = ["std", "derive"] }
 rand = "0.7"
 smallvec = "1.4"
+unsigned-varint = { version = "0.5", features = ["std", "futures"] }
 wasm-timer = "0.2"
 
 [dev-dependencies]

protocols/request-response/src/codec.rs

@@ -64,3 +64,4 @@ pub trait RequestResponseCodec {
     where
         T: AsyncWrite + Unpin + Send;
 }
+

protocols/request-response/src/handler.rs

@@ -47,6 +47,7 @@ use smallvec::SmallVec;
 use std::{
     collections::VecDeque,
     io,
+    sync::{atomic::{AtomicU64, Ordering}, Arc},
     time::Duration,
     task::{Context, Poll}
 };
@@ -79,9 +80,10 @@ where
     /// Inbound upgrades waiting for the incoming request.
     inbound: FuturesUnordered<BoxFuture<'static,
         Result<
-            (TCodec::Request, oneshot::Sender<TCodec::Response>),
+            ((RequestId, TCodec::Request), oneshot::Sender<TCodec::Response>),
             oneshot::Canceled
         >>>,
+    inbound_request_id: Arc<AtomicU64>
 }
 
 impl<TCodec> RequestResponseHandler<TCodec>
@@ -93,6 +95,7 @@ where
         codec: TCodec,
         keep_alive_timeout: Duration,
         substream_timeout: Duration,
+        inbound_request_id: Arc<AtomicU64>
     ) -> Self {
         Self {
             inbound_protocols,
@@ -104,6 +107,7 @@ where
             inbound: FuturesUnordered::new(),
             pending_events: VecDeque::new(),
             pending_error: None,
+            inbound_request_id
         }
     }
 }
@@ -117,6 +121,7 @@ where
 {
     /// An inbound request.
     Request {
+        request_id: RequestId,
         request: TCodec::Request,
         sender: oneshot::Sender<TCodec::Response>
     },
@@ -130,9 +135,9 @@ where
     /// An outbound request failed to negotiate a mutually supported protocol.
     OutboundUnsupportedProtocols(RequestId),
     /// An inbound request timed out.
-    InboundTimeout,
+    InboundTimeout(RequestId),
     /// An inbound request failed to negotiate a mutually supported protocol.
-    InboundUnsupportedProtocols,
+    InboundUnsupportedProtocols(RequestId),
 }
 
 impl<TCodec> ProtocolsHandler for RequestResponseHandler<TCodec>
@@ -145,7 +150,7 @@ where
     type InboundProtocol = ResponseProtocol<TCodec>;
     type OutboundProtocol = RequestProtocol<TCodec>;
     type OutboundOpenInfo = RequestId;
-    type InboundOpenInfo = ();
+    type InboundOpenInfo = RequestId;
 
     fn listen_protocol(&self) -> SubstreamProtocol<Self::InboundProtocol, Self::InboundOpenInfo> {
         // A channel for notifying the handler when the inbound
@@ -156,6 +161,8 @@ where
         // response is sent.
         let (rs_send, rs_recv) = oneshot::channel();
 
+        let request_id = RequestId(self.inbound_request_id.fetch_add(1, Ordering::Relaxed));
+
         // By keeping all I/O inside the `ResponseProtocol` and thus the
         // inbound substream upgrade via above channels, we ensure that it
         // is all subject to the configured timeout without extra bookkeeping
@@ -167,23 +174,22 @@ where
             codec: self.codec.clone(),
             request_sender: rq_send,
             response_receiver: rs_recv,
+            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());
 
-        SubstreamProtocol::new(proto, ()).with_timeout(self.substream_timeout)
+        SubstreamProtocol::new(proto, request_id).with_timeout(self.substream_timeout)
     }
 
     fn inject_fully_negotiated_inbound(
         &mut self,
         (): (),
-        (): ()
+        _: RequestId
     ) {
-        // Nothing to do, as the response has already been sent
-        // as part of the upgrade.
     }
 
     fn inject_fully_negotiated_outbound(
@@ -231,13 +237,12 @@ where
 
     fn inject_listen_upgrade_error(
         &mut self,
-        (): Self::InboundOpenInfo,
+        info: RequestId,
         error: ProtocolsHandlerUpgrErr<io::Error>
     ) {
         match error {
             ProtocolsHandlerUpgrErr::Timeout => {
-                self.pending_events.push_back(
-                    RequestResponseHandlerEvent::InboundTimeout);
+                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.
@@ -246,7 +251,7 @@ 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);
+                    RequestResponseHandlerEvent::InboundUnsupportedProtocols(info));
             }
             _ => {
                 // Anything else is considered a fatal error or misbehaviour of
@@ -282,12 +287,12 @@ where
         // Check for inbound requests.
         while let Poll::Ready(Some(result)) = self.inbound.poll_next_unpin(cx) {
             match result {
-                Ok((rq, rs_sender)) => {
+                Ok(((id, rq), rs_sender)) => {
                     // We received an inbound request.
                     self.keep_alive = KeepAlive::Yes;
                     return Poll::Ready(ProtocolsHandlerEvent::Custom(
                         RequestResponseHandlerEvent::Request {
-                            request: rq, sender: rs_sender
+                            request_id: id, request: rq, sender: rs_sender
                         }))
                 }
                 Err(oneshot::Canceled) => {

protocols/request-response/src/handler/protocol.rs

@@ -71,8 +71,10 @@ where
 {
     pub(crate) codec: TCodec,
     pub(crate) protocols: SmallVec<[TCodec::Protocol; 2]>,
-    pub(crate) request_sender: oneshot::Sender<TCodec::Request>,
-    pub(crate) response_receiver: oneshot::Receiver<TCodec::Response>
+    pub(crate) request_sender: oneshot::Sender<(RequestId, TCodec::Request)>,
+    pub(crate) response_receiver: oneshot::Receiver<TCodec::Response>,
+    pub(crate) request_id: RequestId
+
 }
 
 impl<TCodec> UpgradeInfo for ResponseProtocol<TCodec>
@@ -99,7 +101,7 @@ where
         async move {
             let read = self.codec.read_request(&protocol, &mut io);
             let request = read.await?;
-            if let Ok(()) = self.request_sender.send(request) {
+            if let Ok(()) = self.request_sender.send((self.request_id, request)) {
                 if let Ok(response) = self.response_receiver.await {
                     let write = self.codec.write_response(&protocol, &mut io, response);
                     write.await?;