http2: reduce per-request allocations

Cut several sources of per-stream/per-request overhead on the hot
path:

- Track 'priority'/'frameError' stream listeners by overriding the
  EventEmitter methods on Http2Stream instead of subscribing to
  'newListener'/'removeListener', which made every listener add and
  remove on every stream emit an extra tracking event.
- Replace the per-call SafeSet and sensitive-header mapping in
  buildNgHeaderString with a lazily allocated array and an
  empty-array fast path, and skip the HTTP token regex and
  connection-specific header checks for well-known single-value
  header names.
- Replace per-call closures with shared named handlers in
  onStreamClose, afterShutdown and Http2Stream._destroy.
- Skip the pendingStreams Set add/delete for streams that are
  created with their native handle already available (all server
  streams).
- Hoist the per-request onStreamTimeout closure factories in the
  compat layer to module-level handlers, and avoid a once() wrapper
  allocation per server stream.

h2load, 1 KiB response payload, -c 4 -m 100, mean of 6 alternating
runs: core API 60.2k -> 69.3k req/s (+15%), compat API 43.6k ->
46.2k req/s (+5.9%).

Signed-off-by: Matteo Collina <hello@matteocollina.com>
PR-URL: #64265
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: Gürgün Dayıoğlu <hey@gurgun.day>
Reviewed-By: Tim Perry <pimterry@gmail.com>

Author: mcollina

lib/internal/http2/compat.js

@@ -300,11 +300,12 @@ function onStreamCloseRequest() {
   req.emit('close');
 }
 
-function onStreamTimeout(kind) {
-  return function onStreamTimeout() {
-    const obj = this[kind];
-    obj.emit('timeout');
-  };
+function onStreamTimeoutRequest() {
+  this[kRequest].emit('timeout');
+}
+
+function onStreamTimeoutResponse() {
+  this[kResponse].emit('timeout');
 }
 
 class Http2ServerRequest extends Readable {
@@ -332,7 +333,7 @@ class Http2ServerRequest extends Readable {
     stream.on('error', onStreamError);
     stream.on('aborted', onStreamAbortedRequest);
     stream.on('close', onStreamCloseRequest);
-    stream.on('timeout', onStreamTimeout(kRequest));
+    stream.on('timeout', onStreamTimeoutRequest);
     this.on('pause', onRequestPause);
     this.on('resume', onRequestResume);
   }
@@ -488,7 +489,7 @@ class Http2ServerResponse extends Stream {
     stream.on('aborted', onStreamAbortedResponse);
     stream.on('close', onStreamCloseResponse);
     stream.on('wantTrailers', onStreamTrailersReady);
-    stream.on('timeout', onStreamTimeout(kResponse));
+    stream.on('timeout', onStreamTimeoutResponse);
   }
 
   // User land modules such as finalhandler just check truthiness of this

lib/internal/http2/core.js

@@ -559,29 +559,19 @@ function sessionListenerRemoved(name) {
 }
 
 // Also keep track of listeners for the Http2Stream instances, as some events
-// are emitted on those objects.
-function streamListenerAdded(name) {
-  const session = this[kSession];
-  if (!session) return;
-  switch (name) {
-    case 'priority':
-      session[kNativeFields][kSessionPriorityListenerCount]++;
-      break;
-    case 'frameError':
-      session[kNativeFields][kSessionFrameErrorListenerCount]++;
-      break;
-  }
-}
-
-function streamListenerRemoved(name) {
-  const session = this[kSession];
+// are emitted on those objects. Instead of subscribing to 'newListener' and
+// 'removeListener' (which makes every listener add/remove on every stream
+// emit an extra tracking event), Http2Stream overrides the EventEmitter
+// methods and updates the counts directly.
+function trackStreamListener(stream, name, delta) {
+  const session = stream[kSession];
   if (!session) return;
   switch (name) {
     case 'priority':
-      session[kNativeFields][kSessionPriorityListenerCount]--;
+      session[kNativeFields][kSessionPriorityListenerCount] += delta;
       break;
     case 'frameError':
-      session[kNativeFields][kSessionFrameErrorListenerCount]--;
+      session[kNativeFields][kSessionFrameErrorListenerCount] += delta;
       break;
   }
 }
@@ -595,6 +585,20 @@ function onPing(payload) {
   session.emit('ping', payload);
 }
 
+function streamNaturalCloseSettled(stream) {
+  return (stream._readableState.endEmitted ||
+          !!stream._readableState.errored) &&
+         (stream._writableState.finished ||
+          !!stream._writableState.errored);
+}
+
+// Shared 'end'/'finish'/'error' listener for the natural-close path of
+// onStreamClose(). Named and reused to avoid per-stream closures.
+function maybeDestroyNaturalClose() {
+  if (!this.destroyed && streamNaturalCloseSettled(this))
+    this.destroy();
+}
+
 // Fired by C++ when nghttp2's on_stream_close fires. `peerReset` is
 // true when the peer sent a RST_STREAM frame - peer RST_STREAM(NO_ERROR)
 // is otherwise indistinguishable from a clean END_STREAM exchange at
@@ -655,22 +659,14 @@ function onStreamClose(code, peerReset) {
   // errored readable won't fire 'end' - and on a Duplex a writable
   // error propagates to readable.errored, blocking 'end' too. Treat
   // either side's errored state as settled.
-  const readDone = () => stream._readableState.endEmitted ||
-                         !!stream._readableState.errored;
-  const writeDone = () => stream._writableState.finished ||
-                          !!stream._writableState.errored;
-  if (readDone() && writeDone()) {
+  if (streamNaturalCloseSettled(stream)) {
     stream.destroy();
     return true;
   }
 
-  const maybeDestroy = () => {
-    if (!stream.destroyed && readDone() && writeDone())
-      stream.destroy();
-  };
-  stream.once('end', maybeDestroy);
-  stream.once('finish', maybeDestroy);
-  stream.once('error', maybeDestroy);
+  stream.on('end', maybeDestroyNaturalClose);
+  stream.on('finish', maybeDestroyNaturalClose);
+  stream.on('error', maybeDestroyNaturalClose);
   if (stream[kSession][kType] === NGHTTP2_SESSION_SERVER &&
       !stream[kState].didRead &&
       stream.readableFlowing === null) {
@@ -2015,12 +2011,14 @@ function streamOnPause() {
     this[kHandle].readStop();
 }
 
+function streamOnFinishMaybeDestroy() {
+  this[kMaybeDestroy]();
+}
+
 function afterShutdown(status) {
   const stream = this.handle[kOwner];
   if (stream) {
-    stream.on('finish', () => {
-      stream[kMaybeDestroy]();
-    });
+    stream.on('finish', streamOnFinishMaybeDestroy);
   }
   // Currently this status value is unused
   this.callback();
@@ -2132,7 +2130,7 @@ function finishCloseStream(code) {
 // An Http2Stream is a Duplex stream that is backed by a
 // node::http2::Http2Stream handle implementing StreamBase.
 class Http2Stream extends Duplex {
-  constructor(session, options) {
+  constructor(session, options, hasHandle = false) {
     options.allowHalfOpen = true;
     options.decodeStrings = false;
     options.autoDestroy = false;
@@ -2144,7 +2142,10 @@ class Http2Stream extends Duplex {
     // been assigned.
     this.cork();
     this[kSession] = session;
-    session[kState].pendingStreams.add(this);
+    // Streams constructed with their native handle already available (e.g.
+    // server streams) are initialized immediately and never become pending.
+    if (!hasHandle)
+      session[kState].pendingStreams.add(this);
 
     // Allow our logic for determining whether any reads have happened to
     // work in all situations. This is similar to what we do in _http_incoming.
@@ -2166,9 +2167,53 @@ class Http2Stream extends Duplex {
     this[kProxySocket] = null;
 
     this.on('pause', streamOnPause);
+  }
 
-    this.on('newListener', streamListenerAdded);
-    this.on('removeListener', streamListenerRemoved);
+  addListener(name, listener) {
+    const ret = super.addListener(name, listener);
+    if (name === 'priority' || name === 'frameError')
+      trackStreamListener(this, name, 1);
+    return ret;
+  }
+
+  on(name, listener) {
+    const ret = super.on(name, listener);
+    if (name === 'priority' || name === 'frameError')
+      trackStreamListener(this, name, 1);
+    return ret;
+  }
+
+  prependListener(name, listener) {
+    const ret = super.prependListener(name, listener);
+    if (name === 'priority' || name === 'frameError')
+      trackStreamListener(this, name, 1);
+    return ret;
+  }
+
+  removeListener(name, listener) {
+    if (name === 'priority' || name === 'frameError') {
+      const before = this.listenerCount(name);
+      const ret = super.removeListener(name, listener);
+      if (this.listenerCount(name) !== before)
+        trackStreamListener(this, name, -1);
+      return ret;
+    }
+    return super.removeListener(name, listener);
+  }
+
+  removeAllListeners(name) {
+    let priority = 0;
+    let frameError = 0;
+    if (name === undefined || name === 'priority')
+      priority = this.listenerCount('priority');
+    if (name === undefined || name === 'frameError')
+      frameError = this.listenerCount('frameError');
+    const ret = super.removeAllListeners(name);
+    if (priority !== 0)
+      trackStreamListener(this, 'priority', -priority);
+    if (frameError !== 0)
+      trackStreamListener(this, 'frameError', -frameError);
+    return ret;
   }
 
   [kUpdateTimer]() {
@@ -2562,9 +2607,7 @@ class Http2Stream extends Duplex {
     // will destroy if it has been closed and there are no other open or
     // pending streams. Delay with setImmediate so we don't do it on the
     // nghttp2 stack.
-    setImmediate(() => {
-      session[kMaybeDestroy]();
-    });
+    setImmediate(sessionMaybeDestroy, session);
     if (err) {
       if (session[kType] === NGHTTP2_SESSION_CLIENT) {
         if (onClientStreamErrorChannel.hasSubscribers) {
@@ -2615,6 +2658,8 @@ class Http2Stream extends Duplex {
   }
 }
 
+Http2Stream.prototype.off = Http2Stream.prototype.removeListener;
+
 // TODO(aduh95): remove this in future semver-major
 Http2Stream.prototype.priority = deprecate(function priority(options) {
   if (this.destroyed)
@@ -2649,6 +2694,10 @@ function callStreamClose(stream) {
   stream.close();
 }
 
+function sessionMaybeDestroy(session) {
+  session[kMaybeDestroy]();
+}
+
 function prepareResponseHeaders(stream, headersParam, options) {
   let headers;
   let statusCode;
@@ -2961,12 +3010,14 @@ function afterOpen(session, options, headers, streamOptions, err, fd) {
 
 class ServerHttp2Stream extends Http2Stream {
   constructor(session, handle, id, options, headers) {
-    super(session, options);
+    super(session, options, true);
     handle.owner = this;
     this[kInit](id, handle);
     this[kProtocol] = headers[HTTP2_HEADER_SCHEME];
     this[kAuthority] = getAuthority(headers);
-    this.once('finish', autoDrainReadable);
+    // 'finish' is only emitted once, so a regular listener is safe here and
+    // avoids allocating a once() wrapper for every stream.
+    this.on('finish', autoDrainReadable);
   }
 
   // True if the remote peer accepts push streams
@@ -3307,7 +3358,7 @@ ServerHttp2Stream.prototype[kProceed] = ServerHttp2Stream.prototype.respond;
 
 class ClientHttp2Stream extends Http2Stream {
   constructor(session, handle, id, options) {
-    super(session, options);
+    super(session, options, id !== undefined);
     this[kState].flags |= STREAM_FLAGS_HEADERS_SENT;
     if (id !== undefined)
       this[kInit](id, handle);

lib/internal/http2/util.js

@@ -770,14 +770,16 @@ function buildNgHeaderString(arrayOrMap,
   let pseudoHeaders = '';
   let count = 0;
 
-  const singles = new SafeSet();
+  let singles;
   const sensitiveHeaders = arrayOrMap[kSensitiveHeaders] || emptyArray;
-  const neverIndex = sensitiveHeaders.map((v) => v.toLowerCase());
+  const neverIndex = sensitiveHeaders.length === 0 ?
+    emptyArray : sensitiveHeaders.map((v) => v.toLowerCase());
 
   function processHeader(key, value) {
     key = key.toLowerCase();
+    const isSingleValueField = kSingleValueFields.has(key);
     const isStrictSingleValueField = strictSingleValueFields &&
-      kSingleValueFields.has(key);
+      isSingleValueField;
     let isArray = ArrayIsArray(value);
     if (isArray) {
       switch (value.length) {
@@ -795,11 +797,15 @@ function buildNgHeaderString(arrayOrMap,
       value = String(value);
     }
     if (isStrictSingleValueField) {
-      if (singles.has(key))
+      if (singles === undefined) {
+        singles = [key];
+      } else if (singles.includes(key)) {
         throw new ERR_HTTP2_HEADER_SINGLE_VALUE(key);
-      singles.add(key);
+      } else {
+        singles.push(key);
+      }
     }
-    const flags = neverIndex.includes(key) ?
+    const flags = neverIndex.length !== 0 && neverIndex.includes(key) ?
       kNeverIndexFlag :
       kNoHeaderFlags;
     if (key[0] === ':') {
@@ -810,11 +816,15 @@ function buildNgHeaderString(arrayOrMap,
       count++;
       return;
     }
-    if (!checkIsHttpToken(key)) {
-      throw new ERR_INVALID_HTTP_TOKEN('Header name', key);
-    }
-    if (isIllegalConnectionSpecificHeader(key, value)) {
-      throw new ERR_HTTP2_INVALID_CONNECTION_HEADERS(key);
+    // Well-known single-value fields are all valid HTTP tokens and none of
+    // them is a connection-specific header, so both checks can be skipped.
+    if (!isSingleValueField) {
+      if (!checkIsHttpToken(key)) {
+        throw new ERR_INVALID_HTTP_TOKEN('Header name', key);
+      }
+      if (isIllegalConnectionSpecificHeader(key, value)) {
+        throw new ERR_HTTP2_INVALID_CONNECTION_HEADERS(key);
+      }
     }
     if (isArray) {
       for (let j = 0; j < value.length; ++j) {