Class Semaphore

packages/effect/src/internal/effect/circular.ts:36–126

@internal

/** @internal */
class Semaphore {
  public waiters = new Set<() => void>()
  public taken = 0

  constructor(public permits: number) {}

  get free() {
    return this.permits - this.taken
  }

  readonly take = (n: number): Effect.Effect<number> =>
    core.asyncInterrupt<number>((resume) => {
      if (this.free < n) {
        const observer = () => {
          if (this.free < n) return
          this.waiters.delete(observer)
          resume(core.suspend(() => {
            if (this.free < n) return this.take(n)
            this.taken += n
            return core.succeed(n)
          }))
        }
        this.waiters.add(observer)
        return core.sync(() => {
          this.waiters.delete(observer)
        })
      }
      resume(core.suspend(() => {
        if (this.free < n) return this.take(n)
        this.taken += n
        return core.succeed(n)
      }))
    })

  updateTakenUnsafe(fiber: Fiber.RuntimeFiber<any, any>, f: (n: number) => number): Effect.Effect<number> {
    this.taken = f(this.taken)
    if (this.waiters.size > 0) {
      fiber.getFiberRef(currentScheduler).scheduleTask(
        () => {
          const iter = this.waiters.values()
          let item = iter.next()
          while (item.done === false && this.free > 0) {
            item.value()
            item = iter.next()
          }
        },
        fiber.getFiberRef(core.currentSchedulingPriority),
        fiber
      )
    }
    return core.succeed(this.free)
  }

  updateTaken(f: (n: number) => number): Effect.Effect<number> {
    return core.withFiberRuntime((fiber) => this.updateTakenUnsafe(fiber, f))
  }

  readonly resize = (permits: number) =>