Read multiple async iterators simultaneously in a single thread

Question

Lets assume that i have two separate simple "base" async generators:

async function* base1(){
  while(true) yield 1
}
async function* base2(){
  while(true) yield 2
}

And on top of them i want to create another async generator that will accumulate both iterator's values produced by "base" generators and yield the resault further. But the problem is that yield statement must be in the generator's main thread and so i cannot iterate over couple of iterators in the same thread. For example possible solutions below will not work:

async function* accumulator(){
  let base1Yield,base2Yield
  //main thread will be forever blocked by only this iterator
  for await(const yieldValue of base1()){
    base1Yield=yieldValue
    yield base1Yield+base2Yield
  }
  //and this one will never execute
  for await(const yieldValue of base2()){
    base2Yield=yieldValue
    yield base1Yield+base2Yield
  }
}
async function* accumulator(){
  let base1Yield,base2Yield
  //now we put our iterations in a different threads
  queueMicrotask(async()=>{
    for await(const yieldValue of base1()){
      base1Yield=yieldValue
      //but yield statement cannot be executed outside of generator's main thread
      yield base1Yield+base2Yield //throws exception
    }
  })
  //but now we atleast start to read two iterators simultaneously
  queueMicrotask(async()=>{
    for await(const yieldValue of base2()){
      base2Yield=yieldValue
      yield base1Yield+base2Yield
    }
  })
}

So as far as i can see this problem have no native solution (maybe im wrong) and to achieve my goal i need to write some custom complex async iterator that will be able to listen to an array of iterators and yield with an array of values so i can use it in the main thread of a generator:

function complexIterator(...arrayOfIterators){
  return {
    [Symbol.asyncIterator](){
      const resault=new Array(arrayOfIterators.length).fill(null)
      let resolve
      arrayOfIterators.forEach(async(iterator,index)=>{
        for await(const yieldValue of iterator){
          resault[index]=yieldValue
          resolve({value:resault,done:false})
        }
      })
      return {
        next:()=>new Promise(innerResolve=>{
          resolve=innerResolve
        })
        //also "return" and "throw" methods here to stop iterations
      }
    }
  }
}

And then use it in my "accumulator" generator like this:

async function* accumulator(){
  for await(const [base1Yield,base2Yield] of complexIterator(base1(),base2())){
    yield base1Yield+base2Yield
  }
}

But i do really want to have a native approach for such a simple and common task. Am i missing something?

Answer 1

async/await isn't about threading, it's just a means of handling asynchronous work. Other than worker threads, all of your JavaScript code runs single-threaded. (And worker threads would use their own realms, so you wouldn't directly consume generators from them.) See the answers to this question and this one for more on that. (Disclosure: I have answers posted to both of those.)

If you want to work through the sequences from base1 and base2 at the same time, you can't use for-await-of, but you can use their generator objects directly.

async function* accumulator() {
    // Get the generators from each generator function
    const it1 = base1();
    const it2 = base2();

    while (true) {
        // Wait until we have values from both. (You could await `it1.next()`
        // then await `it2.next()`, but we may as well just wait for both.)
        const [r1, r2] = await Promise.all([it1.next(), it2.next()]);
        if (r1.done && r2.done) {
            return;
        }
        // Yield the combined result
        yield r1.value + r2.value;
    }
}

Live example:

async function* base1() {
    while (true) yield 1;
}

async function* base2() {
    while (true) yield 2;
}

async function* accumulator() {
    // Get the generators from each generator function
    const it1 = base1();
    const it2 = base2();

    while (true) {
        // Wait until we have values from both. (You could await `it1.next()`
        // then await `it2.next()`, but we may as well just wait for both.)
        const [r1, r2] = await Promise.all([it1.next(), it2.next()]);
        if (r1.done && r2.done) {
            return;
        }
        // Yield the combined result
        yield r1.value + r2.value;
    }
}

(async () => {
    let counter = 0;
    for await (const value of accumulator()) {
        console.log(value);
        if (++counter === 10) {
            break;
        }
    }
})();

If you like, you can generalize that into an aggregator:

async function* aggregate(...iterators) {
    while (true) {
        const results = await Promise.all(iterators.map((it) => it.next()));
        if (results.every(({ done }) => done)) {
            return;
        }
        yield results.map(({ value }) => value);
    }
}

Then using it:

for await (const results of aggregate(base1(), base2())) {
    // ...
}

Live example:

async function* base1() {
    while (true) yield 1;
}

async function* base2() {
    while (true) yield 2;
}

async function* aggregate(...iterators) {
    while (true) {
        const results = await Promise.all(iterators.map((it) => it.next()));
        if (results.every(({ done }) => done)) {
            return;
        }
        yield results.map(({ value }) => value);
    }
}

(async () => {
    let counter = 0;
    for await (const results of aggregate(base1(), base2())) {
        console.log(results);
        if (++counter === 10) {
            break;
        }
    }
})();

.as-console-wrapper {
    max-height: 100% !important;
 }