Skip to content

Commit 71b4eff

Browse files
author
shartung
committed
http: flush buffered chunks before uncorking
1 parent 8a1ca0f commit 71b4eff

3 files changed

Lines changed: 284 additions & 23 deletions

File tree

‎benchmark/http/cork.js‎

Lines changed: 35 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,35 @@
1+
'use strict';
2+
3+
const common = require('../common.js');
4+
5+
const bench = common.createBenchmark(main, {
6+
type: ['bytes', 'buffer'],
7+
len: [64, 1024],
8+
chunks: [4, 16],
9+
c: [50],
10+
duration: 5,
11+
});
12+
13+
function main({ type, len, chunks, c, duration }) {
14+
const http = require('http');
15+
const chunk = type === 'bytes' ? 'a'.repeat(len) : Buffer.alloc(len, 'a');
16+
17+
const server = http.createServer((req, res) => {
18+
res.cork();
19+
for (let i = 0; i < chunks; i++) {
20+
res.write(chunk);
21+
}
22+
res.uncork();
23+
res.end();
24+
});
25+
26+
server.listen(0, () => {
27+
bench.http({
28+
connections: c,
29+
duration,
30+
port: server.address().port,
31+
}, () => {
32+
server.close();
33+
});
34+
});
35+
}

‎lib/_http_outgoing.js‎

Lines changed: 50 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -290,45 +290,59 @@ OutgoingMessage.prototype.cork = function cork() {
290290
}
291291
};
292292

293-
OutgoingMessage.prototype.uncork = function uncork() {
294-
this[kCorked]--;
295-
if (this[kSocket]) {
296-
this[kSocket].uncork();
297-
}
298-
299-
if (this[kCorked] || this[kChunkedBuffer].length === 0) {
300-
return;
301-
}
293+
function flushChunkedBuffer(msg) {
294+
const buf = msg[kChunkedBuffer];
295+
const len = msg[kChunkedLength];
302296

303-
const len = this[kChunkedLength];
304-
const buf = this[kChunkedBuffer];
305-
306-
assert(this.chunkedEncoding);
297+
assert(msg.chunkedEncoding);
307298

308299
let callbacks;
309-
this._send(len.toString(16), 'latin1', null);
310-
this._send(crlf_buf, null, null);
300+
msg._send(len.toString(16), 'latin1', null);
301+
msg._send(crlf_buf, null, null);
311302
for (let n = 0; n < buf.length; n += 3) {
312-
this._send(buf[n + 0], buf[n + 1], null);
303+
msg._send(buf[n + 0], buf[n + 1], null);
313304
if (buf[n + 2]) {
314305
callbacks ??= [];
315306
callbacks.push(buf[n + 2]);
316307
}
317308
}
318-
this._send(crlf_buf, null, callbacks.length ? (err) => {
309+
msg._send(crlf_buf, null, callbacks.length ? (err) => {
319310
for (const callback of callbacks) {
320311
callback(err);
321312
}
322313
} : null);
323314

324-
this[kChunkedBuffer].length = 0;
325-
this[kChunkedLength] = 0;
315+
buf.length = 0;
316+
msg[kChunkedLength] = 0;
317+
}
326318

327-
// If we had a pending drain and flushed all data, emit the drain event.
328-
if (this[kNeedDrain] && this.writableLength === 0) {
329-
this[kNeedDrain] = false;
330-
this.emit('drain');
319+
function emitDrainIfNeeded(msg) {
320+
if (msg[kNeedDrain] && msg.writableLength === 0) {
321+
msg[kNeedDrain] = false;
322+
msg.emit('drain');
323+
}
324+
}
325+
326+
OutgoingMessage.prototype.uncork = function uncork() {
327+
this[kCorked]--;
328+
329+
const flushed = !this[kCorked] && this[kChunkedBuffer].length !== 0;
330+
try {
331+
if (flushed) {
332+
flushChunkedBuffer(this);
333+
}
334+
} finally {
335+
if (this[kSocket]) {
336+
this[kSocket].uncork();
337+
}
331338
}
339+
340+
if (!flushed) {
341+
return;
342+
}
343+
344+
// If we had a pending drain and flushed all data, emit the drain event.
345+
emitDrainIfNeeded(this);
332346
};
333347

334348
OutgoingMessage.prototype.setTimeout = function setTimeout(msecs, callback) {
@@ -1131,6 +1145,13 @@ OutgoingMessage.prototype.end = function end(chunk, encoding, callback) {
11311145
throw new ERR_HTTP_CONTENT_LENGTH_MISMATCH(this[kBytesWritten], this._contentLength);
11321146
}
11331147

1148+
// Flush message-level corked data before the terminating chunk. Keep the
1149+
// socket corked so all HTTP framing can be written as a single batch.
1150+
const flushed = this[kChunkedBuffer].length !== 0;
1151+
if (flushed) {
1152+
flushChunkedBuffer(this);
1153+
}
1154+
11341155
const finish = onFinish.bind(undefined, this);
11351156

11361157
if (this._hasBody && this.chunkedEncoding) {
@@ -1149,8 +1170,14 @@ OutgoingMessage.prototype.end = function end(chunk, encoding, callback) {
11491170
this[kCorked] = 1;
11501171
this.uncork();
11511172

1173+
// Mark the message as ended before emitting drain. A synchronous drain
1174+
// listener must not be able to write after the terminating chunk.
11521175
this.finished = true;
11531176

1177+
if (flushed) {
1178+
emitDrainIfNeeded(this);
1179+
}
1180+
11541181
// There is the first message on the outgoing queue, and we've sent
11551182
// everything to the socket.
11561183
debug('outgoing message end.');

‎test/parallel/test-http-outgoing-corked-end.js‎

Lines changed: 199 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,199 @@
1+
/* eslint-disable node-core/crypto-check */
2+
3+
'use strict';
4+
5+
const common = require('../common');
6+
const assert = require('assert');
7+
const http = require('http');
8+
const net = require('net');
9+
10+
function runRoundTrip(transport, serverOptions, requestOptions = {}) {
11+
return new Promise((resolve, reject) => {
12+
const server = serverOptions === undefined ?
13+
transport.createServer(onRequest) :
14+
transport.createServer(serverOptions, onRequest);
15+
16+
function onRequest(req, res) {
17+
let body = '';
18+
req.setEncoding('utf8');
19+
req.on('data', (chunk) => body += chunk);
20+
req.on('end', common.mustCall(() => {
21+
assert.strictEqual(body, 'ABCD');
22+
23+
const callbacks = [];
24+
res.setHeader('Trailer', 'x-test');
25+
res.flushHeaders();
26+
27+
// Exercise an explicit flush while the socket remains corked.
28+
res.cork();
29+
res.cork();
30+
res.write('E', common.mustCall(() => callbacks.push('E')));
31+
res.write('F', common.mustCall(() => callbacks.push('F')));
32+
33+
const originalSend = res._send;
34+
res._send = common.mustCall(function(...args) {
35+
assert.notStrictEqual(this.socket.writableCorked, 0);
36+
return originalSend.apply(this, args);
37+
}, 5);
38+
try {
39+
res.uncork();
40+
res.uncork();
41+
} finally {
42+
res._send = originalSend;
43+
}
44+
45+
// end() must flush this buffer before the terminating chunk.
46+
res.cork();
47+
res.cork();
48+
res.write('G', common.mustCall(() => callbacks.push('G')));
49+
res.addTrailers({ 'x-test': 'yes' });
50+
res.once('finish', common.mustCall(() => callbacks.push('finish')));
51+
res.end('H', common.mustCall(() => {
52+
callbacks.push('end');
53+
assert.deepStrictEqual(callbacks, ['E', 'F', 'G', 'finish', 'end']);
54+
}));
55+
assert.strictEqual(res.writableCorked, 0);
56+
}));
57+
}
58+
59+
server.on('error', reject);
60+
server.listen(0, common.localhostIPv4, common.mustCall(() => {
61+
const callbacks = [];
62+
const req = transport.request({
63+
host: common.localhostIPv4,
64+
port: server.address().port,
65+
method: 'POST',
66+
...requestOptions,
67+
}, common.mustCall((res) => {
68+
let body = '';
69+
res.setEncoding('utf8');
70+
res.on('data', (chunk) => body += chunk);
71+
res.on('end', common.mustCall(() => {
72+
assert.strictEqual(body, 'EFGH');
73+
assert.strictEqual(res.trailers['x-test'], 'yes');
74+
server.close(common.mustCall(resolve));
75+
}));
76+
}));
77+
78+
req.on('error', reject);
79+
req.cork();
80+
req.cork();
81+
req.write('A', common.mustCall(() => callbacks.push('A')));
82+
req.write('B', common.mustCall(() => callbacks.push('B')));
83+
req.uncork();
84+
req.uncork();
85+
req.cork();
86+
req.cork();
87+
req.write('C', common.mustCall(() => callbacks.push('C')));
88+
req.once('finish', common.mustCall(() => callbacks.push('finish')));
89+
req.end('D', common.mustCall(() => {
90+
callbacks.push('end');
91+
assert.deepStrictEqual(callbacks, ['A', 'B', 'C', 'finish', 'end']);
92+
}));
93+
assert.strictEqual(req.writableCorked, 0);
94+
}));
95+
});
96+
}
97+
98+
function runPipelined() {
99+
return new Promise((resolve, reject) => {
100+
let firstResponse;
101+
const server = http.createServer(common.mustCall((req, res) => {
102+
if (req.url === '/first') {
103+
firstResponse = res;
104+
return;
105+
}
106+
107+
assert.strictEqual(req.url, '/second');
108+
assert.strictEqual(res.socket, null);
109+
res.cork();
110+
res.cork();
111+
res.write('B');
112+
res.write('C');
113+
res.end();
114+
assert.strictEqual(res.writableCorked, 0);
115+
firstResponse.end('A');
116+
}, 2));
117+
118+
server.on('error', reject);
119+
server.listen(0, common.localhostIPv4, common.mustCall(() => {
120+
const socket = net.createConnection({
121+
host: common.localhostIPv4,
122+
port: server.address().port,
123+
});
124+
let response = '';
125+
126+
socket.setEncoding('latin1');
127+
socket.on('error', reject);
128+
socket.on('data', (chunk) => response += chunk);
129+
socket.on('end', common.mustCall(() => {
130+
assert.match(response, /\r\n\r\nAHTTP\/1\.1 200 OK\r\n/);
131+
assert.match(response, /\r\n\r\n1\r\nB\r\n1\r\nC\r\n0\r\n\r\n$/);
132+
server.close(common.mustCall(resolve));
133+
}));
134+
socket.on('connect', common.mustCall(() => {
135+
socket.end(
136+
'GET /first HTTP/1.1\r\nHost: localhost\r\n\r\n' +
137+
'GET /second HTTP/1.1\r\nHost: localhost\r\n' +
138+
'Connection: close\r\n\r\n',
139+
);
140+
}));
141+
}));
142+
});
143+
}
144+
145+
function runDrainOnEnd() {
146+
return new Promise((resolve, reject) => {
147+
const server = http.createServer(common.mustCall((req, res) => {
148+
res.cork();
149+
assert.strictEqual(res.write('1'.repeat(10)), true);
150+
assert.strictEqual(res.write('2'.repeat(1000)), false);
151+
assert.strictEqual(res.writableNeedDrain, true);
152+
153+
res.once('drain', common.mustCall(() => {
154+
assert.strictEqual(res.finished, true);
155+
assert.strictEqual(res.writableNeedDrain, false);
156+
assert.strictEqual(res.writableLength, 0);
157+
}));
158+
res.end();
159+
}));
160+
161+
server.on('connection', common.mustCall((socket) => {
162+
socket._writableState.highWaterMark = 1000;
163+
}));
164+
server.on('error', reject);
165+
server.listen(0, common.localhostIPv4, common.mustCall(() => {
166+
const req = http.get({
167+
host: common.localhostIPv4,
168+
port: server.address().port,
169+
}, common.mustCall((res) => {
170+
let body = '';
171+
res.setEncoding('utf8');
172+
res.on('data', (chunk) => body += chunk);
173+
res.on('end', common.mustCall(() => {
174+
assert.strictEqual(body, '1'.repeat(10) + '2'.repeat(1000));
175+
server.close(common.mustCall(resolve));
176+
}));
177+
}));
178+
req.on('error', reject);
179+
}));
180+
});
181+
}
182+
183+
async function main() {
184+
await runRoundTrip(http);
185+
186+
if (common.hasCrypto) {
187+
const fixtures = require('../common/fixtures');
188+
const https = require('https');
189+
await runRoundTrip(https, {
190+
key: fixtures.readKey('agent1-key.pem'),
191+
cert: fixtures.readKey('agent1-cert.pem'),
192+
}, { rejectUnauthorized: false });
193+
}
194+
195+
await runPipelined();
196+
await runDrainOnEnd();
197+
}
198+
199+
main().then(common.mustCall());

0 commit comments

Comments
 (0)