Commit c42fbcc
stream: avoid duplicated endReadableNT scheduling
Calling read() on an ended stream multiple times before the microtask
queue drains scheduled one endReadableNT tick per call, as the only
guard was endEmitted, which is set inside the tick itself. A hello-world
HTTP server was scheduling it four times per request while dumping the
unread request body.
Introduce a kEndScheduled flag armed when the tick is scheduled and
cleared when it runs. Clearing it unconditionally matters for reused
sockets: undestroy() resets endEmitted through the state descriptors
but cannot reach this flag, and a stale value would block the 'end'
event after a net.Socket reconnect.
Signed-off-by: Matteo Collina <hello@matteocollina.com>
PR-URL: #65310
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: Robert Nagy <ronagy@icloud.com>
Reviewed-By: Paolo Insogna <paolo@cowtech.it>1 parent acaf986 commit c42fbcc
1 file changed
Lines changed: 9 additions & 2 deletions
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -137,6 +137,7 @@ const kFlowing = 1 << 24; | |||
| 137 | 137 | const kHasPaused = 1 << 25; | |
| 138 | 138 | const kPaused = 1 << 26; | |
| 139 | 139 | const kDataListening = 1 << 27; | |
| 140 | + const kEndScheduled = 1 << 28; | ||
| 140 | 141 | ||
| 141 | 142 | // TODO(benjamingr) it is likely slower to do it this way than with free functions | |
| 142 | 143 | function makeBitMapDescriptor(bit) { | |
@@ -1944,15 +1945,21 @@ function endReadable(stream) { | |||
| 1944 | 1945 | const state = stream._readableState; | |
| 1945 | 1946 | ||
| 1946 | 1947 | debug('endReadable'); | |
| 1947 | - if ((state[kState] & kEndEmitted) === 0) { | ||
| 1948 | - state[kState] |= kEnded; | ||
| 1948 | + if ((state[kState] & (kEndEmitted | kEndScheduled)) === 0) { | ||
| 1949 | + state[kState] |= kEnded | kEndScheduled; | ||
| 1949 | 1950 | process.nextTick(endReadableNT, state, stream); | |
| 1950 | 1951 | } | |
| 1951 | 1952 | } | |
| 1952 | 1953 | ||
| 1953 | 1954 | function endReadableNT(state, stream) { | |
| 1954 | 1955 | debug('endReadableNT'); | |
| 1955 | 1956 | ||
| 1957 | + // The scheduled tick is running; allow endReadable() to schedule again. | ||
| 1958 | + // This matters both when the 'end' emission is skipped below (e.g. after | ||
| 1959 | + // an unshift()) and when the stream is later reset for reuse | ||
| 1960 | + // (see undestroy()), which clears kEndEmitted but not this flag. | ||
| 1961 | + state[kState] &= ~kEndScheduled; | ||
| 1962 | + | ||
| 1956 | 1963 | // Check that we didn't get one last unshift. | |
| 1957 | 1964 | if ((state[kState] & (kErrored | kCloseEmitted | kEndEmitted)) === 0 && state.length === 0) { | |
| 1958 | 1965 | state[kState] |= kEndEmitted; | |
0 commit comments