http: add highWaterMark opt in http.createServer

Add highWaterMark option when creating a new HTTP server.
This option will override the default (readable|writable)
highWaterMark values on sockets created.

Fixes: #46606

doc/api/http.md

@@ -3192,6 +3192,9 @@ Found'`.
 <!-- YAML
 added: v0.1.13
 changes:
+  - version: REPLACEME
+    pr-url: https://github.com/nodejs/node/pull/47405
+    description: The `highWaterMark` option is supported now.
   - version: v18.0.0
     pr-url: https://github.com/nodejs/node/pull/41263
     description: The `requestTimeout`, `headersTimeout`, `keepAliveTimeout`, and
@@ -3229,6 +3232,10 @@ changes:
     the complete HTTP headers from the client.
     See [`server.headersTimeout`][] for more information.
     **Default:** `60000`.
+  * `highWaterMark` {number} Optionally overrides all `socket`s'
+    `readableHighWaterMark` and `writableHighWaterMark`. This affects
+    `highWaterMark` property of both `IncomingMessage` and `ServerResponse`.
+    **Default:** See [`stream.getDefaultHighWaterMark()`][].
   * `insecureHTTPParser` {boolean} Use an insecure HTTP parser that accepts
     invalid HTTP headers when `true`. Using the insecure parser should be
     avoided. See [`--insecure-http-parser`][] for more information.
@@ -3898,6 +3905,7 @@ Set the maximum number of idle HTTP parsers.
 [`socket.setNoDelay()`]: net.md#socketsetnodelaynodelay
 [`socket.setTimeout()`]: net.md#socketsettimeouttimeout-callback
 [`socket.unref()`]: net.md#socketunref
+[`stream.getDefaultHighWaterMark()`]: stream.md#streamgetdefaulthighwatermarkobjectmode
 [`url.parse()`]: url.md#urlparseurlstring-parsequerystring-slashesdenotehost
 [`writable.cork()`]: stream.md#writablecork
 [`writable.destroy()`]: stream.md#writabledestroyerror

doc/api/net.md

@@ -1539,6 +1539,9 @@ then returns the `net.Socket` that starts the connection.
 <!-- YAML
 added: v0.5.0
 changes:
+  - version: REPLACEME
+    pr-url: https://github.com/nodejs/node/pull/47405
+    description: The `highWaterMark` option is supported now.
   - version:
     - v17.7.0
     - v16.15.0
@@ -1551,6 +1554,9 @@ changes:
   * `allowHalfOpen` {boolean} If set to `false`, then the socket will
     automatically end the writable side when the readable side ends.
     **Default:** `false`.
+  * `highWaterMark` {number} Optionally overrides all [`net.Socket`][]s'
+    `readableHighWaterMark` and `writableHighWaterMark`.
+    **Default:** See [`stream.getDefaultHighWaterMark()`][].
   * `pauseOnConnect` {boolean} Indicates whether the socket should be
     paused on incoming connections. **Default:** `false`.
   * `noDelay` {boolean} If set to `true`, it disables the use of Nagle's algorithm immediately
@@ -1780,6 +1786,7 @@ net.isIPv6('fhqwhgads'); // returns false
 [`socket.setKeepAlive(enable, initialDelay)`]: #socketsetkeepaliveenable-initialdelay
 [`socket.setTimeout()`]: #socketsettimeouttimeout-callback
 [`socket.setTimeout(timeout)`]: #socketsettimeouttimeout-callback
+[`stream.getDefaultHighWaterMark()`]: stream.md#streamgetdefaulthighwatermarkobjectmode
 [`writable.destroy()`]: stream.md#writabledestroyerror
 [`writable.destroyed`]: stream.md#writabledestroyed
 [`writable.end()`]: stream.md#writableendchunk-encoding-callback

lib/_http_outgoing.js

@@ -101,7 +101,7 @@ function isContentDispositionField(s) {
   return s.length === 19 && StringPrototypeToLowerCase(s) === 'content-disposition';
 }
 
-function OutgoingMessage() {
+function OutgoingMessage(options) {
   Stream.call(this);
 
   // Queue that holds all currently pending data, until the response will be
@@ -149,7 +149,7 @@ function OutgoingMessage() {
   this._onPendingData = nop;
 
   this[kErrored] = null;
-  this[kHighWaterMark] = getDefaultHighWaterMark();
+  this[kHighWaterMark] = options?.highWaterMark ?? getDefaultHighWaterMark();
 }
 ObjectSetPrototypeOf(OutgoingMessage.prototype, Stream.prototype);
 ObjectSetPrototypeOf(OutgoingMessage, Stream);
@@ -1171,6 +1171,7 @@ function(err, event) {
 };
 
 module.exports = {
+  kHighWaterMark,
   kUniqueHeaders,
   parseUniqueHeadersOption,
   validateHeaderName,

lib/_http_server.js

@@ -188,8 +188,8 @@ class HTTPServerAsyncResource {
   }
 }
 
-function ServerResponse(req) {
-  OutgoingMessage.call(this);
+function ServerResponse(req, options) {
+  OutgoingMessage.call(this, options);
 
   if (req.method === 'HEAD') this._hasBody = false;
 
@@ -513,7 +513,8 @@ function Server(options, requestListener) {
     this,
     { allowHalfOpen: true, noDelay: options.noDelay ?? true,
       keepAlive: options.keepAlive,
-      keepAliveInitialDelay: options.keepAliveInitialDelay });
+      keepAliveInitialDelay: options.keepAliveInitialDelay,
+      highWaterMark: options.highWaterMark });
 
   if (requestListener) {
     this.on('request', requestListener);
@@ -1019,7 +1020,7 @@ function parserOnIncoming(server, socket, state, req, keepAlive) {
     }
   }
 
-  const res = new server[kServerResponse](req);
+  const res = new server[kServerResponse](req, { highWaterMark: socket.writableHighWaterMark });
   res._keepAliveTimeout = server.keepAliveTimeout;
   res._maxRequestsPerSocket = server.maxRequestsPerSocket;
   res._onPendingData = updateOutgoingData.bind(undefined,

lib/http.js

@@ -54,6 +54,7 @@ let maxHeaderSize;
  *   maxHeaderSize?: number;
  *   requireHostHeader?: boolean;
  *   joinDuplicateHeaders?: boolean;
+ *   highWaterMark?: number;
  *   }} [opts]
  * @param {Function} [requestListener]
  * @returns {Server}

lib/net.js

@@ -154,6 +154,7 @@ const {
   startPerf,
   stopPerf,
 } = require('internal/perf/observe');
+const { getDefaultHighWaterMark } = require('internal/streams/state');
 
 function getFlags(ipv6Only) {
   return ipv6Only === true ? TCPConstants.UV_TCP_IPV6ONLY : 0;
@@ -1682,6 +1683,15 @@ function Server(options, connectionListener) {
       options.keepAliveInitialDelay = 0;
     }
   }
+  if (typeof options.highWaterMark !== 'undefined') {
+    validateNumber(
+      options.highWaterMark, 'options.highWaterMark',
+    );
+
+    if (options.highWaterMark < 0) {
+      options.highWaterMark = getDefaultHighWaterMark();
+    }
+  }
 
   this._connections = 0;
 
@@ -1696,6 +1706,7 @@ function Server(options, connectionListener) {
   this.noDelay = Boolean(options.noDelay);
   this.keepAlive = Boolean(options.keepAlive);
   this.keepAliveInitialDelay = ~~(options.keepAliveInitialDelay / 1000);
+  this.highWaterMark = options.highWaterMark ?? getDefaultHighWaterMark();
 }
 ObjectSetPrototypeOf(Server.prototype, EventEmitter.prototype);
 ObjectSetPrototypeOf(Server, EventEmitter);
@@ -2077,6 +2088,8 @@ function onconnection(err, clientHandle) {
     pauseOnCreate: self.pauseOnConnect,
     readable: true,
     writable: true,
+    readableHighWaterMark: self.highWaterMark,
+    writableHighWaterMark: self.highWaterMark,
   });
 
   if (self.noDelay && clientHandle.setNoDelay) {

test/parallel/test-http-server-options-highwatermark.js

@@ -0,0 +1,47 @@
+// Flags: --expose-internals
+'use strict';
+const common = require('../common');
+const assert = require('assert');
+const http = require('http');
+const { kHighWaterMark } = require('_http_outgoing');
+
+const { getDefaultHighWaterMark } = require('internal/streams/state');
+
+function listen(server) {
+  server.listen(0, common.mustCall(() => {
+    http.get({
+      port: server.address().port,
+    }, (res) => {
+      assert.strictEqual(res.statusCode, 200);
+      res.resume().on('end', common.mustCall(() => {
+        server.close();
+      }));
+    });
+  }));
+}
+
+{
+  const server = http.createServer({
+    highWaterMark: getDefaultHighWaterMark() * 2,
+  }, common.mustCall((req, res) => {
+    assert.strictEqual(req._readableState.highWaterMark, getDefaultHighWaterMark() * 2);
+    assert.strictEqual(res[kHighWaterMark], getDefaultHighWaterMark() * 2);
+    res.statusCode = 200;
+    res.end();
+  }));
+
+  listen(server);
+}
+
+{
+  const server = http.createServer(
+    common.mustCall((req, res) => {
+      assert.strictEqual(req._readableState.highWaterMark, getDefaultHighWaterMark());
+      assert.strictEqual(res[kHighWaterMark], getDefaultHighWaterMark());
+      res.statusCode = 200;
+      res.end();
+    })
+  );
+
+  listen(server);
+}