Simplify delay reset queue (#56)

Author: fafhrd91

CHANGELOG.md

@@ -1,5 +1,9 @@
 # Changes
 
+## [1.8.6] - 2025-03-12
+
+* Simplify delay reset queue
+
 ## [1.8.5] - 2025-02-12
 
 * Fix handle for REFUSED_STREAM reset

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "ntex-h2"
-version = "1.8.5"
+version = "1.8.6"
 license = "MIT OR Apache-2.0"
 authors = ["Nikolay Kim <[email protected]>"]
 description = "An HTTP/2 client and server"

src/config.rs

@@ -25,6 +25,8 @@ pub(crate) struct ConfigInner {
     pub(crate) window_sz_threshold: Cell<WindowSize>,
     /// How long a locally reset stream should ignore frames
     pub(crate) reset_duration: Cell<Duration>,
+    /// Maximum number of locally reset streams to keep at a time
+    pub(crate) reset_max: Cell<usize>,
     /// Initial window size for new connections.
     pub(crate) connection_window_sz: Cell<WindowSize>,
     pub(crate) connection_window_sz_threshold: Cell<WindowSize>,
@@ -89,6 +91,7 @@ impl Config {
             connection_window_sz_threshold,
             dispatcher_config,
             settings: Cell::new(settings),
+            reset_max: Cell::new(consts::DEFAULT_RESET_STREAM_MAX),
             reset_duration: Cell::new(consts::DEFAULT_RESET_STREAM_SECS.into()),
             remote_max_concurrent_streams: Cell::new(None),
             max_header_continuations: Cell::new(consts::DEFAULT_MAX_COUNTINUATIONS),
@@ -232,10 +235,9 @@ impl Config {
     /// received for that stream will result in a connection level protocol
     /// error, forcing the connection to terminate.
     ///
-    /// The default value is 30.
-    #[doc(hidden)]
-    #[deprecated]
-    pub fn max_concurrent_reset_streams(&self, _: usize) -> &Self {
+    /// The default value is 32.
+    pub fn max_concurrent_reset_streams(&self, val: usize) -> &Self {
+        self.0.reset_max.set(val);
         self
     }
 
@@ -355,6 +357,7 @@ impl fmt::Debug for Config {
             .field("window_sz", &self.0.window_sz.get())
             .field("window_sz_threshold", &self.0.window_sz_threshold.get())
             .field("reset_duration", &self.0.reset_duration.get())
+            .field("reset_max", &self.0.reset_max.get())
             .field("connection_window_sz", &self.0.connection_window_sz.get())
             .field(
                 "connection_window_sz_threshold",
@@ -368,23 +371,3 @@ impl fmt::Debug for Config {
             .finish()
     }
 }
-
-impl fmt::Debug for ConfigInner {
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        f.debug_struct("Config")
-            .field("window_sz", &self.window_sz.get())
-            .field("window_sz_threshold", &self.window_sz_threshold.get())
-            .field("reset_duration", &self.reset_duration.get())
-            .field("connection_window_sz", &self.connection_window_sz.get())
-            .field(
-                "connection_window_sz_threshold",
-                &self.connection_window_sz_threshold.get(),
-            )
-            .field(
-                "remote_max_concurrent_streams",
-                &self.remote_max_concurrent_streams.get(),
-            )
-            .field("settings", &self.settings.get())
-            .finish()
-    }
-}

src/connection.rs

@@ -1,5 +1,5 @@
 use std::{cell::Cell, cell::RefCell, fmt, mem, rc::Rc};
-use std::{collections::VecDeque, time::Duration, time::Instant};
+use std::{collections::VecDeque, time::Instant};
 
 use ntex_bytes::{ByteString, Bytes};
 use ntex_http::{HeaderMap, Method};
@@ -408,7 +408,7 @@ impl Connection {
         }
 
         // Add ids to pending queue
-        self.0.local_pending_reset.add(id);
+        self.0.local_pending_reset.add(id, &self.0.local_config);
     }
 
     pub(crate) fn recv_half(&self) -> RecvHalfConnection {
@@ -722,7 +722,7 @@ impl RecvHalfConnection {
                 stream,
                 StreamError::Reset(frm.reason()),
             )))
-        } else if self.0.local_pending_reset.is_pending(id) {
+        } else if self.0.local_pending_reset.remove(id) {
             self.update_rst_count()
         } else {
             self.update_rst_count()?;
@@ -866,79 +866,71 @@ async fn ping(st: Connection, timeout: time::Seconds, io: IoRef) {
     }
 }
 
-const BLOCKS: usize = 5;
-const LAST_BLOCK: usize = 4;
+struct Pending(Cell<Option<Box<PendingInner>>>);
 
-#[cfg(not(test))]
-const SECS: u64 = 30;
-#[cfg(test)]
-const SECS: u64 = 1;
-
-const BLOCK_SIZE: Duration = Duration::from_secs(SECS);
-const ALL_BLOCKS: Duration = Duration::from_secs((BLOCKS as u64) * SECS);
-
-#[derive(Default)]
-struct Pending {
-    idx: Cell<u8>,
-    blocks: RefCell<[Block; BLOCKS]>,
-}
-
-#[derive(Debug)]
-struct Block {
-    start_time: Instant,
+struct PendingInner {
     ids: HashSet<StreamId>,
+    queue: VecDeque<(StreamId, Instant)>,
 }
 
-impl Pending {
-    fn add(&self, id: StreamId) {
-        let cur = now();
-        let idx = self.idx.get() as usize;
-        let mut blocks = self.blocks.borrow_mut();
-
-        // check if we need to insert new block
-        if blocks[idx].start_time < (cur - BLOCK_SIZE) {
-            // shift blocks
-            let idx = if idx == 0 { LAST_BLOCK } else { idx - 1 };
-            // insert new item
-            blocks[idx].start_time = cur;
-            blocks[idx].ids.clear();
-            blocks[idx].ids.insert(id);
-            self.idx.set(idx as u8);
-        } else {
-            blocks[idx].ids.insert(id);
-        }
+impl Default for Pending {
+    fn default() -> Self {
+        Self(Cell::new(Some(Box::new(PendingInner {
+            ids: HashSet::default(),
+            queue: VecDeque::with_capacity(16),
+        }))))
     }
+}
 
-    fn is_pending(&self, id: StreamId) -> bool {
-        let blocks = self.blocks.borrow_mut();
+impl Pending {
+    fn add(&self, id: StreamId, config: &Config) {
+        let mut inner = self.0.take().unwrap();
 
-        let max = now() - ALL_BLOCKS;
-        let mut idx = self.idx.get() as usize;
+        let current_time = now();
 
-        loop {
-            let item = &blocks[idx];
-            if item.start_time < max {
+        // remove old ids
+        let max_time = current_time - config.0.reset_duration.get();
+        while let Some(item) = inner.queue.front() {
+            if item.1 < max_time {
+                inner.ids.remove(&item.0);
+                inner.queue.pop_front();
+            } else {
                 break;
-            } else if item.ids.contains(&id) {
-                return true;
             }
-            idx += 1;
-            if idx == BLOCKS {
-                idx = 0;
-            } else if idx == self.idx.get() as usize {
-                break;
+        }
+
+        // shrink size of ids
+        while inner.queue.len() >= config.0.reset_max.get() {
+            if let Some((id, _)) = inner.queue.pop_front() {
+                inner.ids.remove(&id);
             }
         }
-        false
+
+        inner.ids.insert(id);
+        inner.queue.push_back((id, current_time));
+        self.0.set(Some(inner));
     }
-}
 
-impl Default for Block {
-    fn default() -> Self {
-        Self {
-            ids: HashSet::default(),
-            start_time: now() - ALL_BLOCKS,
+    fn remove(&self, id: StreamId) -> bool {
+        let mut inner = self.0.take().unwrap();
+        let removed = inner.ids.remove(&id);
+        if removed {
+            for idx in 0..inner.queue.len() {
+                if inner.queue[idx].0 == id {
+                    inner.queue.remove(idx);
+                    break;
+                }
+            }
         }
+        self.0.set(Some(inner));
+        removed
+    }
+
+    fn is_pending(&self, id: StreamId) -> bool {
+        let inner = self.0.take().unwrap();
+        let pending = inner.ids.contains(&id);
+        self.0.set(Some(inner));
+        pending
     }
 }
 
@@ -1017,11 +1009,16 @@ mod tests {
 
     #[ntex::test]
     async fn test_delay_reset_queue() {
+        let _ = env_logger::try_init();
+
         let srv = test_server(|| {
             fn_service(|io: Io<_>| async move {
                 let _ = h2::server::handle_one(
                     io.into(),
-                    h2::Config::server().ping_timeout(Seconds::ZERO).clone(),
+                    h2::Config::server()
+                        .ping_timeout(Seconds::ZERO)
+                        .reset_stream_duration(Seconds(1))
+                        .clone(),
                     fn_service(|msg: h2::ControlMessage<h2::StreamError>| async move {
                         Ok::<_, ()>(msg.ack())
                     }),
@@ -1037,7 +1034,7 @@ mod tests {
         });
 
         let addr = ntex::connect::Connect::new("localhost").set_addr(Some(srv.addr()));
-        let io = ntex::connect::connect(addr).await.unwrap();
+        let io = ntex::connect::connect(addr.clone()).await.unwrap();
         let codec = Codec::default();
         let _ = io.with_write_buf(|buf| buf.extend_from_slice(&PREFACE));
 
@@ -1081,14 +1078,45 @@ mod tests {
         assert_eq!(res.reason(), Reason::NO_ERROR);
 
         // prev closed stream
-        io.send(pl.clone().into(), &codec).await.unwrap();
+        io.send(pl.into(), &codec).await.unwrap();
+        let res = goaway(io.recv(&codec).await.unwrap().unwrap());
+        assert_eq!(res.reason(), Reason::PROTOCOL_ERROR);
+
+        // SECOND connection
+        let io = ntex::connect::connect(addr).await.unwrap();
+        let codec = Codec::default();
+        let _ = io.with_write_buf(|buf| buf.extend_from_slice(&PREFACE));
+
+        let settings = frame::Settings::default();
+        io.encode(settings.into(), &codec).unwrap();
+
+        // settings & window
+        let _ = io.recv(&codec).await;
+        let _ = io.recv(&codec).await;
+        let _ = io.recv(&codec).await;
+
+        let id = frame::StreamId::CLIENT;
+        let pseudo = frame::PseudoHeaders {
+            method: Some(Method::GET),
+            scheme: Some("HTTPS".into()),
+            authority: Some("localhost".into()),
+            path: Some("/".into()),
+            ..Default::default()
+        };
+        let hdrs = frame::Headers::new(id, pseudo.clone(), HeaderMap::new(), false);
+        io.send(hdrs.into(), &codec).await.unwrap();
+
+        // server resets stream
         let res = get_reset(io.recv(&codec).await.unwrap().unwrap());
-        assert_eq!(res.reason(), Reason::STREAM_CLOSED);
+        assert_eq!(res.reason(), Reason::NO_ERROR);
 
-        sleep(Millis(5100)).await;
+        // after server receives remote reset, any next frame cause protocol error
+        io.send(frame::Reset::new(id, Reason::NO_ERROR).into(), &codec)
+            .await
+            .unwrap();
 
-        // prev closed stream
-        io.send(pl.into(), &codec).await.unwrap();
+        let pl = frame::Data::new(id, Bytes::from_static(b"data"));
+        io.send(pl.clone().into(), &codec).await.unwrap();
         let res = goaway(io.recv(&codec).await.unwrap().unwrap());
         assert_eq!(res.reason(), Reason::PROTOCOL_ERROR);
     }

src/consts.rs

@@ -5,7 +5,8 @@ use crate::frame::WindowSize;
 
 // Constants
 pub(crate) const MAX_WINDOW_SIZE: WindowSize = (1 << 31) - 1;
-pub(crate) const DEFAULT_RESET_STREAM_SECS: Seconds = Seconds(10);
+pub(crate) const DEFAULT_RESET_STREAM_MAX: usize = 32;
+pub(crate) const DEFAULT_RESET_STREAM_SECS: Seconds = Seconds(30);
 pub(crate) const DEFAULT_CONNECTION_WINDOW_SIZE: WindowSize = 1_048_576;
 pub(crate) const DEFAULT_SETTINGS_MAX_HEADER_LIST_SIZE: u32 = 48 * 1024;
 pub(crate) const DEFAULT_MAX_COUNTINUATIONS: usize = 5;