stream: improve from perf

PR-URL: #50359 Reviewed-By: Vinícius Lourenço Claro Cardoso <[email protected]> Reviewed-By: Robert Nagy <[email protected]> Reviewed-By: Benjamin Gruenbaum <[email protected]>

Author: rluvaton

lib/internal/streams/from.js

@@ -36,6 +36,7 @@ function from(Readable, iterable, opts){
 thrownewERR_INVALID_ARG_TYPE('iterable',['Iterable'],iterable);
 }
 
+
 constreadable=newReadable({
 objectMode: true,
 highWaterMark: 1,
@@ -46,11 +47,19 @@ function from(Readable, iterable, opts){
 // Flag to protect against _read
 // being called before last iteration completion.
 letreading=false;
+letisAsyncValues=false;
 
 readable._read=function(){
 if(!reading){
 reading=true;
-next();
+
+if(isAsync){
+nextAsync();
+}elseif(isAsyncValues){
+nextSyncWithAsyncValues();
+}else{
+nextSyncWithSyncValues();
+}
 }
 };
 
@@ -78,29 +87,115 @@ function from(Readable, iterable, opts){
 }
 }
 
-asyncfunctionnext(){
+// There are a lot of duplication here, it's done on purpose for performance
+// reasons - avoid await when not needed.
+
+functionnextSyncWithSyncValues(){
+for(;;){
+try{
+const{ value, done }=iterator.next();
+
+if(done){
+readable.push(null);
+return;
+}
+
+if(value&&
+typeofvalue.then==='function'){
+returnchangeToAsyncValues(value);
+}
+
+if(value===null){
+reading=false;
+thrownewERR_STREAM_NULL_VALUES();
+}
+
+if(readable.push(value)){
+continue;
+}
+
+reading=false;
+}catch(err){
+readable.destroy(err);
+}
+break;
+}
+}
+
+asyncfunctionchangeToAsyncValues(value){
+isAsyncValues=true;
+
+try{
+constres=awaitvalue;
+
+if(res===null){
+reading=false;
+thrownewERR_STREAM_NULL_VALUES();
+}
+
+if(readable.push(res)){
+nextSyncWithAsyncValues();
+return;
+}
+
+reading=false;
+}catch(err){
+readable.destroy(err);
+}
+}
+
+asyncfunctionnextSyncWithAsyncValues(){
 for(;;){
 try{
-const{ value, done }=isAsync ?
-awaititerator.next() :
-iterator.next();
+const{ value, done }=iterator.next();
 
 if(done){
 readable.push(null);
-}else{
-constres=(value&&
-typeofvalue.then==='function') ?
-awaitvalue :
-value;
-if(res===null){
-reading=false;
-thrownewERR_STREAM_NULL_VALUES();
-}elseif(readable.push(res)){
-continue;
-}else{
-reading=false;
-}
+return;
+}
+
+constres=(value&&
+typeofvalue.then==='function') ?
+awaitvalue :
+value;
+
+if(res===null){
+reading=false;
+thrownewERR_STREAM_NULL_VALUES();
 }
+
+if(readable.push(res)){
+continue;
+}
+
+reading=false;
+}catch(err){
+readable.destroy(err);
+}
+break;
+}
+}
+
+asyncfunctionnextAsync(){
+for(;;){
+try{
+const{ value, done }=awaititerator.next();
+
+if(done){
+readable.push(null);
+return;
+}
+
+if(value===null){
+reading=false;
+thrownewERR_STREAM_NULL_VALUES();
+}
+
+if(readable.push(value)){
+continue;
+}
+
+reading=false;
 }catch(err){
 readable.destroy(err);
 }