async-await-queue

Promise-based priority queues for throttling, rate- and concurrency limiting of Node.js or browser tasks

Zero-dependency, total size: 2.93 kB uncompressed and 1.16 kB gzip-compressed

There is a medium story about using this package to parallelize download loops : Parallelizing download loops in JS with async-await-queue

This is an interesting solution to the priority queues problem.

There are other Promise-based queues out there but they are not async/await compatible and do not support priorities.

It guarantees order and never wakes up contexts that won't run.

I use it with tens of thousands of jobs on the queue. O(log(n)) on the number of jobs, O(log(n)) on the number of different priorities. Just make sure to always call Queue.end(). Or, since 1.2, there is a safer, but less versatile method, Queue.run().

Typical uses:

• Rate-limit expensive external API requests - especially on ban-happy servers
• Avoiding to launch all the tasks in an async loop at the same time while allowing some degree of controlled concurrency

The queues keep references to the Promise resolve() function and resolve it from outside of the Promise constructor. This is a very unusual use of Promises to implement locks that I find interesting (this is what the medium story is about).

2024 Update: My technique is on track to become an official ECMAScript language: ES Promise.withResolvers

Install

npm install --save async-await-queue

Typical usage

Require as CJS

1const { Queue } = require('async-await-queue');

Import as ES6 Module

1import { Queue } from 'async-await-queue';

(or read the jsdoc)

IMPORTANT Keep in mind that when running asynchronous code without explicitly awaiting it, you should always handle the eventual Promise rejections by a .catch() statement.

Examples

Basic example

1const { Queue } = require('async-await-queue');
2/**
3 * No more than 2 concurrent tasks with
4 * at least 100ms between two tasks
5 * (measured from task start to task start)
6 */
7const myq = new Queue(2, 100);
8const myPriority = -1;
9
10/**
11 * This function will launch all tasks and will
12 * wait for them to be scheduled, returning
13 * only when all tasks have finished
14 */
15async function downloadTheInternet() {
16  for (let site of Internet) {
17    /**
18     * The third call will wait for the previous two to complete
19     * plus the time needed to make this at least 100ms
20     * after the second call
21     * The first argument needs to be unique for every
22     * task on the queue
23     */
24    const me = Symbol();
25    /* We wait in the line here */
26    await myq.wait(me, myPriority);
27
28    /**
29     * Do your expensive async task here
30     * Queue will schedule it at
31     * no more than 2 requests running in parallel
32     * launched at least 100ms apart
33     */
34    download(site)
35      /* Signal that we are finished */
36      /* Do not forget to handle the exceptions! */
37      .catch((e) => console.error(e))
38      .finally(() => myq.end(me));
39  }
40  return await myq.flush();
41}

Using a function

1/**
2 * This is the new style API introduced in 1.2
3 * It is equivalent to the previous example
4 */
5async function downloadTheInternet() {
6  const q = [];
7  for (let site of Internet) {
8     /** The third call will wait for the previous two to complete
9      * plus the time needed to make this at least 100ms
10      * after the second call
11      */
12    q.push(myq.run(() => download(site).catch((e) => console.error(e))));
13  }
14  return Promise.all(q);
15}

Running sequentially

1/**
2 * This function will execute a single task at a time
3 * waiting for its place in the queue
4 */
5async function downloadTheInternet() {
6  let p;
7  /**
8   * The third call will wait for the previous two to complete
9   * plus the time needed to make this at least 100ms
10   * after the second call
11   * The first argument needs to be unique for every
12   * task on the queue
13   */
14  const me = Symbol();
15  /* We wait in the line here */
16  await myq.wait(me, myPriority);
17
18  /**
19   * Do your expensive async task here
20   * Queue will schedule it at
21   * no more than 2 requests running in parallel
22   * launched at least 100ms apart
23   */
24  try {
25    await download(site);
26  } catch (e) {
27    console.error(e);
28  } finally {
29    /* Signal that we are finished */
30    /* Do not forget to handle the exceptions! */
31    myq.end(me);
32  }
33}

Fire-and-forget

1/**
2 * This function will schedule all the tasks and
3 * then will return immediately a single Promise
4 * that can be awaited upon
5 */
6async function downloadTheInternet() {
7  const q = [];
8  for (let site of Internet) {
9    /**
10     * The third call will wait for the previous two to complete
11     * plus the time needed to make this at least 100ms
12     * after the second call
13     * The first argument needs to be unique for every
14     * task on the queue
15     */
16    const me = Symbol();
17    q.push(
18      myq
19        .wait(me, myPriority)
20        .then(() => download(site))
21        .catch((e) => console.error(e))
22        .finally(() => myq.end(me))
23    );
24  }
25  return Promise.all(q);
26}

Unresolvable Promises in Node.js

When using this package, something that you should be aware of is that Node.js has a very particular behavior when dealing with unresolvable Promises: https://github.com/nodejs/node/issues/43162

When awaiting an unresolvable Promise, Node.js will simply exit - instead of blocking indefinitely - which would probably be what most people expect.

If you are using this package in Node.js and it seems to simply unexpectedly exit without reaching the program's normal end and without reporting any errors, you most probably have an unresolvable Promise.