concurrent.test.ts

import { CountingGovernor } from "./governor.ts";

const tests: { name: string; fn: () => Promise<void> }[] = [];

function test(name: string, fn: () => Promise<void>) {
  tests.push({ name, fn });
}

test("test 1", async () => {
  await new Promise((resolve) => setTimeout(resolve, 100));
});

test("test 2", async () => {
  await new Promise((resolve) => setTimeout(resolve, 100));
});

test("test 3", async () => {
  await new Promise((resolve) => setTimeout(resolve, 100));
});

const governor = new CountingGovernor(2);

Deno.test("concurrent tests", async (t) => {
  const promises: Promise<boolean>[] = [];
  for (const { name, fn } of tests) {
    const token = await governor.acquire();
    promises.push(
      t.step({
        name,
        sanitizeOps: false,
        sanitizeResources: false,
        sanitizeExit: false,
        fn: async () => {
          try {
            await fn();
          } finally {
            token.release();
          }
        },
      }),
    );
  }
  await Promise.all(promises);
});

governor.ts

export abstract class Governor {
  abstract acquire(): Promise<GovernorToken>;

  with<R>(fn: (...args: []) => R): Promise<Awaited<R>> {
    return this.wrap(fn)();
  }

  wrap<T, A extends unknown[], R>(
    fn: (this: T, ...args: A) => R,
  ): (this: T, ...args: A) => Promise<Awaited<R>> {
    const _this = this;
    return async function (...args): Promise<Awaited<R>> {
      const token = await _this.acquire();
      try {
        return await fn.apply(this, args);
      } finally {
        token[Symbol.dispose]();
      }
    };
  }

  wrapIterator<T>(iter: Iterator<T> | AsyncIterator<T>): AsyncIterator<T> {
    return {
      next: async (n) =>
        await this.wrap(iter.next as Iterator<T>["next"]).call(iter, n),
      return: async () =>
        typeof iter.return === "function"
          ? iter.return()
          : { done: true, value: undefined },
    };
  }
}

export interface GovernorToken {
  release(): void;
  [Symbol.dispose](): void;
}

export class CountingGovernor extends Governor {
  #capacity: number;
  #acquired: number = 0;
  #wait: PromiseWithResolvers<void> | null = null;

  constructor(capacity: number) {
    if ((capacity >>> 0) !== capacity) {
      throw new TypeError("capacity must be an integer");
    }
    if (capacity < 0) {
      throw new RangeError("capacity must be non-negative");
    }
    super();

    this.#capacity = capacity;
  }

  async acquire() {
    while (this.#acquired >= this.#capacity) {
      if (!this.#wait) {
        this.#wait = Promise.withResolvers<void>();
      }
      await this.#wait.promise;
    }
    ++this.#acquired;

    let hasReleased = false;

    const dispose = () => {
      if (hasReleased) {
        throw new Error("Already released");
      }
      hasReleased = true;
      --this.#acquired;
      if (this.#wait) {
        this.#wait.resolve();
        this.#wait = null;
      }
    };

    return {
      release: dispose,
      [Symbol.dispose]: dispose,
    };
  }
}