Limit promise concurrency

The question

The realistic version of "why is Promise.all(urls.map(fetch)) a problem". Firing a thousand requests at once exhausts sockets, trips rate limits and helps nobody. The question is how to cap it, and the follow-ups are:

1. Does the output stay in input order?
2. Is the limit actually never exceeded?
3. What happens when one callback rejects?

The solution

The worker-pool shape is much easier to reason about than batching, and it keeps the pool saturated instead of idling at the end of each batch.

async function mapAsyncLimit(iterable, callbackFn, size) {
  const items = Array.from(iterable);
  const results = new Array(items.length);
  let next = 0;

  // Each worker pulls the next index until the list is exhausted, so no more
  // than `size` callbacks are ever in flight.
  async function worker() {
    while (next < items.length) {
      const index = next;
      next += 1;
      results[index] = await callbackFn(items[index], index, items);
    }
  }

  const workers = [];
  for (let i = 0; i < Math.max(1, Math.min(size, items.length)); i += 1) {
    workers.push(worker());
  }

  // Promise.all rejects on the first failing worker, matching the callback's error.
  await Promise.all(workers);
  return results;
}

How it works

• A shared cursor, not a partition. Each worker takes the next index and moves on, so a worker that finishes early immediately picks up more work.
• Index is captured before awaiting, which is what keeps results in input order however long individual callbacks take.
• The pool size is clamped to at least one and at most the number of items, so a limit larger than the input doesn’t create idle workers.
• Promise.all over the workers propagates the first rejection, which is what makes a failing callback reject the whole operation.

The tests it has to pass

This is the part most explanations leave out. "It works" is not a claim you can check by reading. These are the 7 cases Practice Pad runs against your solution, and the implementation above passes all of them.

const doubled = await mapAsyncLimit([1, 2, 3, 4], async (value) => value * 2, 2);
expectEqual(doubled, [2, 4, 6, 8]);

const upper = await mapAsyncLimit(
  ["a", "b", "c"],
  async (value) => {
    await delay(5);
    return value.toUpperCase();
  },
  2,
);
expectEqual(upper, ["A", "B", "C"]);

expectEqual(await mapAsyncLimit([], async (value) => value, 3), []);

const order = [];
const results = await mapAsyncLimit(
  [30, 5, 10],
  async (ms) => {
    order.push("start:" + String(ms));
    await delay(ms);
    order.push("end:" + String(ms));
    return ms;
  },
  1,
);
expectEqual(results, [30, 5, 10]);
expectEqual(order, ["start:30", "end:30", "start:5", "end:5", "start:10", "end:10"]);

let message;
try {
  await mapAsyncLimit([1, 2], async () => { throw new Error("mapper failed"); }, 2);
} catch (error) {
  message = error && error.message;
}
expectEqual(message, "mapper failed");

let active = 0;
let peak = 0;
const results = await mapAsyncLimit(
  [40, 10, 30, 20],
  async (ms) => {
    active += 1;
    peak = Math.max(peak, active);
    await delay(ms);
    active -= 1;
    return ms;
  },
  2,
);
expectEqual(results, [40, 10, 30, 20]);
expectEqual(peak <= 2, true);

const ordered = await mapAsyncLimit(
  [30, 5],
  async (ms) => {
    await delay(ms);
    return ms;
  },
  2,
);
expectEqual(ordered, [30, 5]);

Edge cases to watch out for

Batching instead of pooling

Splitting into chunks of size and awaiting each chunk is simpler, and wastes time: the whole batch waits for its slowest member before the next one starts. A pool keeps every slot busy.

Pushing results instead of assigning by index

With varying durations, push order is completion order. The last hard test uses a slow first item and a fast second precisely to catch this.

Reading the cursor after the await

Capturing index after awaiting means every worker sees the final value. Take the index and increment the cursor before any await.

Follow up questions

See all 24 questions