Promise.allSettled() instead of Promise.all() option

[alvaro-escalante]

Promise.allSettled() instead of Promise.all() option

[alvaro-escalante]

I think I have managed to add the funcionality with this

export async function asyncPool(poolLimit, array, iteratorFn, exception) {
  const ret = [];
  const executing = [];

  for (const [index, item] of array.entries()) {
    process.stdout.write(`\rProcessing ${index + 1}/${array.length} ...`);
    const p = Promise.resolve().then(() => iteratorFn(item, array));
    ret.push(p);

    if (poolLimit <= array.length) {
      const e = p.then(() => executing.splice(executing.indexOf(e), 1)).catch(() => executing.splice(executing.indexOf(e), 1))

      executing.push(e)
      if (executing.length >= poolLimit) {
        await Promise.race(executing)
      }
    }
  }

  const results = await Promise.allSettled(ret)

  const rejected = results
    .filter(({ status, reason }) => status === 'rejected' && reason.name === exception)
    .map(({ reason }) => reason.error)

  if (rejected.length) {
    await asyncPool(poolLimit, rejected, iteratorFn, exception)
  }
}

[rxaviers]

Hi @alvaro-escalante thanks for your request.

I am afraid the change would break the original behavior of this library: "It rejects immediately as soon as one of the promises rejects. It resolves when all the promises completes".

Can't your use case be solved at app side? For example, by having your iteratorFn to take care of retries before actually rejecting? Such as this #26 (comment)

Also note Promise.allSettled() is fairly new, so the change would impose a significant difference in browser support (probably requiring polyfill)

[alvaro-escalante]

Hi Rafael, thanks for you response.

I can deal with the errors at app level and in fact in the code I presented I actually do that by throwing the item I want to run again on the iteratorFn, I am not so invested in Promise.allSettled() as much as I just want a way to recursively run a failing request.

Like I do on both my functions, I think it would be an interesting proposal to under some circustances, controlled at app level, re-run/ push the 'failing' item. This could be a url that just timesout and can be safetly push back in to the array to re-run,.

Whatever the error is, it could be modified and thrown at app level to the asyncPool in my example I run asyncPool rescursively.

Many thanks.
Alvaro.

[rxaviers]

I think it would be an interesting proposal to under some circustances, controlled at app level, re-run/ push the 'failing' item

Please, what it would look like to use it if asyncPool had such feature implemented?

[alvaro-escalante]

I already did up on the previous code snippet, just checking with you, if you think it makes sense

[rxaviers]

Above I see the potential asyncPool implementation changes/updates. I am wondering what it would look like at the app level to use the modified-asyncPool. How simple/easy it would be to use it in the case you mention. Thanks

[rxaviers]

FWIW, my understanding is that you believe this proposal would be a good addition to the library and the purpose of my question is to understand your point of view.

[alvaro-escalante]

This is my how I implemented it at the app level.

So say I have an array of urls, on this example, I am just removing the https:// part, and if one of the urls matches a condition I reject it and throw it so the asyncPool can detect it and re run, obviously in this example, I then modify the url so it does not go on an infinite loop.

If for example the error was a 500 and I know the API just timed out, then I could just send the same url again, this is what I mean the control is on the app level, so the app decides when to send the error item to the asyncPool again.

This is the whole thing working recursively

asyncPools.js Modified to re run cretain items

export async function asyncPool(poolLimit, array, iteratorFn, exception) {
  const ret = [];
  const executing = [];

  for (const [index, item] of array.entries()) {
    process.stdout.write(`\rProcessing ${index + 1}/${array.length} ...`);
    const p = Promise.resolve().then(() => iteratorFn(item, array));
    ret.push(p);

    if (poolLimit <= array.length) {  
      // I know this is horrible but is the only way I know to let Promise.race ignore the failing url
      const e = p.then(() => executing.splice(executing.indexOf(e), 1)).catch(() => executing.splice(executing.indexOf(e), 1))

      executing.push(e)
      if (executing.length >= poolLimit) {
        await Promise.race(executing)
      }
    }
  }

  const results = await Promise.allSettled(ret)

  const rejected = results
    .filter(({ status, reason }) => status === 'rejected' && reason.name === exception)
    .map(({ reason }) => reason.error)

  if (rejected.length) {
    await asyncPool(poolLimit, rejected, iteratorFn, exception)
  }
}

index.js App level control

import { asyncPool } from './asyncPool.js'
const data = ['many urls', '....'];
const concurrent = 3;
const res = [];

// Mock API response - Let's suppose is a list of urls and if one of then is apple.co.uk add foo and re run by rejecting
const API = (item) =>
  new Promise((resolve, reject) => {
    if (item === 'https://www.apple.co.uk/') reject(item + 'foo')
    setTimeout(() => resolve(item), 1000)
  });

// Iterator Function 
const fetchAPI = async (item) => {
  try {
    const request = await API(item);
    res.push(request.replace(/^https?:\/\//, ''));
  } catch (error) {
    // The rejection is picked up by asyncPool with the name timeout, our `expection`
    throw { name: 'timeout', error };
  }
};

// Call function
(async () => {
  await asyncPool(concurrent, data, fetchAPI, 'timeout');
  console.log(res);
})();

This is just a solution I came up with cause I have this problem, I am not saying is the most elegant solution or the best, it just works for what I need, and just checking with you, maybe you see a better way.

[rxaviers]

The new 2.x API doesn't use Promise.all anymore, so I am closing this issue. If you still believe there's anything to be done in the new API please feel free to reopen. Thanks