From 71b4effc864ed3f80c2518a6756ebeb60ffa63ae Mon Sep 17 00:00:00 2001 From: shartung Date: Sat, 1 Aug 2026 01:21:58 +0200 Subject: [PATCH 1/2] http: flush buffered chunks before uncorking --- benchmark/http/cork.js | 35 +++ lib/_http_outgoing.js | 73 +++++-- .../parallel/test-http-outgoing-corked-end.js | 199 ++++++++++++++++++ 3 files changed, 284 insertions(+), 23 deletions(-) create mode 100644 benchmark/http/cork.js create mode 100644 test/parallel/test-http-outgoing-corked-end.js diff --git a/benchmark/http/cork.js b/benchmark/http/cork.js new file mode 100644 index 000000000000..d49748673fe6 --- /dev/null +++ b/benchmark/http/cork.js @@ -0,0 +1,35 @@ +'use strict'; + +const common = require('../common.js'); + +const bench = common.createBenchmark(main, { + type: ['bytes', 'buffer'], + len: [64, 1024], + chunks: [4, 16], + c: [50], + duration: 5, +}); + +function main({ type, len, chunks, c, duration }) { + const http = require('http'); + const chunk = type === 'bytes' ? 'a'.repeat(len) : Buffer.alloc(len, 'a'); + + const server = http.createServer((req, res) => { + res.cork(); + for (let i = 0; i < chunks; i++) { + res.write(chunk); + } + res.uncork(); + res.end(); + }); + + server.listen(0, () => { + bench.http({ + connections: c, + duration, + port: server.address().port, + }, () => { + server.close(); + }); + }); +} diff --git a/lib/_http_outgoing.js b/lib/_http_outgoing.js index 7694fe4b5b3f..a2134757509a 100644 --- a/lib/_http_outgoing.js +++ b/lib/_http_outgoing.js @@ -290,45 +290,59 @@ OutgoingMessage.prototype.cork = function cork() { } }; -OutgoingMessage.prototype.uncork = function uncork() { - this[kCorked]--; - if (this[kSocket]) { - this[kSocket].uncork(); - } - - if (this[kCorked] || this[kChunkedBuffer].length === 0) { - return; - } +function flushChunkedBuffer(msg) { + const buf = msg[kChunkedBuffer]; + const len = msg[kChunkedLength]; - const len = this[kChunkedLength]; - const buf = this[kChunkedBuffer]; - - assert(this.chunkedEncoding); + assert(msg.chunkedEncoding); let callbacks; - this._send(len.toString(16), 'latin1', null); - this._send(crlf_buf, null, null); + msg._send(len.toString(16), 'latin1', null); + msg._send(crlf_buf, null, null); for (let n = 0; n < buf.length; n += 3) { - this._send(buf[n + 0], buf[n + 1], null); + msg._send(buf[n + 0], buf[n + 1], null); if (buf[n + 2]) { callbacks ??= []; callbacks.push(buf[n + 2]); } } - this._send(crlf_buf, null, callbacks.length ? (err) => { + msg._send(crlf_buf, null, callbacks.length ? (err) => { for (const callback of callbacks) { callback(err); } } : null); - this[kChunkedBuffer].length = 0; - this[kChunkedLength] = 0; + buf.length = 0; + msg[kChunkedLength] = 0; +} - // If we had a pending drain and flushed all data, emit the drain event. - if (this[kNeedDrain] && this.writableLength === 0) { - this[kNeedDrain] = false; - this.emit('drain'); +function emitDrainIfNeeded(msg) { + if (msg[kNeedDrain] && msg.writableLength === 0) { + msg[kNeedDrain] = false; + msg.emit('drain'); + } +} + +OutgoingMessage.prototype.uncork = function uncork() { + this[kCorked]--; + + const flushed = !this[kCorked] && this[kChunkedBuffer].length !== 0; + try { + if (flushed) { + flushChunkedBuffer(this); + } + } finally { + if (this[kSocket]) { + this[kSocket].uncork(); + } } + + if (!flushed) { + return; + } + + // If we had a pending drain and flushed all data, emit the drain event. + emitDrainIfNeeded(this); }; OutgoingMessage.prototype.setTimeout = function setTimeout(msecs, callback) { @@ -1131,6 +1145,13 @@ OutgoingMessage.prototype.end = function end(chunk, encoding, callback) { throw new ERR_HTTP_CONTENT_LENGTH_MISMATCH(this[kBytesWritten], this._contentLength); } + // Flush message-level corked data before the terminating chunk. Keep the + // socket corked so all HTTP framing can be written as a single batch. + const flushed = this[kChunkedBuffer].length !== 0; + if (flushed) { + flushChunkedBuffer(this); + } + const finish = onFinish.bind(undefined, this); if (this._hasBody && this.chunkedEncoding) { @@ -1149,8 +1170,14 @@ OutgoingMessage.prototype.end = function end(chunk, encoding, callback) { this[kCorked] = 1; this.uncork(); + // Mark the message as ended before emitting drain. A synchronous drain + // listener must not be able to write after the terminating chunk. this.finished = true; + if (flushed) { + emitDrainIfNeeded(this); + } + // There is the first message on the outgoing queue, and we've sent // everything to the socket. debug('outgoing message end.'); diff --git a/test/parallel/test-http-outgoing-corked-end.js b/test/parallel/test-http-outgoing-corked-end.js new file mode 100644 index 000000000000..c51fdae90a81 --- /dev/null +++ b/test/parallel/test-http-outgoing-corked-end.js @@ -0,0 +1,199 @@ +/* eslint-disable node-core/crypto-check */ + +'use strict'; + +const common = require('../common'); +const assert = require('assert'); +const http = require('http'); +const net = require('net'); + +function runRoundTrip(transport, serverOptions, requestOptions = {}) { + return new Promise((resolve, reject) => { + const server = serverOptions === undefined ? + transport.createServer(onRequest) : + transport.createServer(serverOptions, onRequest); + + function onRequest(req, res) { + let body = ''; + req.setEncoding('utf8'); + req.on('data', (chunk) => body += chunk); + req.on('end', common.mustCall(() => { + assert.strictEqual(body, 'ABCD'); + + const callbacks = []; + res.setHeader('Trailer', 'x-test'); + res.flushHeaders(); + + // Exercise an explicit flush while the socket remains corked. + res.cork(); + res.cork(); + res.write('E', common.mustCall(() => callbacks.push('E'))); + res.write('F', common.mustCall(() => callbacks.push('F'))); + + const originalSend = res._send; + res._send = common.mustCall(function(...args) { + assert.notStrictEqual(this.socket.writableCorked, 0); + return originalSend.apply(this, args); + }, 5); + try { + res.uncork(); + res.uncork(); + } finally { + res._send = originalSend; + } + + // end() must flush this buffer before the terminating chunk. + res.cork(); + res.cork(); + res.write('G', common.mustCall(() => callbacks.push('G'))); + res.addTrailers({ 'x-test': 'yes' }); + res.once('finish', common.mustCall(() => callbacks.push('finish'))); + res.end('H', common.mustCall(() => { + callbacks.push('end'); + assert.deepStrictEqual(callbacks, ['E', 'F', 'G', 'finish', 'end']); + })); + assert.strictEqual(res.writableCorked, 0); + })); + } + + server.on('error', reject); + server.listen(0, common.localhostIPv4, common.mustCall(() => { + const callbacks = []; + const req = transport.request({ + host: common.localhostIPv4, + port: server.address().port, + method: 'POST', + ...requestOptions, + }, common.mustCall((res) => { + let body = ''; + res.setEncoding('utf8'); + res.on('data', (chunk) => body += chunk); + res.on('end', common.mustCall(() => { + assert.strictEqual(body, 'EFGH'); + assert.strictEqual(res.trailers['x-test'], 'yes'); + server.close(common.mustCall(resolve)); + })); + })); + + req.on('error', reject); + req.cork(); + req.cork(); + req.write('A', common.mustCall(() => callbacks.push('A'))); + req.write('B', common.mustCall(() => callbacks.push('B'))); + req.uncork(); + req.uncork(); + req.cork(); + req.cork(); + req.write('C', common.mustCall(() => callbacks.push('C'))); + req.once('finish', common.mustCall(() => callbacks.push('finish'))); + req.end('D', common.mustCall(() => { + callbacks.push('end'); + assert.deepStrictEqual(callbacks, ['A', 'B', 'C', 'finish', 'end']); + })); + assert.strictEqual(req.writableCorked, 0); + })); + }); +} + +function runPipelined() { + return new Promise((resolve, reject) => { + let firstResponse; + const server = http.createServer(common.mustCall((req, res) => { + if (req.url === '/first') { + firstResponse = res; + return; + } + + assert.strictEqual(req.url, '/second'); + assert.strictEqual(res.socket, null); + res.cork(); + res.cork(); + res.write('B'); + res.write('C'); + res.end(); + assert.strictEqual(res.writableCorked, 0); + firstResponse.end('A'); + }, 2)); + + server.on('error', reject); + server.listen(0, common.localhostIPv4, common.mustCall(() => { + const socket = net.createConnection({ + host: common.localhostIPv4, + port: server.address().port, + }); + let response = ''; + + socket.setEncoding('latin1'); + socket.on('error', reject); + socket.on('data', (chunk) => response += chunk); + socket.on('end', common.mustCall(() => { + assert.match(response, /\r\n\r\nAHTTP\/1\.1 200 OK\r\n/); + assert.match(response, /\r\n\r\n1\r\nB\r\n1\r\nC\r\n0\r\n\r\n$/); + server.close(common.mustCall(resolve)); + })); + socket.on('connect', common.mustCall(() => { + socket.end( + 'GET /first HTTP/1.1\r\nHost: localhost\r\n\r\n' + + 'GET /second HTTP/1.1\r\nHost: localhost\r\n' + + 'Connection: close\r\n\r\n', + ); + })); + })); + }); +} + +function runDrainOnEnd() { + return new Promise((resolve, reject) => { + const server = http.createServer(common.mustCall((req, res) => { + res.cork(); + assert.strictEqual(res.write('1'.repeat(10)), true); + assert.strictEqual(res.write('2'.repeat(1000)), false); + assert.strictEqual(res.writableNeedDrain, true); + + res.once('drain', common.mustCall(() => { + assert.strictEqual(res.finished, true); + assert.strictEqual(res.writableNeedDrain, false); + assert.strictEqual(res.writableLength, 0); + })); + res.end(); + })); + + server.on('connection', common.mustCall((socket) => { + socket._writableState.highWaterMark = 1000; + })); + server.on('error', reject); + server.listen(0, common.localhostIPv4, common.mustCall(() => { + const req = http.get({ + host: common.localhostIPv4, + port: server.address().port, + }, common.mustCall((res) => { + let body = ''; + res.setEncoding('utf8'); + res.on('data', (chunk) => body += chunk); + res.on('end', common.mustCall(() => { + assert.strictEqual(body, '1'.repeat(10) + '2'.repeat(1000)); + server.close(common.mustCall(resolve)); + })); + })); + req.on('error', reject); + })); + }); +} + +async function main() { + await runRoundTrip(http); + + if (common.hasCrypto) { + const fixtures = require('../common/fixtures'); + const https = require('https'); + await runRoundTrip(https, { + key: fixtures.readKey('agent1-key.pem'), + cert: fixtures.readKey('agent1-cert.pem'), + }, { rejectUnauthorized: false }); + } + + await runPipelined(); + await runDrainOnEnd(); +} + +main().then(common.mustCall()); From a50195a6eae1447d2f679fc3cece0806fbd84818 Mon Sep 17 00:00:00 2001 From: shartung Date: Sat, 1 Aug 2026 01:23:30 +0200 Subject: [PATCH 2/2] http: cache maxHeaderPairs per header section --- benchmark/http/bench-parser.js | 2 + src/node_http_parser.cc | 31 +++++--- ...test-http-parser-max-header-pairs-cache.js | 79 +++++++++++++++++++ 3 files changed, 100 insertions(+), 12 deletions(-) create mode 100644 test/parallel/test-http-parser-max-header-pairs-cache.js diff --git a/benchmark/http/bench-parser.js b/benchmark/http/bench-parser.js index 0a1e8f7b5e8a..72cb2b6feb18 100644 --- a/benchmark/http/bench-parser.js +++ b/benchmark/http/bench-parser.js @@ -31,6 +31,8 @@ function main({ len, n }) { function newParser(type) { const parser = new HTTPParser(); parser.initialize(type, {}); + // Direct parsers bypass cleanParser(); use its production default. + parser.maxHeaderPairs = 2000; parser.headers = []; diff --git a/src/node_http_parser.cc b/src/node_http_parser.cc index 62e83074bf88..bab9cdc45cd7 100644 --- a/src/node_http_parser.cc +++ b/src/node_http_parser.cc @@ -323,6 +323,7 @@ class Parser : public AsyncWrap, public StreamListener { allocator_.Reset(); url_.Reset(); status_message_.Reset(); + max_header_pairs_cached_ = false; if (connectionsList_ != nullptr) { connectionsList_->Push(this); @@ -465,6 +466,7 @@ class Parser : public AsyncWrap, public StreamListener { num_fields_ = 0; num_values_ = 0; header_pairs_ = 0; + max_header_pairs_cached_ = false; // METHOD if (parser_.type == HTTP_REQUEST) { @@ -1034,6 +1036,7 @@ class Parser : public AsyncWrap, public StreamListener { headers_completed_ = false; max_http_header_size_ = max_http_header_size; header_pairs_ = 0; + max_header_pairs_cached_ = false; } @@ -1053,21 +1056,23 @@ class Parser : public AsyncWrap, public StreamListener { header_pairs_ += 2; - Local max_header_pairs_v; - if (!object() - ->Get(env()->context(), - FIXED_ONE_BYTE_STRING(env()->isolate(), "maxHeaderPairs")) - .ToLocal(&max_header_pairs_v)) { - got_exception_ = true; - return -1; - } + if (!max_header_pairs_cached_) { + Local max_header_pairs_v; + if (!object() + ->Get(env()->context(), + FIXED_ONE_BYTE_STRING(env()->isolate(), "maxHeaderPairs")) + .ToLocal(&max_header_pairs_v)) { + got_exception_ = true; + return -1; + } - if (!max_header_pairs_v->IsNumber()) { - return 0; + max_header_pairs_ = max_header_pairs_v->IsNumber() + ? max_header_pairs_v.As()->Value() + : 0; + max_header_pairs_cached_ = true; } - const double max_header_pairs = max_header_pairs_v.As()->Value(); - if (max_header_pairs > 0 && header_pairs_ > max_header_pairs) { + if (max_header_pairs_ > 0 && header_pairs_ > max_header_pairs_) { llhttp_set_error_reason(&parser_, "HPE_HEADER_OVERFLOW:Header overflow"); return HPE_USER; } @@ -1109,6 +1114,8 @@ class Parser : public AsyncWrap, public StreamListener { const char* current_buffer_data_; bool headers_completed_ = false; size_t header_pairs_ = 0; + double max_header_pairs_ = 0; + bool max_header_pairs_cached_ = false; bool pending_pause_ = false; bool received_data_ = false; uint64_t header_nread_ = 0; diff --git a/test/parallel/test-http-parser-max-header-pairs-cache.js b/test/parallel/test-http-parser-max-header-pairs-cache.js new file mode 100644 index 000000000000..a8d88b798c89 --- /dev/null +++ b/test/parallel/test-http-parser-max-header-pairs-cache.js @@ -0,0 +1,79 @@ +'use strict'; + +const common = require('../common'); +const assert = require('assert'); +const { HTTPParser } = require('_http_common'); + +const { REQUEST } = HTTPParser; +const kOnHeaders = HTTPParser.kOnHeaders | 0; +const kOnHeadersComplete = HTTPParser.kOnHeadersComplete | 0; +const kOnBody = HTTPParser.kOnBody | 0; +const kOnMessageComplete = HTTPParser.kOnMessageComplete | 0; + +function createParser() { + const parser = new HTTPParser(); + parser.initialize(REQUEST, {}); + parser[kOnHeaders] = () => {}; + parser[kOnHeadersComplete] = () => {}; + parser[kOnBody] = common.mustNotCall(); + parser[kOnMessageComplete] = () => {}; + return parser; +} + +// maxHeaderPairs is cached once for each independent header section. Main +// headers, trailers, the next message, and a reinitialized parser must each +// observe a fresh value. +{ + const parser = createParser(); + const limits = [2, 4, 2, 2]; + + Object.defineProperty(parser, 'maxHeaderPairs', { + configurable: true, + get: common.mustCall(() => limits.shift(), limits.length), + }); + + parser[kOnHeadersComplete] = common.mustCall(undefined, 3); + parser[kOnMessageComplete] = common.mustCall(undefined, 3); + + const pipelined = Buffer.from( + 'POST /first HTTP/1.1\r\n' + + 'Transfer-Encoding: chunked\r\n' + + '\r\n' + + '0\r\n' + + 'X-A: a\r\n' + + 'X-B: b\r\n' + + '\r\n' + + 'GET /second HTTP/1.1\r\n' + + 'X-C: c\r\n' + + '\r\n' + ); + assert.strictEqual(parser.execute(pipelined, 0, pipelined.length), + pipelined.length); + + parser.initialize(REQUEST, {}); + const reused = Buffer.from('GET /reused HTTP/1.1\r\nX-D: d\r\n\r\n'); + assert.strictEqual(parser.execute(reused, 0, reused.length), reused.length); + assert.deepStrictEqual(limits, []); +} + +// Preserve the existing exception behavior for the first property lookup. +{ + const parser = createParser(); + const expected = new Error('maxHeaderPairs getter'); + Object.defineProperty(parser, 'maxHeaderPairs', { + get: common.mustCall(() => { throw expected; }), + }); + const request = Buffer.from('GET / HTTP/1.1\r\nX-A: a\r\n\r\n'); + assert.throws(() => parser.execute(request, 0, request.length), expected); +} + +// Non-positive and non-number values continue to mean unlimited. +for (const maxHeaderPairs of [undefined, null, NaN, 0, -1, new Number(2)]) { + const parser = createParser(); + parser.maxHeaderPairs = maxHeaderPairs; + const request = Buffer.from( + 'GET / HTTP/1.1\r\nX-A: a\r\nX-B: b\r\nX-C: c\r\n\r\n' + ); + assert.strictEqual(parser.execute(request, 0, request.length), + request.length); +}