Pool

__init__(acquire, context, clock, config, state) -> None

def __init__(
    self,
    acquire: Effect[A, E, Any],
    context: Context,
    clock: ClockService,
    config: PoolConfig[A, E],
    state: PoolState[A, E],
) -> None:
    self._acquire = acquire
    self._context = context
    self._clock = clock
    self.config = config
    self.state = state

make(acquire, *, size, concurrency=1, target_utilization=1.0) -> Effect[Pool[T, X], Never, Scope | R]

@staticmethod
def make[T, X = Never, R = Never](
    acquire: Effect[T, X, R],
    *,
    size: int,
    concurrency: int = 1,
    target_utilization: float = 1.0,
) -> Effect[Pool[T, X], Never, Scope | R]:
    """A pool of ``size`` items, acquired in the background at once.
    Each item serves up to ``concurrency`` borrowers at a time."""
    return Pool.make_with_strategy(
        acquire,
        min=size,
        max=size,
        strategy=_NOOP,
        concurrency=concurrency,
        target_utilization=target_utilization,
    )

make_with_ttl(acquire, *, min, max, time_to_live, time_to_live_strategy='usage', concurrency=1, target_utilization=1.0) -> Effect[Pool[T, X], Never, Scope | R]

@staticmethod
def make_with_ttl[T, X = Never, R = Never](
    acquire: Effect[T, X, R],
    *,
    min: int,
    max: int,
    time_to_live: Duration,
    time_to_live_strategy: TimeToLiveStrategy = "usage",
    concurrency: int = 1,
    target_utilization: float = 1.0,
) -> Effect[Pool[T, X], Never, Scope | R]:
    """A pool between ``min`` and ``max`` items, growing with demand.
    Items above ``min`` are released once they have been idle
    (``"usage"``) or alive (``"creation"``) for ``time_to_live``."""
    from pyfect.pool._ttl import creation_ttl, usage_ttl

    require_effect(acquire, "acquire")
    _check_sizes(min, max, concurrency)
    ttl = to_seconds(time_to_live)
    make_strategy = creation_ttl if time_to_live_strategy == "creation" else usage_ttl
    strategy: Effect[Strategy[T, X]] = make_strategy(ttl)
    return strategy.flat_map(
        lambda s: Pool.make_with_strategy(
            acquire,
            min=min,
            max=max,
            strategy=s,
            concurrency=concurrency,
            target_utilization=target_utilization,
        )
    )

make_with_strategy(acquire, *, min, max, strategy, concurrency=1, target_utilization=1.0) -> Effect[Pool[T, X], Never, Scope | R]

@staticmethod
def make_with_strategy[T, X = Never, R = Never](
    acquire: Effect[T, X, R],
    *,
    min: int,
    max: int,
    strategy: Strategy[T, X],
    concurrency: int = 1,
    target_utilization: float = 1.0,
) -> Effect[Pool[T, X], Never, Scope | R]:
    """A pool between ``min`` and ``max`` items, resized by
    ``strategy``."""
    require_effect(acquire, "acquire")
    _check_sizes(min, max, concurrency)
    bounds = (min, max)

    def body(restore: Restore) -> Effect[Pool[T, X]]:
        def with_fiber(fiber: Fiber[Any, Any]) -> Effect[Pool[T, X]]:
            context = fiber._context
            scope = context.unsafe_get(Scope)
            config = PoolConfig(
                acquire=_Locally(acquire, context.merge),
                concurrency=concurrency,
                is_fixed=bounds[0] == bounds[1],
                min_size=bounds[0],
                max_size=bounds[1],
                strategy=strategy,
                target_utilization=_clamp(target_utilization, 0.1, 1.0),
            )
            state: PoolState[T, X] = PoolState(scope, Semaphore(1))
            pool = Pool(acquire, context, context.lookup(Clock)[1], config, state)
            return _Gen(lambda: pool._start(scope, restore))

        return _WithFiber(with_fiber)

    return uninterruptible_mask(body)

get() -> Effect[A, E, Scope]

def get(self) -> Effect[A, E, Scope]:
    """Borrow an item until the current scope closes, waiting for one
    if none is free. Fails with the acquisition's error if acquiring
    the item failed; interrupted once the pool is shutting down."""

    def body(fiber: Fiber[Any, Any]) -> Effect[A, E, Scope]:
        fast = self._fast_item()
        if fast is not None:
            item, value = fast
            return self._lease_scoped(item, value, fiber)
        return self._get_slow(
            lambda item, value, fiber, _restore: self._lease_scoped(item, value, fiber)
        )

    return _WithFiber(body)

use(f) -> Effect[B, E | E2, R2]

def use[B, E2 = Never, R2 = Never](
    self, f: Callable[[A], Effect[B, E2, R2]]
) -> Effect[B, E | E2, R2]:
    """Borrow an item for as long as ``f(item)`` runs."""

    def body(fiber: Fiber[Any, Any]) -> Effect[B, E | E2, R2]:
        fast = self._fast_item()
        if fast is not None:
            item, value = fast
            return self._use_item(item, value, f, fiber, None)
        return self._get_slow(
            lambda item, value, fiber, restore: self._use_item(item, value, f, fiber, restore)
        )

    return _WithFiber(body)

invalidate(item) -> Effect[None]

def invalidate(self, item: A) -> Effect[None]:
    """Release ``item`` once nobody borrows it, and acquire a
    replacement. The item is found by identity (``is``), as Effect's
    ``===``."""

    def body() -> Effect[None]:
        if self.state.is_shutting_down:
            return _VOID
        pool_item = self._find(item)
        if pool_item is None:
            return _VOID
        pool_item.disable_reclaim = True
        return self._invalidate_item(pool_item).uninterruptible()

    return _Suspend(body)

reserve(item) -> Effect[None, Never, Scope]

def reserve(self, item: A) -> Effect[None, Never, Scope]:
    """Keep ``item`` from new borrowers until the current scope closes,
    counting it as fully used, without extending existing leases. A
    no-op when each item serves one borrower. The item is found by
    identity (``is``)."""
    state = self.state
    concurrency = self.config.concurrency

    def take() -> PoolItem[A, E] | None:
        if concurrency == 1:
            return None
        pool_item = self._find(item)
        if pool_item is None:
            return None
        if pool_item.reservations == 0:
            state.usage += concurrency - 1
        pool_item.reservations += 1
        state._remove_available(pool_item)
        return pool_item

    def give_back(pool_item: PoolItem[A, E] | None) -> Effect[None]:
        def run(fiber: Fiber[Any, Any]) -> Effect[None]:
            if pool_item is None:
                return _VOID
            pool_item.reservations -= 1
            if pool_item.reservations > 0:
                return _VOID
            state.usage -= concurrency - 1
            if (
                not state.is_shutting_down
                and pool_item in state.items
                and pool_item not in state.invalidated
                and pool_item.ref_count < concurrency
            ):
                state._add_available_front(pool_item)
                self._wake_waiters(fiber, concurrency - pool_item.ref_count)
            return _VOID

        return _WithFiber(run)

    return acquire_release(_Sync(take), lambda pool_item, _: give_back(pool_item)).as_void()