From a28a4902422c4fd488990180437f9772d9af48c4 Mon Sep 17 00:00:00 2001 From: GetThatCookie Date: Tue, 4 Aug 2026 22:57:23 +0200 Subject: [PATCH 1/3] http: coalesce chunked writes during auto-corking Signed-off-by: GetThatCookie --- benchmark/http/cork.js | 31 ++++ lib/_http_outgoing.js | 170 ++++++++++++++---- test/parallel/test-http-1.0.js | 6 +- test/parallel/test-http-outgoing-auto-cork.js | 130 ++++++++++++++ 4 files changed, 298 insertions(+), 39 deletions(-) create mode 100644 benchmark/http/cork.js create mode 100644 test/parallel/test-http-outgoing-auto-cork.js diff --git a/benchmark/http/cork.js b/benchmark/http/cork.js new file mode 100644 index 000000000000..a8675336a458 --- /dev/null +++ b/benchmark/http/cork.js @@ -0,0 +1,31 @@ +'use strict'; + +const common = require('../common.js'); + +const bench = common.createBenchmark(main, { + type: ['string', 'buffer'], + chunks: [4, 16], + len: [64], + c: [50], + duration: [5] +}); + +function main({ type, chunks, len, c, duration }) { + const http = require('http'); + const chunk = type === 'string' ? 'a'.repeat(len) : Buffer.alloc(len, 'a'); + + const server = http.createServer((req, res) => { + for (let n = 0; n < chunks; n++) { + res.write(chunk); + } + 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 6bf7a1f9f68d..8251df6a85e2 100644 --- a/lib/_http_outgoing.js +++ b/lib/_http_outgoing.js @@ -67,6 +67,7 @@ const { ERR_STREAM_DESTROYED, ERR_STREAM_NULL_VALUES, ERR_STREAM_WRITE_AFTER_END, + ERR_UNKNOWN_ENCODING, }, hideStackFrames, } = require('internal/errors'); @@ -82,6 +83,7 @@ let debug = require('internal/util/debuglog').debuglog('http', (fn) => { }); const kCorked = Symbol('corked'); +const kAutoCorked = Symbol('autoCorked'); const kSocket = Symbol('kSocket'); const kChunkedBuffer = Symbol('kChunkedBuffer'); const kChunkedLength = Symbol('kChunkedLength'); @@ -147,6 +149,7 @@ function OutgoingMessage(options) { this.finished = false; this._headerSent = false; this[kCorked] = 0; + this[kAutoCorked] = false; this[kChunkedBuffer] = []; this[kChunkedLength] = 0; this._closed = false; @@ -225,10 +228,19 @@ ObjectDefineProperty(OutgoingMessage.prototype, 'writableObjectMode', { }, }); +function chunkedBufferLength(msg) { + const len = msg[kChunkedLength]; + if (len === 0) { + return 0; + } + + return len + len.toString(16).length + 4 + (!msg._headerSent && msg._header !== null ? msg._header.length : 0); +} + ObjectDefineProperty(OutgoingMessage.prototype, 'writableLength', { __proto__: null, get() { - return this.outputSize + this[kChunkedLength] + (this[kSocket] ? this[kSocket].writableLength : 0); + return this.outputSize + chunkedBufferLength(this) + (this[kSocket] ? this[kSocket].writableLength : 0); }, }); @@ -297,44 +309,85 @@ OutgoingMessage.prototype.cork = function cork() { } }; -OutgoingMessage.prototype.uncork = function uncork() { - this[kCorked]--; - if (this[kSocket]) { - this[kSocket].uncork(); +function callChunkedCallbacks(callbacks, error) { + for (let n = 0; n < callbacks.length; n++) { + callbacks[n](error); } +} - if (this[kCorked] || this[kChunkedBuffer].length === 0) { +function destroyChunkedBuffer(msg, error) { + const buf = msg[kChunkedBuffer]; + if (buf.length === 0) { return; } - const len = this[kChunkedLength]; - const buf = this[kChunkedBuffer]; + const callbacks = []; + for (let n = 2; n < buf.length; n += 3) { + if (buf[n] !== nop) { + callbacks.push(buf[n]); + } + } + + buf.length = 0; + msg[kChunkedLength] = 0; + if (callbacks.length !== 0) { + process.nextTick(callChunkedCallbacks, callbacks, error || new ERR_STREAM_DESTROYED('write')); + } +} + +function flushChunkedBuffer(msg) { + if (msg.destroyed || msg[kSocket]?.destroyed) { + destroyChunkedBuffer(msg, msg[kErrored] || msg[kSocket]?._writableState?.errored); + return false; + } - assert(this.chunkedEncoding); + const buf = msg[kChunkedBuffer]; + const len = msg[kChunkedLength]; - let callbacks; - this._send(len.toString(16), 'latin1', null); - this._send(crlf_buf, null, null); + assert(msg.chunkedEncoding); + + const callbacks = []; + 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); - if (buf[n + 2]) { - callbacks ??= []; + msg._send(buf[n], buf[n + 1], null); + if (buf[n + 2] !== nop) { callbacks.push(buf[n + 2]); } } - this._send(crlf_buf, null, callbacks.length ? (err) => { - for (const callback of callbacks) { - callback(err); - } - } : null); + msg._send(crlf_buf, null, callbacks.length === 0 ? null : (error) => callChunkedCallbacks(callbacks, error)); - this[kChunkedBuffer].length = 0; - this[kChunkedLength] = 0; + buf.length = 0; + msg[kChunkedLength] = 0; + return true; +} + +function emitDrainIfNeeded(msg) { + if (msg[kNeedDrain] && msg.writableLength === 0) { + msg[kNeedDrain] = false; + msg.emit('drain'); + } +} - // 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'); +OutgoingMessage.prototype.uncork = function uncork() { + if (this[kCorked] === 0) { + return; + } + this[kCorked]--; + + const hasBufferedChunks = this[kCorked] === 0 && this[kChunkedBuffer].length !== 0; + let flushed = false; + try { + if (hasBufferedChunks) { + flushed = flushChunkedBuffer(this); + } + } finally { + this[kSocket]?.uncork(); + } + + if (flushed) { + // If we had a pending drain and flushed all data, emit the drain event. + emitDrainIfNeeded(this); } }; @@ -1006,21 +1059,39 @@ function write_(msg, chunk, encoding, callback, fromEnd) { if (!fromEnd && msg.socket && !msg.socket.writableCorked) { msg.socket.cork(); - process.nextTick(connectionCorkNT, msg.socket); + msg[kAutoCorked] = true; + process.nextTick(connectionCorkNT, msg, msg.socket); } let ret; - if (msg.chunkedEncoding && chunk.length !== 0) { - len ??= typeof chunk === 'string' ? Buffer.byteLength(chunk, encoding) : chunk.byteLength; - if (msg[kCorked] && msg._headerSent) { + if (msg.chunkedEncoding) { + const buf = msg[kChunkedBuffer]; + const buffering = (msg[kAutoCorked] || msg[kCorked]) && (chunk.length !== 0 || buf.length !== 0); + if (buffering) { + if (encoding && (encoding === 'buffer' ? typeof chunk === 'string' : !Buffer.isEncoding(encoding))) { + throw new ERR_UNKNOWN_ENCODING(encoding); + } + + if (chunk.length !== 0) { + len ??= typeof chunk === 'string' ? Buffer.byteLength(chunk, encoding) : chunk.byteLength; + if (typeof chunk !== 'string' && !(chunk instanceof Buffer)) { + chunk = Stream._uint8ArrayToBuffer(chunk); + } + msg[kChunkedLength] += len; + } msg[kChunkedBuffer].push(chunk, encoding, callback); - msg[kChunkedLength] += len; - ret = msg[kChunkedLength] < msg[kHighWaterMark]; - } else { + ret = msg.writableLength < msg.writableHighWaterMark; + if (msg[kAutoCorked] && msg[kCorked] === 0 && chunkedBufferLength(msg) >= msg.writableHighWaterMark) { + flushChunkedBuffer(msg); + } + } else if (chunk.length !== 0) { + len ??= typeof chunk === 'string' ? Buffer.byteLength(chunk, encoding) : chunk.byteLength; msg._send(len.toString(16), 'latin1', null); msg._send(crlf_buf, null, null); msg._send(chunk, encoding, null, len); ret = msg._send(crlf_buf, null, callback); + } else { + ret = msg._send(chunk, encoding, callback, len); } } else { ret = msg._send(chunk, encoding, callback, len); @@ -1031,8 +1102,26 @@ function write_(msg, chunk, encoding, callback, fromEnd) { } -function connectionCorkNT(conn) { - conn.uncork(); +function connectionCorkNT(msg, conn) { + if (!msg[kAutoCorked]) { + return; + } + + msg[kAutoCorked] = false; + let flushed = false; + try { + if (msg.destroyed || conn.destroyed) { + destroyChunkedBuffer(msg, msg[kErrored] || conn._writableState?.errored); + } else if (msg[kCorked] === 0 && msg[kChunkedBuffer].length !== 0) { + flushed = flushChunkedBuffer(msg); + } + } finally { + conn.uncork(); + } + + if (flushed) { + emitDrainIfNeeded(msg); + } } OutgoingMessage.prototype.addTrailers = function addTrailers(headers) { @@ -1138,6 +1227,11 @@ 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 hasBufferedChunks = this[kChunkedBuffer].length !== 0; + const flushed = hasBufferedChunks && flushChunkedBuffer(this); + const finish = onFinish.bind(undefined, this); if (this._hasBody && this.chunkedEncoding) { @@ -1148,6 +1242,7 @@ OutgoingMessage.prototype.end = function end(chunk, encoding, callback) { process.nextTick(finish); } + this[kAutoCorked] = false; if (this[kSocket]) { // Fully uncork connection on end(). this[kSocket]._writableState.corked = 1; @@ -1156,8 +1251,13 @@ OutgoingMessage.prototype.end = function end(chunk, encoding, callback) { this[kCorked] = 1; this.uncork(); + // A synchronous drain listener must not 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-1.0.js b/test/parallel/test-http-1.0.js index 639bd228df00..fb35fad26cd2 100644 --- a/test/parallel/test-http-1.0.js +++ b/test/parallel/test-http-1.0.js @@ -148,10 +148,8 @@ function test(handler, request_generator, response_validator) { 'Connection: close\r\n' + 'Transfer-Encoding: chunked\r\n' + '\r\n' + - '7\r\n' + - 'Hello, \r\n' + - '6\r\n' + - 'world!\r\n' + + 'd\r\n' + + 'Hello, world!\r\n' + '0\r\n' + '\r\n'; diff --git a/test/parallel/test-http-outgoing-auto-cork.js b/test/parallel/test-http-outgoing-auto-cork.js new file mode 100644 index 000000000000..30f403edeccc --- /dev/null +++ b/test/parallel/test-http-outgoing-auto-cork.js @@ -0,0 +1,130 @@ +'use strict'; + +const common = require('../common'); +const assert = require('assert'); +const http = require('http'); +const net = require('net'); + +function responseBody(response) { + return response.slice(response.indexOf('\r\n\r\n') + 4); +} + +function getRawResponse(onRequest) { + return new Promise((resolve, reject) => { + const server = http.createServer(common.mustCall((req, res) => { + onRequest(res); + })); + + server.on('error', reject); + server.listen(0, common.localhostIPv4, common.mustCall(() => { + const socket = net.connect({ + 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(() => { + server.close(common.mustCall(() => resolve(response))); + })); + socket.on('connect', common.mustCall(() => { + socket.write( + 'GET / HTTP/1.1\r\nHost: localhost\r\n' + + 'Connection: close\r\n\r\n', + ); + })); + })); + }); +} + +async function testAutomaticCork() { + const callbacks = []; + const response = await getRawResponse(common.mustCall((res) => { + assert.throws( + () => res.write('ignored', 'invalid'), + { code: 'ERR_UNKNOWN_ENCODING' }, + ); + const chunk = new Uint8Array([0x41]); + res.write(chunk, common.mustCall(() => callbacks.push('A'))); + res.write('', common.mustCall(() => callbacks.push('empty'))); + res.end('BC', common.mustCall(() => { + callbacks.push('end'); + assert.deepStrictEqual(callbacks, ['A', 'empty', 'end']); + })); + })); + + assert.strictEqual(responseBody(response), '3\r\nABC\r\n0\r\n\r\n'); +} + +async function testDetachedUint8Array() { + const response = await getRawResponse(common.mustCall((res) => { + const chunk = new Uint8Array([0x41]); + res.write(chunk, common.mustCall()); + structuredClone(chunk.buffer, { transfer: [chunk.buffer] }); + res.end(); + })); + + assert.match(response, /^HTTP\/1\.1 200 OK\r\n/); +} + +async function testExplicitCorkedEnd() { + const response = await getRawResponse(common.mustCall((res) => { + res.flushHeaders(); + res.cork(); + res.cork(); + res.write('D'); + res.write('E'); + res.uncork(); + res.end('F'); + assert.strictEqual(res.writableCorked, 0); + })); + + assert.strictEqual(responseBody(response), '3\r\nDEF\r\n0\r\n\r\n'); +} + +async function testTickBoundary() { + const response = await getRawResponse(common.mustCall((res) => { + res.write('G'); + process.nextTick(() => res.end('H')); + })); + + assert.strictEqual( + responseBody(response), + '1\r\nG\r\n1\r\nH\r\n0\r\n\r\n', + ); +} + +function testDestroyedWrite() { + return new Promise((resolve, reject) => { + const server = http.createServer(common.mustCall((req, res) => { + res.write('I', common.mustCall((error) => { + assert.strictEqual(error.code, 'ERR_STREAM_DESTROYED'); + server.close(common.mustCall(resolve)); + })); + res.destroy(); + })); + + server.on('error', reject); + server.listen(0, common.localhostIPv4, common.mustCall(() => { + const req = http.get({ + host: common.localhostIPv4, + port: server.address().port, + }); + req.on('error', common.mustCall((error) => { + assert.strictEqual(error.code, 'ECONNRESET'); + })); + })); + }); +} + +async function main() { + await testAutomaticCork(); + await testDetachedUint8Array(); + await testExplicitCorkedEnd(); + await testTickBoundary(); + await testDestroyedWrite(); +} + +main().then(common.mustCall()); From e952878b37004964591a2cca8105a24e1fc6c1c0 Mon Sep 17 00:00:00 2001 From: GetThatCookie Date: Thu, 6 Aug 2026 22:41:04 +0200 Subject: [PATCH 2/3] http: simplify chunked write coalescing Signed-off-by: GetThatCookie --- benchmark/http/cork.js | 31 --- lib/_http_outgoing.js | 197 ++++++------------ test/parallel/test-http-1.0.js | 9 +- test/parallel/test-http-outgoing-auto-cork.js | 130 ------------ test/parallel/test-http-outgoing-destroyed.js | 3 + 5 files changed, 79 insertions(+), 291 deletions(-) delete mode 100644 benchmark/http/cork.js delete mode 100644 test/parallel/test-http-outgoing-auto-cork.js diff --git a/benchmark/http/cork.js b/benchmark/http/cork.js deleted file mode 100644 index a8675336a458..000000000000 --- a/benchmark/http/cork.js +++ /dev/null @@ -1,31 +0,0 @@ -'use strict'; - -const common = require('../common.js'); - -const bench = common.createBenchmark(main, { - type: ['string', 'buffer'], - chunks: [4, 16], - len: [64], - c: [50], - duration: [5] -}); - -function main({ type, chunks, len, c, duration }) { - const http = require('http'); - const chunk = type === 'string' ? 'a'.repeat(len) : Buffer.alloc(len, 'a'); - - const server = http.createServer((req, res) => { - for (let n = 0; n < chunks; n++) { - res.write(chunk); - } - 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 8251df6a85e2..6fdf50250b4f 100644 --- a/lib/_http_outgoing.js +++ b/lib/_http_outgoing.js @@ -83,7 +83,6 @@ let debug = require('internal/util/debuglog').debuglog('http', (fn) => { }); const kCorked = Symbol('corked'); -const kAutoCorked = Symbol('autoCorked'); const kSocket = Symbol('kSocket'); const kChunkedBuffer = Symbol('kChunkedBuffer'); const kChunkedLength = Symbol('kChunkedLength'); @@ -149,7 +148,6 @@ function OutgoingMessage(options) { this.finished = false; this._headerSent = false; this[kCorked] = 0; - this[kAutoCorked] = false; this[kChunkedBuffer] = []; this[kChunkedLength] = 0; this._closed = false; @@ -228,19 +226,14 @@ ObjectDefineProperty(OutgoingMessage.prototype, 'writableObjectMode', { }, }); -function chunkedBufferLength(msg) { - const len = msg[kChunkedLength]; - if (len === 0) { - return 0; - } - - return len + len.toString(16).length + 4 + (!msg._headerSent && msg._header !== null ? msg._header.length : 0); -} - ObjectDefineProperty(OutgoingMessage.prototype, 'writableLength', { __proto__: null, get() { - return this.outputSize + chunkedBufferLength(this) + (this[kSocket] ? this[kSocket].writableLength : 0); + let bufferedLength = this[kChunkedLength]; + if (bufferedLength !== 0) { + bufferedLength += bufferedLength.toString(16).length + 4 + (!this._headerSent && this._header !== null ? this._header.length : 0); + } + return this.outputSize + bufferedLength + (this[kSocket] ? this[kSocket].writableLength : 0); }, }); @@ -309,85 +302,54 @@ OutgoingMessage.prototype.cork = function cork() { } }; -function callChunkedCallbacks(callbacks, error) { - for (let n = 0; n < callbacks.length; n++) { - callbacks[n](error); - } -} - -function destroyChunkedBuffer(msg, error) { - const buf = msg[kChunkedBuffer]; - if (buf.length === 0) { +OutgoingMessage.prototype.uncork = function uncork() { + this[kCorked]--; + const socket = this[kSocket]; + if (this[kCorked] || this[kChunkedBuffer].length === 0) { + socket?.uncork(); return; } - const callbacks = []; - for (let n = 2; n < buf.length; n += 3) { - if (buf[n] !== nop) { - callbacks.push(buf[n]); + const len = this[kChunkedLength]; + const buf = this[kChunkedBuffer]; + if (this.destroyed || socket?.destroyed) { + const error = this[kErrored] || socket?._writableState?.errored || new ERR_STREAM_DESTROYED('write'); + for (let n = 2; n < buf.length; n += 3) { + if (buf[n] !== nop) { + process.nextTick(buf[n], error); + } } + buf.length = 0; + this[kChunkedLength] = 0; + socket?.uncork(); + return; } + assert(this.chunkedEncoding); - buf.length = 0; - msg[kChunkedLength] = 0; - if (callbacks.length !== 0) { - process.nextTick(callChunkedCallbacks, callbacks, error || new ERR_STREAM_DESTROYED('write')); - } -} - -function flushChunkedBuffer(msg) { - if (msg.destroyed || msg[kSocket]?.destroyed) { - destroyChunkedBuffer(msg, msg[kErrored] || msg[kSocket]?._writableState?.errored); - return false; - } - - const buf = msg[kChunkedBuffer]; - const len = msg[kChunkedLength]; - - assert(msg.chunkedEncoding); - - const callbacks = []; - msg._send(len.toString(16), 'latin1', null); - msg._send(crlf_buf, null, null); + let callbacks; + this._send(len.toString(16), 'latin1', null); + this._send(crlf_buf, null, null); for (let n = 0; n < buf.length; n += 3) { - msg._send(buf[n], buf[n + 1], null); - if (buf[n + 2] !== nop) { + this._send(buf[n + 0], buf[n + 1], null); + if (buf[n + 2]) { + callbacks ??= []; callbacks.push(buf[n + 2]); } } - msg._send(crlf_buf, null, callbacks.length === 0 ? null : (error) => callChunkedCallbacks(callbacks, error)); - - buf.length = 0; - msg[kChunkedLength] = 0; - return true; -} - -function emitDrainIfNeeded(msg) { - if (msg[kNeedDrain] && msg.writableLength === 0) { - msg[kNeedDrain] = false; - msg.emit('drain'); - } -} - -OutgoingMessage.prototype.uncork = function uncork() { - if (this[kCorked] === 0) { - return; - } - this[kCorked]--; - - const hasBufferedChunks = this[kCorked] === 0 && this[kChunkedBuffer].length !== 0; - let flushed = false; - try { - if (hasBufferedChunks) { - flushed = flushChunkedBuffer(this); + this._send(crlf_buf, null, callbacks.length ? (err) => { + for (const callback of callbacks) { + callback(err); } - } finally { - this[kSocket]?.uncork(); - } + } : null); + + this[kChunkedBuffer].length = 0; + this[kChunkedLength] = 0; + socket?.uncork(); - if (flushed) { - // If we had a pending drain and flushed all data, emit the drain event. - emitDrainIfNeeded(this); + // 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'); } }; @@ -418,6 +380,10 @@ OutgoingMessage.prototype.destroy = function destroy(error) { this.destroyed = true; this[kErrored] = error; + if (this[kChunkedBuffer].length !== 0) { + this[kCorked] = 1; + this.uncork(); + } if (this[kSocket]) { this[kSocket].destroy(error); @@ -1057,42 +1023,34 @@ function write_(msg, chunk, encoding, callback, fromEnd) { } } + if (msg.chunkedEncoding && typeof chunk === 'string' && encoding && !Buffer.isEncoding(encoding)) { + throw new ERR_UNKNOWN_ENCODING(encoding); + } + if (!fromEnd && msg.socket && !msg.socket.writableCorked) { - msg.socket.cork(); - msg[kAutoCorked] = true; + msg.cork(); process.nextTick(connectionCorkNT, msg, msg.socket); } let ret; - if (msg.chunkedEncoding) { - const buf = msg[kChunkedBuffer]; - const buffering = (msg[kAutoCorked] || msg[kCorked]) && (chunk.length !== 0 || buf.length !== 0); - if (buffering) { - if (encoding && (encoding === 'buffer' ? typeof chunk === 'string' : !Buffer.isEncoding(encoding))) { - throw new ERR_UNKNOWN_ENCODING(encoding); - } - - if (chunk.length !== 0) { - len ??= typeof chunk === 'string' ? Buffer.byteLength(chunk, encoding) : chunk.byteLength; - if (typeof chunk !== 'string' && !(chunk instanceof Buffer)) { - chunk = Stream._uint8ArrayToBuffer(chunk); - } - msg[kChunkedLength] += len; + if (msg.chunkedEncoding && chunk.length !== 0) { + len ??= typeof chunk === 'string' ? Buffer.byteLength(chunk, encoding) : chunk.byteLength; + if (msg[kCorked]) { + if (typeof chunk !== 'string' && !(chunk instanceof Buffer)) { + chunk = Stream._uint8ArrayToBuffer(chunk); } msg[kChunkedBuffer].push(chunk, encoding, callback); + msg[kChunkedLength] += len; ret = msg.writableLength < msg.writableHighWaterMark; - if (msg[kAutoCorked] && msg[kCorked] === 0 && chunkedBufferLength(msg) >= msg.writableHighWaterMark) { - flushChunkedBuffer(msg); - } - } else if (chunk.length !== 0) { - len ??= typeof chunk === 'string' ? Buffer.byteLength(chunk, encoding) : chunk.byteLength; + } else { msg._send(len.toString(16), 'latin1', null); msg._send(crlf_buf, null, null); msg._send(chunk, encoding, null, len); ret = msg._send(crlf_buf, null, callback); - } else { - ret = msg._send(chunk, encoding, callback, len); } + } else if (msg.chunkedEncoding && msg[kCorked] && msg[kChunkedBuffer].length !== 0) { + msg[kChunkedBuffer].push(chunk, encoding, callback); + ret = msg.writableLength < msg.writableHighWaterMark; } else { ret = msg._send(chunk, encoding, callback, len); } @@ -1103,25 +1061,12 @@ function write_(msg, chunk, encoding, callback, fromEnd) { function connectionCorkNT(msg, conn) { - if (!msg[kAutoCorked]) { - return; + if (msg[kCorked]) { + msg.uncork(); } - - msg[kAutoCorked] = false; - let flushed = false; - try { - if (msg.destroyed || conn.destroyed) { - destroyChunkedBuffer(msg, msg[kErrored] || conn._writableState?.errored); - } else if (msg[kCorked] === 0 && msg[kChunkedBuffer].length !== 0) { - flushed = flushChunkedBuffer(msg); - } - } finally { + if (msg[kSocket] !== conn) { conn.uncork(); } - - if (flushed) { - emitDrainIfNeeded(msg); - } } OutgoingMessage.prototype.addTrailers = function addTrailers(headers) { @@ -1227,10 +1172,12 @@ 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 hasBufferedChunks = this[kChunkedBuffer].length !== 0; - const flushed = hasBufferedChunks && flushChunkedBuffer(this); + // Flush buffered chunks before terminating the response. + if (this[kChunkedBuffer].length !== 0) { + this[kSocket]?.cork(); + this[kCorked] = 1; + this.uncork(); + } const finish = onFinish.bind(undefined, this); @@ -1242,7 +1189,6 @@ OutgoingMessage.prototype.end = function end(chunk, encoding, callback) { process.nextTick(finish); } - this[kAutoCorked] = false; if (this[kSocket]) { // Fully uncork connection on end(). this[kSocket]._writableState.corked = 1; @@ -1251,13 +1197,8 @@ OutgoingMessage.prototype.end = function end(chunk, encoding, callback) { this[kCorked] = 1; this.uncork(); - // A synchronous drain listener must not 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-1.0.js b/test/parallel/test-http-1.0.js index fb35fad26cd2..2df00633eafb 100644 --- a/test/parallel/test-http-1.0.js +++ b/test/parallel/test-http-1.0.js @@ -127,9 +127,12 @@ function test(handler, request_generator, response_validator) { assert.strictEqual(req.httpVersionMinor, 1); res.sendDate = false; res.writeHead(200, { 'Content-Type': 'text/plain' }); - res.write('Hello, '); res._send(''); + assert.throws(() => res.write('ignored', 'invalid'), { + code: 'ERR_UNKNOWN_ENCODING', + }); + res.write('Hello, ', common.mustCall()); res._send(''); res.write('world!'); res._send(''); - res.end(); + process.nextTick(() => res.end('X')); } function request_generator() { @@ -150,6 +153,8 @@ function test(handler, request_generator, response_validator) { '\r\n' + 'd\r\n' + 'Hello, world!\r\n' + + '1\r\n' + + 'X\r\n' + '0\r\n' + '\r\n'; diff --git a/test/parallel/test-http-outgoing-auto-cork.js b/test/parallel/test-http-outgoing-auto-cork.js deleted file mode 100644 index 30f403edeccc..000000000000 --- a/test/parallel/test-http-outgoing-auto-cork.js +++ /dev/null @@ -1,130 +0,0 @@ -'use strict'; - -const common = require('../common'); -const assert = require('assert'); -const http = require('http'); -const net = require('net'); - -function responseBody(response) { - return response.slice(response.indexOf('\r\n\r\n') + 4); -} - -function getRawResponse(onRequest) { - return new Promise((resolve, reject) => { - const server = http.createServer(common.mustCall((req, res) => { - onRequest(res); - })); - - server.on('error', reject); - server.listen(0, common.localhostIPv4, common.mustCall(() => { - const socket = net.connect({ - 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(() => { - server.close(common.mustCall(() => resolve(response))); - })); - socket.on('connect', common.mustCall(() => { - socket.write( - 'GET / HTTP/1.1\r\nHost: localhost\r\n' + - 'Connection: close\r\n\r\n', - ); - })); - })); - }); -} - -async function testAutomaticCork() { - const callbacks = []; - const response = await getRawResponse(common.mustCall((res) => { - assert.throws( - () => res.write('ignored', 'invalid'), - { code: 'ERR_UNKNOWN_ENCODING' }, - ); - const chunk = new Uint8Array([0x41]); - res.write(chunk, common.mustCall(() => callbacks.push('A'))); - res.write('', common.mustCall(() => callbacks.push('empty'))); - res.end('BC', common.mustCall(() => { - callbacks.push('end'); - assert.deepStrictEqual(callbacks, ['A', 'empty', 'end']); - })); - })); - - assert.strictEqual(responseBody(response), '3\r\nABC\r\n0\r\n\r\n'); -} - -async function testDetachedUint8Array() { - const response = await getRawResponse(common.mustCall((res) => { - const chunk = new Uint8Array([0x41]); - res.write(chunk, common.mustCall()); - structuredClone(chunk.buffer, { transfer: [chunk.buffer] }); - res.end(); - })); - - assert.match(response, /^HTTP\/1\.1 200 OK\r\n/); -} - -async function testExplicitCorkedEnd() { - const response = await getRawResponse(common.mustCall((res) => { - res.flushHeaders(); - res.cork(); - res.cork(); - res.write('D'); - res.write('E'); - res.uncork(); - res.end('F'); - assert.strictEqual(res.writableCorked, 0); - })); - - assert.strictEqual(responseBody(response), '3\r\nDEF\r\n0\r\n\r\n'); -} - -async function testTickBoundary() { - const response = await getRawResponse(common.mustCall((res) => { - res.write('G'); - process.nextTick(() => res.end('H')); - })); - - assert.strictEqual( - responseBody(response), - '1\r\nG\r\n1\r\nH\r\n0\r\n\r\n', - ); -} - -function testDestroyedWrite() { - return new Promise((resolve, reject) => { - const server = http.createServer(common.mustCall((req, res) => { - res.write('I', common.mustCall((error) => { - assert.strictEqual(error.code, 'ERR_STREAM_DESTROYED'); - server.close(common.mustCall(resolve)); - })); - res.destroy(); - })); - - server.on('error', reject); - server.listen(0, common.localhostIPv4, common.mustCall(() => { - const req = http.get({ - host: common.localhostIPv4, - port: server.address().port, - }); - req.on('error', common.mustCall((error) => { - assert.strictEqual(error.code, 'ECONNRESET'); - })); - })); - }); -} - -async function main() { - await testAutomaticCork(); - await testDetachedUint8Array(); - await testExplicitCorkedEnd(); - await testTickBoundary(); - await testDestroyedWrite(); -} - -main().then(common.mustCall()); diff --git a/test/parallel/test-http-outgoing-destroyed.js b/test/parallel/test-http-outgoing-destroyed.js index b60b6594c765..e6f533d22448 100644 --- a/test/parallel/test-http-outgoing-destroyed.js +++ b/test/parallel/test-http-outgoing-destroyed.js @@ -66,6 +66,9 @@ const { OutgoingMessage } = require('http'); }); })); const err = new Error('Destroy test'); + res.write('x', common.mustCall((writeErr) => { + assert.strictEqual(writeErr, err); + })); res.destroy(err); assert.strictEqual(res.errored, err); })).listen(0, common.mustCall(() => { From bc50bf5c6d9e723703b7273321540c4d39585382 Mon Sep 17 00:00:00 2001 From: GetThatCookie Date: Sat, 8 Aug 2026 01:03:53 +0200 Subject: [PATCH 3/3] http: preserve socket cork count during flush Signed-off-by: GetThatCookie --- lib/_http_outgoing.js | 69 ++++++++++++++++++++++--------------------- 1 file changed, 35 insertions(+), 34 deletions(-) diff --git a/lib/_http_outgoing.js b/lib/_http_outgoing.js index 6fdf50250b4f..8eee5bc4c139 100644 --- a/lib/_http_outgoing.js +++ b/lib/_http_outgoing.js @@ -305,46 +305,47 @@ OutgoingMessage.prototype.cork = function cork() { OutgoingMessage.prototype.uncork = function uncork() { this[kCorked]--; const socket = this[kSocket]; - if (this[kCorked] || this[kChunkedBuffer].length === 0) { - socket?.uncork(); - return; - } + try { + if (this[kCorked] || this[kChunkedBuffer].length === 0) { + return; + } - const len = this[kChunkedLength]; - const buf = this[kChunkedBuffer]; - if (this.destroyed || socket?.destroyed) { - const error = this[kErrored] || socket?._writableState?.errored || new ERR_STREAM_DESTROYED('write'); - for (let n = 2; n < buf.length; n += 3) { - if (buf[n] !== nop) { - process.nextTick(buf[n], error); + const len = this[kChunkedLength]; + const buf = this[kChunkedBuffer]; + if (this.destroyed || socket?.destroyed) { + const error = this[kErrored] || socket?._writableState?.errored || new ERR_STREAM_DESTROYED('write'); + for (let n = 2; n < buf.length; n += 3) { + if (buf[n] !== nop) { + process.nextTick(buf[n], error); + } + } + buf.length = 0; + this[kChunkedLength] = 0; + return; + } + assert(this.chunkedEncoding); + + let callbacks; + this._send(len.toString(16), 'latin1', null); + this._send(crlf_buf, null, null); + for (let n = 0; n < buf.length; n += 3) { + this._send(buf[n + 0], buf[n + 1], null); + if (buf[n + 2]) { + callbacks ??= []; + callbacks.push(buf[n + 2]); } } - buf.length = 0; + this._send(crlf_buf, null, callbacks.length ? (err) => { + for (const callback of callbacks) { + callback(err); + } + } : null); + + this[kChunkedBuffer].length = 0; this[kChunkedLength] = 0; + } finally { socket?.uncork(); - return; } - assert(this.chunkedEncoding); - - let callbacks; - this._send(len.toString(16), 'latin1', null); - this._send(crlf_buf, null, null); - for (let n = 0; n < buf.length; n += 3) { - this._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) => { - for (const callback of callbacks) { - callback(err); - } - } : null); - - this[kChunkedBuffer].length = 0; - this[kChunkedLength] = 0; - socket?.uncork(); // If we had a pending drain and flushed all data, emit the drain event. if (this[kNeedDrain] && this.writableLength === 0) {