FiberMap

__init__() -> None

def __init__(self) -> None:
    self._state: _Open[K, A, E] | None = _Open()  # None once the scope has closed
    # Completed by the first reported failure, or when the scope closes.
    self.deferred: Deferred[None, E] = Deferred()

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

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

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

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

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

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

unsafe_set(key, fiber, *, only_if_missing=False, propagate_interruption=False) -> None

def unsafe_set(
    self,
    key: K,
    fiber: Fiber[A, E],
    *,
    only_if_missing: bool = False,
    propagate_interruption: bool = False,
) -> None:
    """Put a running fiber under ``key``, interrupting the one there
    (or, with ``only_if_missing``, interrupting ``fiber`` instead). Into
    a closed map the fiber is interrupted. Call on the scheduler
    thread."""
    state = self._state
    if state is None:
        fiber.unsafe_interrupt_as(INTERNAL_FIBER_ID)
        return
    entry = state.by_key.get(key)
    if entry is not None and entry.fiber is fiber:
        return
    if entry is not None and only_if_missing:
        fiber.unsafe_interrupt_as(INTERNAL_FIBER_ID)
        return
    state.owned[fiber] = None
    if entry is None:
        state.by_key[key] = _Entry(fiber)
    else:
        # Install the replacement before the interrupt can re-enter the
        # map through a finalizer.
        previous, entry.fiber = entry.fiber, fiber
        previous.unsafe_interrupt_as(INTERNAL_FIBER_ID)

    def observer(ex: Exit[A, E]) -> None:
        state = self._state
        if state is None:
            return
        state.owned.pop(fiber, None)
        entry = state.by_key.get(key)
        if entry is not None and entry.fiber is fiber:
            del state.by_key[key]
        if reports_failure(ex, propagate_interruption):
            self.deferred.unsafe_done(ex)

    fiber.unsafe_add_observer(observer)

set(key, fiber, *, only_if_missing=False, propagate_interruption=False) -> Effect[None]

def set(
    self,
    key: K,
    fiber: Fiber[A, E],
    *,
    only_if_missing: bool = False,
    propagate_interruption: bool = False,
) -> Effect[None]:
    return _Sync(
        lambda: self.unsafe_set(
            key,
            fiber,
            only_if_missing=only_if_missing,
            propagate_interruption=propagate_interruption,
        )
    )

unsafe_get(key) -> Fiber[A, E] | None

def unsafe_get(self, key: K) -> Fiber[A, E] | None:
    entry = None if self._state is None else self._state.by_key.get(key)
    return None if entry is None else entry.fiber

get(key) -> Effect[Fiber[A, E] | None]

def get(self, key: K) -> Effect[Fiber[A, E] | None]:
    return _Sync(lambda: self.unsafe_get(key))

unsafe_has(key) -> bool

def unsafe_has(self, key: K) -> bool:
    return self._state is not None and key in self._state.by_key

has(key) -> Effect[bool]

def has(self, key: K) -> Effect[bool]:
    return _Sync(lambda: self.unsafe_has(key))

remove(key) -> Effect[None]

def remove(self, key: K) -> Effect[None]:
    """Interrupt the fiber under ``key``, if any, and wait for it."""

    def body() -> Effect[None]:
        fiber = self.unsafe_get(key)
        return _Succeed(None) if fiber is None else fiber.interrupt_as(INTERNAL_FIBER_ID)

    return _Suspend(body)

clear() -> Effect[None]

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

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

    return _Suspend(body)

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

def run[R = Never](
    self,
    key: K,
    eff: Effect[A, E, R],
    *,
    only_if_missing: bool = False,
    propagate_interruption: bool = False,
    start_immediately: bool = True,
) -> Effect[Fiber[A, E], Never, R]:
    """Fork ``eff`` under ``key``. It starts at once unless
    ``start_immediately=False``. With ``only_if_missing`` and a fiber
    already there, gives back an interrupted fiber that never runs
    ``eff``. Running into a closed map interrupts the caller."""
    require_effect(eff, "eff")

    def body(parent: Fiber[Any, Any]) -> Effect[Fiber[A, E]]:
        if self._state is None:
            return interrupt()
        if only_if_missing and self.unsafe_has(key):
            return _Succeed(interrupted_fiber(parent._run_scheduler, parent._context))
        fiber: Fiber[A, E] = parent._fork(eff, daemon=True, start_immediately=start_immediately)
        self.unsafe_set(
            key,
            fiber,
            only_if_missing=only_if_missing,
            propagate_interruption=propagate_interruption,
        )
        return _AfterStarted(fiber)

    return _WithFiber(body)

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

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

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

def runtime_async[R = Never](
    self, *tags: TagLike[Any, R]
) -> Effect[FiberMapAsyncRuntime[K, 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(FiberMapAsyncRuntime)

size() -> Effect[int]

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

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 no key holds a fiber (or the map is closed)."""

    def loop():
        while self._state is not None and self._state.by_key:
            yield from next(iter(self._state.by_key.values())).fiber.await_exit()

    return _Gen(loop)