PartitionedSemaphore

__init__(permits) -> None

def __init__(self, permits: int) -> None:
    self.capacity = max(0, permits)
    self._total = self.capacity  # free permits nobody is waiting for
    self._waiting = 0  # permits waiters still miss
    self._pending = 0  # released permits not yet handed out
    self._releasing = False
    # Typed by Hashable, not K, so K only goes in (contravariant, as in Effect).
    self._waiters: RoundRobin[Hashable, _Waiter] = RoundRobin()

make(permits) -> Effect[PartitionedSemaphore[K]]

@classmethod
def make(cls, permits: int) -> Effect[PartitionedSemaphore[K]]:
    """``PartitionedSemaphore[str].make(4)``"""
    return _Sync(lambda: cls(permits))

take(key, n=1) -> Effect[None]

def take(self, key: K, n: int = 1) -> Effect[None]:
    """Wait in ``key``'s queue for ``n`` permits and take them. Prefer
    ``with_permits``, which also gives them back."""
    if n <= 0:
        return _VOID
    if self.capacity < n:
        return never

    def run(fiber: Fiber[Any, Any]) -> Effect[None]:
        guard = _GiveBackOnFailure(lambda: self._release(n))
        fiber._push(guard)

        def arm(_: None) -> None:
            guard.armed = True

        return uninterruptible_mask(lambda restore: self._acquire(key, n, restore).map(arm))

    return _WithFiber(run)

release(n=1) -> Effect[int]

def release(self, n: int = 1) -> Effect[int]:
    """Give back ``n`` permits; succeeds with the number now free."""
    return _Sync(lambda: self._release(n))

with_permits(key, eff, n=1) -> Effect[A, E, R]

def with_permits[A, E, R](self, key: K, eff: Effect[A, E, R], n: int = 1) -> Effect[A, E, R]:
    """Run ``eff`` holding ``n`` permits taken in ``key``'s queue,
    releasing them however it ends."""
    require_effect(eff, "eff")
    if n <= 0:
        return eff
    if self.capacity < n:
        return never
    release = self.release(n)
    return uninterruptible_mask(
        lambda restore: self._acquire(key, n, restore).and_then(restore(eff).ensuring(release))
    )

with_permits_if_available(eff, n=1) -> Effect[Option[A], E, R]

def with_permits_if_available[A, E, R](
    self, eff: Effect[A, E, R], n: int = 1
) -> Effect[Option[A], E, R]:
    """Run ``eff`` holding ``n`` permits if they are free now (``Some``
    of its result); otherwise ``None`` without running it."""
    require_effect(eff, "eff")
    if n <= 0:
        return eff.map(Some)

    def body(restore: Restore) -> Effect[Option[A], E, R]:
        def run() -> Effect[Option[A], E, R]:
            if self._total < n:
                return _Succeed(None)
            self._total -= n
            return restore(eff.map(Some)).ensuring(self.release(n))

        return _Suspend(run)

    return uninterruptible_mask(body)