http,net,stream: optimize write and parser paths

Signed-off-by: GetThatCookie <NimmenKeks@gmx.de>

Author: GetThatCookie

benchmark/common.js

@@ -26,7 +26,7 @@ class Benchmark {
 
     // Parse job-specific configuration from the command line arguments
     const argv = process.argv.slice(2);
-    const parsed_args = this._parseArgs(argv, configs, options);
+    const parsed_args = this._parseArgs([...argv], configs, options);
 
     this.originalOptions = options;
     this.options = parsed_args.cli;
@@ -38,8 +38,10 @@ class Benchmark {
       const groupNames = process.env.NODE_RUN_BENCHMARK_GROUPS?.split(',') ?? Object.keys(configs);
 
       for (const groupName of groupNames) {
-        const config = { ...configs[groupName][0], group: groupName };
-        const parsed_args = this._parseArgs(argv, config, options);
+        const groupConfig = Array.isArray(configs[groupName]) ?
+          configs[groupName][0] : configs[groupName];
+        const config = { ...groupConfig, group: groupName };
+        const parsed_args = this._parseArgs([...argv], config, options);
 
         this.options = parsed_args.cli;
         this.extra_options = parsed_args.extra;
@@ -221,6 +223,9 @@ class Benchmark {
       // function.
       const childEnv = { ...process.env };
       childEnv.NODE_RUN_BENCHMARK_FN = '';
+      if (this.originalOptions.byGroups) {
+        childEnv.NODE_RUN_BENCHMARK_GROUPS = config.group;
+      }
 
       // Create configuration arguments
       const childArgs = [];

benchmark/http/bench-parser.js

@@ -31,6 +31,8 @@ function main({ len, n }) {
   function newParser(type) {
     const parser = new HTTPParser();
     parser.initialize(type, {});
+    // Direct parsers bypass cleanParser(); use its production default.
+    parser.maxHeaderPairs = 2000;
 
     parser.headers = [];
 

benchmark/http/cork.js

@@ -0,0 +1,157 @@
+'use strict';
+
+const common = require('../common.js');
+const protocols = process.versions.openssl ? ['http', 'https'] : ['http'];
+
+const configs = {
+  sameTurn: [{
+    type: ['bytes', 'buffer', 'uint8array'],
+    len: [64, 1024],
+    chunks: [1, 2, 4, 16],
+    mode: ['auto', 'explicit'],
+    transfer: ['chunked', 'length'],
+    protocol: protocols,
+    producer: ['sync'],
+    callback: [0],
+    c: [50],
+    duration: 5,
+  }],
+  streaming: [{
+    type: ['bytes', 'buffer', 'uint8array'],
+    len: [64, 1024],
+    chunks: [4],
+    mode: ['auto'],
+    transfer: ['chunked'],
+    protocol: protocols,
+    producer: ['nextTick', 'microtask', 'immediate'],
+    callback: [0],
+    c: [50],
+    duration: 5,
+  }],
+  callbacks: [{
+    type: ['bytes', 'buffer', 'uint8array'],
+    len: [64],
+    chunks: [4, 16],
+    mode: ['auto', 'explicit'],
+    transfer: ['chunked'],
+    protocol: protocols,
+    producer: ['sync'],
+    callback: [1],
+    c: [50],
+    duration: 5,
+  }],
+  fixedBody: [{
+    type: ['bytes', 'buffer', 'uint8array'],
+    total: [64 * 1024],
+    chunks: [1, 4, 16, 128],
+    mode: ['auto', 'explicit'],
+    transfer: ['chunked'],
+    protocol: protocols,
+    producer: ['sync'],
+    callback: [0],
+    c: [50],
+    duration: 5,
+  }],
+  largeChunks: [{
+    type: ['bytes', 'buffer', 'uint8array'],
+    len: [4 * 1024, 8 * 1024, 16 * 1024, 64 * 1024],
+    chunks: [1, 4],
+    mode: ['auto'],
+    transfer: ['chunked'],
+    protocol: protocols,
+    producer: ['sync'],
+    callback: [0],
+    c: [50],
+    duration: 5,
+  }],
+  concurrency: [{
+    type: ['bytes'],
+    len: [64],
+    chunks: [4],
+    mode: ['auto'],
+    transfer: ['chunked'],
+    protocol: protocols,
+    producer: ['sync'],
+    callback: [0],
+    c: [1, 50, 500],
+    duration: 5,
+  }],
+};
+
+const bench = common.createBenchmark(main, configs, { byGroups: true });
+
+function main({
+  type,
+  len,
+  chunks,
+  mode,
+  transfer,
+  protocol,
+  producer,
+  callback,
+  c,
+  duration,
+  total,
+}) {
+  const transport = require(protocol);
+  len ??= total / chunks;
+  const chunk = type === 'bytes' ? 'a'.repeat(len) :
+    type === 'buffer' ? Buffer.alloc(len, 'a') :
+      new Uint8Array(len).fill(0x61);
+  const writeCallback = callback ? (err) => {
+    if (err) throw err;
+  } : undefined;
+
+  const schedule = producer === 'nextTick' ? process.nextTick :
+    producer === 'microtask' ? queueMicrotask : setImmediate;
+
+  const onRequest = (req, res) => {
+    if (transfer === 'length') {
+      res.setHeader('Content-Length', len * chunks);
+    }
+    if (mode === 'explicit') {
+      res.cork();
+    }
+
+    if (producer === 'sync') {
+      for (let i = 0; i < chunks; i++) {
+        res.write(chunk, writeCallback);
+      }
+      res.end();
+      return;
+    }
+
+    let written = 0;
+    function writeNext() {
+      if (written++ === chunks) {
+        res.end();
+        return;
+      }
+      res.write(chunk, writeCallback);
+      schedule(writeNext);
+    }
+    writeNext();
+  };
+
+  let server;
+  if (protocol === 'https') {
+    const fixtures = require('../../test/common/fixtures');
+    server = transport.createServer({
+      key: fixtures.readKey('rsa_private.pem'),
+      cert: fixtures.readKey('rsa_cert.crt'),
+    }, onRequest);
+  } else {
+    server = transport.createServer(onRequest);
+  }
+
+  server.listen(0, () => {
+    bench.http({
+      connections: c,
+      duration,
+      port: server.address().port,
+      scheme: protocol,
+    }, () => {
+      server.close();
+    });
+  });
+}

lib/_http_client.js

@@ -58,6 +58,7 @@ const {
   parseUniqueHeadersOption,
   OutgoingMessage,
 } = require('_http_outgoing');
+const { kDestroyMessageBuffer } = require('internal/streams/utils');
 const Agent = require('_http_agent');
 const { Buffer } = require('buffer');
 const { defaultTriggerAsyncIdScope } = require('internal/async_hooks');
@@ -699,7 +700,11 @@ ClientRequest.prototype.destroy = function destroy(err) {
   }
 
   this[kError] = err;
-  this.socket?.destroy(err);
+  try {
+    this[kDestroyMessageBuffer](err);
+  } finally {
+    this.socket?.destroy(err);
+  }
 
   return this;
 };
@@ -710,7 +715,7 @@ function emitAbortNT(req) {
 
 function ondrain() {
   const msg = this._httpMessage;
-  if (msg && !msg.finished && msg[kNeedDrain]) {
+  if (msg && !msg.finished && msg[kNeedDrain] && msg.writableLength === 0) {
     msg[kNeedDrain] = false;
     msg.emit('drain');
   }
@@ -726,6 +731,9 @@ function socketCloseListener() {
   const parser = socket.parser;
   const res = req.res;
 
+  req[kDestroyMessageBuffer](
+    req[kError] ?? socket._writableState.errored,
+  );
   req.destroyed = true;
   if (res) {
     // Socket closed before we emitted 'end' below.