http: reduce parts in chunked response when corking

Refs: nodejs/performance#57 PR-URL: #50167 Reviewed-By: Stephen Belanger <[email protected]> Reviewed-By: Matteo Collina <[email protected]> Reviewed-By: Yagiz Nizipli <[email protected]>

Author: ronag

lib/_http_outgoing.js

@@ -82,6 +82,9 @@ let debug = require('internal/util/debuglog').debuglog('http', (fn) =>{
 });
 
 constkCorked=Symbol('corked');
+constkSocket=Symbol('kSocket');
+constkChunkedBuffer=Symbol('kChunkedBuffer');
+constkChunkedLength=Symbol('kChunkedLength');
 constkUniqueHeaders=Symbol('kUniqueHeaders');
 constkBytesWritten=Symbol('kBytesWritten');
 constkErrored=Symbol('errored');
@@ -140,9 +143,11 @@ function OutgoingMessage(options){
 this.finished=false;
 this._headerSent=false;
 this[kCorked]=0;
+this[kChunkedBuffer]=[];
+this[kChunkedLength]=0;
 this._closed=false;
 
-this.socket=null;
+this[kSocket]=null;
 this._header=null;
 this[kOutHeaders]=null;
 
@@ -177,7 +182,7 @@ ObjectDefineProperty(OutgoingMessage.prototype, 'writableFinished',{
 return(
 this.finished&&
 this.outputSize===0&&
-(!this.socket||this.socket.writableLength===0)
+(!this[kSocket]||this[kSocket].writableLength===0)
 );
 },
 });
@@ -192,22 +197,21 @@ ObjectDefineProperty(OutgoingMessage.prototype, 'writableObjectMode',{
 ObjectDefineProperty(OutgoingMessage.prototype,'writableLength',{
 __proto__: null,
 get(){
-returnthis.outputSize+(this.socket ? this.socket.writableLength : 0);
+returnthis.outputSize+this[kChunkedLength]+(this[kSocket] ? this[kSocket].writableLength : 0);
 },
 });
 
 ObjectDefineProperty(OutgoingMessage.prototype,'writableHighWaterMark',{
 __proto__: null,
 get(){
-returnthis.socket ? this.socket.writableHighWaterMark : this[kHighWaterMark];
+returnthis[kSocket] ? this[kSocket].writableHighWaterMark : this[kHighWaterMark];
 },
 });
 
 ObjectDefineProperty(OutgoingMessage.prototype,'writableCorked',{
 __proto__: null,
 get(){
-constcorked=this.socket ? this.socket.writableCorked : 0;
-returncorked+this[kCorked];
+returnthis[kCorked];
 },
 });
 
@@ -235,13 +239,27 @@ ObjectDefineProperty(OutgoingMessage.prototype, '_headers',{
 ObjectDefineProperty(OutgoingMessage.prototype,'connection',{
 __proto__: null,
 get: function(){
-returnthis.socket;
+returnthis[kSocket];
 },
 set: function(val){
 this.socket=val;
 },
 });
 
+ObjectDefineProperty(OutgoingMessage.prototype,'socket',{
+__proto__: null,
+get: function(){
+returnthis[kSocket];
+},
+set: function(val){
+for(letn=0;n<this[kCorked];n++){
+val?.cork();
+this[kSocket]?.uncork();
+}
+this[kSocket]=val;
+},
+});
+
 ObjectDefineProperty(OutgoingMessage.prototype,'_headerNames',{
 __proto__: null,
 get: internalUtil.deprecate(function(){
@@ -299,19 +317,45 @@ OutgoingMessage.prototype._renderHeaders = function _renderHeaders(){
 };
 
 OutgoingMessage.prototype.cork=function(){
-if(this.socket){
-this.socket.cork();
-}else{
-this[kCorked]++;
+this[kCorked]++;
+if(this[kSocket]){
+this[kSocket].cork();
 }
 };
 
 OutgoingMessage.prototype.uncork=function(){
-if(this.socket){
-this.socket.uncork();
-}elseif(this[kCorked]){
-this[kCorked]--;
+this[kCorked]--;
+if(this[kSocket]){
+this[kSocket].uncork();
+}
+
+if(this[kCorked]||this[kChunkedBuffer].length===0){
+return;
 }
+
+constlen=this[kChunkedLength];
+constbuf=this[kChunkedBuffer];
+
+assert(this.chunkedEncoding);
+
+letcallbacks;
+this._send(NumberPrototypeToString(len,16),'latin1',null);
+this._send(crlf_buf,null,null);
+for(letn=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(constcallbackofcallbacks){
+callback(err);
+}
+} : null);
+
+this[kChunkedBuffer].length=0;
+this[kChunkedLength]=0;
 };
 
 OutgoingMessage.prototype.setTimeout=functionsetTimeout(msecs,callback){
@@ -320,12 +364,12 @@ OutgoingMessage.prototype.setTimeout = function setTimeout(msecs, callback){
 this.on('timeout',callback);
 }
 
-if(!this.socket){
+if(!this[kSocket]){
 this.once('socket',functionsocketSetTimeoutOnConnect(socket){
 socket.setTimeout(msecs);
 });
 }else{
-this.socket.setTimeout(msecs);
+this[kSocket].setTimeout(msecs);
 }
 returnthis;
 };
@@ -342,8 +386,8 @@ OutgoingMessage.prototype.destroy = function destroy(error){
 
 this[kErrored]=error;
 
-if(this.socket){
-this.socket.destroy(error);
+if(this[kSocket]){
+this[kSocket].destroy(error);
 }else{
 this.once('socket',functionsocketDestroyOnConnect(socket){
 socket.destroy(error);
@@ -382,7 +426,7 @@ OutgoingMessage.prototype._send = function _send(data, encoding, callback, byteL
 
 OutgoingMessage.prototype._writeRaw=_writeRaw;
 function_writeRaw(data,encoding,callback,size){
-constconn=this.socket;
+constconn=this[kSocket];
 if(conn&&conn.destroyed){
 // The socket was destroyed. If we're still trying to write to it,
 // then we haven't gotten the 'close' event yet.
@@ -938,10 +982,16 @@ function write_(msg, chunk, encoding, callback, fromEnd){
 letret;
 if(msg.chunkedEncoding&&chunk.length!==0){
 len??=typeofchunk==='string' ? Buffer.byteLength(chunk,encoding) : chunk.byteLength;
-msg._send(NumberPrototypeToString(len,16),'latin1',null);
-msg._send(crlf_buf,null,null);
-msg._send(chunk,encoding,null,len);
-ret=msg._send(crlf_buf,null,callback);
+if(msg[kCorked]&&msg._headerSent){
+msg[kChunkedBuffer].push(chunk,encoding,callback);
+msg[kChunkedLength]+=len;
+ret=msg[kChunkedLength]<msg[kHighWaterMark];
+}else{
+msg._send(NumberPrototypeToString(len,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);
 }
@@ -1023,8 +1073,8 @@ OutgoingMessage.prototype.end = function end(chunk, encoding, callback){
 returnthis;
 }
 
-if(this.socket){
-this.socket.cork();
+if(this[kSocket]){
+this[kSocket].cork();
 }
 
 write_(this,chunk,encoding,null,true);
@@ -1038,8 +1088,8 @@ OutgoingMessage.prototype.end = function end(chunk, encoding, callback){
 }
 returnthis;
 }elseif(!this._header){
-if(this.socket){
-this.socket.cork();
+if(this[kSocket]){
+this[kSocket].cork();
 }
 
 this._contentLength=0;
@@ -1063,21 +1113,22 @@ OutgoingMessage.prototype.end = function end(chunk, encoding, callback){
 process.nextTick(finish);
 }
 
-if(this.socket){
+if(this[kSocket]){
 // Fully uncork connection on end().
-this.socket._writableState.corked=1;
-this.socket.uncork();
+this[kSocket]._writableState.corked=1;
+this[kSocket].uncork();
 }
-this[kCorked]=0;
+this[kCorked]=1;
+this.uncork();
 
 this.finished=true;
 
 // There is the first message on the outgoing queue, and we've sent
 // everything to the socket.
 debug('outgoing message end.');
 if(this.outputData.length===0&&
-this.socket&&
-this.socket._httpMessage===this){
+this[kSocket]&&
+this[kSocket]._httpMessage===this){
 this._finish();
 }
 
@@ -1088,7 +1139,7 @@ OutgoingMessage.prototype.end = function end(chunk, encoding, callback){
 // This function is called once all user data are flushed to the socket.
 // Note that it has a chance that the socket is not drained.
 OutgoingMessage.prototype._finish=function_finish(){
-assert(this.socket);
+assert(this[kSocket]);
 this.emit('prefinish');
 };
 
@@ -1113,7 +1164,7 @@ OutgoingMessage.prototype._finish = function _finish(){
 // This function, _flush(), is called by both the Server and Client
 // to attempt to flush any pending messages out to the socket.
 OutgoingMessage.prototype._flush=function_flush(){
-constsocket=this.socket;
+constsocket=this[kSocket];
 
 if(socket&&socket.writable){
 // There might be remaining data in this.output; write it out
@@ -1130,11 +1181,6 @@ OutgoingMessage.prototype._flush = function _flush(){
 };
 
 OutgoingMessage.prototype._flushOutput=function_flushOutput(socket){
-while(this[kCorked]){
-this[kCorked]--;
-socket.cork();
-}
-
 constoutputLength=this.outputData.length;
 if(outputLength<=0)
 returnundefined;