sync h2 code with hyperium/h2 v0.4.8

* Fix reclaiming reserved capacity (fixes <hyperium/h2#607>)
  by @nox in <hyperium/h2#832>
* Fix busy loop on shutdown by @seanmonstar
  in <hyperium/h2#834>
* Fix window size decrement of send-closed streams
  by @nox in <hyperium/h2#830>
* Fix handle implicit resets at the right time
  by @nox in <hyperium/h2#833>
* Fix poll_flush after poll_shutdown
  by @bdbai in <hyperium/h2#836>

Co-authored-by: Sean McArthur <[email protected]>
Co-authored-by: 包布丁 <[email protected]>
Co-authored-by: Anthony Ramine <[email protected]>
Co-authored-by: Samuel Tardieu <[email protected]>

FORK.md

@@ -9,7 +9,7 @@ as a distant relative.
 
 ### hyperium
 
-- <https://github.com/hyperium/h2/tree/v0.4.7>
+- <https://github.com/hyperium/h2/tree/v0.4.8>
 - <https://github.com/hyperium/hyper/tree/v1.5.2>
 - <https://github.com/hyperium/hyper-util/tree/v0.1.10>
 

rama-http-core/src/h2/client.rs

@@ -1441,8 +1441,14 @@ where
 
     fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
         self.inner.maybe_close_connection_if_no_streams();
+        let had_streams_or_refs = self.inner.has_streams_or_other_references();
         let result = self.inner.poll(cx).map_err(Into::into);
-        if result.is_pending() && !self.inner.has_streams_or_other_references() {
+        // if we had streams/refs, and don't anymore, wake up one more time to
+        // ensure proper shutdown
+        if result.is_pending()
+            && had_streams_or_refs
+            && !self.inner.has_streams_or_other_references()
+        {
             tracing::trace!("last stream closed during poll, wake again");
             cx.waker().wake_by_ref();
         }

rama-http-core/src/h2/codec/framed_write.rs

@@ -22,6 +22,7 @@ macro_rules! limited_write_buf {
 pub(super) struct FramedWrite<T, B> {
     /// Upstream `AsyncWrite`
     inner: T,
+    final_flush_done: bool,
 
     encoder: Encoder<B>,
 }
@@ -88,6 +89,7 @@ where
         };
         FramedWrite {
             inner,
+            final_flush_done: false,
             encoder: Encoder {
                 hpack: hpack::Encoder::default(),
                 buf: Cursor::new(BytesMut::with_capacity(DEFAULT_BUFFER_CAPACITY)),
@@ -164,7 +166,10 @@ where
 
     /// Close the codec
     pub(super) fn shutdown(&mut self, cx: &mut Context) -> Poll<io::Result<()>> {
-        ready!(self.flush(cx))?;
+        if !self.final_flush_done {
+            ready!(self.flush(cx))?;
+            self.final_flush_done = true;
+        }
         Pin::new(&mut self.inner).poll_shutdown(cx)
     }
 }

rama-http-core/src/h2/proto/streams/counts.rs

@@ -229,7 +229,7 @@ impl Counts {
                 }
             }
 
-            if stream.is_counted {
+            if !stream.state.is_scheduled_reset() && stream.is_counted {
                 tracing::trace!("dec_num_streams; stream={:?}", stream.id);
                 // Decrement the number of active streams.
                 self.dec_num_streams(&mut stream);

rama-http-core/src/h2/proto/streams/prioritize.rs

@@ -350,13 +350,18 @@ impl Prioritize {
         stream: &mut store::Ptr,
         counts: &mut Counts,
     ) {
-        // only reclaim requested capacity that isn't already buffered
-        if stream.requested_send_capacity as usize > stream.buffered_send_data {
-            let reserved = stream.requested_send_capacity - stream.buffered_send_data as WindowSize;
+        // only reclaim reserved capacity that isn't already buffered
+        if stream.send_flow.available().as_size() as usize > stream.buffered_send_data {
+            let reserved =
+                stream.send_flow.available().as_size() - stream.buffered_send_data as WindowSize;
+
+            // Panic safety: due to how `reserved` is computed it can't be greater
+            // than what's available.
+            stream
+                .send_flow
+                .claim_capacity(reserved)
+                .expect("window size should be greater than reserved");
 
-            // TODO: proper error handling
-            let _res = stream.send_flow.claim_capacity(reserved);
-            debug_assert!(_res.is_ok());
             self.assign_connection_capacity(reserved, stream, counts);
         }
     }
@@ -687,8 +692,11 @@ impl Prioritize {
     }
 
     pub(crate) fn clear_pending_send(&mut self, store: &mut Store, counts: &mut Counts) {
-        while let Some(stream) = self.pending_send.pop(store) {
+        while let Some(mut stream) = self.pending_send.pop(store) {
             let is_pending_reset = stream.is_pending_reset_expiration();
+            if let Some(reason) = stream.state.get_scheduled_reset() {
+                stream.set_reset(reason, Initiator::Library);
+            }
             counts.transition_after(stream, is_pending_reset);
         }
     }

rama-http-core/src/h2/proto/streams/send.rs

@@ -467,6 +467,16 @@ impl Send {
                     store.try_for_each(|mut stream| {
                         let stream = &mut *stream;
 
+                        if stream.state.is_send_closed() && stream.buffered_send_data == 0 {
+                            tracing::trace!(
+                                "skipping send-closed stream; id={:?}; flow={:?}",
+                                stream.id,
+                                stream.send_flow
+                            );
+
+                            return Ok(());
+                        }
+
                         tracing::trace!(
                             "decrementing stream window; id={:?}; decr={}; flow={:?}",
                             stream.id,

rama-http-core/src/h2/proto/streams/streams.rs

@@ -1294,10 +1294,7 @@ impl<B> StreamRef<B> {
 
         let mut stream = me.store.resolve(self.opaque.key);
 
-        me.actions
-            .send
-            .poll_reset(cx, &mut stream, mode)
-            .map_err(From::from)
+        me.actions.send.poll_reset(cx, &mut stream, mode)
     }
 
     pub(crate) fn clone_to_opaque(&self) -> OpaqueStreamRef {

rama-http-core/tests/h2-support/src/frames.rs

@@ -378,6 +378,11 @@ impl Mock<frame::Settings> {
         self
     }
 
+    pub fn max_frame_size(mut self, val: u32) -> Self {
+        self.0.set_max_frame_size(Some(val));
+        self
+    }
+
     pub fn initial_window_size(mut self, val: u32) -> Self {
         self.0.set_initial_window_size(Some(val));
         self