http2: keep reading during small client writes

Signed-off-by: Matteo Collina <[email protected]>

Author: mcollina

benchmark/http2/client-delayed-write.js

@@ -0,0 +1,116 @@
+'use strict';
+
+const common = require('../common.js');
+const net = require('node:net');
+const { Duplex } = require('node:stream');
+
+const WINDOW_SIZE = 32 * 1024 * 1024;
+
+const bench = common.createBenchmark(main, {
+  n: [500],
+  parallel: [25],
+  size: [384 * 1024],
+  delay: [0, 5],
+}, { flags: ['--no-warnings'] });
+
+class DelayedWriteSocket extends Duplex {
+  constructor(port, delay, host = '127.0.0.1') {
+    super();
+    this.delay = delay;
+    this.inner = net.connect(port, host);
+    this.inner.on('data', (chunk) => {
+      if (!this.push(chunk)) this.inner.pause();
+    });
+    this.inner.on('end', () => this.push(null));
+    this.inner.on('error', (err) => this.destroy(err));
+    this.inner.on('close', () => this.destroy());
+  }
+
+  _read() {
+    this.inner.resume();
+  }
+
+  _write(chunk, encoding, callback) {
+    if (!this.inner.write(chunk, encoding)) {
+      this.inner.once('drain', () => setTimeout(callback, this.delay));
+      return;
+    }
+    setTimeout(callback, this.delay);
+  }
+
+  _final(callback) {
+    this.inner.end();
+    setTimeout(callback, this.delay);
+  }
+
+  _destroy(err, callback) {
+    this.inner.destroy();
+    callback(err);
+  }
+}
+
+function once(emitter, event) {
+  return new Promise((resolve, reject) => {
+    emitter.once(event, resolve);
+    emitter.once('error', reject);
+  });
+}
+
+function fetchHttp2(client) {
+  return new Promise((resolve, reject) => {
+    const req = client.request({ ':path': '/' });
+    let total = 0;
+    req.on('data', (chunk) => {
+      total += chunk.length;
+    });
+    req.on('end', () => resolve(total));
+    req.on('error', reject);
+    req.end();
+  });
+}
+
+async function main({ n, parallel, size, delay }) {
+  const http2 = require('node:http2');
+  const payload = Buffer.alloc(size, 'x');
+  const server = http2.createServer();
+
+  server.on('stream', (stream) => {
+    stream.respond({
+      ':status': 200,
+      'content-length': payload.length,
+      'content-type': 'application/octet-stream',
+    });
+    stream.end(payload);
+  });
+
+  server.listen(0, '127.0.0.1');
+  await once(server, 'listening');
+
+  const port = server.address().port;
+  const client = http2.connect(`http://127.0.0.1:${port}`, {
+    settings: { initialWindowSize: WINDOW_SIZE },
+    createConnection: delay === 0 ? undefined : () => {
+      return new DelayedWriteSocket(port, delay);
+    },
+  });
+
+  try {
+    await once(client, 'connect');
+    client.setLocalWindowSize(WINDOW_SIZE);
+
+    // Warm up the session so connection establishment does not dominate.
+    await fetchHttp2(client);
+
+    bench.start();
+    for (let completed = 0; completed < n; completed += parallel) {
+      const batch = Math.min(parallel, n - completed);
+      await Promise.all(
+        Array.from({ length: batch }, () => fetchHttp2(client)),
+      );
+    }
+    bench.end(n);
+  } finally {
+    client.close();
+    server.close();
+  }
+}

src/node_http2.cc

@@ -54,6 +54,11 @@ namespace http2 {
 namespace {
 
 const char zero_bytes_256[256] = {};
+// Small client-side control frame writes (e.g. WINDOW_UPDATE) should not
+// immediately force HTTP/2 input processing to become effectively sequential
+// if the write callback lingers for a short time.
+constexpr size_t kMaxReadWhileWriteInProgress = 1024;
+constexpr uint64_t kMinWriteAgeToKeepReading = 1'000'000;
 
 bool HasHttp2Observer(Environment* env) {
   AliasedUint32Array& observers = env->performance_state()->observers;
@@ -1396,7 +1401,11 @@ int Http2Session::OnDataChunkReceived(nghttp2_session* handle,
 
   // 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()) {
+  if (session->is_write_in_progress() &&
+      (session->session_type_ == NGHTTP2_SESSION_SERVER ||
+       session->current_write_size_ > kMaxReadWhileWriteInProgress ||
+       uv_hrtime() - session->current_write_start_time_ <
+           kMinWriteAgeToKeepReading)) {
     CHECK(session->is_reading_stopped());
     session->set_receive_paused();
     Debug(session, "receive paused");
@@ -1774,6 +1783,8 @@ void Http2Session::OnStreamAfterWrite(WriteWrap* w, int status) {
   MaybeNotifyGracefulCloseComplete();
   CHECK(is_write_in_progress());
   set_write_in_progress(false);
+  current_write_size_ = 0;
+  current_write_start_time_ = 0;
 
   // Inform all pending writes about their completion.
   ClearOutgoing(status);
@@ -1989,10 +2000,14 @@ uint8_t Http2Session::SendPendingData() {
   chunks_sent_since_last_write_++;
 
   CHECK(!is_write_in_progress());
+  current_write_size_ = outgoing_length_;
+  current_write_start_time_ = uv_hrtime();
   set_write_in_progress();
   StreamWriteResult res = underlying_stream()->Write(*bufs, count);
   if (!res.async) {
     set_write_in_progress(false);
+    current_write_size_ = 0;
+    current_write_start_time_ = 0;
     ClearOutgoing(res.err);
     MaybeNotifyGracefulCloseComplete();
   }

src/node_http2.h

@@ -948,6 +948,8 @@ class Http2Session : public AsyncWrap,
   std::vector<NgHttp2StreamWrite> outgoing_buffers_;
   std::vector<uint8_t> outgoing_storage_;
   size_t outgoing_length_ = 0;
+  size_t current_write_size_ = 0;
+  uint64_t current_write_start_time_ = 0;
   std::vector<int32_t> pending_rst_streams_;
   // Count streams that have been rejected while being opened. Exceeding a fixed
   // limit will result in the session being destroyed, as an indication of a