FiberSet

__init__() -> None

def __init__(self) -> None:
    # None once the scope has closed. A dict keeps insertion order.
    self._fibers: dict[Fiber[A, E], None] | None = {}
    # Completed by the first reported failure, or when the scope closes.
    self.deferred: Deferred[None, E] = Deferred()

make() -> Effect[FiberSet[A, E], Never, Scope]

@classmethod
def make(cls) -> Effect[FiberSet[A, E], Never, Scope]:
    """A set whose fibers are interrupted when the current scope closes."""
    return acquire_release(_Sync(cls), lambda s, _: s._close())

make_runtime(*tags) -> Effect[FiberSetRuntime[A, E, R], Never, Scope | R]

@classmethod
def make_runtime[R = Never](
    cls, *tags: TagLike[Any, R]
) -> Effect[FiberSetRuntime[A, E, R], Never, Scope | R]:
    """``make`` and ``runtime`` in one."""
    return cls.make().flat_map(lambda s: s.runtime(*tags))

make_runtime_async(*tags) -> Effect[FiberSetAsyncRuntime[A, E, R], Never, Scope | R]

@classmethod
def make_runtime_async[R = Never](
    cls, *tags: TagLike[Any, R]
) -> Effect[FiberSetAsyncRuntime[A, E, R], Never, Scope | R]:
    """``make`` and ``runtime_async`` in one."""
    return cls.make().flat_map(lambda s: s.runtime_async(*tags))

unsafe_add(fiber, *, propagate_interruption=False) -> None

def unsafe_add(self, fiber: Fiber[A, E], *, propagate_interruption: bool = False) -> None:
    """Add a running fiber; it leaves the set when it completes. Into a
    closed set the fiber is interrupted. Call on the scheduler thread."""
    fibers = self._fibers
    if fibers is None:
        fiber.unsafe_interrupt_as(INTERNAL_FIBER_ID)
        return
    if fiber in fibers:
        return
    fibers[fiber] = None

    def observer(ex: Exit[A, E]) -> None:
        if self._fibers is None:
            return
        self._fibers.pop(fiber, None)
        if reports_failure(ex, propagate_interruption):
            self.deferred.unsafe_done(ex)

    fiber.unsafe_add_observer(observer)

add(fiber, *, propagate_interruption=False) -> Effect[None]

def add(self, fiber: Fiber[A, E], *, propagate_interruption: bool = False) -> Effect[None]:
    return _Sync(lambda: self.unsafe_add(fiber, propagate_interruption=propagate_interruption))

run(eff, *, propagate_interruption=False, start_immediately=True) -> Effect[Fiber[A, E], Never, R]

def run[R = Never](
    self,
    eff: Effect[A, E, R],
    *,
    propagate_interruption: bool = False,
    start_immediately: bool = True,
) -> Effect[Fiber[A, E], Never, R]:
    """Fork ``eff`` into the set. It starts at once unless
    ``start_immediately=False``. A closed set gives back an interrupted
    fiber that never runs ``eff``."""
    require_effect(eff, "eff")

    def body(parent: Fiber[Any, Any]) -> Effect[Fiber[A, E]]:
        if self._fibers is None:
            return _Succeed(interrupted_fiber(parent._run_scheduler, parent._context))
        fiber: Fiber[A, E] = parent._fork(eff, daemon=True, start_immediately=start_immediately)
        self.unsafe_add(fiber, propagate_interruption=propagate_interruption)
        return _AfterStarted(fiber)

    return _WithFiber(body)

runtime(*tags) -> Effect[FiberSetRuntime[A, E, R], Never, R]

def runtime[R = Never](
    self, *tags: TagLike[Any, R]
) -> Effect[FiberSetRuntime[A, E, R], Never, R]:
    """A runner that forks effects needing ``tags`` into the set, with
    the services of the fiber calling this."""
    return _WithFiber(lambda fiber: _Succeed(FiberSetRuntime(self, CapturedRuntime(fiber))))

runtime_async(*tags) -> Effect[FiberSetAsyncRuntime[A, E, R], Never, R]

def runtime_async[R = Never](
    self, *tags: TagLike[Any, R]
) -> Effect[FiberSetAsyncRuntime[A, E, R], Never, R]:
    """``runtime``, but the runner returns a future of the value (needs
    a running asyncio loop when called)."""
    return self.runtime(*tags).map(FiberSetAsyncRuntime)

clear() -> Effect[None]

def clear(self) -> Effect[None]:
    """Interrupt every fiber in the set and wait for them."""

    def body() -> Effect[None]:
        if self._fibers is None:
            return _Succeed(None)
        return Fiber.interrupt_all_as(
            (fiber for fiber, _ in live_items(self._fibers)), INTERNAL_FIBER_ID
        )

    return _Suspend(body)

size() -> Effect[int]

def size(self) -> Effect[int]:
    return _Sync(lambda: 0 if self._fibers is None else len(self._fibers))

join() -> Effect[None, E]

def join(self) -> Effect[None, E]:
    """Wait for the first reported failure and fail with it; succeed
    when the scope closes."""
    return self.deferred.wait()

await_empty() -> Effect[None]

def await_empty(self) -> Effect[None]:
    """Wait until the set is empty (or closed)."""

    def loop():
        while self._fibers:
            yield from next(iter(self._fibers)).await_exit()

    return _Gen(loop)