Fiber

__init__(run_scheduler, context, parent, *, interruptible=True) -> None

def __init__(
    self,
    run_scheduler: Scheduler,
    context: Context,
    parent: Fiber[Any, Any] | None,
    *,
    interruptible: bool = True,
) -> None:
    """A fiber of the run driven by ``run_scheduler``. It runs on
    ``context``'s ``CurrentScheduler``."""
    self.id = next(_fiber_ids)
    self._run_scheduler = run_scheduler
    self._parent = parent
    self._stack: list[Frame] = []
    self._interruptible = interruptible
    self._interrupt: Cause[Never] | None = None
    # Interrupts that arrived while a resume was queued (see _resume).
    self._interrupts_after_resume: list[tuple[int | None, Context | None]] = []
    # A dict, not a set: children are interrupted in fork order, as in Effect.
    self._children: dict[Fiber[Any, Any], None] = {}
    self._observers: list[Callable[[Exit[Any, Any]], None]] = []
    self._exit: Exit[Any, Any] | None = None
    self._waiting: _Wait | None = None
    self._phase: _Phase = "pending"
    self._set_context(context)
    if self._runtime_metrics is not None:
        self._runtime_metrics.record_fiber_start(self._ctx)

await_exit() -> Effect[Exit[A_co, E_co], Never, Never]

def await_exit(self) -> Effect[Exit[A_co, E_co], Never, Never]:
    """Wait for the fiber and succeed with its ``Exit``."""

    def register(resume: Callable[[Effect[Any, Any, Any]], None]) -> Effect[None] | None:
        if self._exit is not None:
            resume(_Succeed(self._exit))
            return None

        def observer(ex: Exit[Any, Any]) -> None:
            resume(_Succeed(ex))

        self._observers.append(observer)
        return _Sync(lambda: self._remove_observer(observer))

    return _Callback(register)

join() -> Effect[A_co, E_co, Never]

def join(self) -> Effect[A_co, E_co, Never]:
    """Wait for the fiber and take on its result: its value, or its
    failure."""
    return self.await_exit().flat_map(_FromExit)

interrupt() -> Effect[None, Never, Never]

def interrupt(self) -> Effect[None, Never, Never]:
    """Interrupt the fiber and wait until it has finished cleaning up.
    Read the outcome with ``await_exit``."""
    return _WithFiber(lambda me: self._interrupt_from(me, me.id))

interrupt_as(by) -> Effect[None, Never, Never]

def interrupt_as(self, by: int | None) -> Effect[None, Never, Never]:
    """``interrupt``, attributed to the fiber with id ``by``."""
    return _WithFiber(lambda me: self._interrupt_from(me, by))

run_in(scope) -> Fiber[A_co, E_co]

def run_in(self, scope: CloseableScope) -> Fiber[A_co, E_co]:
    """Tie the fiber to ``scope``: closing it interrupts the fiber, and
    the fiber leaves the scope when it exits. Registering again moves
    the fiber's finalizer to the end, so it runs first. A fiber in an
    already closed scope is interrupted. Call on the scheduler thread."""
    if self._exit is not None:
        return self
    if scope._closed is not None:
        self.unsafe_interrupt_as(self.id)
    else:
        scope._watch(self)
    return self

get_ref(reference) -> S

def get_ref[S](self, reference: type[Reference[S]]) -> S:
    """The value of ``reference`` in this fiber's current context (its
    default when nobody provided one). Loggers read the log annotations
    and spans with it."""
    return self._ctx.unsafe_get(reference)

get_current() -> Fiber[Any, Any] | None

@staticmethod
def get_current() -> Fiber[Any, Any] | None:
    """The fiber running on this thread, if any."""
    return current_fiber()

await_all(fibers) -> Effect[list[Exit[A, E]], Never, Never]

@staticmethod
def await_all[A, E](fibers: Iterable[Fiber[A, E]]) -> Effect[list[Exit[A, E]], Never, Never]:
    """Wait for every fiber; succeed with their exits in order."""
    from pyfect._fiber_all import await_all

    return await_all(fibers)

join_all(fibers) -> Effect[list[A], E, Never]

@staticmethod
def join_all[A, E](fibers: Iterable[Fiber[A, E]]) -> Effect[list[A], E, Never]:
    """Wait for every fiber and succeed with their values in order, or
    fail with the first failure (the others keep running)."""
    from pyfect._fiber_all import join_all

    return join_all(fibers)

interrupt_all(fibers) -> Effect[None, Never, Never]

@staticmethod
def interrupt_all(fibers: Iterable[Fiber[Any, Any]]) -> Effect[None, Never, Never]:
    """Interrupt every fiber and wait for all of them to finish."""
    from pyfect._fiber_all import interrupt_all

    return interrupt_all(fibers)

interrupt_all_as(fibers, by) -> Effect[None, Never, Never]

@staticmethod
def interrupt_all_as(
    fibers: Iterable[Fiber[Any, Any]], by: int | None
) -> Effect[None, Never, Never]:
    """``interrupt_all``, attributed to the fiber with id ``by``."""
    from pyfect._fiber_all import interrupt_all_as

    return interrupt_all_as(fibers, by)

unsafe_poll() -> Exit[A_co, E_co] | None

def unsafe_poll(self) -> Exit[A_co, E_co] | None:
    """The exit if the fiber is done. Only meaningful on the fiber's
    scheduler thread."""
    return self._exit

unsafe_interrupt_as(by, annotations=None) -> None

def unsafe_interrupt_as(self, by: int | None, annotations: Context | None = None) -> None:
    """Ask the fiber to stop. Safe from any thread; does not wait.

    From a fiber on the same scheduler it takes effect at once (as
    Effect's ``interruptUnsafe``); elsewhere it goes through the
    scheduler. The interruption carries ``annotations`` (the
    interrupting fiber's ``InterruptorStackTrace``) and the frame this
    fiber is in (``StackTrace``)."""
    if current_run_scheduler() is self._run_scheduler:
        self._deliver_interrupt(by, annotations)
    else:
        self._scheduler.schedule(lambda: self._deliver_interrupt(by, annotations))

unsafe_add_observer(observer) -> Callable[[], None]

def unsafe_add_observer(
    self, observer: Callable[[Exit[A_co, E_co]], None]
) -> Callable[[], None]:
    """Call ``observer`` with the exit once the fiber is done (at once if
    it already is). Returns a function that removes the observer. Must
    be called on the scheduler thread."""
    if self._exit is not None:
        observer(self._exit)
        return _no_op
    self._observers.append(observer)
    return lambda: self._remove_observer(observer)