refactor(request-response): don't use upgrade infrastructure

Cargo.toml

@@ -17,6 +17,7 @@ members = [
     "interop-tests",
     "misc/allow-block-list",
     "misc/connection-limits",
+    "misc/futures-bounded",
     "misc/keygen",
     "misc/metrics",
     "misc/multistream-select",
@@ -62,6 +63,7 @@ resolver = "2"
 rust-version = "1.65.0"
 
 [workspace.dependencies]
+futures-bounded = { version = "0.1.0", path = "misc/futures-bounded" }
 libp2p-allow-block-list = { version = "0.2.0", path = "misc/allow-block-list" }
 libp2p-autonat = { version = "0.11.0", path = "protocols/autonat" }
 libp2p-connection-limits = { version = "0.2.1", path = "misc/connection-limits" }
@@ -98,13 +100,12 @@ libp2p-webrtc = { version = "0.6.0-alpha", path = "transports/webrtc" }
 libp2p-websocket = { version = "0.42.0", path = "transports/websocket" }
 libp2p-webtransport-websys = { version = "0.1.0", path = "transports/webtransport-websys" }
 libp2p-yamux = { version = "0.44.1", path = "muxers/yamux" }
+multiaddr = "0.18.0"
+multihash = "0.19.0"
 multistream-select = { version = "0.13.0", path = "misc/multistream-select" }
 quick-protobuf-codec = { version = "0.2.0", path = "misc/quick-protobuf-codec" }
 quickcheck = { package = "quickcheck-ext", path = "misc/quickcheck-ext" }
 rw-stream-sink = { version = "0.4.0", path = "misc/rw-stream-sink" }
-multiaddr = "0.18.0"
-multihash = "0.19.0"
-
 
 [patch.crates-io]
 

misc/futures-bounded/CHANGELOG.md

@@ -0,0 +1,3 @@
+## 0.1.0 - unreleased
+
+Initial release.

misc/futures-bounded/Cargo.toml

@@ -0,0 +1,19 @@
+[package]
+name = "futures-bounded"
+version = "0.1.0"
+edition = "2021"
+rust-version.workspace = true
+license = "MIT"
+repository = "https://github.com/libp2p/rust-libp2p"
+keywords = ["futures", "async", "backpressure"]
+categories = ["data-structures", "asynchronous"]
+description = "Utilities for bounding futures in size and time."
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+futures-util = { version = "0.3.28" }
+futures-timer = "3.0.2"
+
+[dev-dependencies]
+tokio = { version = "1.29.1", features = ["macros", "rt"] }

misc/futures-bounded/src/lib.rs

@@ -0,0 +1,193 @@
+use std::future::Future;
+use std::task::{Context, Poll, Waker};
+use std::time::Duration;
+
+use futures_timer::Delay;
+use futures_util::future::{select, BoxFuture, Either};
+use futures_util::stream::FuturesUnordered;
+use futures_util::{ready, FutureExt, StreamExt};
+
+/// Represents a set of (Worker)-[Future]s.
+///
+/// This wraps [FuturesUnordered] but bounds it by time and size.
+/// In other words, each worker must finish within the specified time and the set never outgrows its capacity.
+pub struct WorkerFutures<K, O> {
+    timeout: Duration,
+    capacity: usize,
+    inner: FuturesUnordered<BoxFuture<'static, (K, Result<O, Timeout>)>>,
+
+    empty_waker: Option<Waker>,
+    full_waker: Option<Waker>,
+}
+
+impl<K, O> WorkerFutures<K, O> {
+    pub fn new(timeout: Duration, capacity: usize) -> Self {
+        Self {
+            timeout,
+            capacity,
+            inner: Default::default(),
+            empty_waker: None,
+            full_waker: None,
+        }
+    }
+}
+
+impl<K, O> WorkerFutures<K, O>
+where
+    K: Send + 'static,
+{
+    pub fn try_push<F>(&mut self, key: K, worker: F) -> Option<F>
+    where
+        F: Future<Output = O> + Send + 'static + Unpin,
+    {
+        if self.inner.len() >= self.capacity {
+            return Some(worker);
+        }
+        let timeout = Delay::new(self.timeout);
+
+        self.inner.push(
+            async move {
+                match select(worker, timeout).await {
+                    Either::Left((out, _)) => (key, Ok(out)),
+                    Either::Right(((), _)) => (key, Err(Timeout::new())),
+                }
+            }
+            .boxed(),
+        );
+
+        if let Some(waker) = self.empty_waker.take() {
+            waker.wake();
+        }
+
+        None
+    }
+
+    pub fn is_empty(&self) -> bool {
+        self.inner.is_empty()
+    }
+
+    pub fn poll_ready_unpin(&mut self, cx: &mut Context<'_>) -> Poll<()> {
+        if self.inner.len() < self.capacity {
+            return Poll::Ready(());
+        }
+
+        self.full_waker = Some(cx.waker().clone());
+        Poll::Pending
+    }
+
+    pub fn poll_unpin(&mut self, cx: &mut Context<'_>) -> Poll<(K, Result<O, Timeout>)> {
+        match ready!(self.inner.poll_next_unpin(cx)) {
+            None => {
+                self.empty_waker = Some(cx.waker().clone());
+                Poll::Pending
+            }
+            Some(result) => {
+                if let Some(waker) = self.full_waker.take() {
+                    waker.wake();
+                }
+
+                Poll::Ready(result)
+            }
+        }
+    }
+}
+
+#[derive(Debug)]
+pub struct Timeout {
+    _priv: (),
+}
+
+impl Timeout {
+    fn new() -> Self {
+        Self { _priv: () }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::future::{pending, poll_fn, ready};
+    use std::pin::Pin;
+    use std::time::Instant;
+
+    #[test]
+    fn cannot_push_more_than_capacity_tasks() {
+        let mut workers = WorkerFutures::new(Duration::from_secs(10), 1);
+
+        assert!(workers.try_push((), ready(())).is_none());
+        assert!(workers.try_push((), ready(())).is_some());
+    }
+
+    #[tokio::test]
+    async fn workers_timeout() {
+        let mut workers = WorkerFutures::new(Duration::from_millis(100), 1);
+
+        let _ = workers.try_push((), pending::<()>());
+        Delay::new(Duration::from_millis(150)).await;
+        let (_, result) = poll_fn(|cx| workers.poll_unpin(cx)).await;
+
+        assert!(result.is_err())
+    }
+
+    // Each worker causes a delay, `Task` only has a capacity of 1, meaning they must be processed in sequence.
+    // We stop after NUM_WORKERS tasks, meaning the overall execution must at least take DELAY * NUM_WORKERS.
+    #[tokio::test]
+    async fn backpressure() {
+        const DELAY: Duration = Duration::from_millis(100);
+        const NUM_WORKERS: u32 = 10;
+
+        let start = Instant::now();
+        Task::new(DELAY, NUM_WORKERS, 1).await;
+        let duration = start.elapsed();
+
+        assert!(duration >= DELAY * NUM_WORKERS);
+    }
+
+    struct Task {
+        worker: Duration,
+        num_workers: usize,
+        num_processed: usize,
+        inner: WorkerFutures<(), ()>,
+    }
+
+    impl Task {
+        fn new(worker: Duration, num_workers: u32, capacity: usize) -> Self {
+            Self {
+                worker,
+                num_workers: num_workers as usize,
+                num_processed: 0,
+                inner: WorkerFutures::new(Duration::from_secs(60), capacity),
+            }
+        }
+    }
+
+    impl Future for Task {
+        type Output = ();
+
+        fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+            let this = self.get_mut();
+
+            while this.num_processed < this.num_workers {
+                if let Poll::Ready(((), result)) = this.inner.poll_unpin(cx) {
+                    if result.is_err() {
+                        panic!("Timeout is great than worker delay")
+                    }
+
+                    this.num_processed += 1;
+                    continue;
+                }
+
+                if let Poll::Ready(()) = this.inner.poll_ready_unpin(cx) {
+                    let maybe_worker = this.inner.try_push((), Delay::new(this.worker));
+                    assert!(maybe_worker.is_none(), "we polled for readiness");
+
+                    continue;
+                }
+
+                return Poll::Pending;
+            }
+
+            Poll::Ready(())
+        }
+    }
+}

protocols/autonat/src/behaviour.rs

@@ -220,9 +220,11 @@ pub struct Behaviour {
 impl Behaviour {
     pub fn new(local_peer_id: PeerId, config: Config) -> Self {
         let protocols = iter::once((DEFAULT_PROTOCOL_NAME, ProtocolSupport::Full));
-        let mut cfg = request_response::Config::default();
-        cfg.set_request_timeout(config.timeout);
-        let inner = request_response::Behaviour::with_codec(AutoNatCodec, protocols, cfg);
+        let inner = request_response::Behaviour::with_codec(
+            AutoNatCodec,
+            protocols,
+            request_response::Config::default().with_request_timeout(config.timeout),
+        );
         Self {
             local_peer_id,
             inner,

protocols/perf/src/client.rs

@@ -58,17 +58,16 @@ pub struct Behaviour {
 
 impl Default for Behaviour {
     fn default() -> Self {
-        let mut req_resp_config = request_response::Config::default();
-        req_resp_config.set_connection_keep_alive(Duration::from_secs(60 * 5));
-        req_resp_config.set_request_timeout(Duration::from_secs(60 * 5));
         Self {
             connected: Default::default(),
             request_response: request_response::Behaviour::new(
                 std::iter::once((
                     crate::PROTOCOL_NAME,
                     request_response::ProtocolSupport::Outbound,
                 )),
-                req_resp_config,
+                request_response::Config::default()
+                    .with_connection_keep_alive(Duration::from_secs(60 * 5))
+                    .with_request_timeout(Duration::from_secs(60 * 5)),
             ),
         }
     }

protocols/perf/src/server.rs

@@ -37,17 +37,15 @@ pub struct Behaviour {
 
 impl Default for Behaviour {
     fn default() -> Self {
-        let mut req_resp_config = request_response::Config::default();
-        req_resp_config.set_connection_keep_alive(Duration::from_secs(60 * 5));
-        req_resp_config.set_request_timeout(Duration::from_secs(60 * 5));
-
         Self {
             request_response: request_response::Behaviour::new(
                 std::iter::once((
                     crate::PROTOCOL_NAME,
                     request_response::ProtocolSupport::Inbound,
                 )),
-                req_resp_config,
+                request_response::Config::default()
+                    .with_connection_keep_alive(Duration::from_secs(60 * 5))
+                    .with_request_timeout(Duration::from_secs(60 * 5)),
             ),
         }
     }

protocols/request-response/CHANGELOG.md

@@ -3,7 +3,12 @@
 - Replace unmaintained `serde_cbor` dependency with `cbor4ii`.
   See [PR 4187].
 
+- Allow at most 100 concurrent inbound + outbound streams.
+  This limit is configurable via `Config::with_max_concurrent_streams`.
+  See [PR 3914].
+
 [PR 4187]: https://github.com/libp2p/rust-libp2p/pull/4187
+[PR 3914]: https://github.com/libp2p/rust-libp2p/pull/3914
 
 ## 0.25.0
 

protocols/request-response/Cargo.toml

@@ -24,6 +24,8 @@ serde_json = { version = "1.0.100", optional = true }
 smallvec = "1.11.0"
 void = "1.0.2"
 log = "0.4.19"
+futures-timer = "3.0.2"
+futures-bounded = { workspace = true }
 
 [features]
 json = ["dep:serde", "dep:serde_json", "libp2p-swarm/macros"]