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 <matteo.collina@gmail.com>
Reviewed-By: Paolo Insogna <paolo@cowtech.it>
Reviewed-By: Moshe Atlow <moshe@atlow.co.il>
Reviewed-By: Luigi Pinca <luigipinca@gmail.com>
Reviewed-By: Benjamin Gruenbaum <benjamingr@gmail.com>
Reviewed-By: Michael Dawson <midawson@redhat.com>
Reviewed-By: Erick Wendel <erick.workspace@gmail.com>

Author: ronag

doc/api/stream.md

@@ -3324,6 +3324,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

@@ -81,12 +81,11 @@ let debug = require('internal/util/debuglog').debuglog('http', (fn) => {
 debug=fn;
 });
 
-constHIGH_WATER_MARK=getDefaultHighWaterMark();
-
 constkCorked=Symbol('corked');
 constkUniqueHeaders=Symbol('kUniqueHeaders');
 constkBytesWritten=Symbol('kBytesWritten');
 constkErrored=Symbol('errored');
+constkHighWaterMark=Symbol('kHighWaterMark');
 
 constnop=()=>{};
 
@@ -151,6 +150,7 @@ function OutgoingMessage() {
 this._onPendingData=nop;
 
 this[kErrored]=null;
+this[kHighWaterMark]=getDefaultHighWaterMark();
 }
 ObjectSetPrototypeOf(OutgoingMessage.prototype,Stream.prototype);
 ObjectSetPrototypeOf(OutgoingMessage,Stream);
@@ -197,7 +197,7 @@ ObjectDefineProperty(OutgoingMessage.prototype, 'writableLength', {
 ObjectDefineProperty(OutgoingMessage.prototype,'writableHighWaterMark',{
 __proto__: null,
 get(){
-returnthis.socket ? this.socket.writableHighWaterMark : HIGH_WATER_MARK;
+returnthis.socket ? this.socket.writableHighWaterMark : this[kHighWaterMark];
 },
 });
 
@@ -404,7 +404,7 @@ function _writeRaw(data, encoding, callback, size) {
 this.outputData.push({ data, encoding, callback });
 this.outputSize+=data.length;
 this._onPendingData(data.length);
-returnthis.outputSize<HIGH_WATER_MARK;
+returnthis.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;
 
+letdefaultHighWaterMarkBytes=16*1024;
+letdefaultHighWaterMarkObjectMode=16;
+
 functionhighWaterMarkFrom(options,isDuplex,duplexKey){
 returnoptions.highWaterMark!=null ? options.highWaterMark :
 isDuplex ? options[duplexKey] : null;
 }
 
 functiongetDefaultHighWaterMark(objectMode){
-returnobjectMode ? 16 : 16*1024;
+returnobjectMode ? defaultHighWaterMarkObjectMode : defaultHighWaterMarkBytes;
+}
+
+functionsetDefaultHighWaterMark(objectMode,value){
+validateInteger(value,'value',0);
+if(objectMode){
+defaultHighWaterMarkObjectMode=value;
+}else{
+defaultHighWaterMarkBytes=value;
+}
 }
 
 functiongetHighWaterMark(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');
 constcompose=require('internal/streams/compose');
+const{ setDefaultHighWaterMark, getDefaultHighWaterMark }=require('internal/streams/state');
 const{ pipeline }=require('internal/streams/pipeline');
 const{ destroyer }=require('internal/streams/destroy');
 consteos=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');
+
+constassert=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);
+
+constw=newWritable({
+write(){}
+});
+assert.strictEqual(w.writableHighWaterMark,32*1000);
+
+constr=newReadable({
+read(){}
+});
+assert.strictEqual(r.readableHighWaterMark,32*1000);
+
+constt=newTransform({
+transform(){}
+});
+assert.strictEqual(t.writableHighWaterMark,32*1000);
+assert.strictEqual(t.readableHighWaterMark,32*1000);