@@ -77,6 +77,7 @@ const kErroredValue = Symbol('kErroredValue'); |
77 | 77 | const kDefaultEncodingValue = Symbol('kDefaultEncodingValue'); |
78 | 78 | const kWriteCbValue = Symbol('kWriteCbValue'); |
79 | 79 | const kAfterWriteTickInfoValue = Symbol('kAfterWriteTickInfoValue'); |
| 80 | +const kBufferedValue = Symbol('kBufferedValue'); |
80 | 81 |
|
81 | 82 | const kObjectMode = 1 << 0; |
82 | 83 | const kEnded = 1 << 1; |
@@ -108,7 +109,7 @@ const kWriteCb = 1 << 26; |
108 | 109 | const kExpectWriteCb = 1 << 27; |
109 | 110 | const kAfterWriteTickInfo = 1 << 28; |
110 | 111 | const kAfterWritePending = 1 << 29; |
111 | | -const kHasBuffer = 1 << 30; |
| 112 | +const kBuffered = 1 << 30; |
112 | 113 |
|
113 | 114 | // TODO(benjamingr) it is likely slower to do it this way than with free functions |
114 | 115 | function makeBitMapDescriptor(bit) { |
@@ -270,6 +271,21 @@ ObjectDefineProperties(WritableState.prototype, { |
270 | 271 | } |
271 | 272 | }, |
272 | 273 | }, |
| 274 | + |
| 275 | + buffered: { |
| 276 | + __proto__: null, |
| 277 | + enumerable: false, |
| 278 | + get() { return (this.state & kBuffered) !== 0 ? this[kBufferedValue] : []; }, |
| 279 | + set(value) { |
| 280 | + this[kBufferedValue] = value; |
| 281 | + if (value) { |
| 282 | + this.state |= kBuffered; |
| 283 | + } else { |
| 284 | + this.state &= ~kBuffered; |
| 285 | + } |
| 286 | + }, |
| 287 | + }, |
| 288 | + |
273 | 289 | }); |
274 | 290 |
|
275 | 291 | function WritableState(options, stream, isDuplex) { |
@@ -338,20 +354,20 @@ function WritableState(options, stream, isDuplex) { |
338 | 354 | } |
339 | 355 |
|
340 | 356 | function resetBuffer(state) { |
341 | | - state.buffered = []; |
| 357 | + state[kBufferedValue] = null; |
342 | 358 | state.bufferedIndex = 0; |
343 | 359 | state.state |= kAllBuffers | kAllNoop; |
344 | | - state.state &= ~kHasBuffer; |
| 360 | + state.state &= ~kBuffered; |
345 | 361 | } |
346 | 362 |
|
347 | 363 | WritableState.prototype.getBuffer = function getBuffer() { |
348 | | - return ArrayPrototypeSlice(this.buffered, this.bufferedIndex); |
| 364 | + return (this.state & kBuffered) === 0 ? [] : ArrayPrototypeSlice(this.buffered, this.bufferedIndex); |
349 | 365 | }; |
350 | 366 |
|
351 | 367 | ObjectDefineProperty(WritableState.prototype, 'bufferedRequestCount', { |
352 | 368 | __proto__: null, |
353 | 369 | get() { |
354 | | - return this.buffered.length - this.bufferedIndex; |
| 370 | + return (this.state & kBuffered) === 0 ? 0 : this[kBufferedValue].length - this.bufferedIndex; |
355 | 371 | }, |
356 | 372 | }); |
357 | 373 |
|
@@ -518,8 +534,12 @@ function writeOrBuffer(stream, state, chunk, encoding, callback) { |
518 | 534 | state.length += len; |
519 | 535 |
|
520 | 536 | if ((state.state & (kWriting | kErrored | kCorked | kConstructed)) !== kConstructed) { |
521 | | - state.buffered.push({ chunk, encoding, callback }); |
522 | | - state.state |= kHasBuffer; |
| 537 | + if ((state.state & kBuffered) === 0) { |
| 538 | + state.state |= kBuffered; |
| 539 | + state[kBufferedValue] = []; |
| 540 | + } |
| 541 | + |
| 542 | + state[kBufferedValue].push({ chunk, encoding, callback }); |
523 | 543 | if ((state.state & kAllBuffers) !== 0 && encoding !== 'buffer') { |
524 | 544 | state.state &= ~kAllBuffers; |
525 | 545 | } |
@@ -611,7 +631,7 @@ function onwrite(stream, er) { |
611 | 631 | onwriteError(stream, state, er, cb); |
612 | 632 | } |
613 | 633 | } else { |
614 | | - if ((state.state & kHasBuffer) !== 0) { |
| 634 | + if ((state.state & kBuffered) !== 0) { |
615 | 635 | clearBuffer(stream, state); |
616 | 636 | } |
617 | 637 |
|
@@ -687,11 +707,13 @@ function errorBuffer(state) { |
687 | 707 | return; |
688 | 708 | } |
689 | 709 |
|
690 | | - for (let n = state.bufferedIndex; n < state.buffered.length; ++n) { |
691 | | - const { chunk, callback } = state.buffered[n]; |
692 | | - const len = (state.state & kObjectMode) !== 0 ? 1 : chunk.length; |
693 | | - state.length -= len; |
694 | | - callback(state.errored ?? new ERR_STREAM_DESTROYED('write')); |
| 710 | + if ((state.state & kBuffered) !== 0) { |
| 711 | + for (let n = state.bufferedIndex; n < state.buffered.length; ++n) { |
| 712 | + const { chunk, callback } = state[kBufferedValue][n]; |
| 713 | + const len = (state.state & kObjectMode) !== 0 ? 1 : chunk.length; |
| 714 | + state.length -= len; |
| 715 | + callback(state.errored ?? new ERR_STREAM_DESTROYED('write')); |
| 716 | + } |
695 | 717 | } |
696 | 718 |
|
697 | 719 |
|
@@ -702,13 +724,12 @@ function errorBuffer(state) { |
702 | 724 |
|
703 | 725 | // If there's something in the buffer waiting, then process it. |
704 | 726 | function clearBuffer(stream, state) { |
705 | | - if ((state.state & (kDestroyed | kBufferProcessing | kCorked | kHasBuffer)) !== kHasBuffer || |
706 | | - (state.state & kConstructed) === 0) { |
| 727 | + if ((state.state & (kDestroyed | kBufferProcessing | kCorked | kBuffered)) !== kBuffered) { |
707 | 728 | return; |
708 | 729 | } |
709 | 730 |
|
710 | 731 | const objectMode = (state.state & kObjectMode) !== 0; |
711 | | - const { buffered, bufferedIndex } = state; |
| 732 | + const { [kBufferedValue]: buffered, bufferedIndex } = state; |
712 | 733 | const bufferedLength = buffered.length - bufferedIndex; |
713 | 734 |
|
714 | 735 | if (!bufferedLength) { |
@@ -838,10 +859,9 @@ function needFinish(state) { |
838 | 859 | kWriting | |
839 | 860 | kErrorEmitted | |
840 | 861 | kCloseEmitted | |
841 | | - kErrored |
842 | | - )) === (kEnding | kConstructed) && |
843 | | - state.length === 0 && |
844 | | - state.buffered.length === 0); |
| 862 | + kErrored | |
| 863 | + kBuffered |
| 864 | + )) === (kEnding | kConstructed) && state.length === 0); |
845 | 865 | } |
846 | 866 |
|
847 | 867 | function callFinal(stream, state) { |
@@ -1083,9 +1103,7 @@ Writable.prototype.destroy = function(err, cb) { |
1083 | 1103 | const state = this._writableState; |
1084 | 1104 |
|
1085 | 1105 | // Invoke pending callbacks. |
1086 | | - if ((state.state & kDestroyed) === 0 && |
1087 | | - (state.bufferedIndex < state.buffered.length || |
1088 | | - (state.state & kOnFinished) !== 0)) { |
| 1106 | + if ((state.state & (kBuffered | kOnFinished | kDestroyed)) !== kDestroyed) { |
1089 | 1107 | process.nextTick(errorBuffer, state); |
1090 | 1108 | } |
1091 | 1109 |
|
|