async-semaphore.ts

/** 
 * Asynchronous Semaphore, made to work with Promises.
 * 
 * ```
 * const delay = (ms: number) =>
 *   new Promise<void>(_ => setTimeout(_, ms))
 * 
 * const semaphore = new AsyncSemaphore(10)
 * const promises: Promise<void>[] = []

 * for (let i = 0; i < 100; i++) {
 *   semaphore.withLock(() => {
 *     return delay(1000)
 *       .then(() => console.log(`Done: ${i}`))
 *   })
 * }
 * ```
 */
class AsyncSemaphore {
  private _available: number
  private _upcoming: Function[]
  private _heads: Function[]

  constructor(public workersCount: number) {
    if (workersCount <= 0) throw new Error("workersCount must be positive")
    this._available = workersCount
    this._upcoming = []
    this._heads = []
  }

  async withLock<A>(f: () => Promise<A>): Promise<A> {
    await this._acquire()
    try {
      return await f()
    } finally {
      this._release()
    }
  }

  private _queue(): Function[] {
    if (!this._heads.length) {
      this._heads = this._upcoming.reverse()
      this._upcoming = []
    }
    return this._heads
  }

  private _acquire(): void | Promise<void> {
    if (this._available > 0) {
      this._available -= 1
      return undefined
    } else {
      let fn: Function = () => {}
      const p = new Promise<void>(ref => { fn = ref })
      this._upcoming.push(fn)
      return p
    }
  }

  private _release() {
    const queue = this._queue()
    if (queue.length) {
      const fn = queue.pop()
      if (fn) fn()
    } else {
      this._available += 1
    }
  }
}