GeezerVison is showing github.com — net,http2: merge after-write code · nodejs/node@24acd53
File treeExpand file treeCollapse file tree Open diff view settings Expand file treeCollapse file tree Open diff view settings Collapse file Copy file name to clipboardExpand all lines: lib/internal/http2/core.js +9-16Lines changed: 9 additions & 16 deletions | Original file line number | Diff line number | Diff line change |
|---|
@@ -108,6 +108,7 @@ const { | 108 | 108 | writeGeneric, | 109 | 109 | writevGeneric, | 110 | 110 | onStreamRead, | | 111 | + kAfterAsyncWrite, | 111 | 112 | kMaybeDestroy, | 112 | 113 | kUpdateTimer | 113 | 114 | } = require('internal/stream_base_commons'); | @@ -1515,21 +1516,6 @@ function trackWriteState(stream, bytes) { | 1515 | 1516 | session[kHandle].chunksSentSinceLastWrite = 0; | 1516 | 1517 | } | 1517 | 1518 |
| 1518 | | -function afterDoStreamWrite(status, handle) { | 1519 | | - const stream = handle[kOwner]; | 1520 | | - const session = stream[kSession]; | 1521 | | - | 1522 | | - stream[kUpdateTimer](); | 1523 | | - | 1524 | | - const { bytes } = this; | 1525 | | - stream[kState].writeQueueSize -= bytes; | 1526 | | - | 1527 | | - if (session !== undefined) | 1528 | | - session[kState].writeQueueSize -= bytes; | 1529 | | - if (typeof this.callback === 'function') | 1530 | | - this.callback(null); | 1531 | | -} | 1532 | | - | 1533 | 1519 | function streamOnResume() { | 1534 | 1520 | if (!this.destroyed) | 1535 | 1521 | this[kHandle].readStart(); | @@ -1782,6 +1768,13 @@ class Http2Stream extends Duplex { | 1782 | 1768 | 'bug in Node.js'); | 1783 | 1769 | } | 1784 | 1770 |
| | 1771 | + [kAfterAsyncWrite]({ bytes }) { | | 1772 | + this[kState].writeQueueSize -= bytes; | | 1773 | + | | 1774 | + if (this.session !== undefined) | | 1775 | + this.session[kState].writeQueueSize -= bytes; | | 1776 | + } | | 1777 | + | 1785 | 1778 | [kWriteGeneric](writev, data, encoding, cb) { | 1786 | 1779 | // When the Http2Stream is first created, it is corked until the | 1787 | 1780 | // handle and the stream ID is assigned. However, if the user calls | @@ -1808,7 +1801,7 @@ class Http2Stream extends Duplex { | 1808 | 1801 | if (!this.headersSent) | 1809 | 1802 | this[kProceed](); | 1810 | 1803 |
| 1811 | | - const req = createWriteWrap(this[kHandle], afterDoStreamWrite); | | 1804 | + const req = createWriteWrap(this[kHandle]); | 1812 | 1805 | req.stream = this[kID]; | 1813 | 1806 |
| 1814 | 1807 | if (writev) | |
Collapse file Copy file name to clipboardExpand all lines: lib/internal/stream_base_commons.js +26-2Lines changed: 26 additions & 2 deletions | Original file line number | Diff line number | Diff line change |
|---|
@@ -16,6 +16,7 @@ const { owner_symbol } = require('internal/async_hooks').symbols; | 16 | 16 |
| 17 | 17 | const kMaybeDestroy = Symbol('kMaybeDestroy'); | 18 | 18 | const kUpdateTimer = Symbol('kUpdateTimer'); | | 19 | +const kAfterAsyncWrite = Symbol('kAfterAsyncWrite'); | 19 | 20 |
| 20 | 21 | function handleWriteReq(req, data, encoding) { | 21 | 22 | const { handle } = req; | @@ -52,11 +53,33 @@ function handleWriteReq(req, data, encoding) { | 52 | 53 | } | 53 | 54 | } | 54 | 55 |
| 55 | | -function createWriteWrap(handle, oncomplete) { | | 56 | +function onWriteComplete(status) { | | 57 | + const stream = this.handle[owner_symbol]; | | 58 | + | | 59 | + if (stream.destroyed) { | | 60 | + if (typeof this.callback === 'function') | | 61 | + this.callback(null); | | 62 | + return; | | 63 | + } | | 64 | + | | 65 | + if (status < 0) { | | 66 | + const ex = errnoException(status, 'write', this.error); | | 67 | + stream.destroy(ex, this.callback); | | 68 | + return; | | 69 | + } | | 70 | + | | 71 | + stream[kUpdateTimer](); | | 72 | + stream[kAfterAsyncWrite](this); | | 73 | + | | 74 | + if (typeof this.callback === 'function') | | 75 | + this.callback(null); | | 76 | +} | | 77 | + | | 78 | +function createWriteWrap(handle) { | 56 | 79 | var req = new WriteWrap(); | 57 | 80 |
| 58 | 81 | req.handle = handle; | 59 | | - req.oncomplete = oncomplete; | | 82 | + req.oncomplete = onWriteComplete; | 60 | 83 | req.async = false; | 61 | 84 | req.bytes = 0; | 62 | 85 | req.buffer = null; | @@ -160,6 +183,7 @@ module.exports = { | 160 | 183 | writevGeneric, | 161 | 184 | writeGeneric, | 162 | 185 | onStreamRead, | | 186 | + kAfterAsyncWrite, | 163 | 187 | kMaybeDestroy, | 164 | 188 | kUpdateTimer, | 165 | 189 | }; |
Collapse file Copy file name to clipboardExpand all lines: lib/net.js +6-34Lines changed: 6 additions & 34 deletions | Original file line number | Diff line number | Diff line change |
|---|
@@ -62,6 +62,7 @@ const { | 62 | 62 | writevGeneric, | 63 | 63 | writeGeneric, | 64 | 64 | onStreamRead, | | 65 | + kAfterAsyncWrite, | 65 | 66 | kUpdateTimer | 66 | 67 | } = require('internal/stream_base_commons'); | 67 | 68 | const { | @@ -685,6 +686,10 @@ protoGetter('localPort', function localPort() { | 685 | 686 | }); | 686 | 687 |
| 687 | 688 |
| | 689 | +Socket.prototype[kAfterAsyncWrite] = function() { | | 690 | + this[kLastWriteQueueSize] = 0; | | 691 | +}; | | 692 | + | 688 | 693 | Socket.prototype._writeGeneric = function(writev, data, encoding, cb) { | 689 | 694 | // If we are still connecting, then buffer this for later. | 690 | 695 | // The Writable logic will buffer up any more writes while | @@ -707,7 +712,7 @@ Socket.prototype._writeGeneric = function(writev, data, encoding, cb) { | 707 | 712 |
| 708 | 713 | this._unrefTimer(); | 709 | 714 |
| 710 | | - var req = createWriteWrap(this._handle, afterWrite); | | 715 | + var req = createWriteWrap(this._handle); | 711 | 716 | if (writev) | 712 | 717 | writevGeneric(this, req, data, cb); | 713 | 718 | else | @@ -771,39 +776,6 @@ protoGetter('bytesWritten', function bytesWritten() { | 771 | 776 | }); | 772 | 777 |
| 773 | 778 |
| 774 | | -function afterWrite(status, handle, err) { | 775 | | - var self = handle[owner_symbol]; | 776 | | - if (self !== process.stderr && self !== process.stdout) | 777 | | - debug('afterWrite', status); | 778 | | - | 779 | | - if (this.async) | 780 | | - self[kLastWriteQueueSize] = 0; | 781 | | - | 782 | | - // callback may come after call to destroy. | 783 | | - if (self.destroyed) { | 784 | | - debug('afterWrite destroyed'); | 785 | | - if (this.callback) | 786 | | - this.callback(null); | 787 | | - return; | 788 | | - } | 789 | | - | 790 | | - if (status < 0) { | 791 | | - var ex = errnoException(status, 'write', this.error); | 792 | | - debug('write failure', ex); | 793 | | - self.destroy(ex, this.callback); | 794 | | - return; | 795 | | - } | 796 | | - | 797 | | - self._unrefTimer(); | 798 | | - | 799 | | - if (self !== process.stderr && self !== process.stdout) | 800 | | - debug('afterWrite call cb'); | 801 | | - | 802 | | - if (this.callback) | 803 | | - this.callback.call(undefined); | 804 | | -} | 805 | | - | 806 | | - | 807 | 779 | function checkBindError(err, port, handle) { | 808 | 780 | // EADDRINUSE may not be reported until we call listen() or connect(). | 809 | 781 | // To complicate matters, a failed bind() followed by listen() or connect() | |
Collapse file Copy file name to clipboardExpand all lines: src/node_http2.cc +6-2Lines changed: 6 additions & 2 deletions | Original file line number | Diff line number | Diff line change |
|---|
@@ -1571,8 +1571,12 @@ void Http2Session::ClearOutgoing(int status) { | 1571 | 1571 | current_outgoing_buffers_.swap(outgoing_buffers_); | 1572 | 1572 | for (const nghttp2_stream_write& wr : current_outgoing_buffers_) { | 1573 | 1573 | WriteWrap* wrap = wr.req_wrap; | 1574 | | - if (wrap != nullptr) | 1575 | | - wrap->Done(status); | | 1574 | + if (wrap != nullptr) { | | 1575 | + // TODO(addaleax): Pass `status` instead of 0, so that we actually error | | 1576 | + // out with the error from the write to the underlying protocol, | | 1577 | + // if one occurred. | | 1578 | + wrap->Done(0); | | 1579 | + } | 1576 | 1580 | } | 1577 | 1581 | } | 1578 | 1582 |
| |
|