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();
+ });
+ });
+}

src/node_http2.cc

@@ -981,45 +981,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);
@@ -1028,14 +1007,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();
@@ -1485,15 +1456,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;
 }
 
@@ -1576,7 +1538,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);
 
@@ -1891,11 +1852,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();
@@ -1942,7 +1898,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();
  }
@@ -2207,7 +2163,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
@@ -2222,40 +2178,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

@@ -85,9 +85,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
@@ -698,7 +697,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)
 
@@ -979,7 +977,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;

test/parallel/test-http2-bidirectional-write-deadlock.js

@@ -0,0 +1,107 @@
+'use strict';
+
+// Regression test against deadlocks between two HTTP/2 peers that are both
+// writing at the same time.
+//
+// To bound how much it buffered while output was backed up, an Http2Session
+// used to stop reading from its socket whenever a write was in flight, and
+// resume only once that write completed. When the peer was itself blocked
+// writing, that write never completed, so the session never read again and
+// the connection hung forever with no error and no timeout.
+//
+// Rather than relying on kernel socket buffers filling up - which depends on
+// the platform and configured window sizes - this models one half of that
+// cycle directly. The client's socket forwards a write but does not report it
+// as complete, substituting for a write blocked because the peer is not
+// reading. Only after that write is stalled does the server send its response
+// body. A session that stops reading while writing never sees it.
+
+const common = require('../common');
+if (!common.hasCrypto)
+ common.skip('missing crypto');
+const assert = require('assert');
+const http2 = require('http2');
+const net = require('net');
+const { Duplex } = require('stream');
+
+const BODY = 'the response body';
+
+const heldCallbacks = [];
+let serverStream;
+
+let stallWrites = false;
+
+// Client-side socket that forwards writes to a real connection, but can leave
+// their completion callbacks pending to model a transport-blocked write.
+class StalledClientSocket extends Duplex {
+ constructor(port) {
+ super();
+ this.inner = net.connect(port, common.localhostIPv4);
+ this.inner.on('data', (chunk) => this.push(chunk));
+ }
+ _read() {
+ // Incoming data is pushed as it arrives.
+ }
+ _write(chunk, encoding, callback) {
+ this.inner.write(chunk, encoding);
+ if (stallWrites) {
+ heldCallbacks.push(callback);
+ // Avoid writing from the server re-entrantly inside _write(). The
+ // ordering is still explicit: this callback is already held.
+ setImmediate(() => serverStream.end(BODY));
+ return;
+ }
+ callback();
+ }
+ _final(callback) {
+ callback();
+ }
+ _destroy(err, callback) {
+ this.inner.destroy();
+ callback(err);
+ }
+}
+
+const server = http2.createServer();
+
+server.on('stream', common.mustCall((stream) => {
+ // Send headers first. Their response event starts the stalled client write.
+ stream.respond();
+ serverStream = stream;
+}));
+
+server.listen(0, common.mustCall(() => {
+ const port = server.address().port;
+
+ const client = http2.connect(`http://${common.localhostIPv4}:${port}`, {
+ createConnection: () => new StalledClientSocket(port),
+ });
+
+ const req = client.request({ ':method': 'POST' });
+
+ let received = '';
+
+ req.on('response', common.mustCall(() => {
+ // _write() will schedule the response body only after it has retained the
+ // callback, guaranteeing that the native write is still in progress.
+ stallWrites = true;
+ req.write(Buffer.alloc(256));
+ }));
+
+ req.on('data', (chunk) => {
+ received += chunk;
+ });
+
+ req.on('end', common.mustCall(() => {
+ assert.strictEqual(received, BODY);
+ assert.ok(heldCallbacks.length > 0,
+ 'test did not actually stall a socket write');
+
+ // Let the stalled writes complete so that everything can shut down.
+ stallWrites = false;
+ for (const callback of heldCallbacks) callback();
+
+ client.destroy();
+ server.close();
+ }));
+}));

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();
+ });
+ });
+}

src/node_http2.cc

@@ -981,45 +981,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);
@@ -1028,14 +1007,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();
@@ -1485,15 +1456,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;
 }
 
@@ -1576,7 +1538,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);
 
@@ -1891,11 +1852,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();
@@ -1942,7 +1898,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();
  }
@@ -2207,7 +2163,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
@@ -2222,40 +2178,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

@@ -85,9 +85,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
@@ -698,7 +697,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)
 
@@ -979,7 +977,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;

test/parallel/test-http2-bidirectional-write-deadlock.js

@@ -0,0 +1,107 @@
+'use strict';
+
+// Regression test against deadlocks between two HTTP/2 peers that are both
+// writing at the same time.
+//
+// To bound how much it buffered while output was backed up, an Http2Session
+// used to stop reading from its socket whenever a write was in flight, and
+// resume only once that write completed. When the peer was itself blocked
+// writing, that write never completed, so the session never read again and
+// the connection hung forever with no error and no timeout.
+//
+// Rather than relying on kernel socket buffers filling up - which depends on
+// the platform and configured window sizes - this models one half of that
+// cycle directly. The client's socket forwards a write but does not report it
+// as complete, substituting for a write blocked because the peer is not
+// reading. Only after that write is stalled does the server send its response
+// body. A session that stops reading while writing never sees it.
+
+const common = require('../common');
+if (!common.hasCrypto)
+ common.skip('missing crypto');
+const assert = require('assert');
+const http2 = require('http2');
+const net = require('net');
+const { Duplex } = require('stream');
+
+const BODY = 'the response body';
+
+const heldCallbacks = [];
+let serverStream;
+
+let stallWrites = false;
+
+// Client-side socket that forwards writes to a real connection, but can leave
+// their completion callbacks pending to model a transport-blocked write.
+class StalledClientSocket extends Duplex {
+ constructor(port) {
+ super();
+ this.inner = net.connect(port, common.localhostIPv4);
+ this.inner.on('data', (chunk) => this.push(chunk));
+ }
+ _read() {
+ // Incoming data is pushed as it arrives.
+ }
+ _write(chunk, encoding, callback) {
+ this.inner.write(chunk, encoding);
+ if (stallWrites) {
+ heldCallbacks.push(callback);
+ // Avoid writing from the server re-entrantly inside _write(). The
+ // ordering is still explicit: this callback is already held.
+ setImmediate(() => serverStream.end(BODY));
+ return;
+ }
+ callback();
+ }
+ _final(callback) {
+ callback();
+ }
+ _destroy(err, callback) {
+ this.inner.destroy();
+ callback(err);
+ }
+}
+
+const server = http2.createServer();
+
+server.on('stream', common.mustCall((stream) => {
+ // Send headers first. Their response event starts the stalled client write.
+ stream.respond();
+ serverStream = stream;
+}));
+
+server.listen(0, common.mustCall(() => {
+ const port = server.address().port;
+
+ const client = http2.connect(`http://${common.localhostIPv4}:${port}`, {
+ createConnection: () => new StalledClientSocket(port),
+ });
+
+ const req = client.request({ ':method': 'POST' });
+
+ let received = '';
+
+ req.on('response', common.mustCall(() => {
+ // _write() will schedule the response body only after it has retained the
+ // callback, guaranteeing that the native write is still in progress.
+ stallWrites = true;
+ req.write(Buffer.alloc(256));
+ }));
+
+ req.on('data', (chunk) => {
+ received += chunk;
+ });
+
+ req.on('end', common.mustCall(() => {
+ assert.strictEqual(received, BODY);
+ assert.ok(heldCallbacks.length > 0,
+ 'test did not actually stall a socket write');
+
+ // Let the stalled writes complete so that everything can shut down.
+ stallWrites = false;
+ for (const callback of heldCallbacks) callback();
+
+ client.destroy();
+ server.close();
+ }));
+}));

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();
+ });
+ });
+}

src/node_http2.cc

@@ -981,45 +981,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);
@@ -1028,14 +1007,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();
@@ -1485,15 +1456,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;
 }
 
@@ -1576,7 +1538,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);
 
@@ -1891,11 +1852,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();
@@ -1942,7 +1898,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();
  }
@@ -2207,7 +2163,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
@@ -2222,40 +2178,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

@@ -85,9 +85,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
@@ -698,7 +697,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)
 
@@ -979,7 +977,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;

test/parallel/test-http2-bidirectional-write-deadlock.js

@@ -0,0 +1,107 @@
+'use strict';
+
+// Regression test against deadlocks between two HTTP/2 peers that are both
+// writing at the same time.
+//
+// To bound how much it buffered while output was backed up, an Http2Session
+// used to stop reading from its socket whenever a write was in flight, and
+// resume only once that write completed. When the peer was itself blocked
+// writing, that write never completed, so the session never read again and
+// the connection hung forever with no error and no timeout.
+//
+// Rather than relying on kernel socket buffers filling up - which depends on
+// the platform and configured window sizes - this models one half of that
+// cycle directly. The client's socket forwards a write but does not report it
+// as complete, substituting for a write blocked because the peer is not
+// reading. Only after that write is stalled does the server send its response
+// body. A session that stops reading while writing never sees it.
+
+const common = require('../common');
+if (!common.hasCrypto)
+ common.skip('missing crypto');
+const assert = require('assert');
+const http2 = require('http2');
+const net = require('net');
+const { Duplex } = require('stream');
+
+const BODY = 'the response body';
+
+const heldCallbacks = [];
+let serverStream;
+
+let stallWrites = false;
+
+// Client-side socket that forwards writes to a real connection, but can leave
+// their completion callbacks pending to model a transport-blocked write.
+class StalledClientSocket extends Duplex {
+ constructor(port) {
+ super();
+ this.inner = net.connect(port, common.localhostIPv4);
+ this.inner.on('data', (chunk) => this.push(chunk));
+ }
+ _read() {
+ // Incoming data is pushed as it arrives.
+ }
+ _write(chunk, encoding, callback) {
+ this.inner.write(chunk, encoding);
+ if (stallWrites) {
+ heldCallbacks.push(callback);
+ // Avoid writing from the server re-entrantly inside _write(). The
+ // ordering is still explicit: this callback is already held.
+ setImmediate(() => serverStream.end(BODY));
+ return;
+ }
+ callback();
+ }
+ _final(callback) {
+ callback();
+ }
+ _destroy(err, callback) {
+ this.inner.destroy();
+ callback(err);
+ }
+}
+
+const server = http2.createServer();
+
+server.on('stream', common.mustCall((stream) => {
+ // Send headers first. Their response event starts the stalled client write.
+ stream.respond();
+ serverStream = stream;
+}));
+
+server.listen(0, common.mustCall(() => {
+ const port = server.address().port;
+
+ const client = http2.connect(`http://${common.localhostIPv4}:${port}`, {
+ createConnection: () => new StalledClientSocket(port),
+ });
+
+ const req = client.request({ ':method': 'POST' });
+
+ let received = '';
+
+ req.on('response', common.mustCall(() => {
+ // _write() will schedule the response body only after it has retained the
+ // callback, guaranteeing that the native write is still in progress.
+ stallWrites = true;
+ req.write(Buffer.alloc(256));
+ }));
+
+ req.on('data', (chunk) => {
+ received += chunk;
+ });
+
+ req.on('end', common.mustCall(() => {
+ assert.strictEqual(received, BODY);
+ assert.ok(heldCallbacks.length > 0,
+ 'test did not actually stall a socket write');
+
+ // Let the stalled writes complete so that everything can shut down.
+ stallWrites = false;
+ for (const callback of heldCallbacks) callback();
+
+ client.destroy();
+ server.close();
+ }));
+}));

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();
+ });
+ });
+}

src/node_http2.cc

@@ -981,45 +981,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);
@@ -1028,14 +1007,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();
@@ -1485,15 +1456,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;
 }
 
@@ -1576,7 +1538,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);
 
@@ -1891,11 +1852,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();
@@ -1942,7 +1898,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();
  }
@@ -2207,7 +2163,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
@@ -2222,40 +2178,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

@@ -85,9 +85,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
@@ -698,7 +697,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)
 
@@ -979,7 +977,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;

test/parallel/test-http2-bidirectional-write-deadlock.js

@@ -0,0 +1,107 @@
+'use strict';
+
+// Regression test against deadlocks between two HTTP/2 peers that are both
+// writing at the same time.
+//
+// To bound how much it buffered while output was backed up, an Http2Session
+// used to stop reading from its socket whenever a write was in flight, and
+// resume only once that write completed. When the peer was itself blocked
+// writing, that write never completed, so the session never read again and
+// the connection hung forever with no error and no timeout.
+//
+// Rather than relying on kernel socket buffers filling up - which depends on
+// the platform and configured window sizes - this models one half of that
+// cycle directly. The client's socket forwards a write but does not report it
+// as complete, substituting for a write blocked because the peer is not
+// reading. Only after that write is stalled does the server send its response
+// body. A session that stops reading while writing never sees it.
+
+const common = require('../common');
+if (!common.hasCrypto)
+ common.skip('missing crypto');
+const assert = require('assert');
+const http2 = require('http2');
+const net = require('net');
+const { Duplex } = require('stream');
+
+const BODY = 'the response body';
+
+const heldCallbacks = [];
+let serverStream;
+
+let stallWrites = false;
+
+// Client-side socket that forwards writes to a real connection, but can leave
+// their completion callbacks pending to model a transport-blocked write.
+class StalledClientSocket extends Duplex {
+ constructor(port) {
+ super();
+ this.inner = net.connect(port, common.localhostIPv4);
+ this.inner.on('data', (chunk) => this.push(chunk));
+ }
+ _read() {
+ // Incoming data is pushed as it arrives.
+ }
+ _write(chunk, encoding, callback) {
+ this.inner.write(chunk, encoding);
+ if (stallWrites) {
+ heldCallbacks.push(callback);
+ // Avoid writing from the server re-entrantly inside _write(). The
+ // ordering is still explicit: this callback is already held.
+ setImmediate(() => serverStream.end(BODY));
+ return;
+ }
+ callback();
+ }
+ _final(callback) {
+ callback();
+ }
+ _destroy(err, callback) {
+ this.inner.destroy();
+ callback(err);
+ }
+}
+
+const server = http2.createServer();
+
+server.on('stream', common.mustCall((stream) => {
+ // Send headers first. Their response event starts the stalled client write.
+ stream.respond();
+ serverStream = stream;
+}));
+
+server.listen(0, common.mustCall(() => {
+ const port = server.address().port;
+
+ const client = http2.connect(`http://${common.localhostIPv4}:${port}`, {
+ createConnection: () => new StalledClientSocket(port),
+ });
+
+ const req = client.request({ ':method': 'POST' });
+
+ let received = '';
+
+ req.on('response', common.mustCall(() => {
+ // _write() will schedule the response body only after it has retained the
+ // callback, guaranteeing that the native write is still in progress.
+ stallWrites = true;
+ req.write(Buffer.alloc(256));
+ }));
+
+ req.on('data', (chunk) => {
+ received += chunk;
+ });
+
+ req.on('end', common.mustCall(() => {
+ assert.strictEqual(received, BODY);
+ assert.ok(heldCallbacks.length > 0,
+ 'test did not actually stall a socket write');
+
+ // Let the stalled writes complete so that everything can shut down.
+ stallWrites = false;
+ for (const callback of heldCallbacks) callback();
+
+ client.destroy();
+ server.close();
+ }));
+}));

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();
+ });
+ });
+}

src/node_http2.cc

@@ -981,45 +981,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);
@@ -1028,14 +1007,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();
@@ -1485,15 +1456,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;
 }
 
@@ -1576,7 +1538,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);
 
@@ -1891,11 +1852,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();
@@ -1942,7 +1898,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();
  }
@@ -2207,7 +2163,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
@@ -2222,40 +2178,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

@@ -85,9 +85,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
@@ -698,7 +697,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)
 
@@ -979,7 +977,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;

test/parallel/test-http2-bidirectional-write-deadlock.js

@@ -0,0 +1,107 @@
+'use strict';
+
+// Regression test against deadlocks between two HTTP/2 peers that are both
+// writing at the same time.
+//
+// To bound how much it buffered while output was backed up, an Http2Session
+// used to stop reading from its socket whenever a write was in flight, and
+// resume only once that write completed. When the peer was itself blocked
+// writing, that write never completed, so the session never read again and
+// the connection hung forever with no error and no timeout.
+//
+// Rather than relying on kernel socket buffers filling up - which depends on
+// the platform and configured window sizes - this models one half of that
+// cycle directly. The client's socket forwards a write but does not report it
+// as complete, substituting for a write blocked because the peer is not
+// reading. Only after that write is stalled does the server send its response
+// body. A session that stops reading while writing never sees it.
+
+const common = require('../common');
+if (!common.hasCrypto)
+ common.skip('missing crypto');
+const assert = require('assert');
+const http2 = require('http2');
+const net = require('net');
+const { Duplex } = require('stream');
+
+const BODY = 'the response body';
+
+const heldCallbacks = [];
+let serverStream;
+
+let stallWrites = false;
+
+// Client-side socket that forwards writes to a real connection, but can leave
+// their completion callbacks pending to model a transport-blocked write.
+class StalledClientSocket extends Duplex {
+ constructor(port) {
+ super();
+ this.inner = net.connect(port, common.localhostIPv4);
+ this.inner.on('data', (chunk) => this.push(chunk));
+ }
+ _read() {
+ // Incoming data is pushed as it arrives.
+ }
+ _write(chunk, encoding, callback) {
+ this.inner.write(chunk, encoding);
+ if (stallWrites) {
+ heldCallbacks.push(callback);
+ // Avoid writing from the server re-entrantly inside _write(). The
+ // ordering is still explicit: this callback is already held.
+ setImmediate(() => serverStream.end(BODY));
+ return;
+ }
+ callback();
+ }
+ _final(callback) {
+ callback();
+ }
+ _destroy(err, callback) {
+ this.inner.destroy();
+ callback(err);
+ }
+}
+
+const server = http2.createServer();
+
+server.on('stream', common.mustCall((stream) => {
+ // Send headers first. Their response event starts the stalled client write.
+ stream.respond();
+ serverStream = stream;
+}));
+
+server.listen(0, common.mustCall(() => {
+ const port = server.address().port;
+
+ const client = http2.connect(`http://${common.localhostIPv4}:${port}`, {
+ createConnection: () => new StalledClientSocket(port),
+ });
+
+ const req = client.request({ ':method': 'POST' });
+
+ let received = '';
+
+ req.on('response', common.mustCall(() => {
+ // _write() will schedule the response body only after it has retained the
+ // callback, guaranteeing that the native write is still in progress.
+ stallWrites = true;
+ req.write(Buffer.alloc(256));
+ }));
+
+ req.on('data', (chunk) => {
+ received += chunk;
+ });
+
+ req.on('end', common.mustCall(() => {
+ assert.strictEqual(received, BODY);
+ assert.ok(heldCallbacks.length > 0,
+ 'test did not actually stall a socket write');
+
+ // Let the stalled writes complete so that everything can shut down.
+ stallWrites = false;
+ for (const callback of heldCallbacks) callback();
+
+ client.destroy();
+ server.close();
+ }));
+}));

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();
+ });
+ });
+}

src/node_http2.cc

@@ -981,45 +981,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);
@@ -1028,14 +1007,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();
@@ -1485,15 +1456,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;
 }
 
@@ -1576,7 +1538,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);
 
@@ -1891,11 +1852,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();
@@ -1942,7 +1898,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();
  }
@@ -2207,7 +2163,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
@@ -2222,40 +2178,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

@@ -85,9 +85,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
@@ -698,7 +697,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)
 
@@ -979,7 +977,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;

test/parallel/test-http2-bidirectional-write-deadlock.js

@@ -0,0 +1,107 @@
+'use strict';
+
+// Regression test against deadlocks between two HTTP/2 peers that are both
+// writing at the same time.
+//
+// To bound how much it buffered while output was backed up, an Http2Session
+// used to stop reading from its socket whenever a write was in flight, and
+// resume only once that write completed. When the peer was itself blocked
+// writing, that write never completed, so the session never read again and
+// the connection hung forever with no error and no timeout.
+//
+// Rather than relying on kernel socket buffers filling up - which depends on
+// the platform and configured window sizes - this models one half of that
+// cycle directly. The client's socket forwards a write but does not report it
+// as complete, substituting for a write blocked because the peer is not
+// reading. Only after that write is stalled does the server send its response
+// body. A session that stops reading while writing never sees it.
+
+const common = require('../common');
+if (!common.hasCrypto)
+ common.skip('missing crypto');
+const assert = require('assert');
+const http2 = require('http2');
+const net = require('net');
+const { Duplex } = require('stream');
+
+const BODY = 'the response body';
+
+const heldCallbacks = [];
+let serverStream;
+
+let stallWrites = false;
+
+// Client-side socket that forwards writes to a real connection, but can leave
+// their completion callbacks pending to model a transport-blocked write.
+class StalledClientSocket extends Duplex {
+ constructor(port) {
+ super();
+ this.inner = net.connect(port, common.localhostIPv4);
+ this.inner.on('data', (chunk) => this.push(chunk));
+ }
+ _read() {
+ // Incoming data is pushed as it arrives.
+ }
+ _write(chunk, encoding, callback) {
+ this.inner.write(chunk, encoding);
+ if (stallWrites) {
+ heldCallbacks.push(callback);
+ // Avoid writing from the server re-entrantly inside _write(). The
+ // ordering is still explicit: this callback is already held.
+ setImmediate(() => serverStream.end(BODY));
+ return;
+ }
+ callback();
+ }
+ _final(callback) {
+ callback();
+ }
+ _destroy(err, callback) {
+ this.inner.destroy();
+ callback(err);
+ }
+}
+
+const server = http2.createServer();
+
+server.on('stream', common.mustCall((stream) => {
+ // Send headers first. Their response event starts the stalled client write.
+ stream.respond();
+ serverStream = stream;
+}));
+
+server.listen(0, common.mustCall(() => {
+ const port = server.address().port;
+
+ const client = http2.connect(`http://${common.localhostIPv4}:${port}`, {
+ createConnection: () => new StalledClientSocket(port),
+ });
+
+ const req = client.request({ ':method': 'POST' });
+
+ let received = '';
+
+ req.on('response', common.mustCall(() => {
+ // _write() will schedule the response body only after it has retained the
+ // callback, guaranteeing that the native write is still in progress.
+ stallWrites = true;
+ req.write(Buffer.alloc(256));
+ }));
+
+ req.on('data', (chunk) => {
+ received += chunk;
+ });
+
+ req.on('end', common.mustCall(() => {
+ assert.strictEqual(received, BODY);
+ assert.ok(heldCallbacks.length > 0,
+ 'test did not actually stall a socket write');
+
+ // Let the stalled writes complete so that everything can shut down.
+ stallWrites = false;
+ for (const callback of heldCallbacks) callback();
+
+ client.destroy();
+ server.close();
+ }));
+}));

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();
+ });
+ });
+}

src/node_http2.cc

@@ -981,45 +981,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);
@@ -1028,14 +1007,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();
@@ -1485,15 +1456,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;
 }
 
@@ -1576,7 +1538,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);
 
@@ -1891,11 +1852,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();
@@ -1942,7 +1898,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();
  }
@@ -2207,7 +2163,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
@@ -2222,40 +2178,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

@@ -85,9 +85,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
@@ -698,7 +697,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)
 
@@ -979,7 +977,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;

test/parallel/test-http2-bidirectional-write-deadlock.js

@@ -0,0 +1,107 @@
+'use strict';
+
+// Regression test against deadlocks between two HTTP/2 peers that are both
+// writing at the same time.
+//
+// To bound how much it buffered while output was backed up, an Http2Session
+// used to stop reading from its socket whenever a write was in flight, and
+// resume only once that write completed. When the peer was itself blocked
+// writing, that write never completed, so the session never read again and
+// the connection hung forever with no error and no timeout.
+//
+// Rather than relying on kernel socket buffers filling up - which depends on
+// the platform and configured window sizes - this models one half of that
+// cycle directly. The client's socket forwards a write but does not report it
+// as complete, substituting for a write blocked because the peer is not
+// reading. Only after that write is stalled does the server send its response
+// body. A session that stops reading while writing never sees it.
+
+const common = require('../common');
+if (!common.hasCrypto)
+ common.skip('missing crypto');
+const assert = require('assert');
+const http2 = require('http2');
+const net = require('net');
+const { Duplex } = require('stream');
+
+const BODY = 'the response body';
+
+const heldCallbacks = [];
+let serverStream;
+
+let stallWrites = false;
+
+// Client-side socket that forwards writes to a real connection, but can leave
+// their completion callbacks pending to model a transport-blocked write.
+class StalledClientSocket extends Duplex {
+ constructor(port) {
+ super();
+ this.inner = net.connect(port, common.localhostIPv4);
+ this.inner.on('data', (chunk) => this.push(chunk));
+ }
+ _read() {
+ // Incoming data is pushed as it arrives.
+ }
+ _write(chunk, encoding, callback) {
+ this.inner.write(chunk, encoding);
+ if (stallWrites) {
+ heldCallbacks.push(callback);
+ // Avoid writing from the server re-entrantly inside _write(). The
+ // ordering is still explicit: this callback is already held.
+ setImmediate(() => serverStream.end(BODY));
+ return;
+ }
+ callback();
+ }
+ _final(callback) {
+ callback();
+ }
+ _destroy(err, callback) {
+ this.inner.destroy();
+ callback(err);
+ }
+}
+
+const server = http2.createServer();
+
+server.on('stream', common.mustCall((stream) => {
+ // Send headers first. Their response event starts the stalled client write.
+ stream.respond();
+ serverStream = stream;
+}));
+
+server.listen(0, common.mustCall(() => {
+ const port = server.address().port;
+
+ const client = http2.connect(`http://${common.localhostIPv4}:${port}`, {
+ createConnection: () => new StalledClientSocket(port),
+ });
+
+ const req = client.request({ ':method': 'POST' });
+
+ let received = '';
+
+ req.on('response', common.mustCall(() => {
+ // _write() will schedule the response body only after it has retained the
+ // callback, guaranteeing that the native write is still in progress.
+ stallWrites = true;
+ req.write(Buffer.alloc(256));
+ }));
+
+ req.on('data', (chunk) => {
+ received += chunk;
+ });
+
+ req.on('end', common.mustCall(() => {
+ assert.strictEqual(received, BODY);
+ assert.ok(heldCallbacks.length > 0,
+ 'test did not actually stall a socket write');
+
+ // Let the stalled writes complete so that everything can shut down.
+ stallWrites = false;
+ for (const callback of heldCallbacks) callback();
+
+ client.destroy();
+ server.close();
+ }));
+}));

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();
+ });
+ });
+}

src/node_http2.cc

@@ -981,45 +981,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);
@@ -1028,14 +1007,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();
@@ -1485,15 +1456,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;
 }
 
@@ -1576,7 +1538,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);
 
@@ -1891,11 +1852,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();
@@ -1942,7 +1898,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();
  }
@@ -2207,7 +2163,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
@@ -2222,40 +2178,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

@@ -85,9 +85,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
@@ -698,7 +697,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)
 
@@ -979,7 +977,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;

test/parallel/test-http2-bidirectional-write-deadlock.js

@@ -0,0 +1,107 @@
+'use strict';
+
+// Regression test against deadlocks between two HTTP/2 peers that are both
+// writing at the same time.
+//
+// To bound how much it buffered while output was backed up, an Http2Session
+// used to stop reading from its socket whenever a write was in flight, and
+// resume only once that write completed. When the peer was itself blocked
+// writing, that write never completed, so the session never read again and
+// the connection hung forever with no error and no timeout.
+//
+// Rather than relying on kernel socket buffers filling up - which depends on
+// the platform and configured window sizes - this models one half of that
+// cycle directly. The client's socket forwards a write but does not report it
+// as complete, substituting for a write blocked because the peer is not
+// reading. Only after that write is stalled does the server send its response
+// body. A session that stops reading while writing never sees it.
+
+const common = require('../common');
+if (!common.hasCrypto)
+ common.skip('missing crypto');
+const assert = require('assert');
+const http2 = require('http2');
+const net = require('net');
+const { Duplex } = require('stream');
+
+const BODY = 'the response body';
+
+const heldCallbacks = [];
+let serverStream;
+
+let stallWrites = false;
+
+// Client-side socket that forwards writes to a real connection, but can leave
+// their completion callbacks pending to model a transport-blocked write.
+class StalledClientSocket extends Duplex {
+ constructor(port) {
+ super();
+ this.inner = net.connect(port, common.localhostIPv4);
+ this.inner.on('data', (chunk) => this.push(chunk));
+ }
+ _read() {
+ // Incoming data is pushed as it arrives.
+ }
+ _write(chunk, encoding, callback) {
+ this.inner.write(chunk, encoding);
+ if (stallWrites) {
+ heldCallbacks.push(callback);
+ // Avoid writing from the server re-entrantly inside _write(). The
+ // ordering is still explicit: this callback is already held.
+ setImmediate(() => serverStream.end(BODY));
+ return;
+ }
+ callback();
+ }
+ _final(callback) {
+ callback();
+ }
+ _destroy(err, callback) {
+ this.inner.destroy();
+ callback(err);
+ }
+}
+
+const server = http2.createServer();
+
+server.on('stream', common.mustCall((stream) => {
+ // Send headers first. Their response event starts the stalled client write.
+ stream.respond();
+ serverStream = stream;
+}));
+
+server.listen(0, common.mustCall(() => {
+ const port = server.address().port;
+
+ const client = http2.connect(`http://${common.localhostIPv4}:${port}`, {
+ createConnection: () => new StalledClientSocket(port),
+ });
+
+ const req = client.request({ ':method': 'POST' });
+
+ let received = '';
+
+ req.on('response', common.mustCall(() => {
+ // _write() will schedule the response body only after it has retained the
+ // callback, guaranteeing that the native write is still in progress.
+ stallWrites = true;
+ req.write(Buffer.alloc(256));
+ }));
+
+ req.on('data', (chunk) => {
+ received += chunk;
+ });
+
+ req.on('end', common.mustCall(() => {
+ assert.strictEqual(received, BODY);
+ assert.ok(heldCallbacks.length > 0,
+ 'test did not actually stall a socket write');
+
+ // Let the stalled writes complete so that everything can shut down.
+ stallWrites = false;
+ for (const callback of heldCallbacks) callback();
+
+ client.destroy();
+ server.close();
+ }));
+}));