http,net,stream: reconcile read, write, and parser paths

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,139 @@
+'use strict';
+
+const common = require('../common.js');
+const protocols = process.versions.openssl ? ['http', 'https'] : ['http'];
+
+const scenarios = {
+  'end-64': {
+    len: 64,
+    chunks: 1,
+    endChunk: true,
+  },
+  'end-1024': {
+    len: 1024,
+    chunks: 1,
+    endChunk: true,
+  },
+  'end-1025': {
+    len: 1025,
+    chunks: 1,
+    endChunk: true,
+  },
+  'auto-4': {
+    len: 64,
+    chunks: 4,
+  },
+  'auto-16': {
+    len: 64,
+    chunks: 16,
+  },
+  'explicit-16': {
+    len: 64,
+    chunks: 16,
+    explicit: true,
+  },
+  'content-length-16': {
+    len: 64,
+    chunks: 16,
+    contentLength: true,
+  },
+  'next-tick-4': {
+    len: 64,
+    chunks: 4,
+    schedule: process.nextTick,
+  },
+  'callbacks-16': {
+    len: 64,
+    chunks: 16,
+    callbacks: true,
+  },
+  'fixed-body-128': {
+    len: 512,
+    chunks: 128,
+  },
+  'large-16k': {
+    len: 16 * 1024,
+    chunks: 4,
+  },
+};
+
+const bench = common.createBenchmark(main, {
+  type: ['string', 'buffer', 'uint8array'],
+  scenario: Object.keys(scenarios),
+  protocol: protocols,
+  c: [50],
+  duration: [5],
+});
+
+function main({ type, scenario, protocol, c, duration }) {
+  const {
+    callbacks,
+    chunks,
+    contentLength,
+    endChunk,
+    explicit,
+    len,
+    schedule,
+  } = scenarios[scenario];
+  const transport = require(protocol);
+  const chunk = type === 'string' ? 'a'.repeat(len) :
+    type === 'buffer' ? Buffer.alloc(len, 'a') :
+      new Uint8Array(len).fill(0x61);
+  const writeCallback = callbacks ? (err) => {
+    if (err) throw err;
+  } : undefined;
+
+  const onRequest = (req, res) => {
+    if (contentLength) {
+      res.setHeader('Content-Length', len * chunks);
+    }
+    if (explicit) {
+      res.cork();
+    }
+    if (endChunk) {
+      res.end(chunk, writeCallback);
+      return;
+    }
+
+    if (schedule === undefined) {
+      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();
+    });
+  });
+}

benchmark/http2/compat.js

@@ -6,17 +6,24 @@ const fs = require('fs');
 const file = path.join(path.resolve(__dirname, '../fixtures'), 'alice.html');
 
 const bench = common.createBenchmark(main, {
+  response: ['end', 'pipe'],
+  size: [64],
   requests: [100, 1000, 5000],
   streams: [1, 10, 20, 40, 100, 200],
   clients: [2],
   benchmarker: ['test-double-http2'],
   duration: 5,
 }, { flags: ['--no-warnings'] });
 
-function main({ requests, streams, clients, duration }) {
+function main({ response, size, requests, streams, clients, duration }) {
   const http2 = require('http2');
+  const body = 'a'.repeat(size);
   const server = http2.createServer();
   server.on('request', (req, res) => {
+    if (response === 'end') {
+      res.end(body);
+      return;
+    }
     const out = fs.createReadStream(file);
     res.setHeader('content-type', 'text/html');
     out.pipe(res);

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.