[v26.x backport] http2: fix write deadlock exposed by larger window sizes

benchmark/http2/full-duplex.js

@@ -0,0 +1,71 @@
+'use strict';
+
+const common = require('../common.js');
+const fixtures = require('../../test/common/fixtures');
+
+const bench = common.createBenchmark(main, {
+  n: [100],
+  streams: [2],
+  size: [4 * 1024 * 1024],
+  // Use the HTTP/2 protocol default.
+  window: [65535],
+}, {
+  test: { size: 128 * 1024, window: 65535 },
+});
+
+function main({ n, streams, size, window }) {
+  const http2 = require('http2');
+  const payload = Buffer.alloc(size);
+  const server = http2.createSecureServer({
+    key: fixtures.readKey('agent1-key.pem'),
+    cert: fixtures.readKey('agent1-cert.pem'),
+    settings: { initialWindowSize: window },
+  });
+
+  let completed = 0;
+  let batches = 0;
+
+  function onTransferComplete() {
+    if (++completed !== streams * 2)
+      return;
+
+    if (++batches === n) {
+      // Report combined upload and download throughput in MiB/s.
+      bench.end(n * streams * size * 2 / (1024 * 1024));
+      client.close();
+      server.close();
+      return;
+    }
+
+    startBatch();
+  }
+
+  server.on('stream', (stream) => {
+    stream.resume();
+    stream.on('end', onTransferComplete);
+    stream.respond();
+    stream.end(payload);
+  });
+
+  let client;
+  function startBatch() {
+    completed = 0;
+    for (let i = 0; i < streams; i++) {
+      const request = client.request({ ':method': 'POST' });
+      request.resume();
+      request.on('end', onTransferComplete);
+      request.end(payload);
+    }
+  }
+
+  server.listen(0, () => {
+    client = http2.connect(`https://localhost:${server.address().port}`, {
+      rejectUnauthorized: false,
+      settings: { initialWindowSize: window },
+    });
+    client.on('connect', () => {
+      bench.start();
+      startBatch();
+    });
+  });
+}

lib/internal/http2/core.js

@@ -591,7 +591,7 @@ function onPing(payload) {
 // point, close them. If there is an open fd for file send, close that also.
 // At this point the underlying node::http2:Http2Stream handle is no
 // longer usable so destroy it also.
-function onStreamClose(code) {
+function onStreamClose(code, peerReset) {
   const stream = this[kOwner];
   if (!stream || stream.destroyed)
     return false;
@@ -610,8 +610,9 @@ function onStreamClose(code) {
     // If errored or ended, we can destroy immediately.
     stream.destroy();
   } else {
-    // Wait for end to destroy.
-    stream.on('end', stream[kMaybeDestroy]);
+    // Preserve buffered reads after a clean reset, but do not wait for
+    // pending writes: the peer reset may prevent them from finishing.
+    stream.on('end', peerReset ? stream.destroy : stream[kMaybeDestroy]);
     // Push a null so the stream can end whenever the client consumes
     // it completely.
     stream.push(null);

src/node_http2.cc

@@ -970,45 +970,24 @@ ssize_t Http2Session::OnMaxFrameSizePadding(size_t frameLen,
 // quite expensive. This is a potential performance optimization target later.
 void Http2Session::ConsumeHTTP2Data() {
   CHECK_NOT_NULL(stream_buf_.base);
-  CHECK_LE(stream_buf_offset_, stream_buf_.len);
-  size_t read_len = stream_buf_.len - stream_buf_offset_;
 
   // multiple side effects.
-  Debug(this, "receiving %d bytes [wants data? %d]",
-        read_len,
+  Debug(this,
+        "receiving %d bytes [wants data? %d]",
+        stream_buf_.len,
         nghttp2_session_want_read(session_.get()));
-  set_receive_paused(false);
   custom_recv_error_code_ = nullptr;
   set_receiving();
   ssize_t ret =
-    nghttp2_session_mem_recv(session_.get(),
-                             reinterpret_cast<uint8_t*>(stream_buf_.base) +
-                                 stream_buf_offset_,
-                             read_len);
+      nghttp2_session_mem_recv(session_.get(),
+                               reinterpret_cast<uint8_t*>(stream_buf_.base),
+                               stream_buf_.len);
   set_receiving(false);
   CHECK_NE(ret, NGHTTP2_ERR_NOMEM);
   CHECK_IMPLIES(custom_recv_error_code_ != nullptr, ret < 0);
 
-  if (is_receive_paused()) {
-    CHECK(is_reading_stopped());
-
-    CHECK_GT(ret, 0);
-    CHECK_LE(static_cast<size_t>(ret), read_len);
-
-    // Mark the remainder of the data as available for later consumption.
-    // Even if all bytes were received, a paused stream may delay the
-    // nghttp2_on_frame_recv_callback which may have an END_STREAM flag.
-    stream_buf_offset_ += ret;
-    // Still complete a Close() deferred during mem_recv; do not fall through
-    // to SendPendingData() here (paused receives historically skip that flush
-    // because a write may already be in progress).
-    MaybeFinishPendingClose();
-    goto done;
-  }
-
   // We are done processing the current input chunk.
   DecrementCurrentSessionMemory(stream_buf_.len);
-  stream_buf_offset_ = 0;
   stream_buf_ab_.Reset();
   stream_buf_allocation_.reset();
   stream_buf_ = uv_buf_init(nullptr, 0);
@@ -1017,14 +996,6 @@ void Http2Session::ConsumeHTTP2Data() {
   // not written after pending RST_STREAM frames.
   MaybeFinishPendingClose();
 
-done:
-  // Finish a Close() deferred above before flushing, so GOAWAY is not written
-  // after pending RST_STREAM frames.
-  if (is_close_pending() && !is_destroyed()) {
-    set_close_pending(false);
-    FinishClose(pending_close_code_, pending_close_socket_closed_);
-  }
-
   // Send any data that was queued up while processing the received data.
   if (ret >= 0 && !is_destroyed()) {
     SendPendingData();
@@ -1147,6 +1118,13 @@ int Http2Session::OnFrameReceive(nghttp2_session* handle,
     case NGHTTP2_HEADERS:
       session->HandleHeadersFrame(frame);
       break;
+    case NGHTTP2_RST_STREAM:
+      // Distinguish a peer reset from a natural close with the same code.
+      if (BaseObjectPtr<Http2Stream> stream =
+              session->FindStream(frame->hd.stream_id)) {
+        stream->set_peer_reset();
+      }
+      break;
     case NGHTTP2_SETTINGS:
       session->HandleSettingsFrame(frame);
       break;
@@ -1356,9 +1334,12 @@ int Http2Session::OnStreamClose(nghttp2_session* handle,
   // ever passed on to the javascript side. If that happens, the callback
   // will return false.
   if (env->can_call_into_js()) {
-    Local<Value> arg = Integer::NewFromUnsigned(isolate, code);
+    Local<Value> argv[] = {
+        Integer::NewFromUnsigned(isolate, code),
+        Boolean::New(isolate, stream->peer_reset()),
+    };
     MaybeLocal<Value> answer = stream->MakeCallback(
-        env->http2session_on_stream_close_function(), 1, &arg);
+        env->http2session_on_stream_close_function(), arraysize(argv), argv);
     if (answer.IsEmpty() || answer.ToLocalChecked()->IsFalse()) {
       // Skip to destroy
       stream->Destroy();
@@ -1461,15 +1442,6 @@ int Http2Session::OnDataChunkReceived(nghttp2_session* handle,
     }
   } while (len != 0);
 
-  // If we are currently waiting for a write operation to finish, we should
-  // tell nghttp2 that we want to wait before we process more input data.
-  if (session->is_write_in_progress()) {
-    CHECK(session->is_reading_stopped());
-    session->set_receive_paused();
-    Debug(session, "receive paused");
-    return NGHTTP2_ERR_PAUSE;
-  }
-
   return 0;
 }
 
@@ -1552,7 +1524,6 @@ void Http2StreamListener::OnStreamRead(ssize_t nread, const uv_buf_t& buf) {
   size_t offset = buf.base - session->stream_buf_.base;
 
   // Verify that the data offset is inside the current read buffer.
-  CHECK_GE(offset, session->stream_buf_offset_);
   CHECK_LE(offset, session->stream_buf_.len);
   CHECK_LE(offset + buf.len, session->stream_buf_.len);
 
@@ -1867,11 +1838,6 @@ void Http2Session::OnStreamAfterWrite(WriteWrap* w, int status) {
     return;
   }
 
-  // If there is more incoming data queued up, consume it.
-  if (stream_buf_offset_ > 0) {
-    ConsumeHTTP2Data();
-  }
-
   if (!is_write_scheduled() && !is_destroyed()) {
     // Schedule a new write if nghttp2 wants to send data.
     MaybeScheduleWrite();
@@ -1918,7 +1884,7 @@ void Http2Session::MaybeStopReading() {
   if (is_reading_stopped() || is_closing()) return;
   int want_read = nghttp2_session_want_read(session_.get());
   Debug(this, "wants read? %d", want_read);
-  if (want_read == 0 || is_write_in_progress()) {
+  if (want_read == 0) {
     set_reading_stopped();
     stream_->ReadStop();
   }
@@ -2178,7 +2144,7 @@ void Http2Session::OnStreamRead(ssize_t nread, const uv_buf_t& buf_) {
   Context::Scope context_scope(env()->context());
   Http2Scope h2scope(this);
   CHECK_NOT_NULL(stream_);
-  Debug(this, "receiving %d bytes, offset %d", nread, stream_buf_offset_);
+  Debug(this, "receiving %d bytes", nread);
   std::unique_ptr<BackingStore> bs = env()->release_managed_buffer(buf_);
 
   // Only pass data on if nread > 0
@@ -2193,40 +2159,18 @@ void Http2Session::OnStreamRead(ssize_t nread, const uv_buf_t& buf_) {
 
   statistics_.data_received += nread;
 
-  if (stream_buf_offset_ == 0 && static_cast<size_t>(nread) != bs->ByteLength())
-      [[likely]] {
+  // ConsumeHTTP2Data() always consumes the whole chunk, so there is never a
+  // partially processed buffer left over from a previous read.
+  DCHECK_NULL(stream_buf_.base);
+
+  if (static_cast<size_t>(nread) != bs->ByteLength()) [[likely]] {
     // Shrink to the actual amount of used data.
     std::unique_ptr<BackingStore> old_bs = std::move(bs);
     bs = ArrayBuffer::NewBackingStore(
         env()->isolate(),
         nread,
         BackingStoreInitializationMode::kUninitialized);
     memcpy(bs->Data(), old_bs->Data(), nread);
-  } else {
-    // This is a very unlikely case, and should only happen if the ReadStart()
-    // call in OnStreamAfterWrite() immediately provides data. If that does
-    // happen, we concatenate the data we received with the already-stored
-    // pending input data, slicing off the already processed part.
-    size_t pending_len = stream_buf_.len - stream_buf_offset_;
-    std::unique_ptr<BackingStore> new_bs = ArrayBuffer::NewBackingStore(
-        env()->isolate(),
-        pending_len + nread,
-        BackingStoreInitializationMode::kUninitialized);
-    memcpy(static_cast<char*>(new_bs->Data()),
-           stream_buf_.base + stream_buf_offset_,
-           pending_len);
-    memcpy(static_cast<char*>(new_bs->Data()) + pending_len,
-           bs->Data(),
-           nread);
-
-    bs = std::move(new_bs);
-    nread = bs->ByteLength();
-    stream_buf_offset_ = 0;
-    stream_buf_ab_.Reset();
-
-    // We have now fully processed the stream_buf_ input chunk (by moving the
-    // remaining part into buf, which will be accounted for below).
-    DecrementCurrentSessionMemory(stream_buf_.len);
   }
 
   IncrementCurrentSessionMemory(nread);

src/node_http2.h

@@ -65,6 +65,7 @@ constexpr int kStreamStateReadPaused = 0x4;
 constexpr int kStreamStateClosed = 0x8;
 constexpr int kStreamStateDestroyed = 0x10;
 constexpr int kStreamStateTrailers = 0x20;
+constexpr int kStreamStatePeerReset = 0x40;
 
 // Http2Session internal states
 constexpr int kSessionStateNone = 0x0;
@@ -75,9 +76,8 @@ constexpr int kSessionStateClosing = 0x8;
 constexpr int kSessionStateSending = 0x10;
 constexpr int kSessionStateWriteInProgress = 0x20;
 constexpr int kSessionStateReadingStopped = 0x40;
-constexpr int kSessionStateReceivePaused = 0x80;
-constexpr int kSessionStateReceiving = 0x100;
-constexpr int kSessionStateClosePending = 0x200;
+constexpr int kSessionStateReceiving = 0x80;
+constexpr int kSessionStateClosePending = 0x100;
 
 // The Padding Strategy determines the method by which extra padding is
 // selected for HEADERS and DATA frames. These are configurable via the
@@ -353,6 +353,10 @@ class Http2Stream : public AsyncWrap,
     return flags_ & kStreamStateClosed;
   }
 
+  bool peer_reset() const { return flags_ & kStreamStatePeerReset; }
+
+  void set_peer_reset() { flags_ |= kStreamStatePeerReset; }
+
   bool has_trailers() const {
     return flags_ & kStreamStateTrailers;
   }
@@ -664,7 +668,6 @@ class Http2Session : public AsyncWrap,
   IS_FLAG(sending, kSessionStateSending)
   IS_FLAG(write_in_progress, kSessionStateWriteInProgress)
   IS_FLAG(reading_stopped, kSessionStateReadingStopped)
-  IS_FLAG(receive_paused, kSessionStateReceivePaused)
   IS_FLAG(receiving, kSessionStateReceiving)
   IS_FLAG(close_pending, kSessionStateClosePending)
 
@@ -945,7 +948,6 @@ class Http2Session : public AsyncWrap,
   // will be set. stream_buf_ab_ is lazily created from stream_buf_allocation_.
   v8::Global<v8::ArrayBuffer> stream_buf_ab_;
   std::unique_ptr<v8::BackingStore> stream_buf_allocation_;
-  size_t stream_buf_offset_ = 0;
   // Custom error code for errors that originated inside one of the callbacks
   // called by nghttp2_session_mem_recv.
   const char* custom_recv_error_code_ = nullptr;