http: coalesce chunked writes during auto-corking

Signed-off-by: GetThatCookie <NimmenKeks@gmx.de>

Author: GetThatCookie

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());
+  });
+}

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.');

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';