stream: writable state bitmap

Author: ronag

lib/internal/streams/writable.js

@@ -73,6 +73,8 @@ ObjectSetPrototypeOf(Writable, Stream);
 function nop() {}
 
 const kOnFinished = Symbol('kOnFinished');
+const kErrored = Symbol('kErrored');
+const kCorked = Symbol('kCorked');
 
 const kObjectMode = 1 << 0;
 const kEnded = 1 << 1;
@@ -94,6 +96,12 @@ const kBufferProcessing = 1 << 16;
 const kPrefinished = 1 << 17;
 const kAllBuffers = 1 << 18;
 const kAllNoop = 1 << 19;
+const kHasOnFinished = 1 << 20;
+const kHasErrored = 1 << 21;
+
+const kCorkedShift = 22;
+const kCorkedMask = 0b1111
+const kCorked = kCorkedMask << kCorkedShift; // 4 bits
 
 // TODO(benjamingr) it is likely slower to do it this way than with free functions
 function makeBitMapDescriptor(bit) {
@@ -176,6 +184,46 @@ ObjectDefineProperties(WritableState.prototype, {
 
   allBuffers: makeBitMapDescriptor(kAllBuffers),
   allNoop: makeBitMapDescriptor(kAllNoop),
+
+  // Indicates whether the stream has errored. When true all write() calls
+  // should return false. This is needed since when autoDestroy
+  // is disabled we need a way to tell whether the stream has failed.
+  // This is/should be a cold path.
+  errored: {
+    enumerable: false,
+    get() { return (this.state & kHasErrored) !== 0 ? this[kErrored] : null; },
+    set(value) {
+      if (value) {
+        this[kErrored] = value;
+        this.state |= kHasErrored;
+      } else {
+        delete this[kErrored];
+        this.state &= ~kHasErrored;
+      }
+    },
+  },
+
+  // When true all writes will be buffered until .uncork() call.
+  // This is/should be a cold path.
+  corked: {
+    enumerable: false,
+    get() {
+      const val = (this.state >>> kCorkedShift) & kCorkedMask;
+      return val < kCorkedMask ? val : this[kCorked];
+    },
+    set(value) {
+      if (value >= kCorkedMask) {
+        this[kCorked] = value;
+        this.state |= kCorkedMask << kCorkedShift;
+      } else {
+        if ((this.state >>> kCorkedShift) === kCorkedMask) {
+          delete this[kCorked];
+        }
+        this.state &= ~kCorked;
+        this.state |= value << kCorkedShift;
+      }
+    },
+  }
 });
 
 function WritableState(options, stream, isDuplex) {
@@ -226,9 +274,6 @@ function WritableState(options, stream, isDuplex) {
   // socket or file.
   this.length = 0;
 
-  // When true all writes will be buffered until .uncork() call.
-  this.corked = 0;
-
   // The callback that's passed to _write(chunk, cb).
   this.onwrite = onwrite.bind(undefined, stream);
 
@@ -247,13 +292,6 @@ function WritableState(options, stream, isDuplex) {
   // Number of pending user-supplied write callbacks
   // this must be 0 before 'finish' can be emitted.
   this.pendingcb = 0;
-
-  // Indicates whether the stream has errored. When true all write() calls
-  // should return false. This is needed since when autoDestroy
-  // is disabled we need a way to tell whether the stream has failed.
-  this.errored = null;
-
-  this[kOnFinished] = [];
 }
 
 function resetBuffer(state) {
@@ -394,13 +432,21 @@ Writable.prototype.write = function(chunk, encoding, cb) {
 };
 
 Writable.prototype.cork = function() {
-  this._writableState.corked++;
+  const state = this._writableState;
+
+  const corked = ((state & kCorked) >>> kCorkedShift) + 1;
+  if (corked < kCorkedMask) {
+    this.state |= corked << kCorkedShift;
+  } else {
+    this._writableState.corked++;
+  }
 };
 
 Writable.prototype.uncork = function() {
   const state = this._writableState;
 
-  if (state.corked) {
+  if ((state.state & kCorked) !== 0) {
+    // TODO: Optimize
     state.corked--;
 
     if ((state.state & kWriting) === 0)
@@ -432,7 +478,7 @@ function writeOrBuffer(stream, state, chunk, encoding, callback) {
   if (!ret)
     state.state |= kNeedDrain;
 
-  if ((state.state & kWriting) !== 0 || state.corked || state.errored || (state.state & kConstructed) === 0) {
+  if ((state.state & kWriting) !== 0 || (state.state & (kHasErrored | kCorked)) || (state.state & kConstructed) === 0) {
     state.buffered.push({ chunk, encoding, callback });
     if ((state.state & kAllBuffers) !== 0 && encoding !== 'buffer') {
       state.state &= ~kAllBuffers;
@@ -450,7 +496,7 @@ function writeOrBuffer(stream, state, chunk, encoding, callback) {
 
   // Return false if errored or destroyed in order to break
   // any synchronous while(stream.write(data)) loops.
-  return ret && !state.errored && (state.state & kDestroyed) === 0;
+  return ret && (state.state & kHasErrored) === 0 && (state.state & kDestroyed) === 0;
 }
 
 function doWrite(stream, state, writev, len, chunk, encoding, cb) {
@@ -498,7 +544,7 @@ function onwrite(stream, er) {
     // Avoid V8 leak, https://github.com/nodejs/node/pull/34103#issuecomment-652002364
     er.stack; // eslint-disable-line no-unused-expressions
 
-    if (!state.errored) {
+    if ((state.state & kHasErrored) === 0) {
       state.errored = er;
     }
 
@@ -573,18 +619,19 @@ function errorBuffer(state) {
     callback(state.errored ?? new ERR_STREAM_DESTROYED('write'));
   }
 
-  const onfinishCallbacks = state[kOnFinished].splice(0);
-  for (let i = 0; i < onfinishCallbacks.length; i++) {
-    onfinishCallbacks[i](state.errored ?? new ERR_STREAM_DESTROYED('end'));
+  if ((state.state & kHasOnFinished) !== 0) {
+    const onfinishCallbacks = state[kOnFinished].splice(0);
+    for (let i = 0; i < onfinishCallbacks.length; i++) {
+      onfinishCallbacks[i](state.errored ?? new ERR_STREAM_DESTROYED('end'));
+    }
   }
 
   resetBuffer(state);
 }
 
 // If there's something in the buffer waiting, then process it.
 function clearBuffer(stream, state) {
-  if (state.corked ||
-      (state.state & (kDestroyed | kBufferProcessing)) !== 0 ||
+  if ((state.state & (kDestroyed | kBufferProcessing | kCorked)) !== 0 ||
       (state.state & kConstructed) === 0) {
     return;
   }
@@ -669,14 +716,14 @@ Writable.prototype.end = function(chunk, encoding, cb) {
   }
 
   // .end() fully uncorks.
-  if (state.corked) {
+  if ((state.state & kCorked) !== 0) {
     state.corked = 1;
     this.uncork();
   }
 
   if (err) {
     // Do nothing...
-  } else if (!state.errored && (state.state & kEnding) === 0) {
+  } else if ((state.state & kErrored) === 0 && (state.state & kEnding) === 0) {
     // This is forgiving in terms of unnecessary calls to end() and can hide
     // logic errors. However, usually such errors are harmless and causing a
     // hard error can be disproportionately destructive. It is not always
@@ -698,6 +745,8 @@ Writable.prototype.end = function(chunk, encoding, cb) {
     } else if ((state.state & kFinished) !== 0) {
       process.nextTick(cb, null);
     } else {
+      state.state |= kHasOnFinished;
+      state[kOnFinished] ??= [];
       state[kOnFinished].push(cb);
     }
   }
@@ -715,10 +764,10 @@ function needFinish(state) {
           kFinished |
           kWriting |
           kErrorEmitted |
-          kCloseEmitted
+          kCloseEmitted |
+          kHasErrored
     )) === (kEnding | kConstructed) &&
           state.length === 0 &&
-          !state.errored &&
           state.buffered.length === 0);
 }
 
@@ -734,9 +783,11 @@ function callFinal(stream, state) {
 
     state.pendingcb--;
     if (err) {
-      const onfinishCallbacks = state[kOnFinished].splice(0);
-      for (let i = 0; i < onfinishCallbacks.length; i++) {
-        onfinishCallbacks[i](err);
+      if ((state.state & kHasOnFinished) !== 0) {
+        const onfinishCallbacks = state[kOnFinished].splice(0);
+        for (let i = 0; i < onfinishCallbacks.length; i++) {
+          onfinishCallbacks[i](err);
+        }
       }
       errorOrDestroy(stream, err, (state.state & kSync) !== 0);
     } else if (needFinish(state)) {
@@ -799,9 +850,11 @@ function finish(stream, state) {
   state.pendingcb--;
   state.state |= kFinished;
 
-  const onfinishCallbacks = state[kOnFinished].splice(0);
-  for (let i = 0; i < onfinishCallbacks.length; i++) {
-    onfinishCallbacks[i](null);
+  if ((state.state & kHasOnFinished) !== 0) {
+    const onfinishCallbacks = state[kOnFinished].splice(0);
+    for (let i = 0; i < onfinishCallbacks.length; i++) {
+      onfinishCallbacks[i](null);
+    }
   }
 
   stream.emit('finish');
@@ -853,8 +906,8 @@ ObjectDefineProperties(Writable.prototype, {
       // where the writable side was disabled upon construction.
       // Compat. The user might manually disable writable side through
       // deprecated setter.
-      return !!w && w.writable !== false && !w.errored &&
-        (w.state & (kEnding | kEnded | kDestroyed)) === 0;
+      return !!w && w.writable !== false &&
+        (w.state & (kEnding | kEnded | kDestroyed | kHasErrored)) === 0;
     },
     set(val) {
       // Backwards compatible.
@@ -928,7 +981,7 @@ ObjectDefineProperties(Writable.prototype, {
     __proto__: null,
     enumerable: false,
     get() {
-      return this._writableState ? this._writableState.errored : null;
+      return this._writableState && (this._writableState.state & kHasErrored) !== 0 ? this._writableState.errored : null;
     },
   },
 
@@ -938,7 +991,7 @@ ObjectDefineProperties(Writable.prototype, {
     get: function() {
       return !!(
         this._writableState.writable !== false &&
-        ((this._writableState.state & kDestroyed) !== 0 || this._writableState.errored) &&
+        (this._writableState.state & (kDestroyed | kHasErrored)) !== 0 &&
         (this._writableState.state & kFinished) === 0
       );
     },
@@ -952,7 +1005,7 @@ Writable.prototype.destroy = function(err, cb) {
   // Invoke pending callbacks.
   if ((state.state & kDestroyed) === 0 &&
     (state.bufferedIndex < state.buffered.length ||
-      state[kOnFinished].length)) {
+      (((state.state & kHasOnFinished) !== 0) && state[kOnFinished].length))) {
     process.nextTick(errorBuffer, state);
   }