Run Concurrent Tasks With a Limit Using Pure JavaScript

You might be familiar with libraries like p-limit, async or bottleneck. They help you run asynchronous tasks with a concurrency limit. This is useful when you don't want to overwhelm an API or want to keep resource usage below the maximum threshold.

Using a library is convenient, but it is yet another addition to an already long list of dependencies that your application relies on.

Besides, how do these libraries implement a concurrency limit anyway?

What if I told you that you can write your own implementation with only a few lines of code?

I'm going to teach you a simple and elegant way to run async tasks with a concurrency limit so you don't have to rely on another library.

Concurrency limit using iterators

It turns out that iterators have a unique property that makes them an ideal choice for this use case.

Looping over an iterator consumes it because generally, it's only possible to do once.

This gives us the guarantee that an iterator's value won't be read more than once.

With that in mind, we can have several loops going through an iterator at the same time knowing that each loop will process a different value.

When we have to run many tasks, it's oftentimes because we have an array that holds some type of value for each task — a list of URLs we want to fetch, or an image collection we want to process. To get a consumable iterator from an array you can use the .values() method on the array.

If we then create an array with size X (= concurrency limit) and fill it with the same iterator, we can map over the array and start off X concurrent loops that go through the iterator.

Here's how that looks in code:

async function doWork(iterator) {
  for (const value of iterator) {
    await delay(1000);
    console.log(value);
  }
}

const iterator = Array.from('abcdefghi').values();

// Run async tasks with a concurrency limit of 3
const workers = new Array(3).fill(iterator).map(doWork);

// Wait until all tasks are done
await Promise.allSettled(workers);

console.log('Done!');

In the above example, we create a workers array with size 3 which is the number of tasks we want to run simultaneously. We then fill it with the iterator obtained using the .values() method. Finally, we map through the workers and kick off concurrent for...of loops that go through the iterator and run async tasks.

This prints out the following:

a
b
c
(1s later)
d
e
f
(1s later)
g
h
i
Done!

The end result is that we simultaneously execute tasks with a specific concurrency limit. By using a consumable iterator we're making sure a task will not run more than once.

Using return values

In practice, async tasks have some type of result that we want to assign to a variable and use later on. When using an array, we want these results to be in the same order as the original array so we know which result belongs to which task.

Because asynchronous tasks can finish at different times, simply returning an array of results from each worker would have us lose the original order. The results will show up in order of completion instead.

We can get around this issue by using the .entries() method instead of .values() to also get the index for each value. We'll then use this index to construct a results array that's in the same order as the original array:

const results = [];

async function doWork(iterator) {
  for (const [index, value] of iterator) {
    await delay(1000);

    // Add result to its original place
    results[index] = value;
  }
}

// Use `.entries()` to get the index and value for each element
const iterator = Array.from('abcdefghi').entries();
const workers = new Array(3).fill(iterator).map(doWork);

await Promise.allSettled(workers);

console.log(results); // ['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h', 'i']

Extracting into a utility function

You can generalize and extract this implementation into a utility function that you can then import and use throughout your project.

Let's create a limit function that takes two arguments:

1. tasks (Array) - An array of asynchronous functions to be executed
2. concurrency (Number) - A concurrency limit for task execution

And returns:

• Array - The result values returned by running the async functions, if any. In case of failure, the result will be of type Error

// utils/limit.js
export default async function limit(tasks, concurrency) {
  const results = [];

  async function runTasks(tasksIterator) {
    for (const [index, task] of tasksIterator) {
      try {
        results[index] = await task();
      } catch (error) {
        results[index] = new Error(`Failed with: ${error.message}`);
      }
    }
  }

  const workers = new Array(concurrency)
    .fill(tasks.entries())
    .map(runTasks);

  await Promise.allSettled(workers);

  return results;
}

You may have noticed there's a try...catch statement that was missing in previous examples. If a task throws an error, it will propagate to the worker running the task which will stop the worker and we effectively end up with one less concurrency. By handling the error, we make sure the worker continues running tasks if a task throws an error.

Elsewhere in your project, you can import the function and pass it an array of async tasks with a concurrency limit:

// main.js
import limit from 'utils/limit.js';

const tasks = [
  () => fetch(url),
  () => fetch(url),
  () => fetch(url),
  // ...
];

const results = await limit(tasks, 3);

And voila! You've just created your own async utility function. The API looks neat, doesn't it? ✨

Conclusion

You've learned a simple and elegant way of executing tasks with a concurrency limit without having to rely on external libraries.

If this is your first time working with iterators, you've learned that they are consumed when iterated since it's generally only possible to do once.

This implementation is great for simple use cases. If you need to do anything more complicated such as cancelling tasks, introspection and pausing, I recommend using a well-established library instead of writing your own. However, if you have a simple use case then this is a great opportunity to remove a dependency from your application.