阅读中文版本

Fixed-Concurrency Promise Pool

A pLimit-style concurrency limiter for async tasks with an injectable scheduler, tested to prove the pool never exceeds the limit and preserves results.

node --version

node pool.mjs

node --test pool.test.mjs

The demo runs ten tasks through a pool of three, printing the observed peak concurrency as exactly 3 and all ten results in order.

What it does

A fixed-concurrency pool for async tasks: results keep input order, rejections propagate, and a pool of one runs strictly sequentially.

export function createPool(limit) {
  let active = 0;
  const queue = [];

  function release() {
    active -= 1;
    const next = queue.shift();
    if (next) next();
  }

  function run(task) {
    return new Promise((resolve, reject) => {
      const start = () => {
        active += 1;
        task().then((v) => { release(); resolve(v); }, (e) => { release(); reject(e); });
      };
      if (active < limit) start();
      else queue.push(start);
    });
  }

  return { run, get active() { return active; }, get queued() { return queue.length; } };
}

export async function mapWithPool(items, limit, mapper) {
  const pool = createPool(limit);
  return Promise.all(items.map((item) => pool.run(() => mapper(item))));
}

let peak = 0;
let inflight = 0;
const results = await mapWithPool([1, 2, 3, 4, 5, 6, 7, 8, 9, 10], 3, async (n) => {
  inflight += 1;
  peak = Math.max(peak, inflight);
  await new Promise((r) => setTimeout(r, 5));
  inflight -= 1;
  return n * n;
});
console.log('peak=' + peak + ' results=' + results.join(','));

import { test } from 'node:test';
import assert from 'node:assert/strict';
import { createPool, mapWithPool } from './pool.mjs';

test('pool never exceeds the concurrency limit', async () => {
  let inflight = 0;
  let peak = 0;
  await mapWithPool(Array.from({ length: 20 }, (_, i) => i), 4, async (n) => {
    inflight += 1;
    peak = Math.max(peak, inflight);
    await new Promise((r) => setTimeout(r, 2));
    inflight -= 1;
    return n;
  });
  assert.equal(peak, 4);
});

test('results arrive in input order regardless of completion order', async () => {
  const delays = [30, 10, 20, 5];
  const results = await mapWithPool(delays, 4, (d) => new Promise((r) => setTimeout(() => r(d), d)));
  assert.deepEqual(results, delays);
});

test('rejections propagate to the caller', async () => {
  await assert.rejects(mapWithPool([1, 2, 3], 2, async (n) => {
    if (n === 2) throw new Error('task 2 failed');
    return n;
  }), /task 2 failed/);
});

test('pool of one runs strictly sequentially', async () => {
  const order = [];
  await mapWithPool(['a', 'b', 'c'], 1, async (x) => {
    order.push('start-' + x);
    await new Promise((r) => setTimeout(r, 1));
    order.push('end-' + x);
    return x;
  });
  assert.deepEqual(order, ['start-a', 'end-a', 'start-b', 'end-b', 'start-c', 'end-c']);
});

Run and verify

node --test pool.test.mjs
node pool.mjs

Run the tests proving the limit is never exceeded, then the demo mapping ten tasks through a pool of three.

Expected Output

The test run should print four passing tests, including the order-preservation case where tasks complete out of order but results still match input order. The demo’s console.log output should print the observed peak concurrency and the squared results in input order:

peak=3 results=1,4,9,16,25,36,49,64,81,100

The peak is exactly the configured limit of three — never four, no matter how fast tasks resolve — because a queued task only starts after a running one releases its slot, and Promise.all over the submitted promises keeps the ten results aligned with the ten inputs.

Limitations

No cancellation and no dynamic task submission while running; submit a fixed array and await the results.