streams: use Array for Readable buffer

Author: ronag

lib/internal/streams/readable.js

@@ -73,6 +73,7 @@ const{
 const{ validateObject }=require('internal/validators');
 
 constkState=Symbol('kState');
+constFastBuffer=Buffer[Symbol.species];
 
 const{ StringDecoder }=require('string_decoder');
 constfrom=require('internal/streams/from');
@@ -278,7 +279,8 @@ function ReadableState(options, stream, isDuplex){
 // A linked list is used to store data chunks instead of an array because the
 // linked list can remove elements from the beginning faster than
 // array.shift().
-this.buffer=newBufferList();
+this.buffer=[];
+this.bufferIndex=0;
 this.length=0;
 this.pipes=[];
 
@@ -546,10 +548,15 @@ function addChunk(stream, state, chunk, addToFront){
 }else{
 // Update the buffer info.
 state.length+=(state[kState]&kObjectMode)!==0 ? 1 : chunk.length;
-if(addToFront)
-state.buffer.unshift(chunk);
-else
+if(addToFront){
+if(state.bufferIndex>0){
+state.buffer[--state.bufferIndex]=chunk;
+}else{
+state.buffer.unshift(chunk);// Slow path
+}
+}else{
 state.buffer.push(chunk);
+}
 
 if((state[kState]&kNeedReadable)!==0)
 emitReadable(stream);
@@ -564,21 +571,24 @@ Readable.prototype.isPaused = function(){
 
 // Backwards compatibility.
 Readable.prototype.setEncoding=function(enc){
+conststate=this._readableState;
+
 constdecoder=newStringDecoder(enc);
-this._readableState.decoder=decoder;
+state.decoder=decoder;
 // If setEncoding(null), decoder.encoding equals utf8.
-this._readableState.encoding=this._readableState.decoder.encoding;
+state.encoding=state.decoder.encoding;
 
-constbuffer=this._readableState.buffer;
 // Iterate over current buffer to convert already stored Buffers:
 letcontent='';
-for(constdataofbuffer){
+for(constdataofstate.buffer.slice(state.bufferIndex)){
 content+=decoder.write(data);
 }
-buffer.clear();
+state.buffer.length=0;
+state.bufferIndex=0;
+
 if(content!=='')
 buffer.push(content);
-this._readableState.length=content.length;
+state.length=content.length;
 returnthis;
 };
 
@@ -611,7 +621,7 @@ function howMuchToRead(n, state){
 if(NumberIsNaN(n)){
 // Only flow one buffer at a time.
 if((state[kState]&kFlowing)!==0&&state.length)
-returnstate.buffer.first().length;
+returnstate.buffer[state.bufferIndex].length;
 returnstate.length;
 }
 if(n<=state.length)
@@ -1550,20 +1560,96 @@ function fromList(n, state){
 returnnull;
 
 letret;
-if(state.objectMode)
-ret=state.buffer.shift();
-elseif(!n||n>=state.length){
+if((state[kState]&kObjectMode)!==0){
+ret=state.buffer[state.bufferIndex++];
+}elseif(!n||n>=state.length){
 // Read it all, truncate the list.
-if(state.decoder)
-ret=state.buffer.join('');
-elseif(state.buffer.length===1)
-ret=state.buffer.first();
-else
-ret=state.buffer.concat(state.length);
-state.buffer.clear();
+if((state[kState]&kDecoder)!==0){
+ret=''
+for(letn=state.bufferIndex;n<state.buffer.length;n++){
+ret+=state.buffer[n];
+}
+}elseif(state.buffer.length-state.bufferIndex===0){
+ret=Buffer.alloc(0)
+}elseif(state.buffer.length-state.bufferIndex===1){
+ret=state.buffer[state.bufferIndex];
+}else{
+ret=Buffer.allocUnsafe(n>>>0);
+leti=0;
+for(letn=state.bufferIndex;n<state.buffer.length;n++){
+constdata=state.buffer[n];
+ret.set(data,i);
+i+=data.length;
+}
+}
+state.buffer.length=0;
+state.bufferIndex=0;
 }else{
 // read part of list.
-ret=state.buffer.consume(n,state.decoder);
+
+constdata=state.buffer[state.bufferIndex];
+
+if(n<data.length){
+// `slice` is the same for buffers and strings.
+constslice=data.slice(0,n);
+state.buffer[state.bufferIndex]=data.slice(n);
+returnslice;
+}
+
+if(n===data.length){
+// First chunk is a perfect match.
+returnstate.buffer[state.bufferIndex++];
+}
+
+if((state[kState]&kDecoder)!==0){
+ret='';
+while(state.bufferIndex<state.buffer.length){
+conststr=state.buffer[state.bufferIndex];
+if(n>str.length){
+ret+=str;
+n-=str.length;
+state.bufferIndex++;
+}else{
+if(n===buf.length){
+ret+=str;
+state.bufferIndex++;
+}else{
+ret+=str.slice(0,n);
+state.buffer[state.bufferIndex]=str.slice(n);
+}
+break;
+}
+}
+}else{
+ret=Buffer.allocUnsafe(n);
+
+constretLen=n;
+while(state.bufferIndex<state.buffer.length){
+constbuf=state.buffer[state.bufferIndex];
+if(n>buf.length){
+ret.set(buf,retLen-n);
+n-=buf.length;
+state.bufferIndex++;
+}else{
+if(n===buf.length){
+ret.set(buf,retLen-n);
+state.bufferIndex++;
+}else{
+ret.set(newFastBuffer(buf.buffer,buf.byteOffset,n),retLen-n);
+state.buffer[state.bufferIndex]=newFastBuffer(buf.buffer,buf.byteOffset+n);
+}
+break;
+}
+}
+}
+
+if(state.bufferIndex===state.buffer.length){
+state.buffer.length=0;
+state.bufferIndex=0
+}elseif(state.bufferIndex>256){
+state.buffer=state.buffer.slice(state.bufferIndex);
+state.bufferIndex=0;
+}
 }
 
 returnret;