pLimit

Creates a scheduler that limits concurrently running async tasks.

Code

pLimit.ts

export type LimitedTask<Return> = (this: void) => Return;
type QueueEntry = { run: () => void; reject: (reason: unknown) => void };

export type Limit = {
  <Return>(task: LimitedTask<Return>): Promise<Awaited<Return>>;
  readonly activeCount: number;
  readonly pendingCount: number;
  clearQueue: (reason?: unknown) => void;
};

/** Creates a scheduler that limits concurrently running promise-returning tasks. */
export const pLimit = (concurrency: number): Limit => {
  if (!Number.isSafeInteger(concurrency) || concurrency < 1) {
    throw new RangeError("concurrency must be a positive safe integer");
  }
  let activeCount = 0;
  const queue: QueueEntry[] = [];

  const next = () => {
    activeCount -= 1;
    queue.shift()?.run();
  };
  const limit = (<Return>(
    task: LimitedTask<Return>
  ): Promise<Awaited<Return>> => {
    return new Promise<Awaited<Return>>((resolve, reject) => {
      const run = () => {
        activeCount += 1;
        Promise.resolve()
          .then(() => task())
          .then((value) => resolve(value as Awaited<Return>), reject)
          .finally(next);
      };
      if (activeCount < concurrency) run();
      else queue.push({ run, reject });
    });
  }) as Limit;

  Object.defineProperties(limit, {
    activeCount: { get: () => activeCount },
    pendingCount: { get: () => queue.length },
  });
  limit.clearQueue = (reason = new Error("Pending task was cleared")) => {
    for (const entry of queue.splice(0)) entry.reject(reason);
  };
  return limit;
};

Download

wget

Terminal

wget -O src/lib/pLimit.ts https://raw.githubusercontent.com/jrTilak/lazykit/HEAD/registry/functions/pLimit.ts

degit

Terminal

temp_dir="$(mktemp -d)" && bunx degit jrTilak/lazykit/registry/functions "$temp_dir" && mv "$temp_dir/pLimit.ts" src/lib/pLimit.ts && rm -r "$temp_dir"

Examples

pLimit.example.ts

import { pLimit } from "./pLimit";

const limit = pLimit(2);
const users = await Promise.all(
  ["ada", "lin", "sam"].map((id) =>
    limit(() => fetch(`/api/users/${id}`).then((response) => response.json()))
  )
);

Props

• concurrency (number) — Positive safe-integer task limit.
• The returned scheduler exposes activeCount, pendingCount, and clearQueue.
• Clearing rejects pending work without stopping active tasks.