1
\$\begingroup\$

A personal project I'm working on needs Worker (thread) pool. The libraries I looked at were too high-level for my needs, as I simply needed access to the Worker itself while the libraries offered task-runner abstractions that were difficult to work around.

I decided to write my own, but I figured I would make it generic so that it would work with not only workers, but any arbitrary resource.

interface ResourceFactory<T, U = T> {
  create(): PromiseLike<T>;
  dispose?(resource: T): PromiseLike<T>;
  access?(resource: T): PromiseLike<U>;
}


class AsyncQueue<T> {
  #elements: T[] = [];
  #waiting: ((el: T) => void)[] = [];
  get size() { return this.#elements.length }

  enqueue(el: T) {
    const next = this.#waiting.shift();
    if (next) {
      next(el);
    } else {
      this.#elements.push(el);
    }
  }
  async dequeue() {
    const next = this.#elements.shift();
    if (next) {
      return next;
    } else {
      const defer = new Promise<T>(res => this.#waiting.push(res))
      return defer;
    }
  }
}

class ResourcePooler<T, U> {
  factory: ResourceFactory<T, U>;
  targetSize: number = 0;
  currentSize: number = 0;
  resourceQueue: AsyncQueue<T> = new AsyncQueue();

  constructor({
    factory,
  }: { factory: ResourceFactory<T, U> }) {
    this.factory = factory;
  }

  async use<O>(task: (resource: U) => Promise<O>, waitForDispose = false): Promise<O> {
    const resource = await this.resourceQueue.dequeue();
    const accessor = this.factory.access ?? (resource => resource);
    const accessed = await accessor(resource) as U;
    const result = await task(accessed);
    
    if (this.currentSize > this.targetSize) {
      // Get rid of excess resources
      const diposing = this.factory.dispose?.(resource);
      if (waitForDispose) await diposing;
      this.currentSize--;
    } else {
      this.resourceQueue.enqueue(resource);
    }

    return result;
  }

  private resizing = false;

  async resize(newSize: number) {
    // prevent conflicting resizes
    if (this.resizing) return;
    this.resizing = true;

    this.targetSize = newSize;

    while (newSize > this.currentSize) {
      this.currentSize++;
      const next = await this.factory.create();
      this.resourceQueue.enqueue(next);
    }

    
    while (newSize < this.currentSize) {
      this.use(async () => {}, true)
    }
  }
}

async function createPool<T, U = T>(factory: ResourceFactory<T, U>, size: number = 8) {
  const pooler = new ResourcePooler<T, U>({
    factory,
  })
  pooler.resize(size);
  return pooler;
}

I wrote some consumer code to test it:

let counter = 0;

class Foo {
  id: number;
  constructor() {
    this.id = ++counter;
  }
  getReady() {
    return new Promise<void>(res => {
      setTimeout(() =>{ 
        res()
        }, 1);
    })
  }
}

async function main() {
   const pooler = await createPool({
    create: async () => new Foo()
   });
   for (let x = 0; x < 200; x++) {
    pooler.use(foo => new Promise<number>(res => {
      console.log("Doing some work with", foo.id);
      setTimeout(() => res(foo.id), 1000*Math.random());
    })).then(result => console.log(result))
   }
}

main();

It seems to behave correctly. I wonder if I could implement any of this better; in particular the pool resizing.

\$\endgroup\$
4
  • \$\begingroup\$ What exactly is the purpose of dequeue()? It appears to just return a promise that resolves to undefined. \$\endgroup\$
    – jfriend00
    Commented May 5, 2023 at 0:54
  • \$\begingroup\$ dequeue() gets the first resource from the elements array if available, OR it returns a promise which will be resolved to the next resource when it becomes available. The Promise is of type T. It never returns undefined. \$\endgroup\$ Commented May 5, 2023 at 6:28
  • \$\begingroup\$ I know what the code for dequeue() does. I'm asking why it's a method? What is it used for? In what circumstance is it useful? \$\endgroup\$
    – jfriend00
    Commented May 5, 2023 at 6:35
  • \$\begingroup\$ I use it in ResourcePooler to access the resources because it guarantees two things: 1) no two invocations of use will access the same resource at the same time 2) each invocation of use is guaranteed to eventually receive access to a resource (in FIFO order). I initially had the logic governing this in ResourcePooler itself, but it seemed natural to separate the two concerns. \$\endgroup\$ Commented May 5, 2023 at 6:40

0

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Browse other questions tagged or ask your own question.