Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

stream: readable use bitmap accessors #50350

Merged
merged 1 commit into from
Oct 25, 2023
+31 −27
Merged

stream: readable use bitmap accessors #50350

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
58 changes: 31 additions & 27 deletions lib/internal/streams/readable.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -435,7 +435,7 @@ function readableAddChunkUnshiftObjectMode(stream, state, chunk) {
function readableAddChunkUnshiftValue(stream, state, chunk) {
if ((state[kState] & kEndEmitted) !== 0)
errorOrDestroy(stream, new ERR_STREAM_UNSHIFT_AFTER_END_EVENT());
else if (state.destroyed || state.errored)
else if ((state[kState] & (kDestroyed | kErrored)) !== 0)
return false;
else
addChunk(stream, state, chunk, true);
Expand Down Expand Up @@ -604,7 +604,7 @@ function computeNewHighWaterMark(n) {
// This function is designed to be inlinable, so please take care when making
// changes to the function body.
function howMuchToRead(n, state) {
if (n <= 0 || (state.length === 0 && state.ended))
if (n <= 0 || (state.length === 0 && (state[kState] & kEnded) !== 0))
return 0;
if ((state[kState] & kObjectMode) !== 0)
return 1;
Expand Down Expand Up @@ -648,7 +648,7 @@ Readable.prototype.read = function(n) {
state.length >= state.highWaterMark :
state.length > 0) ||
(state[kState] & kEnded) !== 0)) {
debug('read: emitReadable', state.length, (state[kState] & kEnded) !== 0);
debug('read: emitReadable');
if (state.length === 0 && (state[kState] & kEnded) !== 0)
endReadable(this);
else
Expand Down Expand Up @@ -806,7 +806,7 @@ function emitReadable(stream) {
function emitReadable_(stream) {
const state = stream._readableState;
debug('emitReadable_');
if ((state[kState] & (kDestroyed | kErrored)) === 0 && (state.length || state.ended)) {
if ((state[kState] & (kDestroyed | kErrored)) === 0 && (state.length || (state[kState] & kEnded) !== 0)) {
stream.emit('readable');
state[kState] &= ~kEmittedReadable;
}
Expand Down Expand Up @@ -887,7 +887,7 @@ Readable.prototype.pipe = function(dest, pipeOpts) {
const state = this._readableState;

if (state.pipes.length === 1) {
if (!state.multiAwaitDrain) {
if ((state[kState] & kMultiAwaitDrain) === 0) {
state[kState] |= kMultiAwaitDrain;
state.awaitDrainWriters = new SafeSet(
state.awaitDrainWriters ? [state.awaitDrainWriters] : [],
Expand All @@ -903,7 +903,7 @@ Readable.prototype.pipe = function(dest, pipeOpts) {
dest !== process.stderr;

const endFn = doEnd ? onend : unpipe;
if (state.endEmitted)
if ((state[kState] & kEndEmitted) !== 0)
process.nextTick(endFn);
else
src.once('end', endFn);
Expand Down Expand Up @@ -962,7 +962,7 @@ Readable.prototype.pipe = function(dest, pipeOpts) {
if (state.pipes.length === 1 && state.pipes[0] === dest) {
debug('false write response, pause', 0);
state.awaitDrainWriters = dest;
state.multiAwaitDrain = false;
state[kState] &= ~kMultiAwaitDrain;
} else if (state.pipes.length > 1 && state.pipes.includes(dest)) {
debug('false write response, pause', state.awaitDrainWriters.size);
state.awaitDrainWriters.add(dest);
Expand Down Expand Up @@ -1034,7 +1034,7 @@ Readable.prototype.pipe = function(dest, pipeOpts) {

if (dest.writableNeedDrain === true) {
pause();
} else if (!state.flowing) {
} else if ((state[kState] & kFlowing) === 0) {
debug('pipe resume');
src.resume();
}
Expand All @@ -1052,7 +1052,7 @@ function pipeOnDrain(src, dest) {
if (state.awaitDrainWriters === dest) {
debug('pipeOnDrain', 1);
state.awaitDrainWriters = null;
} else if (state.multiAwaitDrain) {
} else if ((state[kState] & kMultiAwaitDrain) !== 0) {
debug('pipeOnDrain', state.awaitDrainWriters.size);
state.awaitDrainWriters.delete(dest);
}
Expand Down Expand Up @@ -1107,20 +1107,20 @@ Readable.prototype.on = function(ev, fn) {
if (ev === 'data') {
// Update readableListening so that resume() may be a no-op
// a few lines down. This is needed to support once('readable').
state.readableListening = this.listenerCount('readable') > 0;
state[kState] |= this.listenerCount('readable') > 0 ? kReadableListening : 0;

// Try start flowing on next tick if stream isn't explicitly paused.
if (state.flowing !== false)
if ((state[kState] & (kHasFlowing | kFlowing)) !== kHasFlowing) {
this.resume();
}
} else if (ev === 'readable') {
if (!state.endEmitted && !state.readableListening) {
state.readableListening = state.needReadable = true;
state.flowing = false;
state.emittedReadable = false;
debug('on readable', state.length, state.reading);
if ((state[kState] & (kEndEmitted | kReadableListening)) === 0) {
state[kState] |= kReadableListening | kNeedReadable | kHasFlowing;
state[kState] &= ~(kFlowing | kEmittedReadable);
debug('on readable');
if (state.length) {
emitReadable(this);
} else if (!state.reading) {
} else if ((state[kState] & kReading) === 0) {
process.nextTick(nReadingNextTick, this);
}
}
Expand Down Expand Up @@ -1167,7 +1167,12 @@ Readable.prototype.removeAllListeners = function(ev) {

function updateReadableListening(self) {
const state = self._readableState;
state.readableListening = self.listenerCount('readable') > 0;

if (self.listenerCount('readable') > 0) {
state[kState] |= kReadableListening;
} else {
state[kState] &= ~kReadableListening;
}

if ((state[kState] & (kHasPaused | kPaused | kResumeScheduled)) === (kHasPaused | kResumeScheduled)) {
// Flowing needs to be set to true now, otherwise
Expand Down Expand Up @@ -1197,7 +1202,7 @@ Readable.prototype.resume = function() {
// for readable, but we still have to call
// resume().
state[kState] |= kHasFlowing;
if (!state.readableListening) {
if ((state[kState] & kReadableListening) === 0) {
state[kState] |= kFlowing;
} else {
state[kState] &= ~kFlowing;
Expand All @@ -1210,8 +1215,8 @@ Readable.prototype.resume = function() {
};

function resume(stream, state) {
if (!state.resumeScheduled) {
state.resumeScheduled = true;
if ((state[kState] & kResumeScheduled) === 0) {
state[kState] |= kResumeScheduled;
process.nextTick(resume_, stream, state);
}
}
Expand All @@ -1232,7 +1237,7 @@ function resume_(stream, state) {
Readable.prototype.pause = function() {
const state = this._readableState;
debug('call pause');
if (state.flowing !== false) {
if ((state[kState] & (kHasFlowing | kFlowing)) !== kHasFlowing) {
debug('pause');
state[kState] |= kHasFlowing;
state[kState] &= ~kFlowing;
Expand Down Expand Up @@ -1572,20 +1577,19 @@ function fromList(n, state) {
function endReadable(stream) {
const state = stream._readableState;

debug('endReadable', (state[kState] & kEndEmitted) !== 0);
debug('endReadable');
if ((state[kState] & kEndEmitted) === 0) {
state[kState] |= kEnded;
process.nextTick(endReadableNT, state, stream);
}
}

function endReadableNT(state, stream) {
debug('endReadableNT', state.endEmitted, state.length);
debug('endReadableNT');

// Check that we didn't get one last unshift.
if (!state.errored && !state.closeEmitted &&
!state.endEmitted && state.length === 0) {
state.endEmitted = true;
if ((state[kState] & (kErrored | kCloseEmitted | kEndEmitted)) === 0 && state.length === 0) {
state[kState] |= kEndEmitted;
stream.emit('end');

if (stream.writable && stream.allowHalfOpen === false) {
Expand Down