stream: add setter & getter for default highWaterMark (#46929)

Adds stream.(get|set)DefaultHighWaterMark to read or update
the default hwm.

PR-URL: #46929
Reviewed-By: Matteo Collina <[email protected]>
Reviewed-By: Paolo Insogna <[email protected]>
Reviewed-By: Moshe Atlow <[email protected]>
Reviewed-By: Luigi Pinca <[email protected]>
Reviewed-By: Benjamin Gruenbaum <[email protected]>
Reviewed-By: Michael Dawson <[email protected]>
Reviewed-By: Erick Wendel <[email protected]>

Author: ronag

doc/api/stream.md

@@ -3346,6 +3346,29 @@ reader.read().then(({ value, done }) => {
 });
 ```
 
+### `stream.getDefaultHighWaterMark(objectMode)`
+
+<!-- YAML
+added: REPLACEME
+-->
+
+* {boolean} objectMode
+* Returns: {integer}
+
+Returns the default highWaterMark used by streams.
+Defaults to `16384` (16 KiB), or `16` for `objectMode`.
+
+### `stream.setDefaultHighWaterMark(objectMode, value)`
+
+<!-- YAML
+added: REPLACEME
+-->
+
+* {boolean} objectMode
+* {integer} highWaterMark value
+
+Sets the default highWaterMark used by streams.
+
 ## API for stream implementers
 
 <!--type=misc-->

lib/_http_outgoing.js

@@ -80,12 +80,11 @@ let debug = require('internal/util/debuglog').debuglog('http', (fn) => {
   debug = fn;
 });
 
-const HIGH_WATER_MARK = getDefaultHighWaterMark();
-
 const kCorked = Symbol('corked');
 const kUniqueHeaders = Symbol('kUniqueHeaders');
 const kBytesWritten = Symbol('kBytesWritten');
 const kErrored = Symbol('errored');
+const kHighWaterMark = Symbol('kHighWaterMark');
 
 const nop = () => {};
 
@@ -150,6 +149,7 @@ function OutgoingMessage() {
   this._onPendingData = nop;
 
   this[kErrored] = null;
+  this[kHighWaterMark] = getDefaultHighWaterMark();
 }
 ObjectSetPrototypeOf(OutgoingMessage.prototype, Stream.prototype);
 ObjectSetPrototypeOf(OutgoingMessage, Stream);
@@ -196,7 +196,7 @@ ObjectDefineProperty(OutgoingMessage.prototype, 'writableLength', {
 ObjectDefineProperty(OutgoingMessage.prototype, 'writableHighWaterMark', {
   __proto__: null,
   get() {
-    return this.socket ? this.socket.writableHighWaterMark : HIGH_WATER_MARK;
+    return this.socket ? this.socket.writableHighWaterMark : this[kHighWaterMark];
   },
 });
 
@@ -403,7 +403,7 @@ function _writeRaw(data, encoding, callback, size) {
   this.outputData.push({ data, encoding, callback });
   this.outputSize += data.length;
   this._onPendingData(data.length);
-  return this.outputSize < HIGH_WATER_MARK;
+  return this.outputSize < this[kHighWaterMark];
 }
 
 

lib/internal/streams/state.js

@@ -4,16 +4,29 @@ const {
   MathFloor,
   NumberIsInteger,
 } = primordials;
+const { validateInteger } = require('internal/validators');
 
 const { ERR_INVALID_ARG_VALUE } = require('internal/errors').codes;
 
+let defaultHighWaterMarkBytes = 16 * 1024;
+let defaultHighWaterMarkObjectMode = 16;
+
 function highWaterMarkFrom(options, isDuplex, duplexKey) {
   return options.highWaterMark != null ? options.highWaterMark :
     isDuplex ? options[duplexKey] : null;
 }
 
 function getDefaultHighWaterMark(objectMode) {
-  return objectMode ? 16 : 16 * 1024;
+  return objectMode ? defaultHighWaterMarkObjectMode : defaultHighWaterMarkBytes;
+}
+
+function setDefaultHighWaterMark(objectMode, value) {
+  validateInteger(value, 'value', 0);
+  if (objectMode) {
+    defaultHighWaterMarkObjectMode = value;
+  } else {
+    defaultHighWaterMarkBytes = value;
+  }
 }
 
 function getHighWaterMark(state, options, duplexKey, isDuplex) {
@@ -33,4 +46,5 @@ function getHighWaterMark(state, options, duplexKey, isDuplex) {
 module.exports = {
   getHighWaterMark,
   getDefaultHighWaterMark,
+  setDefaultHighWaterMark,
 };

lib/stream.js

@@ -42,6 +42,7 @@ const {
   },
 } = require('internal/errors');
 const compose = require('internal/streams/compose');
+const { setDefaultHighWaterMark, getDefaultHighWaterMark } = require('internal/streams/state');
 const { pipeline } = require('internal/streams/pipeline');
 const { destroyer } = require('internal/streams/destroy');
 const eos = require('internal/streams/end-of-stream');
@@ -105,6 +106,8 @@ Stream.addAbortSignal = addAbortSignal;
 Stream.finished = eos;
 Stream.destroy = destroyer;
 Stream.compose = compose;
+Stream.setDefaultHighWaterMark = setDefaultHighWaterMark;
+Stream.getDefaultHighWaterMark = getDefaultHighWaterMark;
 
 ObjectDefineProperty(Stream, 'promises', {
   __proto__: null,

test/parallel/test-stream-set-default-hwm.js

@@ -0,0 +1,36 @@
+'use strict';
+
+require('../common');
+
+const assert = require('node:assert');
+const {
+  setDefaultHighWaterMark,
+  getDefaultHighWaterMark,
+  Writable,
+  Readable,
+  Transform
+} = require('stream');
+
+assert.notStrictEqual(getDefaultHighWaterMark(false), 32 * 1000);
+setDefaultHighWaterMark(false, 32 * 1000);
+assert.strictEqual(getDefaultHighWaterMark(false), 32 * 1000);
+
+assert.notStrictEqual(getDefaultHighWaterMark(true), 32);
+setDefaultHighWaterMark(true, 32);
+assert.strictEqual(getDefaultHighWaterMark(true), 32);
+
+const w = new Writable({
+  write() {}
+});
+assert.strictEqual(w.writableHighWaterMark, 32 * 1000);
+
+const r = new Readable({
+  read() {}
+});
+assert.strictEqual(r.readableHighWaterMark, 32 * 1000);
+
+const t = new Transform({
+  transform() {}
+});
+assert.strictEqual(t.writableHighWaterMark, 32 * 1000);
+assert.strictEqual(t.readableHighWaterMark, 32 * 1000);