Concurrency in Practice: Pools, Queues, Atomics and the Event Loop

// Amdahl's law: if a fraction p of the work can run in parallel, n workers give this speedup.
export const speedup = (p, n) => 1 / (1 - p + p / n);

// With unlimited workers only the serial part is left.
export const limit = (p) => 1 / (1 - p);

// How many workers? CPU-bound work: about one per core. Work that waits: more, in
// proportion to wait time over compute time.
export const poolSize = (cores, waitMs, computeMs) => Math.round(cores * (1 + waitMs / computeMs));

// A fixed number of workers pulling from a bounded queue. When both are full,
// reject at once instead of letting the queue (and memory) grow without limit.
export class Pool {
  #running = 0;
  #queue = [];

  constructor(size, maxQueue) {
    this.size = size;
    this.maxQueue = maxQueue;
  }

  submit(task) {
    return new Promise((resolve, reject) => {
      if (this.#running >= this.size && this.#queue.length >= this.maxQueue) {
        reject(new Error('rejected: queue full'));
        return;
      }
      this.#queue.push({ task, resolve, reject });
      this.#next();
    });
  }

  #next() {
    while (this.#running < this.size && this.#queue.length > 0) {
      const { task, resolve, reject } = this.#queue.shift();
      this.#running += 1;
      Promise.resolve()
        .then(task)
        .then(resolve, reject)
        .finally(() => {
          this.#running -= 1;
          this.#next();
        });
    }
  }
}

// Producer-consumer with a bounded buffer. A full buffer makes put() wait (backpressure);
// an empty one makes take() wait. Waiting hands the thread back instead of spinning.
export class BoundedQueue {
  #items = [];
  #takers = []; // consumers waiting for an item
  #putters = []; // producers waiting for room, with the item they hold

  constructor(capacity) {
    this.capacity = capacity;
  }

  async put(item) {
    if (this.#takers.length > 0) {
      this.#takers.shift()(item); // a consumer is already waiting: hand it over
      return;
    }
    if (this.#items.length < this.capacity) {
      this.#items.push(item);
      return;
    }
    await new Promise((resume) => this.#putters.push({ item, resume })); // full: wait for a take()
  }

  async take() {
    if (this.#items.length > 0) {
      const item = this.#items.shift();
      const waiting = this.#putters.shift();
      if (waiting) {
        this.#items.push(waiting.item); // room appeared: the oldest blocked producer goes in
        waiting.resume();
      }
      return item;
    }
    return new Promise((resolve) => this.#takers.push(resolve)); // empty: wait for a put()
  }
}

// Compare-and-swap: write `next` only if the cell still holds `seen`. No lock, no waiting:
// if another thread got there first, read again and retry. `cells` is an Int32Array on a SharedArrayBuffer.
export function update(cells, i, fn) {
  for (;;) {
    const seen = Atomics.load(cells, i);
    const next = fn(seen);
    if (Atomics.compareExchange(cells, i, seen, next) === seen) return next; // nobody changed it
    // someone else wrote in between: our `next` was based on stale data, so loop
  }
}

// Order of execution on a single-threaded event loop: the current code finishes first,
// then queued promise callbacks (microtasks), then the next timer or I/O callback.
export function order() {
  return new Promise((done) => {
    const log = [];
    setTimeout(() => {
      log.push('timer');
      done(log);
    }, 0);
    Promise.resolve().then(() => log.push('promise'));
    queueMicrotask(() => log.push('microtask'));
    log.push('sync');
  });
}

// Busy-waiting on the one thread stops everything else, including a timer due in 0 ms.
export function blockLoop(ms) {
  const end = Date.now() + ms;
  while (Date.now() < end);
}