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.

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

http: outgoing cork #29053

Closed
wants to merge 1 commit into from
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
16 changes: 16 additions & 0 deletions doc/api/http.md
Expand Up @@ -1231,6 +1231,13 @@ deprecated: v13.0.0

See [`response.socket`][].

### response.cork()
<!-- YAML
added: REPLACEME
-->

See [`writable.cork()`][].

### response.end(\[data\[, encoding\]\]\[, callback\])
<!-- YAML
added: v0.1.90
Expand Down Expand Up @@ -1516,6 +1523,13 @@ response.statusMessage = 'Not found';
After response header was sent to the client, this property indicates the
status message which was sent out.

### response.uncork()
<!-- YAML
added: REPLACEME
-->

See [`writable.uncork()`][].

### response.writableEnded
<!-- YAML
added: v12.9.0
Expand Down Expand Up @@ -2356,3 +2370,5 @@ not abort the request or do anything besides add a `'timeout'` event.
[`socket.unref()`]: net.html#net_socket_unref
[`url.parse()`]: url.html#url_url_parse_urlstring_parsequerystring_slashesdenotehost
[`HPE_HEADER_OVERFLOW`]: errors.html#errors_hpe_header_overflow
[`writable.cork()`]: stream.html#stream_writable_cork
[`writable.uncork()`]: stream.html#stream_writable_uncork
45 changes: 39 additions & 6 deletions lib/_http_outgoing.js
Expand Up @@ -56,6 +56,8 @@ const { validateString } = require('internal/validators');
const HIGH_WATER_MARK = getDefaultHighWaterMark();
const { CRLF, debug } = common;

const kCorked = Symbol('corked');

const RE_CONN_CLOSE = /(?:^|\W)close(?:$|\W)/i;
const RE_TE_CHUNKED = common.chunkExpression;

Expand Down Expand Up @@ -99,6 +101,7 @@ function OutgoingMessage() {

this.finished = false;
this._headerSent = false;
this[kCorked] = 0;

this.socket = null;
this._header = null;
Expand Down Expand Up @@ -137,6 +140,13 @@ Object.defineProperty(OutgoingMessage.prototype, 'writableHighWaterMark', {
}
});

Object.defineProperty(OutgoingMessage.prototype, 'writableCorked', {
get() {
const corked = this.socket ? this.socket.writableCorked : 0;
return corked + this[kCorked];
}
});

Object.defineProperty(OutgoingMessage.prototype, '_headers', {
get: internalUtil.deprecate(function() {
return this.getHeaders();
Expand Down Expand Up @@ -213,6 +223,21 @@ OutgoingMessage.prototype._renderHeaders = function _renderHeaders() {
return headers;
};

OutgoingMessage.prototype.cork = function() {
if (this.socket) {
this.socket.cork();
} else {
this[kCorked]++;
}
};

OutgoingMessage.prototype.uncork = function() {
if (this.socket) {
this.socket.uncork();
} else if (this[kCorked]) {
this[kCorked]--;
}
};

OutgoingMessage.prototype.setTimeout = function setTimeout(msecs, callback) {

Expand Down Expand Up @@ -710,7 +735,10 @@ OutgoingMessage.prototype.end = function end(chunk, encoding, callback) {
return this;
}

var uncork;
if (this.socket) {
this.socket.cork();
}

if (chunk) {
if (typeof chunk !== 'string' && !(chunk instanceof Buffer)) {
throw new ERR_INVALID_ARG_TYPE('chunk', ['string', 'Buffer'], chunk);
Expand All @@ -721,10 +749,6 @@ OutgoingMessage.prototype.end = function end(chunk, encoding, callback) {
else
this._contentLength = chunk.length;
}
if (this.socket) {
this.socket.cork();
uncork = true;
}
write_(this, chunk, encoding, null, true);
} else if (!this._header) {
this._contentLength = 0;
Expand All @@ -743,8 +767,12 @@ OutgoingMessage.prototype.end = function end(chunk, encoding, callback) {
this._send('', 'latin1', finish);
}

if (uncork)
if (this.socket) {
// Fully uncork connection on end().
this.socket._writableState.corked = 1;
this.socket.uncork();
}
this[kCorked] = 0;

this.finished = true;

Expand Down Expand Up @@ -805,6 +833,11 @@ OutgoingMessage.prototype._flush = function _flush() {
};

OutgoingMessage.prototype._flushOutput = function _flushOutput(socket) {
while (this[kCorked]) {
this[kCorked]--;
socket.cork();
}

const outputLength = this.outputData.length;
if (outputLength <= 0)
return undefined;
Expand Down
10 changes: 10 additions & 0 deletions lib/_stream_duplex.js
Expand Up @@ -108,6 +108,16 @@ Object.defineProperty(Duplex.prototype, 'writableFinished', {
}
});

Object.defineProperty(Duplex.prototype, 'writableCorked', {
// Making it explicit this property is not enumerable
// because otherwise some prototype manipulation in
// userland will fail
enumerable: false,
get() {
return this._writableState ? this._writableState.corked : 0;
}
});

Object.defineProperty(Duplex.prototype, 'writableEnded', {
// Making it explicit this property is not enumerable
// because otherwise some prototype manipulation in
Expand Down
12 changes: 12 additions & 0 deletions lib/internal/http2/compat.js
Expand Up @@ -503,6 +503,10 @@ class Http2ServerResponse extends Stream {
return this[kState].statusCode;
}

get writableCorked() {
return this[kStream].writableCorked;
}

set statusCode(code) {
code |= 0;
if (code >= 100 && code < 200)
Expand Down Expand Up @@ -627,6 +631,14 @@ class Http2ServerResponse extends Stream {
return this;
}

cork() {
this[kStream].cork();
}

uncork() {
this[kStream].uncork();
}

write(chunk, encoding, cb) {
if (typeof encoding === 'function') {
cb = encoding;
Expand Down
33 changes: 33 additions & 0 deletions test/parallel/test-http-response-cork.js
@@ -0,0 +1,33 @@
'use strict';
const common = require('../common');
const http = require('http');
const assert = require('assert');

const server = http.createServer((req, res) => {
let corked = false;
const originalWrite = res.socket.write;
res.socket.write = common.mustCall((...args) => {
assert.strictEqual(corked, false);
return originalWrite.call(res.socket, ...args);
}, 5);
corked = true;
res.cork();
assert.strictEqual(res.writableCorked, res.socket.writableCorked);
res.cork();
assert.strictEqual(res.writableCorked, res.socket.writableCorked);
res.writeHead(200, { 'a-header': 'a-header-value' });
res.uncork();
assert.strictEqual(res.writableCorked, res.socket.writableCorked);
corked = false;
res.end('asd');
assert.strictEqual(res.writableCorked, res.socket.writableCorked);
});

server.listen(0, () => {
http.get({ port: server.address().port }, (res) => {
res.on('data', common.mustCall());
res.on('end', common.mustCall(() => {
server.close();
}));
});
});