diff --git a/lib/internal/streams/destroy.js b/lib/internal/streams/destroy.js index 3119de5dd9d..e488e18db9a 100644 --- a/lib/internal/streams/destroy.js +++ b/lib/internal/streams/destroy.js @@ -182,6 +182,7 @@ function undestroy() { r.errored = null; r.errorEmitted = false; r.reading = false; + r.fastChunk = null; r.ended = r.readable === false; r.endEmitted = r.readable === false; } diff --git a/lib/internal/streams/readable.js b/lib/internal/streams/readable.js index 1b8f9ab3adb..e584757dd9a 100644 --- a/lib/internal/streams/readable.js +++ b/lib/internal/streams/readable.js @@ -139,6 +139,9 @@ const kPaused = 1 << 26; const kDataListening = 1 << 27; const kEndScheduled = 1 << 28; const kEofReadablePending = 1 << 29; +// Set only while flowSync() is inside _read(). push() then keeps the +// chunk on state.fastChunk instead of the buffer array. +const kFastPush = 1 << 30; // TODO(benjamingr) it is likely slower to do it this way than with free functions function makeBitMapDescriptor(bit) { @@ -296,6 +299,12 @@ function ReadableState(options, stream, isDuplex) { this.buffer = []; this.bufferIndex = 0; this.length = 0; + // Chunk prefetched by flowSync(), kept off the buffer array. + this.fastChunk = null; + // The sole pipe() 'data' listener, when there is exactly one destination. + // flowSync() writes buffers straight to that destination. + this.pipeOnData = null; + this.pipePause = null; this.pipes = []; // Should close be emitted on destroy. Defaults to true. @@ -400,9 +409,29 @@ Readable.prototype[SymbolAsyncDispose] = async function() { // similar to how Writable.write() returns true if you should // write() some more. Readable.prototype.push = function(chunk, encoding) { + const state = this._readableState; + + // flowSync() is inside _read() and wants a single buffer parked on + // fastChunk. A second push, a string, or EOF drops back to the buffer. + if ((state[kState] & kFastPush) !== 0 && encoding == null && + state.fastChunk == null && chunk instanceof Buffer && chunk.length > 0) { + state.fastChunk = chunk; + state.length = chunk.length; + state[kState] &= ~kReading; + return chunk.length < state.highWaterMark; + } + if ((state[kState] & kFastPush) !== 0) { + state[kState] &= ~kFastPush; + if (state.fastChunk != null) { + const first = state.fastChunk; + state.fastChunk = null; + state.length = 0; + readableAddChunkPushByteMode(this, state, first); + } + } + debug('push', chunk); - const state = this._readableState; return (state[kState] & kObjectMode) === 0 ? readableAddChunkPushByteMode(this, state, chunk, encoding) : readableAddChunkPushObjectMode(this, state, chunk, encoding); @@ -601,6 +630,8 @@ Readable.prototype.isPaused = function() { // Backwards compatibility. Readable.prototype.setEncoding = function(enc) { const state = this._readableState; + if (state.fastChunk != null) + materializeFastChunk(state); const decoder = new StringDecoder(enc); state.decoder = decoder; @@ -667,6 +698,11 @@ function howMuchToRead(n, state) { // You can override either this method, or the async _read(n) below. Readable.prototype.read = function(n) { + // A nested read() during flowSync()'s 'data' event must see the + // prefetched chunk. Null for every read that is not inside that loop. + if (this._readableState.fastChunk != null) + materializeFastChunk(this._readableState); + debug('read', n); // Same as parseInt(undefined, 10), however V8 7.3 performance regressed // in this scenario, so we are doing it manually. @@ -953,6 +989,16 @@ Readable.prototype.pipe = function(dest, pipeOpts) { } state.pipes.push(dest); + // Only a single pipe destination can skip emit('data') and call write() + // directly. A second destination, or an extra 'data' listener, must go + // through emit so every listener still runs. + if (state.pipes.length === 1) { + state.pipeOnData = ondata; + state.pipePause = pause; + } else { + state.pipeOnData = null; + state.pipePause = null; + } debug('pipe count=%d opts=%j', state.pipes.length, pipeOpts); const doEnd = (!pipeOpts || pipeOpts.end !== false) && @@ -1143,6 +1189,8 @@ Readable.prototype.unpipe = function(dest) { // remove all. const dests = state.pipes; state.pipes = []; + state.pipeOnData = null; + state.pipePause = null; this.pause(); for (let i = 0; i < dests.length; i++) @@ -1156,6 +1204,8 @@ Readable.prototype.unpipe = function(dest) { return this; state.pipes.splice(index, 1); + state.pipeOnData = null; + state.pipePause = null; if (state.pipes.length === 0) this.pause(); @@ -1336,7 +1386,115 @@ Readable.prototype.pause = function() { function flow(stream) { const state = stream._readableState; debug('flow'); + // Byte-mode pipe sits in read() to pull one already-buffered chunk and + // refill. That read is most of the per-chunk cost. flowSync() keeps the + // same prefetch order without the buffer array or the general read path. + if (flowSync(stream, state)) + return; + while ((state[kState] & kFlowing) !== 0 && stream.read() !== null); +} + +const kFastFlowNeed = kConstructed | kFlowing | kDataListening; +const kFastFlowBlock = kObjectMode | kDecoder | kEnded | kDestroyed | + kErrored | kPaused | kReading | kSync; + +let writeKnownBuffer; + +// Pipe's only listener is ondata(), which calls dest.write(). Skip emit +// and the general write() checks for a single Buffer in that steady state. +function deliverFlowChunk(stream, state, chunk) { + const pipeOnData = state.pipeOnData; + const events = stream._events; + if (pipeOnData !== null && events !== undefined && events.data === pipeOnData) { + writeKnownBuffer ??= require('internal/streams/writable').writeKnownBuffer; + const dest = state.pipes[0]; + let ret; + try { + ret = writeKnownBuffer(dest, chunk); + } catch (error) { + dest.destroy(error); + return; + } + if (ret === undefined) + stream.emit('data', chunk); + else if (ret === false && state.pipePause !== null) + state.pipePause(); + return; + } + stream.emit('data', chunk); +} + +// Returns true when this call owned the flowing loop, including any +// fallback to read() after the fast path stops. +function flowSync(stream, state) { + const bits = state[kState]; + if ((bits & kFastFlowNeed) !== kFastFlowNeed || + (bits & kFastFlowBlock) !== 0 || + !(state.highWaterMark > 0) || + state.fastChunk != null) { + return false; + } + + if (state.length !== 0) { + const buf = state.buffer; + const idx = state.bufferIndex; + const chunk = buf[idx]; + // Only the one-chunk prefetch left by read(0) / the previous read. + if (buf.length !== idx + 1 || chunk == null || chunk.length !== state.length) + return false; + buf.length = 0; + state.bufferIndex = 0; + state.fastChunk = chunk; + } + + while ((state[kState] & kFlowing) !== 0) { + const current = state.fastChunk; + if (current == null) + break; + if ((state[kState] & kFastFlowBlock) !== 0) + break; + + // _read() of the next chunk runs before 'data', matching read(). + state.fastChunk = null; + state.length = 0; + state[kState] |= kReading | kSync | kFastPush; + try { + stream._read(state.highWaterMark); + } catch (err) { + state[kState] &= ~(kSync | kFastPush); + errorOrDestroy(stream, err); + break; + } + state[kState] &= ~(kSync | kFastPush); + + if ((state[kState] & (kErrorEmitted | kCloseEmitted)) === 0) { + state[kState] |= kDataEmitted; + deliverFlowChunk(stream, state, current); + } + + // Nested read() moved fastChunk into the buffer and may have refilled. + if (state.fastChunk == null && state.length !== 0) + break; + } + + if (state.fastChunk != null) + materializeFastChunk(state); + while ((state[kState] & kFlowing) !== 0 && stream.read() !== null); + return true; +} + +// Put fastChunk at the head of the buffer. state.length already counts it. +function materializeFastChunk(state) { + const chunk = state.fastChunk; + if (chunk == null) + return; + state.fastChunk = null; + if (state.bufferIndex > 0) { + state.buffer[--state.bufferIndex] = chunk; + } else { + state.buffer.unshift(chunk); + } } // Wrap an old-style stream as the async data source. @@ -1724,7 +1882,16 @@ ObjectDefineProperties(Readable.prototype, { __proto__: null, enumerable: false, get: function() { - return this._readableState?.buffer; + const state = this._readableState; + if (state == null) + return undefined; + if (state.fastChunk == null) + return state.buffer; + if (state.bufferIndex === state.buffer.length) + return [state.fastChunk]; + const out = state.buffer.slice(state.bufferIndex); + out.push(state.fastChunk); + return out; }, }, diff --git a/lib/internal/streams/writable.js b/lib/internal/streams/writable.js index 09d713f8a1f..f8c61f81961 100644 --- a/lib/internal/streams/writable.js +++ b/lib/internal/streams/writable.js @@ -586,6 +586,43 @@ function writeOrBuffer(stream, state, chunk, encoding, callback) { return ret && (state[kState] & (kDestroyed | kErrored)) === 0; } +// Steady state of a flowing pipe into a byte-mode Writable: one Buffer, +// nothing queued, and no user write callback. Returns undefined when the +// caller must use write() instead. Otherwise the same boolean as write(). +const kWriteFlowBlock = kObjectMode | kDestroyed | kErrored | kSync | + kEnding | kFinished | kWriting | kCorked | kBuffered | kEnded | + kNeedDrain | kWriteCb | kExpectWriteCb | kBufferProcessing | + kFinalCalled | kPrefinished | kOnFinished | kErrorEmitted; + +function writeKnownBuffer(stream, chunk) { + const state = stream._writableState; + if (state == null || state.length !== 0 || !(chunk instanceof Buffer)) + return undefined; + + const bits = state[kState]; + if ((bits & kConstructed) === 0 || (bits & kWriteFlowBlock) !== 0) + return undefined; + + const len = chunk.length; + state.pendingcb++; + state.length = len; + state.writelen = len; + state[kState] = bits | kWriting | kSync | kExpectWriteCb; + stream._write(chunk, 'buffer', state.onwrite); + state[kState] &= ~kSync; + + const ret = state.length < state.highWaterMark || state.length === 0; + if (!ret) + state[kState] |= kNeedDrain; + return ret && (state[kState] & (kDestroyed | kErrored)) === 0; +} + +ObjectDefineProperty(Writable, 'writeKnownBuffer', { + __proto__: null, + value: writeKnownBuffer, + enumerable: false, +}); + function doWrite(stream, state, writev, len, chunk, encoding, cb) { state.writelen = len; if (cb !== nop) {