Class PoolImpl

packages/effect/src/internal/pool.ts:127–344

}

class PoolImpl<A, E> extends Effectable.Class<A, E, Scope> implements Pool<A, E> {
  readonly [PoolTypeId]: Pool.Variance<A, E>[PoolTypeId_]

  isShuttingDown = false
  readonly semaphore: Semaphore
  readonly items = new Set<PoolItem<A, E>>()
  readonly available = new Set<PoolItem<A, E>>()
  readonly availableLatch = circular.unsafeMakeLatch(false)
  readonly invalidated = new Set<PoolItem<A, E>>()
  waiters = 0

  constructor(
    readonly scope: Scope,
    readonly acquire: Effect<A, E, Scope>,
    readonly concurrency: number,
    readonly minSize: number,
    readonly maxSize: number,
    readonly strategy: Strategy<A, E>,
    readonly targetUtilization: number
  ) {
    super()
    this[PoolTypeId] = poolVariance
    this.semaphore = circular.unsafeMakeSemaphore(concurrency * maxSize)
  }

  readonly allocate: Effect<PoolItem<A, E>> = core.acquireUseRelease(
    fiberRuntime.scopeMake(),
    (scope) =>
      this.acquire.pipe(
        fiberRuntime.scopeExtend(scope),
        core.exit,
        core.flatMap((exit) => {
          const item: PoolItem<A, E> = {
            exit,
            finalizer: core.catchAllCause(scope.close(exit), reportUnhandledError),
            refCount: 0,
            disableReclaim: false
          }
          this.items.add(item)
          this.available.add(item)
          return core.as(
            exit._tag === "Success"
              ? this.strategy.onAcquire(item)
              : core.zipRight(item.finalizer, this.strategy.onAcquire(item)),
            item
          )
        })
      ),
    (scope, exit) => exit._tag === "Failure" ? scope.close(exit) : core.void
  )

  get currentUsage() {
    let count = this.waiters
    for (const item of this.items) {
      count += item.refCount
    }
    return count
  }