🚀 Concurrency Implementation in JavaScript

Photo by Nicolas Hoizey on Unsplash
Hello everyone, I am Biao Zhu.
I hope you enjoy reading this article.

Today, let’s learn about concurrency control in JavaScript. In your everyday development, you might encounter scenarios where concurrency control is necessary, such as controlling the concurrency of requests. So, how can we implement concurrency control in JavaScript? But before answering this question, let’s briefly introduce concurrency control.

***01.What is concurrency control？***🤔

Assume there are 5 tasks to be executed. We want to limit the number of tasks executed simultaneously to 2, meaning only a maximum of 2 tasks can be executed concurrently. When any one task in the list of tasks being executed is completed, the program will automatically fetch a new task from the list of pending tasks and add it to the list of tasks being executed. To help everyone understand the process more intuitively, Biao specially drew the following 3 diagrams:

stage1

Images created by the author

stage2

Images created by the author

stage3

Images created by the author

***02.How to implement concurrency control？***🧐

After introducing concurrency control, I will demonstrate the specific implementation of asynchronous task concurrency control using the async-pool library on GitHub.

2.1 The usage of asyncPool 😎

const timeout = i => new Promise(resolve => 
    setTimeout(() => resolve(i), i)
);
await asyncPool(2, [1000, 5000, 3000, 2000], timeout);

In the above code, we utilize the ‘asyncPool’ function provided by the asyncPool library to implement concurrency control for asynchronous tasks. The signature of the asyncPool the function is as follows :

function asyncPool(poolLimit, array, iteratorFn){ ... }

The function takes 3 parameters:

• poolLimit (number type): Indicates the concurrency limit.

• array (array type): Represents the array of tasks.

• iteratorFn (function type): Represents the iterator function used to process each task item, which returns a Promise object or an asynchronous function.

For the example above, after using the asyncPool function, the corresponding execution process is as follows:

const timeout = i => new Promise(resolve => setTimeout(() => resolve(i), i));
await asyncPool(2, [1000, 5000, 3000, 2000], timeout);
// Call iterator (i = 1000)
// Call iterator (i = 5000)
// Pool limit of 2 reached, wait for the quicker one to complete...
// 1000 finishes
// Call iterator (i = 3000)
// Pool limit of 2 reached, wait for the quicker one to complete...
// 3000 finishes
// Call iterator (i = 2000)
// Itaration is complete, wait until running ones complete...
// 5000 finishes
// 2000 finishes
// Resolves, results are passed in given array order `[1000, 5000, 3000, 2000]`.

By observing the comments above, we can roughly understand the control flow inside the asyncPool function. Now let’s analyze the ES7 implementation of the asyncPool function.

2.2 asyncPool ES7 🫢

async function asyncPool(poolLimit, array, iteratorFn) {
  const ret = []; //  Store all asynchronous tasks
  const executing = []; //  Store currently executing asynchronous tasks
  for (const item of array) {
    //  Call the iteratorFn function to create asynchronous tasks
    const p = Promise.resolve().then(() => iteratorFn(item, array));
    ret.push(p); // Save the new asynchronous task
    //  When poolLimit is less than or equal to the total number of 
    //  tasks, perform concurrency control.
    if (poolLimit <= array.length) {
      // After the task is completed, 
      // remove the completed task from the array of executing tasks.
      const e = p.then(() => executing.splice(executing.indexOf(e), 1));
      executing.push(e); // Save the executing asynchronous task.
      if (executing.length >= poolLimit) {
        // Wait for the fastest task to complete execution.
        await Promise.race(executing); 
      }
    }
  }
  return Promise.all(ret);
}

In the above code, the characteristics of Promise.all and Promise.race functions are fully utilized, combined with the async await feature provided in ES7, to ultimately achieve concurrency control. By using the statement await await Promise.race(executing);, we wait for the fastest task in the list of executing tasks to complete before proceeding to the next iteration.

The async-pool library also provides an ES6 implementation approach, which interested individuals can explore and learn more about.

Conclusion 🚀

This article meticulously analyzes the core code of async-pool to help readers better understand the specific implementation of asynchronous task concurrency control using async-pool.

Thank you for reading until the end. Before you go:

• Please consider following me 👏 and you can subscribe to my medium.

• I am looking forward to your reply and communicating with you!💬