Map with a concurrency limit

Write mapWithLimit(items, limit, fn). It calls the async function fn(item, index) for every item in the array items, with at most limit calls running at the same time, and returns a promise of the results, in input order.

Start a new call as soon as one finishes, so limit calls are running for as long as work remains: no batches.

If a call rejects, reject with that error and don't start any more calls. If limit is below 1, reject with a RangeError without calling fn. An empty array fulfills with [].

const pages = await mapWithLimit(urls, 3, (url) => fetchJson(url));
// at most 3 requests at once; pages[i] is the response for urls[i]
 
const wait = (ms) => new Promise((resolve) => setTimeout(resolve, ms, "waited " + ms));
await mapWithLimit([300, 100, 200], 2, wait);
// at 0 ms: wait(300) and wait(100) start
// at 100 ms: wait(100) is done, so wait(200) starts right away
// at 300 ms: wait(300) and wait(200) are done, and it fulfills with
// ["waited 300", "waited 100", "waited 200"]

What the tests check

1. 1fulfills with the results in input order

   const clock = useFakeTimers();
   const fn = (ms) => new Promise((resolve) => setTimeout(resolve, ms, "done in " + ms));
   let result;
   mapWithLimit([300, 100, 200], 2, fn).then((value) => {
     result = value;
   });
   await clock.tickAsync(1000);
   expect(result).toEqual(["done in 300", "done in 100", "done in 200"]);
   

2. 2never runs more than limit calls at once

   const clock = useFakeTimers();
   let running = 0;
   let most = 0;
   const fn = (ms) => {
     running += 1;
     most = Math.max(most, running);
     return new Promise((resolve) =>
       setTimeout(() => {
         running -= 1;
         resolve(ms);
       }, ms),
     );
   };
   mapWithLimit([50, 100, 30, 80, 20, 60], 2, fn);
   await clock.tickAsync(1000);
   expect(most).toBeLessThanOrEqual(2);
   

3. 3starts the next call as soon as one finishes, not in batches

   const clock = useFakeTimers();
   const start = clock.now();
   const startedAt = [];
   const fn = (ms) => {
     startedAt.push(clock.now() - start);
     return new Promise((resolve) => setTimeout(resolve, ms, ms));
   };
   mapWithLimit([100, 300, 100, 100], 2, fn);
   await clock.tickAsync(1000);
   expect(startedAt).toEqual([0, 0, 100, 200]);
   

4. 4passes each item and its index to fn

   const fn = spy(async (item) => item);
   await mapWithLimit(["a", "b", "c"], 2, fn);
   expect(fn.calls.map((call) => call.args)).toEqual([
     ["a", 0],
     ["b", 1],
     ["c", 2],
   ]);
   

5. 5rejects with the error of a call that fails

   const clock = useFakeTimers();
   const error = new Error("item 1 failed");
   const fn = (ms, index) =>
     new Promise((resolve, reject) => setTimeout(() => (index === 1 ? reject(error) : resolve(ms)), ms));
   let reason;
   mapWithLimit([100, 50, 100, 100], 2, fn).catch((e) => {
     reason = e;
   });
   await clock.tickAsync(1000);
   expect(reason).toBe(error);
   

6. 6starts no more calls after one fails

   const clock = useFakeTimers();
   const fn = spy((ms, index) =>
     new Promise((resolve, reject) => setTimeout(() => (index === 1 ? reject(new Error("failed")) : resolve(ms)), ms)),
   );
   mapWithLimit([100, 50, 100, 100, 100], 2, fn).catch(() => {}); // it rejects; this test only counts the calls
   await clock.tickAsync(1000);
   expect(fn).toHaveBeenCalledTimes(2);
   

7. 7fulfills with [] for an empty list

   let result;
   mapWithLimit([], 3, async (item) => item).then((value) => {
     result = value;
   });
   await flush();
   expect(result).toEqual([]);
   

8. 8rejects with a RangeError when limit is below 1

   const fn = spy(async (item) => item);
   await expect(mapWithLimit([1, 2], 0, fn)).rejects.toThrow(RangeError);
   expect(fn).not.toHaveBeenCalled();
   

9. 9calls fn once per item when limit is larger than the list

   const fn = spy(async (item) => item * 2);
   await expect(mapWithLimit([1, 2], 10, fn)).resolves.toEqual([2, 4]);
   expect(fn).toHaveBeenCalledTimes(2);
   

Learn more

• Async patternsLesson
• Sequential vs parallelLesson
• Write mapWithLimit(items, limit, fn): run an async function over a list, at most limit calls at a time.Interview question