@@ -1991,6 +1991,47 @@ function shutdownWritable(callback) { |
1991 | 1991 | return afterShutdown.call(req, 0); |
1992 | 1992 | } |
1993 | 1993 |
|
| 1994 | +// Completes one of the two halves of a dispatched write (the write callback |
| 1995 | +// itself and the end-of-stream check); the stream machinery callback runs |
| 1996 | +// once both have finished. The state lives on stream[kState] because only a |
| 1997 | +// single write may be in flight at any given time. |
| 1998 | +function finishWrite(stream) { |
| 1999 | + const state = stream[kState]; |
| 2000 | + if (--state.writePending !== 0) |
| 2001 | + return; |
| 2002 | + const cb = state.writeCb; |
| 2003 | + state.writeCb = null; |
| 2004 | + const err = aggregateTwoErrors(state.endErr, state.writeErr); |
| 2005 | + state.writeErr = null; |
| 2006 | + state.endErr = null; |
| 2007 | + // writeGeneric does not destroy on error and |
| 2008 | + // we cannot enable autoDestroy, |
| 2009 | + // so make sure to destroy on error. |
| 2010 | + if (err) { |
| 2011 | + stream.destroy(err); |
| 2012 | + } |
| 2013 | + cb(err); |
| 2014 | +} |
| 2015 | + |
| 2016 | +// Runs on the tick after a write was dispatched: if the write turned out to |
| 2017 | +// be the last chunk of an ending writable, shut the writable side down right |
| 2018 | +// away so the final DATA frame can include the END_STREAM flag. |
| 2019 | +function endCheckNT(stream) { |
| 2020 | + const state = stream[kState]; |
| 2021 | + if (state.writeErr || |
| 2022 | + !stream._writableState.ending || |
| 2023 | + stream._writableState.buffered.length || |
| 2024 | + (state.flags & STREAM_FLAGS_HAS_TRAILERS)) { |
| 2025 | + finishWrite(stream); |
| 2026 | + return; |
| 2027 | + } |
| 2028 | + debugStreamObj(stream, 'shutting down writable on last write'); |
| 2029 | + shutdownWritable.call(stream, (err) => { |
| 2030 | + state.endErr = err; |
| 2031 | + finishWrite(stream); |
| 2032 | + }); |
| 2033 | +} |
| 2034 | + |
1994 | 2035 | function finishSendTrailers(stream, headersList) { |
1995 | 2036 | // The stream might be destroyed and in that case |
1996 | 2037 | // there is nothing to do. |
@@ -2098,6 +2139,12 @@ class Http2Stream extends Duplex { |
2098 | 2139 | writeQueueSize: 0, |
2099 | 2140 | trailersReady: false, |
2100 | 2141 | endAfterHeaders: false, |
| 2142 | + writeCb: null, |
| 2143 | + writeErr: null, |
| 2144 | + endErr: null, |
| 2145 | + writePending: 0, |
| 2146 | + shutdownWritableCalled: false, |
| 2147 | + fd: -1, |
2101 | 2148 | }; |
2102 | 2149 |
|
2103 | 2150 | // Fields used by the compat API to avoid megamorphisms. |
@@ -2285,45 +2332,34 @@ class Http2Stream extends Duplex { |
2285 | 2332 | if (!this.headersSent) |
2286 | 2333 | this[kProceed](); |
2287 | 2334 |
|
2288 | | - let req; |
| 2335 | + // The stream machinery dispatches at most one _write()/_writev() at a |
| 2336 | + // time, so the coordination state between the write callback and the |
| 2337 | + // end-of-stream check below can live on the stream state instead of |
| 2338 | + // being captured by per-write closures. |
| 2339 | + const state = this[kState]; |
| 2340 | + state.writeCb = cb; |
| 2341 | + state.writeErr = null; |
| 2342 | + state.endErr = null; |
| 2343 | + |
| 2344 | + if (state.flags & STREAM_FLAGS_HAS_TRAILERS) { |
| 2345 | + // Trailers are pending, so the writable side cannot be shut down |
| 2346 | + // early anyway; there is no point in scheduling the end check. |
| 2347 | + state.writePending = 1; |
| 2348 | + } else { |
| 2349 | + state.writePending = 2; |
| 2350 | + // Shutdown write stream right after last chunk is sent |
| 2351 | + // so final DATA frame can include END_STREAM flag |
| 2352 | + process.nextTick(endCheckNT, this); |
| 2353 | + } |
2289 | 2354 |
|
2290 | | - let waitingForWriteCallback = true; |
2291 | | - let waitingForEndCheck = true; |
2292 | | - let writeCallbackErr; |
2293 | | - let endCheckCallbackErr; |
2294 | | - const done = () => { |
2295 | | - if (waitingForEndCheck || waitingForWriteCallback) return; |
2296 | | - const err = aggregateTwoErrors(endCheckCallbackErr, writeCallbackErr); |
2297 | | - // writeGeneric does not destroy on error and |
2298 | | - // we cannot enable autoDestroy, |
2299 | | - // so make sure to destroy on error. |
2300 | | - if (err) { |
2301 | | - this.destroy(err); |
2302 | | - } |
2303 | | - cb(err); |
2304 | | - }; |
| 2355 | + // This is invoked both as a method on the write req and as a plain |
| 2356 | + // call, so the stream has to be captured here. |
2305 | 2357 | const writeCallback = (err) => { |
2306 | | - waitingForWriteCallback = false; |
2307 | | - writeCallbackErr = err; |
2308 | | - done(); |
2309 | | - }; |
2310 | | - const endCheckCallback = (err) => { |
2311 | | - waitingForEndCheck = false; |
2312 | | - endCheckCallbackErr = err; |
2313 | | - done(); |
| 2358 | + state.writeErr = err; |
| 2359 | + finishWrite(this); |
2314 | 2360 | }; |
2315 | | - // Shutdown write stream right after last chunk is sent |
2316 | | - // so final DATA frame can include END_STREAM flag |
2317 | | - process.nextTick(() => { |
2318 | | - if (writeCallbackErr || |
2319 | | - !this._writableState.ending || |
2320 | | - this._writableState.buffered.length || |
2321 | | - (this[kState].flags & STREAM_FLAGS_HAS_TRAILERS)) |
2322 | | - return endCheckCallback(); |
2323 | | - debugStreamObj(this, 'shutting down writable on last write'); |
2324 | | - shutdownWritable.call(this, endCheckCallback); |
2325 | | - }); |
2326 | 2361 |
|
| 2362 | + let req; |
2327 | 2363 | if (writev) |
2328 | 2364 | req = writevGeneric(this, data, writeCallback); |
2329 | 2365 | else |
|