stream: use bit fields for construct/destroy

PR-URL: #50408

Author: ronag

lib/internal/streams/destroy.js

@@ -15,6 +15,15 @@ const {
   isDestroyed,
   isFinished,
   isServerRequest,
+  kState,
+  kErrorEmitted,
+  kEmitClose,
+  kClosed,
+  kCloseEmitted,
+  kConstructed,
+  kDestroyed,
+  kAutoDestroy,
+  kErrored,
 } = require('internal/streams/utils');
 
 const kDestroy = Symbol('kDestroy');
@@ -42,7 +51,10 @@ function destroy(err, cb) {
   // With duplex streams we use the writable side for state.
   const s = w || r;
 
-  if (w?.destroyed || r?.destroyed) {
+  if (
+    (w && (w[kState] ? (w[kState] & kDestroyed) !== 0 : w.destroyed)) ||
+    (r && (r[kState] ? (r[kState] & kDestroyed) !== 0 : r.destroyed))
+  ) {
     if (typeof cb === 'function') {
       cb();
     }
@@ -56,14 +68,14 @@ function destroy(err, cb) {
   checkError(err, w, r);
 
   if (w) {
-    w.destroyed = true;
+    w[kState] |= kDestroyed;
   }
   if (r) {
-    r.destroyed = true;
+    r[kState] |= kDestroyed;
   }
 
   // If still constructing then defer calling _destroy.
-  if (!s.constructed) {
+  if ((s[kState] & kConstructed) === 0) {
     this.once(kDestroy, function(er) {
       _destroy(this, aggregateTwoErrors(er, err), cb);
     });
@@ -89,10 +101,10 @@ function _destroy(self, err, cb) {
     checkError(err, w, r);
 
     if (w) {
-      w.closed = true;
+      w[kState] |= kClosed;
     }
     if (r) {
-      r.closed = true;
+      r[kState] |= kClosed;
     }
 
     if (typeof cb === 'function') {
@@ -122,13 +134,16 @@ function emitCloseNT(self) {
   const w = self._writableState;
 
   if (w) {
-    w.closeEmitted = true;
+    w[kState] |= kCloseEmitted;
   }
   if (r) {
-    r.closeEmitted = true;
+    r[kState] |= kCloseEmitted;
   }
 
-  if (w?.emitClose || r?.emitClose) {
+  if (
+    (w && (w[kState] & kEmitClose) !== 0) ||
+    (r && (r[kState] & kEmitClose) !== 0)
+  ) {
     self.emit('close');
   }
 }
@@ -137,15 +152,18 @@ function emitErrorNT(self, err) {
   const r = self._readableState;
   const w = self._writableState;
 
-  if (w?.errorEmitted || r?.errorEmitted) {
+  if (
+    (w && (w[kState] & kErrorEmitted) !== 0) ||
+    (r && (r[kState] & kErrorEmitted) !== 0)
+  ) {
     return;
   }
 
   if (w) {
-    w.errorEmitted = true;
+    w[kState] |= kErrorEmitted;
   }
   if (r) {
-    r.errorEmitted = true;
+    r[kState] |= kErrorEmitted;
   }
 
   self.emit('error', err);
@@ -192,20 +210,26 @@ function errorOrDestroy(stream, err, sync) {
   const r = stream._readableState;
   const w = stream._writableState;
 
-  if (w?.destroyed || r?.destroyed) {
+  if (
+    (w && (w[kState] ? (w[kState] & kDestroyed) !== 0 : w.destroyed)) ||
+    (r && (r[kState] ? (r[kState] & kDestroyed) !== 0 : r.destroyed))
+  ) {
     return this;
   }
 
-  if (r?.autoDestroy || w?.autoDestroy)
+  if (
+    (r && (r[kState] & kAutoDestroy) !== 0) ||
+    (w && (w[kState] & kAutoDestroy) !== 0)
+  ) {
     stream.destroy(err);
-  else if (err) {
+  } else if (err) {
     // Avoid V8 leak, https://github.com/nodejs/node/pull/34103#issuecomment-652002364
     err.stack; // eslint-disable-line no-unused-expressions
 
-    if (w && !w.errored) {
+    if (w && (w[kState] & kErrored) === 0) {
       w.errored = err;
     }
-    if (r && !r.errored) {
+    if (r && (r[kState] & kErrored) === 0) {
       r.errored = err;
     }
     if (sync) {
@@ -225,10 +249,10 @@ function construct(stream, cb) {
   const w = stream._writableState;
 
   if (r) {
-    r.constructed = false;
+    r[kState] &= ~kConstructed;
   }
   if (w) {
-    w.constructed = false;
+    w[kState] &= ~kConstructed;
   }
 
   stream.once(kConstruct, cb);
@@ -256,10 +280,10 @@ function constructNT(stream) {
     const s = w || r;
 
     if (r) {
-      r.constructed = true;
+      r[kState] |= kConstructed;
     }
     if (w) {
-      w.constructed = true;
+      w[kState] |= kConstructed;
     }
 
     if (s.destroyed) {

lib/internal/streams/readable.js

@@ -58,6 +58,20 @@ const {
   getHighWaterMark,
   getDefaultHighWaterMark,
 } = require('internal/streams/state');
+const {
+  kState,
+  // bitfields
+  kObjectMode,
+  kErrorEmitted,
+  kAutoDestroy,
+  kEmitClose,
+  kDestroyed,
+  kClosed,
+  kCloseEmitted,
+  kErrored,
+  kConstructed,
+  kOnConstructed,
+} = require('internal/streams/utils');
 
 const {
   aggregateTwoErrors,
@@ -72,9 +86,7 @@ const {
   AbortError,
 } = require('internal/errors');
 const { validateObject } = require('internal/validators');
-const { kOnConstructed } = require('internal/streams/utils');
 
-const kState = Symbol('kState');
 const FastBuffer = Buffer[SymbolSpecies];
 
 const { StringDecoder } = require('string_decoder');
@@ -91,26 +103,17 @@ const kDefaultEncodingValue = Symbol('kDefaultEncodingValue');
 const kDecoderValue = Symbol('kDecoderValue');
 const kEncodingValue = Symbol('kEncodingValue');
 
-const kObjectMode = 1 << 0;
-const kEnded = 1 << 1;
-const kEndEmitted = 1 << 2;
-const kReading = 1 << 3;
-const kConstructed = 1 << 4;
-const kSync = 1 << 5;
-const kNeedReadable = 1 << 6;
-const kEmittedReadable = 1 << 7;
-const kReadableListening = 1 << 8;
-const kResumeScheduled = 1 << 9;
-const kErrorEmitted = 1 << 10;
-const kEmitClose = 1 << 11;
-const kAutoDestroy = 1 << 12;
-const kDestroyed = 1 << 13;
-const kClosed = 1 << 14;
-const kCloseEmitted = 1 << 15;
-const kMultiAwaitDrain = 1 << 16;
-const kReadingMore = 1 << 17;
-const kDataEmitted = 1 << 18;
-const kErrored = 1 << 19;
+const kEnded = 1 << 9;
+const kEndEmitted = 1 << 10;
+const kReading = 1 << 11;
+const kSync = 1 << 12;
+const kNeedReadable = 1 << 13;
+const kEmittedReadable = 1 << 14;
+const kReadableListening = 1 << 15;
+const kResumeScheduled = 1 << 16;
+const kMultiAwaitDrain = 1 << 17;
+const kReadingMore = 1 << 18;
+const kDataEmitted = 1 << 19;
 const kDefaultUTF8Encoding = 1 << 20;
 const kDecoder = 1 << 21;
 const kEncoding = 1 << 22;

lib/internal/streams/utils.js

@@ -22,6 +22,17 @@ const kOnConstructed = Symbol('kOnConstructed');
 const kIsClosedPromise = SymbolFor('nodejs.webstream.isClosedPromise');
 const kControllerErrorFunction = SymbolFor('nodejs.webstream.controllerErrorFunction');
 
+const kState = Symbol('kState');
+const kObjectMode = 1 << 0;
+const kErrorEmitted = 1 << 1;
+const kAutoDestroy = 1 << 2;
+const kEmitClose = 1 << 3;
+const kDestroyed = 1 << 4;
+const kClosed = 1 << 5;
+const kCloseEmitted = 1 << 6;
+const kErrored = 1 << 7;
+const kConstructed = 1 << 8;
+
 function isReadableNodeStream(obj, strict = false) {
   return !!(
     obj &&
@@ -339,4 +350,15 @@ module.exports = {
   isServerResponse,
   willEmitClose,
   isTransformStream,
+  kState,
+  // bitfields
+  kObjectMode,
+  kErrorEmitted,
+  kAutoDestroy,
+  kEmitClose,
+  kDestroyed,
+  kClosed,
+  kCloseEmitted,
+  kErrored,
+  kConstructed
 };

lib/internal/streams/writable.js

@@ -44,7 +44,6 @@ const EE = require('events');
 const Stream = require('internal/streams/legacy').Stream;
 const { Buffer } = require('buffer');
 const destroyImpl = require('internal/streams/destroy');
-const { kOnConstructed } = require('internal/streams/utils');
 
 const {
   addAbortSignal,
@@ -65,6 +64,20 @@ const {
   ERR_STREAM_WRITE_AFTER_END,
   ERR_UNKNOWN_ENCODING,
 } = require('internal/errors').codes;
+const {
+  kState,
+  // bitfields
+  kObjectMode,
+  kErrorEmitted,
+  kAutoDestroy,
+  kEmitClose,
+  kDestroyed,
+  kClosed,
+  kCloseEmitted,
+  kErrored,
+  kConstructed,
+  kOnConstructed,
+} = require('internal/streams/utils');
 
 const { errorOrDestroy } = destroyImpl;
 
@@ -79,18 +92,8 @@ const kDefaultEncodingValue = Symbol('kDefaultEncodingValue');
 const kWriteCbValue = Symbol('kWriteCbValue');
 const kAfterWriteTickInfoValue = Symbol('kAfterWriteTickInfoValue');
 const kBufferedValue = Symbol('kBufferedValue');
-const kState = Symbol('kState');
-
-const kObjectMode = 1 << 0;
-const kEnded = 1 << 1;
-const kConstructed = 1 << 2;
-const kSync = 1 << 3;
-const kErrorEmitted = 1 << 4;
-const kEmitClose = 1 << 5;
-const kAutoDestroy = 1 << 6;
-const kDestroyed = 1 << 7;
-const kClosed = 1 << 8;
-const kCloseEmitted = 1 << 9;
+
+const kSync = 1 << 9;
 const kFinalCalled = 1 << 10;
 const kNeedDrain = 1 << 11;
 const kEnding = 1 << 12;
@@ -102,16 +105,16 @@ const kPrefinished = 1 << 17;
 const kAllBuffers = 1 << 18;
 const kAllNoop = 1 << 19;
 const kOnFinished = 1 << 20;
-const kErrored = 1 << 21;
-const kHasWritable = 1 << 22;
-const kWritable = 1 << 23;
-const kCorked = 1 << 24;
-const kDefaultUTF8Encoding = 1 << 25;
-const kWriteCb = 1 << 26;
-const kExpectWriteCb = 1 << 27;
-const kAfterWriteTickInfo = 1 << 28;
-const kAfterWritePending = 1 << 29;
-const kBuffered = 1 << 30;
+const kHasWritable = 1 << 21;
+const kWritable = 1 << 22;
+const kCorked = 1 << 23;
+const kDefaultUTF8Encoding = 1 << 24;
+const kWriteCb = 1 << 25;
+const kExpectWriteCb = 1 << 26;
+const kAfterWriteTickInfo = 1 << 27;
+const kAfterWritePending = 1 << 28;
+const kBuffered = 1 << 29;
+const kEnded = 1 << 30;
 
 // TODO(benjamingr) it is likely slower to do it this way than with free functions
 function makeBitMapDescriptor(bit) {