mmomtchev / queue

async/await-compatible Promise-based priority queues

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

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

Import as ES6 Module

import { 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

const { Queue } = require('async-await-queue');
/**
 * No more than 2 concurrent tasks with
 * at least 100ms between two tasks
 * (measured from task start to task start)
 */
const myq = new Queue(2, 100);
const myPriority = -1;

/**
 * This function will launch all tasks and will
 * wait for them to be scheduled, returning
 * only when all tasks have finished
 */
async function downloadTheInternet() {
  for (let site of Internet) {
    /**
     * The third call will wait for the previous two to complete
     * plus the time needed to make this at least 100ms
     * after the second call
     * The first argument needs to be unique for every
     * task on the queue
     */
    const me = Symbol();
    /* We wait in the line here */
    await myq.wait(me, myPriority);

    /**
     * Do your expensive async task here
     * Queue will schedule it at
     * no more than 2 requests running in parallel
     * launched at least 100ms apart
     */
    download(site)
      /* Signal that we are finished */
      /* Do not forget to handle the exceptions! */
      .catch((e) => console.error(e))
      .finally(() => myq.end(me));
  }
  return await myq.flush();
}

Using a function

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

Running sequentially

/**
 * This function will execute a single task at a time
 * waiting for its place in the queue
 */
async function downloadTheInternet() {
  let p;
  /**
   * The third call will wait for the previous two to complete
   * plus the time needed to make this at least 100ms
   * after the second call
   * The first argument needs to be unique for every
   * task on the queue
   */
  const me = Symbol();
  /* We wait in the line here */
  await myq.wait(me, myPriority);

  /**
   * Do your expensive async task here
   * Queue will schedule it at
   * no more than 2 requests running in parallel
   * launched at least 100ms apart
   */
  try {
    await download(site);
  } catch (e) {
    console.error(e);
  } finally {
    /* Signal that we are finished */
    /* Do not forget to handle the exceptions! */
    myq.end(me);
  }
}

Fire-and-forget

/**
 * This function will schedule all the tasks and
 * then will return immediately a single Promise
 * that can be awaited upon
 */
async function downloadTheInternet() {
  const q = [];
  for (let site of Internet) {
    /**
     * The third call will wait for the previous two to complete
     * plus the time needed to make this at least 100ms
     * after the second call
     * The first argument needs to be unique for every
     * task on the queue
     */
    const me = Symbol();
    q.push(
      myq
        .wait(me, myPriority)
        .then(() => download(site))
        .catch((e) => console.error(e))
        .finally(() => myq.end(me))
    );
  }
  return Promise.all(q);
}

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: nodejs/node#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.

queue's Issues

Q/A is there a way to remove a currently waiting task after adding to queue?

Hi! This project looks super promising for my needs, and I love the minimum dependencies, but I have a feeling it's missing 1 crucial piece

The project I'm working on right now allows for users to request a batch of PDFs, package them up into a zip, and get later for download

I can make great use of this package to simplify the very asynchronous nature of this workflow on our server, but I don't see anything about removing a waiting request in the queue

say I have a queue with 10 requests, which won't resolve in an hour or so and decide 10 minutes after making my request I don't actually want that export anymore so I send another request to cancel my last request

wait() and end() make use of a hash to keep track of a request in the queue. Could this hash be used to remove that request from the queue later, assuming it's not currently executing?

TypeError: Queue is not a constructor

while using v2.1.4

const {Queue} = require('async-await-queue');
const myq = new Queue(1, 100);

TypeError: Queue is not a constructor

"typescript": "~4.9.5"
Works with v1.2.1

saw diff in index.d.ts
in 2.1.4 export declare class Queue .... is used
in 1.2.1 export class Queue ... is used
saw Andre Pena's answer. He thinks its of no use
https://stackoverflow.com/questions/35019987/what-does-declare-do-in-export-declare-class-actions

thanks

A way to wait indefinitely until a job is completed

Hi, this is such an amazing project, thank you!

I wanted to know if there was a way to wait indefinitely until a job is completed before moving to the next in queue? Currently just setting it to a really large number does the job, but it just seems there is probably a better way that I'm missing.

I need this because I use this to write to a file on any state change, and it can result in overwriting issues sometimes if multiple file saves are called at the same time.

Again thank you for an amazing project!

Incorrect typings for the QueueStats interface

It looks like the type definition for QueueStats interface is incorrect.

index.d.ts contains the following definition:

export interface QueueStats {running: {number}, waiting: {number}, last: {number}}

this type means that the running, waiting and last property are all objects with a property inside them called number and a type of any. However looking at the code I think each of those should simply be of type number, like so:

export interface QueueStats {
    running: number;
    waiting: number;
    last: number;
}

Nice work!

Not an issue. Just wanted to say thanks. This is a really great algorithm you've published here! Exactly what I needed and it worked right out of the box.

How can i use this in browser?

How can i use this in browser

Doc issue

Thank you for this package.
The example code "running sequentially" seems incomplete to me. Could you take a look ?
Thanks !

RangeError: Too many elements passed to Promise.all

Your readme excited me because it says, I use it with tens of thousands of jobs on the queue. However, I tried adding 10,400 items to the queue and got RangeError: Too many elements passed to Promise.all. I am using the 1.2 syntax (second example in your README). Is there a different approach required for larger queues?