caller

Classes:

Caller –

A flexible interface for scheduling work across threads and asynchronous backends.
StartStopTask –

A class which provides start/stop functionality to run a coroutine function.

Caller ¶

A flexible interface for scheduling work across threads and asynchronous backends.

Each caller is associated with an asyncio or trio event loop and can dispatch functions from any thread. When a async_kernel.pending.Pending is returned from a caller method (call_soon, to_thread, etc), the result can be awaited or cancelled from anywhere.

CPython: There is one caller per thread.
Pyodide: Pyodide does not currently implement threads and trio is not supported. A context variable is used to provide the caller_id enabling multiple caller support.

In CPython multi-eventloop management is supported including:

host: zero or one host gui event loop.
backend: One backend (asyncio or trio) that can run as a guest in the host if a host is in use.
guest backend: The opposite of the backend that runs as a guest of the host or backend when there is no host.

A Caller instance can be created in any thread where a backend (asyncio or trio) is already running provided the Caller is instantiated inside the thread (using Caller()).

A new caller with its own managed thread can be created using one of:

Caller("NewThread"): A new caller instance that will run until it is stopped.
caller.get: As a child of an existing caller. The child will stop when the parent is stopped furthermore, the same child can be accessed by passing the string argument name.

An async context is provided for lifecycle management. The async-context of a caller can be entered multiple times from any thread. Once entered, the caller is marked as 'protected'. The first entered context will wait for all other contexts to exit and then stop itself, only leaving the context once it has stopped.

High level methods

Caller.call_soon: Schedule a function call in the caller's thread.
Caller.call_later: Schedule a function call in the caller's thread after a delay.
Caller.to_thread: Schedule a function call using a worker caller (separate thread).
Caller.call_using_backend: Schedule a function call using the backend in the caller's thread.
Caller.as_completed: An async iterator a collection of pending as they complete.
Caller.wait: A method to wait for a collection of pending to complete.
Caller.create_pending_group: Create a new pending group to use as an asynchronous context.
Caller.get: Get a new caller instance (child).

Low level methods

Caller.schedule_call: Schedule a function call in the caller's thread - configurable (used by high level methods).
Caller.call_direct: Call a function directly in the scheduler in the caller's thread.
Caller.queue_call: Execute a function in the caller's thread using a queue (sequential).
Caller.queue_get: Get the pending associated with the queue call.
Caller.queue_close: Close the queue associated with the function.
Caller.create_start_stop_task: Run a coroutine function in a Task.

Class methods

Caller.current_pending: Get the active pending in the current context.
Caller.id_current: Get the id of the caller in the current thread/context.
Caller.get_existing: Get the caller by id.
Caller.all_callers: Get a list of all callers.

Referenced by:

Reference
async-kernel How it works

Returned by:

Reference

Used by:

Reference

Methods:

__new__ –

Create or retrieve a caller.
id_current –

The id of a caller given the current thread in CPython or context in Pyodide.
get_existing –

A classmethod to get the caller instance if it exists.
current_pending –

A classmethod that returns the pending for the current context.
all_callers –

A classmethod to get a list of all callers.
stop –

Stop the caller.
get –

Get an existing caller by name or create a new one.
schedule_call –

A low-level method to schedule func to be run as a task running in the caller's backend event loop.
call_later –

Schedule func to be called in the caller's backend event loop with a delay using a copy of the current context.
call_soon –

Schedule func to be called in the caller's backend event loop using a copy of the current context.
call_using_backend –

Schedule func to be executed using the specified backend event loop using a copy of the current context.
call_direct –

A low-level method to schedule func to be executed directly in caller's backend scheduler (no context copy).
to_thread –

Call func in a child 'worker' that has the same type of backend as the caller (parent).
queue_get –

Returns the pending where with the queue_call is running.
queue_call –

A low-level method to queue the execution of func in a queue unique to it and the caller.
queue_close –

Close the Caller.queue_call execution queue associated with func.
create_start_stop_task –

Wrap the coroutine function func with a StartStopTask.
as_completed –

An async iterator which yields a pending for each awaitable in items as they complete.
wait –

Wait for one or more of the awaitable items to complete.
create_pending_group –

Create a new PendingGroup.

Attributes:

MAX_IDLE_POOL_INSTANCES –

The number of child workers to keep in the pool (See also to_thread).
IDLE_WORKER_SHUTDOWN_DURATION –

The duration in seconds for a worker to remain in the pool before it is shutdown.
started –

A ProtectedPending that is set when the caller has started.
stopping –

A ProtectedPending that is set done when the caller is stopping.
stopped –

A ProtectedPending that is set done when the caller has stopped.
name (str) –

The name of the caller.
id (int) –

The id of the caller, which is the thread's ident where threading is being used otherwise it is a unique value.
backend (Backend) –

The primary backend of the caller.
backend_options (dict | None) –

Options used to create the backend event loop when the caller is started.
host (Hosts | None) –

The name of the host if the caller is started with a host.
host_options (dict | None) –

Options used to create the host.
protected (bool) –

Returns True if the caller is protected from stopping.
children (set[Self]) –

The caller's children that are not stopping.
thread (Thread) –

The thread where the caller is running.
parent (Self | None) –

The caller's parent if it has one.

Source code in src/async_kernel/caller.py

@final
class Caller:
    """A flexible interface for scheduling work across threads and asynchronous backends.

    Each caller is associated with an asyncio or trio event loop and can dispatch functions
    from any thread. When a [async_kernel.pending.Pending][] is returned from a caller method
    (`call_soon`, `to_thread`, etc), the result can be awaited or cancelled from anywhere.

    - CPython: There is one caller per thread.
    - Pyodide: Pyodide does not currently implement threads and trio is not supported.
        A context variable is used to provide the `caller_id` enabling multiple caller support.

    In CPython multi-eventloop management is supported including:

    - host: zero or one host gui event loop.
    - backend: One backend (asyncio or trio) that can run as a guest in the host if a host is in use.
    - guest backend: The opposite of the backend that runs as a guest of the host or backend
        when there is no host.

    A `Caller` instance can be created in any thread where a backend (asyncio or trio)
    is already running provided the `Caller` is instantiated inside the thread (using `Caller()`).

    A new caller with its own managed thread can be created using one of:

    1. `Caller("NewThread")`: A new caller instance that will run until it is stopped.
    2. `caller.get`: As a child of an existing caller. The child will stop when the parent
        is stopped furthermore, the same child can be accessed by passing the string argument
        `name`.

    An async context is provided for lifecycle management. The async-context of a caller
    can be entered multiple times from any thread.  Once entered, the caller is marked as
    'protected'. The first entered context will wait for all other contexts to exit and then
    stop itself, only leaving the context once it has stopped.

    **High level methods**

    - [Caller.call_soon][]: Schedule a function call in the caller's thread.
    - [Caller.call_later][]: Schedule a function call in the caller's thread after a delay.
    - [Caller.to_thread][]: Schedule a function call using a worker caller (separate thread).
    - [Caller.call_using_backend][]: Schedule a function call using the backend in the caller's thread.
    - [Caller.as_completed][]: An async iterator a collection of pending as they complete.
    - [Caller.wait][]: A method to wait for a collection of pending to complete.
    - [Caller.create_pending_group][]: Create a new pending group to use as an asynchronous context.
    - [Caller.get][]: Get a new caller instance (child).

    **Low level methods**

    - [Caller.schedule_call][]: Schedule a function call in the caller's thread - configurable (used by high level methods).
    - [Caller.call_direct][]: Call a function directly in the scheduler in the caller's thread.
    - [Caller.queue_call][]: Execute a function in the caller's thread using a queue (sequential).
    - [Caller.queue_get][]: Get the pending associated with the queue call.
    - [Caller.queue_close][]: Close the queue associated with the function.
    - [Caller.create_start_stop_task][]: Run a coroutine function in a Task.

    **Class methods**

    - [Caller.current_pending][]: Get the active pending in the current context.
    - [Caller.id_current][]: Get the id of the caller in the current thread/context.
    - [Caller.get_existing][]: Get the caller by id.
    - [Caller.all_callers][]: Get a list of all callers.
    """

    MAX_IDLE_POOL_INSTANCES = 10
    "The number of child workers to keep in the pool (See also [to_thread][async_kernel.caller.Caller.to_thread])."

    IDLE_WORKER_SHUTDOWN_DURATION = 0 if "pytest" in sys.modules else 60
    """The duration in seconds for a worker to remain in the pool before it is shutdown.

    Set to 0 to disable (default when running tests).
    """

    CALLER_MAIN_THREAD_ID: int = int(threading.main_thread().ident)  # pyright: ignore[reportArgumentType]

    _caller_token = contextvars.ContextVar("caller_tokens", default=CALLER_MAIN_THREAD_ID)
    _instances: ClassVar[weakref.WeakValueDictionary[int, Self]] = weakref.WeakValueDictionary()
    _lock: ClassVar = threading.Lock()

    _thread: threading.Thread
    _caller_id: int
    _name: str
    _idle_time: float = 0.0
    _backend: Backend
    _backend_options: dict[str, Any] | None
    _host: Hosts | None
    _host_options: dict[str, Any] | None
    _protected = False
    _use_safe_checkpoint = False
    _state: CallerState = CallerState.initial
    _enter_count = 0
    _wait_exit = False
    _state_reprs: ClassVar[dict] = {
        CallerState.initial: "❗ not running",
        CallerState.start_sync: "pre-start",
        CallerState.starting: "starting",
        CallerState.running: "🏃 running",
        CallerState.stopping: "🏁 stopping",
        CallerState.stopped: "🏁 stopped",
    }

    # Fixed
    _worker_pool: Fixed[Self, deque[Self]] = Fixed(deque)

    # Private
    _inst_lock: threading.Lock
    _children: set[Self]
    _tasks: set[asyncio.Task]
    _queue_map: dict[int, Pending]
    _scheduler_queue: SingleAsyncQueue[Pending | tuple[Callable, tuple, dict]]
    _guest_queues: dict[Backend, SingleAsyncQueue[Pending | tuple[Callable, tuple, dict]]]
    _children_countdown: CountdownEvent
    _pen_stop: list[Pending]

    started = Fixed(ProtectedPending)
    """A `ProtectedPending` that is set when the caller has started."""

    stopping = Fixed(ProtectedPending)
    """A `ProtectedPending` that is set done when the caller is stopping."""

    stopped = Fixed(ProtectedPending)
    """A `ProtectedPending` that is set done when the caller has stopped."""

    _pending_var: contextvars.ContextVar[Pending | None] = contextvars.ContextVar("_pending_var", default=None)

    log: logging.LoggerAdapter

    @property
    def name(self) -> str:
        """The name of the caller."""
        return self._name

    @property
    def id(self) -> int:
        """The id of the caller, which is the thread's ident where threading is being used otherwise it is a unique value."""
        return self._caller_id

    @property
    def backend(self) -> Backend:
        """The primary backend of the caller."""
        return self._backend

    @property
    def backend_options(self) -> dict | None:
        """Options used to create the backend event loop when the caller is started."""
        return self._backend_options

    @property
    def host(self) -> Hosts | None:
        """The [name][async_kernel.typing.Hosts] of the host if the caller is started with a host."""
        return self._host

    @property
    def host_options(self) -> dict | None:
        """Options used to create the host."""
        return self._host_options

    @property
    def protected(self) -> bool:
        """Returns `True` if the caller is protected from stopping.

        This can be set to `True` at any time.
        """
        return self._protected

    @protected.setter
    def protected(self, value=True) -> None:
        assert value is True
        self._protected = True

    @property
    def children(self) -> set[Self]:
        """The caller's children that are not stopping.

        Notes:
            - When the parent is stopped, all children are stopped.
            - All children are stopped before the parent's stopped `ProtectedPending` is set as done.
        """
        return {c for c in self._children.copy() if not c.stopping.done()}

    @property
    def thread(self) -> threading.Thread:
        """The thread where the caller is running."""
        return self._thread

    @property
    def parent(self) -> Self | None:
        """The caller's parent if it has one."""
        return self._parent_ref()

    def _parent_ref(self) -> Self | None:
        return None

    def _get_info(self) -> dict[str, Any]:
        if self._state.value < CallerState.running.value:
            return {"name": self._name, "backend": str(self._backend), "host": self._host}
        return {
            "name": self._name,
            "backend": str(self._backend),
            "host": self._host,
            "id": self._caller_id,
            "parent": self.parent.id if self.parent else None,
        }

    @override
    def __repr__(self) -> str:
        info = " ".join(f"{k}={v!r}" for k, v in self._get_info().items())
        current = "⚪" if self.id_current() == getattr(self, "_caller_id", None) else "⚫"
        protected = "🔐 " if self.protected else " "
        n = len(self._children)
        children = "" if not n else (" 1 child" if n == 1 else f" {n} children")
        if self._wait_exit and self._enter_count:
            info = f"{info} waiting context {self._enter_count} remain"
        return f"<Caller {current}{self._state_reprs[self._state]}{protected}{info}{children}>"

    def __new__(
        cls,
        modifier: Literal["CurrentThread", "MainThread", "NewThread"] = "CurrentThread",
        /,
        **kwargs: Unpack[CallerCreateOptions],
    ) -> Self:
        """Create or retrieve a caller.

        Args:
            modifier: Specifies the caller instance to retrieve.

                - CurrentThread: The caller for the current thread.
                - MainThread: The Caller associated with the main thread.
                - NewThread: Create a caller with a new thread. The lifecycle is not
                    It is recommended to used this as an async context.
                    [Caller.get][] and [Caller.to_thread][] are preferred alternatives.

            **kwargs: Additional options for Caller creation, such as:
                - name: The name to use.
                - backend: The async backend to use.
                - backend_options: Options for the backend.

        Returns:
            Self: The created or retrieved Caller instance.

        Raises:
            RuntimeError: If the backend is not provided and backend can't be determined.
            ValueError: If the thread and caller's name do not match.
        """
        if modifier != "NewThread":
            cls._lock.acquire()
        try:
            name, backend = kwargs.get("name", ""), kwargs.get("backend")
            match modifier:
                case "MainThread":
                    caller_id, thread = cls.CALLER_MAIN_THREAD_ID, threading.main_thread()
                case "NewThread":
                    caller_id, thread = None, None
                case _:
                    caller_id, thread = cls.id_current(), threading.current_thread()

            # Locate existing
            if caller_id is not None and (caller := cls._instances.get(caller_id)):
                if name and name != caller.name:
                    msg = f"The thread and caller's name do not match! {name=} {caller=}"
                    raise ValueError(msg)
                if backend and backend != caller.backend:
                    msg = f"The backend does not match! {backend=} {caller.backend=}"
                    raise ValueError(msg)
                return caller

            host = Hosts(host) if (host := kwargs.get("host")) else None
            if (backend := Backend(backend or current_async_library())) is Backend.trio:
                trio.sleep  # noqa: B018 # Check trio is available.

            # create a new instance
            inst = super().__new__(cls)
            if sys.platform == "emscripten":
                if caller_id is None:
                    caller_id = id(inst)
                thread = threading.current_thread()

            # Add private objects
            inst._inst_lock = threading.Lock()
            inst._children = set()
            inst._tasks = set()
            inst._queue_map = {}
            inst._scheduler_queue = SingleAsyncQueue(reject=inst._reject)
            inst._guest_queues = {}
            inst._children_countdown = CountdownEvent()
            inst._pen_stop = []

            # Apply settings
            inst._name = name
            inst._backend = backend
            inst._host = host
            inst._backend_options = kwargs.get("backend_options")
            inst._host_options = kwargs.get("host_options")
            inst.log = logging.LoggerAdapter(logging.getLogger())

            # Start the instance.
            inst._start((caller_id, thread) if caller_id and thread else None, kwargs.get("no_debug", False))
            assert inst._caller_id is not None
            assert inst._caller_id not in cls._instances
            cls._instances[inst._caller_id] = inst
            return inst
        finally:
            if modifier != "NewThread":
                cls._lock.release()

    async def __aenter__(self) -> Self:
        self._protected = True
        await self.started.wait(result=False)
        ctx = utils.get_context_hash()
        with self._inst_lock:
            if not (stopping := self.stopping.done()):
                if self._enter_count == 0:
                    # Identify the first context.
                    self._first_ctx_and_count = ctx, 0
                    # Special handling if the context is run inside a caller managed task.
                    self._first_ctx_stop = bool(self.current_pending())
                elif (c := self._first_ctx_and_count)[0] == ctx:
                    # The same context, re-entry
                    self._first_ctx_and_count = c[0], c[1] + 1
                elif (pen := self.current_pending()) is not None:
                    # A different context
                    self._pen_stop.append(pen)
                self._enter_count = self._enter_count + 1
        if stopping:
            await self.stopped
            msg = f"The caller is stopped {self}"
            raise RuntimeError(msg)
        return self

    async def __aexit__(self, type, value, traceback) -> Literal[False]:
        ctx = utils.get_context_hash()
        with self._inst_lock:
            self._enter_count = self._enter_count - 1
            if (c := self._first_ctx_and_count)[0] == ctx:
                self._first_ctx_and_count = c[0], c[1] - 1
            elif (pen := self.current_pending()) and pen in self._pen_stop:
                self._pen_stop.remove(pen)
            wait_exit = c == (ctx, 0)
        if self._enter_count == 0:
            self.stop(force=True)
        if wait_exit:
            self._wait_exit = True
            if self._first_ctx_stop:
                for pen in self._pen_stop:
                    pen.cancel("At exit force shutdown")
                await self.wait(self._pen_stop, return_when="ALL_COMPLETED")
                self.stop(force=True)
            if (self.current_pending() is None) or self.get_existing() is not self:
                self.log.debug("Waiting for %r", self)
                await self.stopped.wait(shield=True)
        return False

    def _start(self, caller_id_thread: tuple[int, threading.Thread] | None, no_debug: bool = False) -> None:
        """Start the scheduler."""
        self._set_state(CallerState.start_sync)

        async def run_scheduler() -> None:
            if self._set_state(CallerState.starting) is not CallerState.start_sync:
                return
            if not self._name:
                self._name = self._thread.name
            if no_debug:
                utils.mark_thread_pydev_do_not_trace()
            try:
                async with task_factory() as create_task:
                    create_task(contextvars.Context(), self._scheduler, self._scheduler_queue)
                    self._set_state(CallerState.running)
                    await self.stopping
                    await self.wait(self._pen_stop, return_when="ALL_COMPLETED")
                    await self._children_countdown
                    await async_checkpoint(force=True)
                    self._scheduler_queue.stop()
                    await async_checkpoint(force=True)
            finally:
                self._set_state(CallerState.stopping)
                self._set_state(CallerState.stopped)

        if caller_id_thread:
            assert self.host is None, "host is not valid"
            self._caller_id, self._thread = caller_id_thread

            if self.backend == Backend.asyncio:
                self._tasks.add(asyncio.create_task(run_scheduler()))
            else:
                # Use another thread to schedule a trio Task
                trio_token = trio.lowlevel.current_trio_token()

                def to_thread() -> None:
                    utils.mark_thread_pydev_do_not_trace()
                    try:
                        trio.from_thread.run(run_scheduler, trio_token=trio_token)
                    except (BaseExceptionGroup, BaseException) as e:  # pragma: no cover
                        if not "shutdown" not in str(e):
                            raise

                threading.Thread(target=to_thread).start()
        else:

            def async_kernel_caller() -> None:
                self._thread, self._caller_id = threading.current_thread(), threading.get_ident()
                settings = RunSettings(
                    backend=self.backend,
                    host=self.host,
                    backend_options=self.backend_options,
                    host_options=self.host_options,
                )
                async_kernel.event_loop.run(run_scheduler, (), settings)

            thread = threading.Thread(target=async_kernel_caller, name=self._name or None)
            if no_debug:
                utils.mark_thread_pydev_do_not_trace(thread)
            thread.start()
            assert thread.ident
            self._thread, self._caller_id = thread, thread.ident

    def _set_state(self, state: CallerState) -> CallerState:
        with self._inst_lock:
            old_state = self._state
            if state.value > old_state.value:
                self._state = state
                self.log.debug("%s %s", state.name.capitalize(), self)
                match state:
                    case CallerState.running:
                        self.started.set_result(None)
                    case CallerState.stopping:
                        self.started.cancel("Stopping")
                        for child in self._children.copy():
                            child.stop(force=True)
                        # Shutdown queue_call
                        for func in self._queue_map.copy():
                            self.queue_close(func)
                    case CallerState.stopped:
                        self._children_countdown.wait()
                        self._instances.pop(self._caller_id)
                        self._scheduler_queue.stop()
                        self.stopped.set_result(None)
                        if parent := self.parent:
                            parent._children.discard(self)
                            parent._children_countdown.down()
                    case _:
                        pass
        return old_state

    async def _scheduler(self, queue: SingleAsyncQueue[tuple[Callable, tuple, dict] | Pending]) -> None:
        """A function that async iterates the queue scheduling execution of items as they arrive.

        It handles two types of items:
            - tuple: A tuple of `func`, `args`, `kwargs` which is called directly in the scheduler.
            - Pending: A pending to start as a 'task'. The pending must include `func`,
                `args` and `kwargs` as metadata. The pending's context is used to start the task,
                and the pending is set as the `active_pending` for that context.

        Args:
            queue: The queue to access the items for scheduling.
        """
        backend = Backend(current_async_library())

        async def run_pending_function(pen: Pending[Any]) -> None:
            if pen.done():
                return  # pragma: no cover
            md = pen.metadata
            token_pending = self._pending_var.set(pen)
            token_ident = self._caller_token.set(self._caller_id)
            e = None
            try:
                result = md["func"](*md["args"], **md["kwargs"])
                if iscoroutinelike(result):
                    if backend is Backend.asyncio:
                        task = asyncio.current_task()
                        assert task
                        pen.set_canceller(lambda msg: queue.append((task.cancel, (msg,), {})))
                        try:
                            pen.set_result(await result)
                        except asyncio.CancelledError:
                            pen.cancel("Task was cancelled")
                            raise
                    else:
                        with trio.CancelScope() as scope:
                            pen.set_canceller(lambda msg: queue.append((scope.cancel, (msg,), {})))
                            try:
                                pen.set_result(await result)
                            except trio.Cancelled:
                                pen.cancel("Task was cancelled")
                                raise
                else:
                    pen.set_result(result)
            except (Exception, KernelInterrupt) as exc:
                pen.set_exception(exc)
            except BaseException as exc:
                e = exc
            finally:
                if not pen.done():
                    pen.cancel("Unable to finish execution")
                    pen.set_result(None)
                del pen
                self._pending_var.reset(token_pending)
                self._caller_token.reset(token_ident)
                if e:
                    raise e from None

        if not queue.stopped:
            async with task_factory() as create_task:
                async for item in queue:
                    if isinstance(item, Pending):
                        if not item.done():
                            create_task(item.context, run_pending_function, item)
                    else:
                        try:
                            item[0](*item[1], **item[2])
                        except Exception as e:
                            self.log.exception("Direct call failed func:%s args:%s kwargs:%s", *item, exc_info=e)
                    del item

    @staticmethod
    def _reject(item: tuple | Pending) -> None:
        if isinstance(item, Pending):
            item.cancel("The caller has been closed")

    async def _idle_worker_cleanup(self) -> None:
        """Periodically check the worker pool and stop workers when their idle duration is reached."""
        while (t := self.IDLE_WORKER_SHUTDOWN_DURATION) and self._worker_pool:
            await anyio.sleep(t)
            for worker in self._worker_pool.copy():
                if (time.monotonic() - worker._idle_time) > t:
                    worker.stop(force=True)

    def _start_guest(self, backend: Backend, queue: SingleAsyncQueue[tuple[Callable, tuple, dict] | Pending]) -> None:
        """Start a guest backend."""
        assert self.get_existing() is self
        # Thread: caller

        def guest_done_callback(value: Any):
            with self._inst_lock:
                self._guest_queues.pop(backend)
                self._pen_stop.remove(pen)
            pen.set_result(None)

        with self._inst_lock:
            if self._state is CallerState.running:
                self._pen_stop.append(pen := Pending())
                pen.set_canceller(lambda _: queue.stop())
                self.stopping.add_done_callback(lambda _: queue.stop())
                host: Host[Any] | None = Host.current(self.thread)
                get_start_guest_run(backend)(
                    self._scheduler,
                    queue,
                    done_callback=guest_done_callback,
                    run_sync_soon_threadsafe=host.run_sync_soon_threadsafe if host else self.call_direct,
                    run_sync_soon_not_threadsafe=host.run_sync_soon_not_threadsafe if host else None,
                    host_uses_signal_set_wakeup_fd=host.host_uses_signal_set_wakeup_fd if host else True,
                )

    @classmethod
    def id_current(cls) -> int:
        """The id of a caller given the current thread in CPython or context in Pyodide.

        Note:
            This does not mean a caller actually exists.
        """
        return cls._caller_token.get() if sys.platform == "emscripten" else threading.get_ident()

    @classmethod
    def get_existing(cls, caller_id: int | None = None, /) -> Self | None:
        """A [classmethod][] to get the caller instance if it exists.

        Args:
            caller_id: The id of the caller.
        """
        with cls._lock:
            return cls._instances.get(caller_id or cls.id_current())

    @classmethod
    def current_pending(cls) -> Pending[Any] | None:
        """A [classmethod][] that returns the pending for the current context."""
        return cls._pending_var.get()

    @classmethod
    def all_callers(cls) -> list[Caller]:
        """A [classmethod][] to get a list of all callers."""
        return list(Caller._instances.values())

    def stop(self, *, force: bool = False) -> ProtectedPending | None:
        """Stop the caller.

        Args:
            force: If the caller is protected the call is a no-op unless `force=True`.
        """
        if self._protected and not force:
            self.log.debug("Non-force stop ignored for  %s", self)
            return None
        self.stopping.set_result(None)
        if (parent := self.parent) and self in parent._worker_pool:
            with contextlib.suppress(IndexError):
                parent._worker_pool.remove(self)
        if (old_state := self._set_state(CallerState.stopping)) and old_state.value < CallerState.starting.value:
            self._set_state(CallerState.stopped)
        return self.stopped

    def get(self, **kwargs: Unpack[CallerCreateOptions]) -> Self:
        """Get an existing caller by name or create a new one.

        If `name` is provided and a child with that name exists, the child will be returned,
        otherwise a new caller is returned.

        Args:
            **kwargs: Options for creating or retrieving a caller.

        Returns:
            Self: The child caller.

        Raises:
            RuntimeError: If a caller with the specified name exists but the backend does not match.

        Notes:
            - The returned caller is added to `children` and stopped with the caller.
            - If 'backend' or 'zmq_context' are not specified they are copied from the caller.
        """
        with self._inst_lock:
            if self.stopping.done():
                msg = f"Caller is stopping or stopped {self}"
                raise RuntimeError(msg)
            if name := kwargs.get("name"):
                for child in self._children:
                    if child.name == name:
                        if (backend := kwargs.get("backend")) and child.backend != backend:
                            msg = f"Backend mismatch! {backend=} {child.backend=}"
                            raise RuntimeError(msg)
                        if (host := kwargs.get("host")) and child.host != host:
                            msg = f"Host mismatch! {host=} {child.host=}"
                            raise RuntimeError(msg)
                        return child
            if name is None:
                msg = "Name must be specified"
                raise ValueError(msg)
            if "backend" not in kwargs:
                kwargs["backend"] = self._backend
                kwargs["backend_options"] = self.backend_options
            child = self.__class__("NewThread", **kwargs)
            child._parent_ref = weakref.ref(self)
            self._children.add(child)
            self._children_countdown.up()
            return child

    def schedule_call(
        self,
        func: Callable[..., CoroutineType[Any, Any, T] | T],
        args: tuple,
        kwargs: dict,
        context: contextvars.Context | None = None,
        trackers: type[PendingTracker] | tuple[type[PendingTracker], ...] = PendingTracker,
        backend: NoValue | Backend = NoValue,
        /,
        **metadata: Any,
    ) -> Pending[T]:
        """A low-level method to schedule `func` to be run as a task running in the caller's backend event loop.

        Args:
            func: The function to be called.
            args: Arguments to use when calling `func`.
            kwargs: Keyword arguments to use when calling `func`.
            context: A context to copy, if not provided the current context is copied.
            trackers: A tuple of tracker subclasses where the pending should be registered.
            backend: An optional backend to require `func`.
            **metadata: Additional metadata to store in the pending.

        Returns:
            Pending: A pending that can be awaited to obtain the result of func.
        """
        pen = Pending(context, trackers, func=func, args=args, kwargs=kwargs, caller=self, **metadata)
        if self._state is CallerState.stopped:
            pen.cancel(f"The caller has been stopped: {self}")
            return pen
        if backend is NoValue or (backend := Backend(backend)) is self.backend:
            queue = self._scheduler_queue
        elif not (queue := self._guest_queues.get(backend)):
            with self._inst_lock:
                if not self._protected:
                    # For guest backends we need a managed shutdown.
                    msg = "Async context must be acquired prior to using a guest backend!"
                    raise RuntimeError(msg)
                if backend is Backend.trio:
                    trio.sleep  # noqa: B018 # Check trio is available.
                if not (queue := self._guest_queues.get(backend)):
                    self.call_direct(self._start_guest, backend, queue := SingleAsyncQueue(reject=self._reject))
                    self._guest_queues[backend] = queue
        queue.append(pen)
        return pen

    def call_later(
        self,
        delay: float,
        func: Callable[P, T | CoroutineType[Any, Any, T]],
        /,
        *args: P.args,
        **kwargs: P.kwargs,
    ) -> Pending[T]:
        """Schedule `func` to be called in the caller's backend event loop with a delay using a copy of the current context.

        Args:
            delay: The minimum delay to add between submission and execution.
            func: The function.
            *args: Arguments to use with `func`.
            **kwargs: Keyword arguments to use with `func`.
        """

        async def _call_later(*args: P.args, **kwargs: P.kwargs) -> T:
            if (delay_ := start_time - time.monotonic() + delay) >= 0:
                await anyio.sleep(delay_)
            result = func(*args, **kwargs)
            if iscoroutinelike(result):
                result = await result
            return result  # pyright: ignore[reportReturnType]

        start_time = time.monotonic()
        return self.schedule_call(_call_later, args, kwargs)

    def call_soon(
        self,
        func: Callable[P, T | CoroutineType[Any, Any, T]],
        /,
        *args: P.args,
        **kwargs: P.kwargs,
    ) -> Pending[T]:
        """Schedule `func` to be called in the caller's backend event loop using a copy of the current context.

        Args:
            func: The function.
            *args: Arguments to use with `func`.
            **kwargs: Keyword arguments to use with `func`.
        """
        return self.schedule_call(func, args, kwargs)

    def call_using_backend(
        self,
        backend: Backend | Literal["asyncio", "trio"],
        func: Callable[P, T | CoroutineType[Any, Any, T]],
        /,
        *args: P.args,
        **kwargs: P.kwargs,
    ) -> Pending[T]:
        """Schedule `func` to be executed using the specified `backend` event loop using a copy of the current context.

        A guest backend is used when the `backend` is different to the caller's backend.

        Args:
            backend: The backend to use for execution.
            func: The function.
            *args: Arguments to use with `func`.
            **kwargs: Keyword arguments to use with `func`.

        **See Also**

            - [Caller.get][]
        """
        return self.schedule_call(func, args, kwargs, None, PendingTracker, Backend(backend))

    def call_direct(
        self,
        func: Callable[P, T],
        /,
        *args: P.args,
        **kwargs: P.kwargs,
    ) -> None:
        """A low-level method to schedule `func` to be executed directly in caller's backend scheduler (no context copy).

        Args:
            func: The function.
            *args: Arguments to use with `func`.
            **kwargs: Keyword arguments to use with `func`.
        """
        self._scheduler_queue.append((func, args, kwargs))

    def to_thread(
        self,
        func: Callable[P, T | CoroutineType[Any, Any, T]],
        /,
        *args: P.args,
        **kwargs: P.kwargs,
    ) -> Pending[T]:
        """Call `func` in a child 'worker' that has the same type of backend as the caller (parent).

        Args:
            func: The function.
            *args: Arguments to use with `func`.
            **kwargs: Keyword arguments to use with `func`.

        Notes:
            - A pool of workers are retained to enable quick dispatch.
        """

        def _to_thread_on_done(_) -> None:
            if (
                not self.stopping.done()
                and not caller.stopping.done()
                and len(self._worker_pool) < self.MAX_IDLE_POOL_INSTANCES
            ):
                caller._idle_time = time.monotonic()
                self._worker_pool.append(caller)
                if self.IDLE_WORKER_SHUTDOWN_DURATION > 0 and not self.queue_get(self._idle_worker_cleanup):
                    self.queue_call(self._idle_worker_cleanup)
            else:
                caller.stop(force=True)

        try:
            while (caller := self._worker_pool.popleft()) and caller.stopping.done():
                pass
        except IndexError:
            caller = self.get(name="")
            caller._name = "async-kernel worker"
        pen = caller.call_soon(func, *args, **kwargs)
        pen.add_done_callback(_to_thread_on_done)
        return pen

    def queue_get(self, func: Callable) -> Pending[None] | None:
        """Returns the pending where with the `queue_call` is running.

        Notes:
            - `queue_close` is the preferred means to shutdown the queue.
        """
        return self._queue_map.get(hash(func))

    def queue_call(
        self,
        func: Callable[P, Any | Awaitable[Any]],
        /,
        *args: P.args,
        **kwargs: P.kwargs,
    ) -> None:
        """A low-level method to queue the execution of `func` in a queue unique to it and the caller.

        This sets up a long-lived task to provide a fast pathway for repetitive calls to a function.

        Args:
            func: The function.
            *args: Arguments to use with `func`.
            **kwargs: Keyword arguments to use with `func`.

        Notes:
            - The queue runs inside a task that remains running until one of the following occurs:
                1. The queue is stopped.
                2. The method [Caller.queue_close][] is called with `func` or `func`'s hash.
                3. `func` is deleted (utilising [weakref.finalize][]).
            - The [context][contextvars.Context] of the initial call is used for subsequent queue calls.
            - Exceptions are logged to caller.log but not propagated.
            - The pending created on the first call will only be registered with [async_kernel.pending.PendingManager][]
                subclassed trackers and **not** PendingGroup.
        """
        if not (pen_ := self._queue_map.get(key := hash(func))):
            with self._inst_lock:
                if not (pen_ := self._queue_map.get(key)):
                    queue = SingleAsyncQueue[tuple[Callable, tuple, dict]](reject=self._reject)
                    with contextlib.suppress(TypeError):
                        weakref.finalize(func.__self__ if inspect.ismethod(func) else func, self.queue_close, key)

                    async def queue_loop() -> None:
                        pen = self.current_pending()
                        assert pen is not None
                        try:
                            async for item in queue:
                                try:
                                    result = item[0](*item[1], **item[2])
                                    if iscoroutinelike(result):
                                        await result
                                    del result
                                except anyio.get_cancelled_exc_class():
                                    self.log.debug("queue_call task cancelled: %s args:%s kwargs:%s", *item)
                                    raise
                                except (anyio.get_cancelled_exc_class(), Exception) as e:
                                    self.log.exception("Execution of %s failed! args:%s kwargs:%s", *item, exc_info=e)
                                finally:
                                    del item
                        finally:
                            self._queue_map.pop(key)

                    pen_ = self.schedule_call(queue_loop, (), {}, None, PendingManager, key=key, queue=queue)
                    self._queue_map[key] = pen_
        pen_.metadata["queue"].append((func, args, kwargs))

    def queue_close(self, func: Callable | int) -> None:
        """Close the [Caller.queue_call][async_kernel.caller.Caller.queue_call] execution queue associated with `func`.

        Args:
            func: The queue of the function to close.
        """
        if pen := self._queue_map.pop(func if isinstance(func, int) else hash(func), None):
            pen.metadata["queue"].stop()

    def create_start_stop_task(
        self,
        func: Callable[Concatenate[Callable[[], None], ProtectedPending[None], P], CoroutineType[Any, Any, T]],
        /,
    ) -> StartStopTask[P, T]:
        """Wrap the coroutine function `func` with a `StartStopTask`.

        The method [StartStopTask.start][] must be called to start the task. args and kwargs
        can be passed in the function call. Subsequent calls are noop.

        `func` must accept two position only arguments:

        1. A callback to indicate that `func` has started.
        2. A `ProtectedPending` that is used as a signal for func to stop.

        Usage:
            ```python
            async def func(started, stopped):
                started()
                await stopped


            # Non-blocking
            task = caller.create_start_stop_task(func).start()
            task.stop()

            # or as an async context
            async with caller.create_start_stop_task(func).start():
                pass
            ```
        """
        return StartStopTask().set_task_function(func, caller=self)

    async def as_completed(
        self,
        items: Iterable[Awaitable[T]] | AsyncGenerator[Awaitable[T]],
        *,
        max_concurrent: NoValue | int = NoValue,
        cancel_unfinished: bool = True,
    ) -> AsyncGenerator[Pending[T], Any]:
        """An async iterator which yields a pending for each awaitable in `items` as they complete.

        Args:
            items: A container or a generator that yields awaitables.
            max_concurrent: The maximum number of pending to monitor at a time if `items` is a generator.
            cancel_unfinished: Cancel any `pending` when exiting.

        Tip:
            - Pass a generator if you wish to limit the number result jobs when calling to_thread/to_task etc.
            - Pass a container with all results when the limiter is not relevant.
            -  `Caller.MAX_IDLE_POOL_INSTANCES`
        """

        def noop() -> None:
            pass

        resume = noop
        done: SingleAsyncQueue[Pending[T]] = SingleAsyncQueue()
        unfinished: set[Pending[T]] = set()
        pen_current = self.current_pending()
        if isinstance(items, set | list | tuple):
            max_concurrent_ = 0
        else:
            max_concurrent_ = self.MAX_IDLE_POOL_INSTANCES if max_concurrent is NoValue else int(max_concurrent)

        async def scheduler():
            nonlocal resume
            gen = items if isinstance(items, AsyncGenerator) else iter(items)
            is_async = isinstance(gen, AsyncGenerator)
            while not done.stopped and (pen := await anext(gen, None) if is_async else next(gen, None)) is not None:
                if pen is pen_current:
                    done.stop()
                    msg = "Waiting for the pending in which it is running would result in deadlock!"
                    raise RuntimeError(msg)
                if not isinstance(pen, Pending):
                    pen = self.call_soon(await_for, pen)
                if not pen.done():
                    unfinished.add(pen)
                    pen.add_done_callback(done.append)
                    if max_concurrent_ and len(unfinished) == max_concurrent_:
                        event = create_async_event()
                        resume = event.set
                        if len(unfinished) == max_concurrent_:
                            await event
                        resume = noop
                else:
                    done.append(pen)
            if len(done) == 0 and not unfinished:
                done.stop()

        pen_ = self.call_soon(scheduler)
        pen_.add_done_callback(lambda pen: pen.cancelled() and done.stop())
        try:
            async for pen in done:
                unfinished.discard(pen)
                yield pen
                if pen_.done() and not unfinished and len(done) == 0:
                    break
                elif max_concurrent_ and len(unfinished) < max_concurrent_:
                    resume()
            pen_.result()
        finally:
            done.stop()
            for pen in unfinished:
                pen.remove_done_callback(done.append)
                if cancel_unfinished:
                    pen.cancel("Cancelled by as_completed")
            await pen_.cancel_wait(shield=True)

    async def wait(
        self,
        items: Iterable[Awaitable[T]],
        *,
        shield: bool = False,
        timeout: float | None = None,
        return_when: Literal["FIRST_COMPLETED", "FIRST_EXCEPTION", "ALL_COMPLETED"] = "ALL_COMPLETED",
    ) -> tuple[set[Pending[T]], set[Pending[T]]]:
        """Wait for one or more of the awaitable items to complete.

        Returns two sets of the results: (done, pending).

        Args:
            items: An iterable of results to wait for.
            shield: Shield from external cancellation.
            timeout: The maximum time before returning.
            return_when: The same options as available for [asyncio.wait][].

        Example:
            ```python
            done, pending = await asyncio.wait(items)
            ```
        Info:
            - This does NOT raise a [TimeoutError]!
            - Pendings that aren't done when the timeout occurs are returned in the second set.
        """
        pending: set[Pending[T]] = set()
        done = set()
        for item in items:
            if isinstance(item, Pending):
                done.add(item) if item.done() else pending.add(item)
            else:
                assert inspect.isawaitable(item)
                pending.add(self.call_soon(await_for, item))
        if done:
            if return_when == "FIRST_COMPLETED":
                return done, pending
            if return_when == "FIRST_EXCEPTION":
                for pen in done:
                    if pen.cancelled() or pen.exception():
                        return done, pending
        if pending:
            deadline = anyio.current_time() + timeout if timeout is not None else math.inf
            with anyio.CancelScope(deadline=deadline, shield=shield):
                async for pen in self.as_completed(pending.copy(), cancel_unfinished=False):
                    pending.discard(pen)
                    done.add(pen)
                    if return_when == "FIRST_COMPLETED":
                        break
                    if return_when == "FIRST_EXCEPTION" and (pen.cancelled() or pen.exception()):
                        break
        return done, pending

    def create_pending_group(
        self,
        *,
        shield: bool = False,
        mode: Literal[0, 1, 2, 3] = 0,
        timeout: float | None = None,
    ) -> PendingGroup:
        """Create a new [PendingGroup][async_kernel.pending.PendingGroup].

        This automatically tracks [Pending][async_kernel.pending.Pending] created in the context
        that have opt-in (by including the class `PendingTracker` as a 'tracker'). This includes
        all [Caller][] methods that return a pending.

        The context will not exit until all registered pending are complete. The exit and cancellation behaviour
        is determined by the `mode`.

        Args:
            shield: Shield the pending group from external cancellation.
            mode: The mode.
                - 0: Ignore cancellation of pending, if any pending is cancelled - exit quietly.
                - 1: Cancel if any pending is cancelled - raise PendingCancelled on exit.
                - 2: Cancel if any pending is cancelled - exit quietly.
                - 3: Ignore cancellation of pending, if any pending is cancelled - raise PendingCancelled on exit.
            timeout: An approximate time limit for the context to remain open before cancelling unfinished pending.

        Usage:
            ```python
            async with Caller().create_pending_group() as pg:
                pg.caller.to_thread(my_func)
            ```
        """
        return PendingGroup(shield=shield, mode=mode, timeout=timeout)

MAX_IDLE_POOL_INSTANCES `class-attribute` `instance-attribute` ¶

MAX_IDLE_POOL_INSTANCES = 10

The number of child workers to keep in the pool (See also to_thread).

IDLE_WORKER_SHUTDOWN_DURATION `class-attribute` `instance-attribute` ¶

IDLE_WORKER_SHUTDOWN_DURATION = 0 if 'pytest' in sys.modules else 60

The duration in seconds for a worker to remain in the pool before it is shutdown.

Set to 0 to disable (default when running tests).

started `class-attribute` `instance-attribute` ¶

started = Fixed(ProtectedPending)

A ProtectedPending that is set when the caller has started.

stopping `class-attribute` `instance-attribute` ¶

stopping = Fixed(ProtectedPending)

A ProtectedPending that is set done when the caller is stopping.

stopped `class-attribute` `instance-attribute` ¶

stopped = Fixed(ProtectedPending)

A ProtectedPending that is set done when the caller has stopped.

name `property` ¶

name: str

The name of the caller.

id `property` ¶

id: int

The id of the caller, which is the thread's ident where threading is being used otherwise it is a unique value.

backend `property` ¶

backend: Backend

The primary backend of the caller.

backend_options `property` ¶

backend_options: dict | None

Options used to create the backend event loop when the caller is started.

host `property` ¶

host: Hosts | None

The name of the host if the caller is started with a host.

host_options `property` ¶

host_options: dict | None

Options used to create the host.

protected `property` `writable` ¶

protected: bool

Returns True if the caller is protected from stopping.

This can be set to True at any time.

children `property` ¶

children: set[Self]

The caller's children that are not stopping.

Notes

When the parent is stopped, all children are stopped.
All children are stopped before the parent's stopped ProtectedPending is set as done.

thread `property` ¶

thread: Thread

The thread where the caller is running.

parent `property` ¶

parent: Self | None

The caller's parent if it has one.

new ¶

__new__(
    modifier: Literal["CurrentThread", "MainThread", "NewThread"] = "CurrentThread",
    /,
    **kwargs: Unpack[CallerCreateOptions],
) -> Self

Create or retrieve a caller.

Parameters:

modifier ¶
(Literal['CurrentThread', 'MainThread', 'NewThread'], default: 'CurrentThread' ) –
Specifies the caller instance to retrieve.
- CurrentThread: The caller for the current thread.
- MainThread: The Caller associated with the main thread.
- NewThread: Create a caller with a new thread. The lifecycle is not It is recommended to used this as an async context. Caller.get and Caller.to_thread are preferred alternatives.
**kwargs ¶
(Unpack[CallerCreateOptions], default: {} ) –

Additional options for Caller creation, such as: - name: The name to use. - backend: The async backend to use. - backend_options: Options for the backend.

Returns:

Self ( Self ) –

The created or retrieved Caller instance.

Raises:

RuntimeError –

If the backend is not provided and backend can't be determined.
ValueError –

If the thread and caller's name do not match.

Source code in src/async_kernel/caller.py

def __new__(
    cls,
    modifier: Literal["CurrentThread", "MainThread", "NewThread"] = "CurrentThread",
    /,
    **kwargs: Unpack[CallerCreateOptions],
) -> Self:
    """Create or retrieve a caller.

    Args:
        modifier: Specifies the caller instance to retrieve.

            - CurrentThread: The caller for the current thread.
            - MainThread: The Caller associated with the main thread.
            - NewThread: Create a caller with a new thread. The lifecycle is not
                It is recommended to used this as an async context.
                [Caller.get][] and [Caller.to_thread][] are preferred alternatives.

        **kwargs: Additional options for Caller creation, such as:
            - name: The name to use.
            - backend: The async backend to use.
            - backend_options: Options for the backend.

    Returns:
        Self: The created or retrieved Caller instance.

    Raises:
        RuntimeError: If the backend is not provided and backend can't be determined.
        ValueError: If the thread and caller's name do not match.
    """
    if modifier != "NewThread":
        cls._lock.acquire()
    try:
        name, backend = kwargs.get("name", ""), kwargs.get("backend")
        match modifier:
            case "MainThread":
                caller_id, thread = cls.CALLER_MAIN_THREAD_ID, threading.main_thread()
            case "NewThread":
                caller_id, thread = None, None
            case _:
                caller_id, thread = cls.id_current(), threading.current_thread()

        # Locate existing
        if caller_id is not None and (caller := cls._instances.get(caller_id)):
            if name and name != caller.name:
                msg = f"The thread and caller's name do not match! {name=} {caller=}"
                raise ValueError(msg)
            if backend and backend != caller.backend:
                msg = f"The backend does not match! {backend=} {caller.backend=}"
                raise ValueError(msg)
            return caller

        host = Hosts(host) if (host := kwargs.get("host")) else None
        if (backend := Backend(backend or current_async_library())) is Backend.trio:
            trio.sleep  # noqa: B018 # Check trio is available.

        # create a new instance
        inst = super().__new__(cls)
        if sys.platform == "emscripten":
            if caller_id is None:
                caller_id = id(inst)
            thread = threading.current_thread()

        # Add private objects
        inst._inst_lock = threading.Lock()
        inst._children = set()
        inst._tasks = set()
        inst._queue_map = {}
        inst._scheduler_queue = SingleAsyncQueue(reject=inst._reject)
        inst._guest_queues = {}
        inst._children_countdown = CountdownEvent()
        inst._pen_stop = []

        # Apply settings
        inst._name = name
        inst._backend = backend
        inst._host = host
        inst._backend_options = kwargs.get("backend_options")
        inst._host_options = kwargs.get("host_options")
        inst.log = logging.LoggerAdapter(logging.getLogger())

        # Start the instance.
        inst._start((caller_id, thread) if caller_id and thread else None, kwargs.get("no_debug", False))
        assert inst._caller_id is not None
        assert inst._caller_id not in cls._instances
        cls._instances[inst._caller_id] = inst
        return inst
    finally:
        if modifier != "NewThread":
            cls._lock.release()

id_current `classmethod` ¶

id_current() -> int

The id of a caller given the current thread in CPython or context in Pyodide.

Note

This does not mean a caller actually exists.

Referenced by:

Reference caller Caller

Source code in src/async_kernel/caller.py

@classmethod
def id_current(cls) -> int:
    """The id of a caller given the current thread in CPython or context in Pyodide.

    Note:
        This does not mean a caller actually exists.
    """
    return cls._caller_token.get() if sys.platform == "emscripten" else threading.get_ident()

get_existing `classmethod` ¶

get_existing(caller_id: int | None = None) -> Self | None

A classmethod to get the caller instance if it exists.

Parameters:

caller_id ¶
(int | None, default: None ) –

The id of the caller.

Referenced by:

Reference caller Caller

Source code in src/async_kernel/caller.py

@classmethod
def get_existing(cls, caller_id: int | None = None, /) -> Self | None:
    """A [classmethod][] to get the caller instance if it exists.

    Args:
        caller_id: The id of the caller.
    """
    with cls._lock:
        return cls._instances.get(caller_id or cls.id_current())

current_pending `classmethod` ¶

current_pending() -> Pending[Any] | None

A classmethod that returns the pending for the current context.

Referenced by:

Reference
- caller Caller
- event_loop ZMQPoll event_handler

Source code in src/async_kernel/caller.py

@classmethod
def current_pending(cls) -> Pending[Any] | None:
    """A [classmethod][] that returns the pending for the current context."""
    return cls._pending_var.get()

all_callers `classmethod` ¶

all_callers() -> list[Caller]

A classmethod to get a list of all callers.

Referenced by:

Reference caller Caller

Source code in src/async_kernel/caller.py

@classmethod
def all_callers(cls) -> list[Caller]:
    """A [classmethod][] to get a list of all callers."""
    return list(Caller._instances.values())

stop ¶

stop(*, force: bool = False) -> ProtectedPending | None

Stop the caller.

Parameters:

force ¶
(bool, default: False ) –

If the caller is protected the call is a no-op unless force=True.

Source code in src/async_kernel/caller.py

def stop(self, *, force: bool = False) -> ProtectedPending | None:
    """Stop the caller.

    Args:
        force: If the caller is protected the call is a no-op unless `force=True`.
    """
    if self._protected and not force:
        self.log.debug("Non-force stop ignored for  %s", self)
        return None
    self.stopping.set_result(None)
    if (parent := self.parent) and self in parent._worker_pool:
        with contextlib.suppress(IndexError):
            parent._worker_pool.remove(self)
    if (old_state := self._set_state(CallerState.stopping)) and old_state.value < CallerState.starting.value:
        self._set_state(CallerState.stopped)
    return self.stopped

get ¶

get(**kwargs: Unpack[CallerCreateOptions]) -> Self

Get an existing caller by name or create a new one.

If name is provided and a child with that name exists, the child will be returned, otherwise a new caller is returned.

Parameters:

**kwargs ¶
(Unpack[CallerCreateOptions], default: {} ) –

Options for creating or retrieving a caller.

Returns:

Self ( Self ) –

The child caller.

Raises:

RuntimeError –

If a caller with the specified name exists but the backend does not match.

Notes

The returned caller is added to children and stopped with the caller.
If 'backend' or 'zmq_context' are not specified they are copied from the caller.

Referenced by:

Reference caller Caller

Source code in src/async_kernel/caller.py

def get(self, **kwargs: Unpack[CallerCreateOptions]) -> Self:
    """Get an existing caller by name or create a new one.

    If `name` is provided and a child with that name exists, the child will be returned,
    otherwise a new caller is returned.

    Args:
        **kwargs: Options for creating or retrieving a caller.

    Returns:
        Self: The child caller.

    Raises:
        RuntimeError: If a caller with the specified name exists but the backend does not match.

    Notes:
        - The returned caller is added to `children` and stopped with the caller.
        - If 'backend' or 'zmq_context' are not specified they are copied from the caller.
    """
    with self._inst_lock:
        if self.stopping.done():
            msg = f"Caller is stopping or stopped {self}"
            raise RuntimeError(msg)
        if name := kwargs.get("name"):
            for child in self._children:
                if child.name == name:
                    if (backend := kwargs.get("backend")) and child.backend != backend:
                        msg = f"Backend mismatch! {backend=} {child.backend=}"
                        raise RuntimeError(msg)
                    if (host := kwargs.get("host")) and child.host != host:
                        msg = f"Host mismatch! {host=} {child.host=}"
                        raise RuntimeError(msg)
                    return child
        if name is None:
            msg = "Name must be specified"
            raise ValueError(msg)
        if "backend" not in kwargs:
            kwargs["backend"] = self._backend
            kwargs["backend_options"] = self.backend_options
        child = self.__class__("NewThread", **kwargs)
        child._parent_ref = weakref.ref(self)
        self._children.add(child)
        self._children_countdown.up()
        return child

schedule_call ¶

schedule_call(
    func: Callable[..., CoroutineType[Any, Any, T] | T],
    args: tuple,
    kwargs: dict,
    context: Context | None = None,
    trackers: type[PendingTracker] | tuple[type[PendingTracker], ...] = PendingTracker,
    backend: NoValue | Backend = NoValue,
    /,
    **metadata: Any,
) -> Pending[T]

A low-level method to schedule func to be run as a task running in the caller's backend event loop.

Parameters:

func ¶
(Callable[..., CoroutineType[Any, Any, T] | T]) –

The function to be called.
args ¶
(tuple) –

Arguments to use when calling func.
kwargs ¶
(dict) –

Keyword arguments to use when calling func.
context ¶
(Context | None, default: None ) –

A context to copy, if not provided the current context is copied.
trackers ¶
(type[PendingTracker] | tuple[type[PendingTracker], ...], default: PendingTracker ) –

A tuple of tracker subclasses where the pending should be registered.
backend ¶
(NoValue | Backend, default: NoValue ) –

An optional backend to require func.
**metadata ¶
(Any, default: {} ) –

Additional metadata to store in the pending.

Returns:

Pending ( Pending[T] ) –

A pending that can be awaited to obtain the result of func.

Referenced by:

Reference caller Caller

Source code in src/async_kernel/caller.py

def schedule_call(
    self,
    func: Callable[..., CoroutineType[Any, Any, T] | T],
    args: tuple,
    kwargs: dict,
    context: contextvars.Context | None = None,
    trackers: type[PendingTracker] | tuple[type[PendingTracker], ...] = PendingTracker,
    backend: NoValue | Backend = NoValue,
    /,
    **metadata: Any,
) -> Pending[T]:
    """A low-level method to schedule `func` to be run as a task running in the caller's backend event loop.

    Args:
        func: The function to be called.
        args: Arguments to use when calling `func`.
        kwargs: Keyword arguments to use when calling `func`.
        context: A context to copy, if not provided the current context is copied.
        trackers: A tuple of tracker subclasses where the pending should be registered.
        backend: An optional backend to require `func`.
        **metadata: Additional metadata to store in the pending.

    Returns:
        Pending: A pending that can be awaited to obtain the result of func.
    """
    pen = Pending(context, trackers, func=func, args=args, kwargs=kwargs, caller=self, **metadata)
    if self._state is CallerState.stopped:
        pen.cancel(f"The caller has been stopped: {self}")
        return pen
    if backend is NoValue or (backend := Backend(backend)) is self.backend:
        queue = self._scheduler_queue
    elif not (queue := self._guest_queues.get(backend)):
        with self._inst_lock:
            if not self._protected:
                # For guest backends we need a managed shutdown.
                msg = "Async context must be acquired prior to using a guest backend!"
                raise RuntimeError(msg)
            if backend is Backend.trio:
                trio.sleep  # noqa: B018 # Check trio is available.
            if not (queue := self._guest_queues.get(backend)):
                self.call_direct(self._start_guest, backend, queue := SingleAsyncQueue(reject=self._reject))
                self._guest_queues[backend] = queue
    queue.append(pen)
    return pen

call_later ¶

call_later(
    delay: float,
    func: Callable[P, T | CoroutineType[Any, Any, T]],
    /,
    *args: args,
    **kwargs: kwargs,
) -> Pending[T]

Schedule func to be called in the caller's backend event loop with a delay using a copy of the current context.

Parameters:

delay ¶
(float) –

The minimum delay to add between submission and execution.
func ¶
(Callable[P, T | CoroutineType[Any, Any, T]]) –

The function.
*args ¶
(args, default: () ) –

Arguments to use with func.
**kwargs ¶
(kwargs, default: {} ) –

Keyword arguments to use with func.

Referenced by:

Reference caller Caller

Source code in src/async_kernel/caller.py

def call_later(
    self,
    delay: float,
    func: Callable[P, T | CoroutineType[Any, Any, T]],
    /,
    *args: P.args,
    **kwargs: P.kwargs,
) -> Pending[T]:
    """Schedule `func` to be called in the caller's backend event loop with a delay using a copy of the current context.

    Args:
        delay: The minimum delay to add between submission and execution.
        func: The function.
        *args: Arguments to use with `func`.
        **kwargs: Keyword arguments to use with `func`.
    """

    async def _call_later(*args: P.args, **kwargs: P.kwargs) -> T:
        if (delay_ := start_time - time.monotonic() + delay) >= 0:
            await anyio.sleep(delay_)
        result = func(*args, **kwargs)
        if iscoroutinelike(result):
            result = await result
        return result  # pyright: ignore[reportReturnType]

    start_time = time.monotonic()
    return self.schedule_call(_call_later, args, kwargs)

call_soon ¶

call_soon(
    func: Callable[P, T | CoroutineType[Any, Any, T]],
    /,
    *args: args,
    **kwargs: kwargs,
) -> Pending[T]

Schedule func to be called in the caller's backend event loop using a copy of the current context.

Parameters:

func ¶
(Callable[P, T | CoroutineType[Any, Any, T]]) –

The function.
*args ¶
(args, default: () ) –

Arguments to use with func.
**kwargs ¶
(kwargs, default: {} ) –

Keyword arguments to use with func.

Referenced by:

Reference
async-kernel How it works

Source code in src/async_kernel/caller.py

def call_soon(
    self,
    func: Callable[P, T | CoroutineType[Any, Any, T]],
    /,
    *args: P.args,
    **kwargs: P.kwargs,
) -> Pending[T]:
    """Schedule `func` to be called in the caller's backend event loop using a copy of the current context.

    Args:
        func: The function.
        *args: Arguments to use with `func`.
        **kwargs: Keyword arguments to use with `func`.
    """
    return self.schedule_call(func, args, kwargs)

call_using_backend ¶

call_using_backend(
    backend: Backend | Literal["asyncio", "trio"],
    func: Callable[P, T | CoroutineType[Any, Any, T]],
    /,
    *args: args,
    **kwargs: kwargs,
) -> Pending[T]

Schedule func to be executed using the specified backend event loop using a copy of the current context.

A guest backend is used when the backend is different to the caller's backend.

Parameters:

backend ¶
(Backend | Literal['asyncio', 'trio']) –

The backend to use for execution.
func ¶
(Callable[P, T | CoroutineType[Any, Any, T]]) –

The function.
*args ¶
(args, default: () ) –

Arguments to use with func.
**kwargs ¶
(kwargs, default: {} ) –

Keyword arguments to use with func.

See Also

- [Caller.get][]

Referenced by:

Reference caller Caller

Source code in src/async_kernel/caller.py

def call_using_backend(
    self,
    backend: Backend | Literal["asyncio", "trio"],
    func: Callable[P, T | CoroutineType[Any, Any, T]],
    /,
    *args: P.args,
    **kwargs: P.kwargs,
) -> Pending[T]:
    """Schedule `func` to be executed using the specified `backend` event loop using a copy of the current context.

    A guest backend is used when the `backend` is different to the caller's backend.

    Args:
        backend: The backend to use for execution.
        func: The function.
        *args: Arguments to use with `func`.
        **kwargs: Keyword arguments to use with `func`.

    **See Also**

        - [Caller.get][]
    """
    return self.schedule_call(func, args, kwargs, None, PendingTracker, Backend(backend))

call_direct ¶

call_direct(func: Callable[P, T], /, *args: args, **kwargs: kwargs) -> None

A low-level method to schedule func to be executed directly in caller's backend scheduler (no context copy).

Parameters:

func ¶
(Callable[P, T]) –

The function.
*args ¶
(args, default: () ) –

Arguments to use with func.
**kwargs ¶
(kwargs, default: {} ) –

Keyword arguments to use with func.

Referenced by:

Reference caller Caller

Source code in src/async_kernel/caller.py

def call_direct(
    self,
    func: Callable[P, T],
    /,
    *args: P.args,
    **kwargs: P.kwargs,
) -> None:
    """A low-level method to schedule `func` to be executed directly in caller's backend scheduler (no context copy).

    Args:
        func: The function.
        *args: Arguments to use with `func`.
        **kwargs: Keyword arguments to use with `func`.
    """
    self._scheduler_queue.append((func, args, kwargs))

to_thread ¶

to_thread(
    func: Callable[P, T | CoroutineType[Any, Any, T]],
    /,
    *args: args,
    **kwargs: kwargs,
) -> Pending[T]

Call func in a child 'worker' that has the same type of backend as the caller (parent).

Parameters:

func ¶
(Callable[P, T | CoroutineType[Any, Any, T]]) –

The function.
*args ¶
(args, default: () ) –

Arguments to use with func.
**kwargs ¶
(kwargs, default: {} ) –

Keyword arguments to use with func.

Notes

A pool of workers are retained to enable quick dispatch.

Referenced by:

Reference
- caller Caller __new__
- typing RunMode thread
async-kernel How it works

Source code in src/async_kernel/caller.py

def to_thread(
    self,
    func: Callable[P, T | CoroutineType[Any, Any, T]],
    /,
    *args: P.args,
    **kwargs: P.kwargs,
) -> Pending[T]:
    """Call `func` in a child 'worker' that has the same type of backend as the caller (parent).

    Args:
        func: The function.
        *args: Arguments to use with `func`.
        **kwargs: Keyword arguments to use with `func`.

    Notes:
        - A pool of workers are retained to enable quick dispatch.
    """

    def _to_thread_on_done(_) -> None:
        if (
            not self.stopping.done()
            and not caller.stopping.done()
            and len(self._worker_pool) < self.MAX_IDLE_POOL_INSTANCES
        ):
            caller._idle_time = time.monotonic()
            self._worker_pool.append(caller)
            if self.IDLE_WORKER_SHUTDOWN_DURATION > 0 and not self.queue_get(self._idle_worker_cleanup):
                self.queue_call(self._idle_worker_cleanup)
        else:
            caller.stop(force=True)

    try:
        while (caller := self._worker_pool.popleft()) and caller.stopping.done():
            pass
    except IndexError:
        caller = self.get(name="")
        caller._name = "async-kernel worker"
    pen = caller.call_soon(func, *args, **kwargs)
    pen.add_done_callback(_to_thread_on_done)
    return pen

queue_get ¶

queue_get(func: Callable) -> Pending[None] | None

Returns the pending where with the queue_call is running.

Notes

queue_close is the preferred means to shutdown the queue.

Referenced by:

Reference caller Caller

Source code in src/async_kernel/caller.py

def queue_get(self, func: Callable) -> Pending[None] | None:
    """Returns the pending where with the `queue_call` is running.

    Notes:
        - `queue_close` is the preferred means to shutdown the queue.
    """
    return self._queue_map.get(hash(func))

queue_call ¶

queue_call(
    func: Callable[P, Any | Awaitable[Any]], /, *args: args, **kwargs: kwargs
) -> None

A low-level method to queue the execution of func in a queue unique to it and the caller.

This sets up a long-lived task to provide a fast pathway for repetitive calls to a function.

Parameters:

func ¶
(Callable[P, Any | Awaitable[Any]]) –

The function.
*args ¶
(args, default: () ) –

Arguments to use with func.
**kwargs ¶
(kwargs, default: {} ) –

Keyword arguments to use with func.

Notes

The queue runs inside a task that remains running until one of the following occurs:
1. The queue is stopped.
2. The method Caller.queue_close is called with func or func's hash.
3. func is deleted (utilising weakref.finalize).
The context of the initial call is used for subsequent queue calls.
Exceptions are logged to caller.log but not propagated.
The pending created on the first call will only be registered with async_kernel.pending.PendingManager subclassed trackers and not PendingGroup.

Referenced by:

Source code in src/async_kernel/caller.py

def queue_call(
    self,
    func: Callable[P, Any | Awaitable[Any]],
    /,
    *args: P.args,
    **kwargs: P.kwargs,
) -> None:
    """A low-level method to queue the execution of `func` in a queue unique to it and the caller.

    This sets up a long-lived task to provide a fast pathway for repetitive calls to a function.

    Args:
        func: The function.
        *args: Arguments to use with `func`.
        **kwargs: Keyword arguments to use with `func`.

    Notes:
        - The queue runs inside a task that remains running until one of the following occurs:
            1. The queue is stopped.
            2. The method [Caller.queue_close][] is called with `func` or `func`'s hash.
            3. `func` is deleted (utilising [weakref.finalize][]).
        - The [context][contextvars.Context] of the initial call is used for subsequent queue calls.
        - Exceptions are logged to caller.log but not propagated.
        - The pending created on the first call will only be registered with [async_kernel.pending.PendingManager][]
            subclassed trackers and **not** PendingGroup.
    """
    if not (pen_ := self._queue_map.get(key := hash(func))):
        with self._inst_lock:
            if not (pen_ := self._queue_map.get(key)):
                queue = SingleAsyncQueue[tuple[Callable, tuple, dict]](reject=self._reject)
                with contextlib.suppress(TypeError):
                    weakref.finalize(func.__self__ if inspect.ismethod(func) else func, self.queue_close, key)

                async def queue_loop() -> None:
                    pen = self.current_pending()
                    assert pen is not None
                    try:
                        async for item in queue:
                            try:
                                result = item[0](*item[1], **item[2])
                                if iscoroutinelike(result):
                                    await result
                                del result
                            except anyio.get_cancelled_exc_class():
                                self.log.debug("queue_call task cancelled: %s args:%s kwargs:%s", *item)
                                raise
                            except (anyio.get_cancelled_exc_class(), Exception) as e:
                                self.log.exception("Execution of %s failed! args:%s kwargs:%s", *item, exc_info=e)
                            finally:
                                del item
                    finally:
                        self._queue_map.pop(key)

                pen_ = self.schedule_call(queue_loop, (), {}, None, PendingManager, key=key, queue=queue)
                self._queue_map[key] = pen_
    pen_.metadata["queue"].append((func, args, kwargs))

queue_close ¶

queue_close(func: Callable | int) -> None

Close the Caller.queue_call execution queue associated with func.

Parameters:

func ¶
(Callable | int) –

The queue of the function to close.

Referenced by:

Reference caller Caller queue_call

Source code in src/async_kernel/caller.py

def queue_close(self, func: Callable | int) -> None:
    """Close the [Caller.queue_call][async_kernel.caller.Caller.queue_call] execution queue associated with `func`.

    Args:
        func: The queue of the function to close.
    """
    if pen := self._queue_map.pop(func if isinstance(func, int) else hash(func), None):
        pen.metadata["queue"].stop()

create_start_stop_task ¶

create_start_stop_task(
    func: Callable[
        Concatenate[Callable[[], None], ProtectedPending[None], P],
        CoroutineType[Any, Any, T],
    ],
) -> StartStopTask[P, T]

Wrap the coroutine function func with a StartStopTask.

The method StartStopTask.start must be called to start the task. args and kwargs can be passed in the function call. Subsequent calls are noop.

func must accept two position only arguments:

A callback to indicate that func has started.
A ProtectedPending that is used as a signal for func to stop.

Usage

async def func(started, stopped):
    started()
    await stopped


# Non-blocking
task = caller.create_start_stop_task(func).start()
task.stop()

# or as an async context
async with caller.create_start_stop_task(func).start():
    pass

Referenced by:

Reference caller Caller

Source code in src/async_kernel/caller.py

def create_start_stop_task(
    self,
    func: Callable[Concatenate[Callable[[], None], ProtectedPending[None], P], CoroutineType[Any, Any, T]],
    /,
) -> StartStopTask[P, T]:
    """Wrap the coroutine function `func` with a `StartStopTask`.

    The method [StartStopTask.start][] must be called to start the task. args and kwargs
    can be passed in the function call. Subsequent calls are noop.

    `func` must accept two position only arguments:

    1. A callback to indicate that `func` has started.
    2. A `ProtectedPending` that is used as a signal for func to stop.

    Usage:
        ```python
        async def func(started, stopped):
            started()
            await stopped


        # Non-blocking
        task = caller.create_start_stop_task(func).start()
        task.stop()

        # or as an async context
        async with caller.create_start_stop_task(func).start():
            pass
        ```
    """
    return StartStopTask().set_task_function(func, caller=self)

as_completed `async` ¶

as_completed(
    items: Iterable[Awaitable[T]] | AsyncGenerator[Awaitable[T]],
    *,
    max_concurrent: NoValue | int = NoValue,
    cancel_unfinished: bool = True,
) -> AsyncGenerator[Pending[T], Any]

An async iterator which yields a pending for each awaitable in items as they complete.

Parameters:

items ¶
(Iterable[Awaitable[T]] | AsyncGenerator[Awaitable[T]]) –

A container or a generator that yields awaitables.
max_concurrent ¶
(NoValue | int, default: NoValue ) –

The maximum number of pending to monitor at a time if items is a generator.
cancel_unfinished ¶
(bool, default: True ) –

Cancel any pending when exiting.

Tip

Pass a generator if you wish to limit the number result jobs when calling to_thread/to_task etc.
Pass a container with all results when the limiter is not relevant.
Caller.MAX_IDLE_POOL_INSTANCES

Referenced by:

Reference caller Caller

Source code in src/async_kernel/caller.py

async def as_completed(
    self,
    items: Iterable[Awaitable[T]] | AsyncGenerator[Awaitable[T]],
    *,
    max_concurrent: NoValue | int = NoValue,
    cancel_unfinished: bool = True,
) -> AsyncGenerator[Pending[T], Any]:
    """An async iterator which yields a pending for each awaitable in `items` as they complete.

    Args:
        items: A container or a generator that yields awaitables.
        max_concurrent: The maximum number of pending to monitor at a time if `items` is a generator.
        cancel_unfinished: Cancel any `pending` when exiting.

    Tip:
        - Pass a generator if you wish to limit the number result jobs when calling to_thread/to_task etc.
        - Pass a container with all results when the limiter is not relevant.
        -  `Caller.MAX_IDLE_POOL_INSTANCES`
    """

    def noop() -> None:
        pass

    resume = noop
    done: SingleAsyncQueue[Pending[T]] = SingleAsyncQueue()
    unfinished: set[Pending[T]] = set()
    pen_current = self.current_pending()
    if isinstance(items, set | list | tuple):
        max_concurrent_ = 0
    else:
        max_concurrent_ = self.MAX_IDLE_POOL_INSTANCES if max_concurrent is NoValue else int(max_concurrent)

    async def scheduler():
        nonlocal resume
        gen = items if isinstance(items, AsyncGenerator) else iter(items)
        is_async = isinstance(gen, AsyncGenerator)
        while not done.stopped and (pen := await anext(gen, None) if is_async else next(gen, None)) is not None:
            if pen is pen_current:
                done.stop()
                msg = "Waiting for the pending in which it is running would result in deadlock!"
                raise RuntimeError(msg)
            if not isinstance(pen, Pending):
                pen = self.call_soon(await_for, pen)
            if not pen.done():
                unfinished.add(pen)
                pen.add_done_callback(done.append)
                if max_concurrent_ and len(unfinished) == max_concurrent_:
                    event = create_async_event()
                    resume = event.set
                    if len(unfinished) == max_concurrent_:
                        await event
                    resume = noop
            else:
                done.append(pen)
        if len(done) == 0 and not unfinished:
            done.stop()

    pen_ = self.call_soon(scheduler)
    pen_.add_done_callback(lambda pen: pen.cancelled() and done.stop())
    try:
        async for pen in done:
            unfinished.discard(pen)
            yield pen
            if pen_.done() and not unfinished and len(done) == 0:
                break
            elif max_concurrent_ and len(unfinished) < max_concurrent_:
                resume()
        pen_.result()
    finally:
        done.stop()
        for pen in unfinished:
            pen.remove_done_callback(done.append)
            if cancel_unfinished:
                pen.cancel("Cancelled by as_completed")
        await pen_.cancel_wait(shield=True)

wait `async` ¶

wait(
    items: Iterable[Awaitable[T]],
    *,
    shield: bool = False,
    timeout: float | None = None,
    return_when: Literal[
        "FIRST_COMPLETED", "FIRST_EXCEPTION", "ALL_COMPLETED"
    ] = "ALL_COMPLETED",
) -> tuple[set[Pending[T]], set[Pending[T]]]

Wait for one or more of the awaitable items to complete.

Returns two sets of the results: (done, pending).

Parameters:

items ¶
(Iterable[Awaitable[T]]) –

An iterable of results to wait for.
shield ¶
(bool, default: False ) –

Shield from external cancellation.
timeout ¶
(float | None, default: None ) –

The maximum time before returning.
return_when ¶
(Literal['FIRST_COMPLETED', 'FIRST_EXCEPTION', 'ALL_COMPLETED'], default: 'ALL_COMPLETED' ) –

The same options as available for asyncio.wait.

Example

done, pending = await asyncio.wait(items)

Info: - This does NOT raise a [TimeoutError]! - Pendings that aren't done when the timeout occurs are returned in the second set.

Referenced by:

Reference caller Caller

Source code in src/async_kernel/caller.py

async def wait(
    self,
    items: Iterable[Awaitable[T]],
    *,
    shield: bool = False,
    timeout: float | None = None,
    return_when: Literal["FIRST_COMPLETED", "FIRST_EXCEPTION", "ALL_COMPLETED"] = "ALL_COMPLETED",
) -> tuple[set[Pending[T]], set[Pending[T]]]:
    """Wait for one or more of the awaitable items to complete.

    Returns two sets of the results: (done, pending).

    Args:
        items: An iterable of results to wait for.
        shield: Shield from external cancellation.
        timeout: The maximum time before returning.
        return_when: The same options as available for [asyncio.wait][].

    Example:
        ```python
        done, pending = await asyncio.wait(items)
        ```
    Info:
        - This does NOT raise a [TimeoutError]!
        - Pendings that aren't done when the timeout occurs are returned in the second set.
    """
    pending: set[Pending[T]] = set()
    done = set()
    for item in items:
        if isinstance(item, Pending):
            done.add(item) if item.done() else pending.add(item)
        else:
            assert inspect.isawaitable(item)
            pending.add(self.call_soon(await_for, item))
    if done:
        if return_when == "FIRST_COMPLETED":
            return done, pending
        if return_when == "FIRST_EXCEPTION":
            for pen in done:
                if pen.cancelled() or pen.exception():
                    return done, pending
    if pending:
        deadline = anyio.current_time() + timeout if timeout is not None else math.inf
        with anyio.CancelScope(deadline=deadline, shield=shield):
            async for pen in self.as_completed(pending.copy(), cancel_unfinished=False):
                pending.discard(pen)
                done.add(pen)
                if return_when == "FIRST_COMPLETED":
                    break
                if return_when == "FIRST_EXCEPTION" and (pen.cancelled() or pen.exception()):
                    break
    return done, pending

create_pending_group ¶

create_pending_group(
    *, shield: bool = False, mode: Literal[0, 1, 2, 3] = 0, timeout: float | None = None
) -> PendingGroup

Create a new PendingGroup.

This automatically tracks Pending created in the context that have opt-in (by including the class PendingTracker as a 'tracker'). This includes all Caller methods that return a pending.

The context will not exit until all registered pending are complete. The exit and cancellation behaviour is determined by the mode.

Parameters:

shield ¶
(bool, default: False ) –

Shield the pending group from external cancellation.
mode ¶
(Literal[0, 1, 2, 3], default: 0 ) –

The mode. - 0: Ignore cancellation of pending, if any pending is cancelled - exit quietly. - 1: Cancel if any pending is cancelled - raise PendingCancelled on exit. - 2: Cancel if any pending is cancelled - exit quietly. - 3: Ignore cancellation of pending, if any pending is cancelled - raise PendingCancelled on exit.
timeout ¶
(float | None, default: None ) –

An approximate time limit for the context to remain open before cancelling unfinished pending.

Usage

async with Caller().create_pending_group() as pg:
    pg.caller.to_thread(my_func)

Referenced by:

Reference caller Caller

Source code in src/async_kernel/caller.py

def create_pending_group(
    self,
    *,
    shield: bool = False,
    mode: Literal[0, 1, 2, 3] = 0,
    timeout: float | None = None,
) -> PendingGroup:
    """Create a new [PendingGroup][async_kernel.pending.PendingGroup].

    This automatically tracks [Pending][async_kernel.pending.Pending] created in the context
    that have opt-in (by including the class `PendingTracker` as a 'tracker'). This includes
    all [Caller][] methods that return a pending.

    The context will not exit until all registered pending are complete. The exit and cancellation behaviour
    is determined by the `mode`.

    Args:
        shield: Shield the pending group from external cancellation.
        mode: The mode.
            - 0: Ignore cancellation of pending, if any pending is cancelled - exit quietly.
            - 1: Cancel if any pending is cancelled - raise PendingCancelled on exit.
            - 2: Cancel if any pending is cancelled - exit quietly.
            - 3: Ignore cancellation of pending, if any pending is cancelled - raise PendingCancelled on exit.
        timeout: An approximate time limit for the context to remain open before cancelling unfinished pending.

    Usage:
        ```python
        async with Caller().create_pending_group() as pg:
            pg.caller.to_thread(my_func)
        ```
    """
    return PendingGroup(shield=shield, mode=mode, timeout=timeout)

StartStopTask ¶

Bases: AsyncContextManagerMixin, Generic[P, T]

A class which provides start/stop functionality to run a coroutine function.

The func is 'stopped' rather being 'cancelled'. This means that the function will run to completion, so that try - finally blocks and shielded CancelScopes can be omitted (Pending cancelled errors can still occur).

The function that is set with set_task_function must accept two positional arguments:

A callable to indicate the function has 'started'.
A ProtectedPending that should be awaited to stop the task.

The sequencing is:

set_task_function: Set func and the caller to use.
start: Start the task.
started: func indicates it has started.
stop: has been called which sets stopping indicating to func that it should commence it's shutdown.
stopped: The task is finished.

For convenience, this class is awaitable and provides an async context manager. Both of which are only available after start has been called. The async context can only be entered once, and will call stop when the context begins to exit. The conxext will return once the task has stopped.

Usage

async def func(started, stop, /, msg: str):
    started()
    print(msg)
    print("Waiting for stop")
    await stop
    print("stopped")


task = StartStopTask().set_task_function(func).start("Hello world")
await task.stop()
# or
async with StartStopTask().set_task_function(func).start():
    do_something
    # When the context is exited, stop will be called.

Returned by:

Reference caller Caller create_start_stop_task

Subclassed by:

Reference
- interface
  - Interface
  - Interface
- messaging BaseMessage

Methods:

set_task_function –

Set the func and optionally specify the caller that runs it.
start –

Schedule func (that was set via set_task_function) to run as a task.
stop –

Stop the task.

Attributes:

started (Fixed[Self, ProtectedPending[None]]) –

A ProtectedPending that is set once func indicates it has started.
stopping (Fixed[Self, ProtectedPending[None]]) –

A ProtectedPending that is set when the method stop is called.
stopped (Fixed[Self, ProtectedPending[T]]) –

A ProtectedPending that is set when the task has stopped.
caller (Caller) –

The caller where the task is running.

Source code in src/async_kernel/caller.py

class StartStopTask(anyio.AsyncContextManagerMixin, Generic[P, T]):
    """A class which provides [start][]/[stop][] functionality to run a coroutine function.

    The `func` is 'stopped' rather being 'cancelled'. This means that the function
    will run to completion, so that `try` - `finally` blocks and shielded CancelScopes can
    be omitted (Pending cancelled errors can still occur).

    The function that is set with [set_task_function][] must accept two positional arguments:

    1. A callable to indicate the function has 'started'.
    2. A `ProtectedPending` that should be awaited to stop the task.

    The sequencing is:

    1. [set_task_function][]: Set `func` and the caller to use.
    2. [start][]: Start the task.
    3. [started][]: `func` indicates it has started.
    4. [stop][StartStopTask.stop]: has been called which sets [stopping][] indicating to `func` that it should commence it's shutdown.
    5. [stopped][]: The task is finished.

    For convenience, this class is awaitable and provides an async context manager.
    Both of which are only available after [start][] has been called. The async context can only
    be entered once, and will call [stop][] when the context begins to exit. The conxext will
    return once the task has stopped.

    Usage:
        ```python
        async def func(started, stop, /, msg: str):
            started()
            print(msg)
            print("Waiting for stop")
            await stop
            print("stopped")


        task = StartStopTask().set_task_function(func).start("Hello world")
        await task.stop()
        # or
        async with StartStopTask().set_task_function(func).start():
            do_something
            # When the context is exited, stop will be called.
        ```
    """

    started: Fixed[Self, ProtectedPending[None]] = Fixed(lambda c: ProtectedPending(info=c["owner"]._info(c["name"])))
    """A `ProtectedPending` that is set once `func` indicates it has started."""

    stopping: Fixed[Self, ProtectedPending[None]] = Fixed(lambda c: ProtectedPending(info=c["owner"]._info(c["name"])))
    """A `ProtectedPending` that is set when the method `stop` is called."""

    stopped: Fixed[Self, ProtectedPending[T]] = Fixed(lambda c: ProtectedPending(info=c["owner"]._info(c["name"])))
    """A `ProtectedPending` that is set when the task has stopped."""

    @property
    def caller(self) -> Caller:
        """The caller where the task is running."""
        try:
            return self._caller_ref()  # pyright: ignore[reportReturnType]
        except AttributeError:
            msg = "The caller has not been set. Call `set_task_function` first!"
            raise RuntimeError(msg) from None

    def set_task_function(
        self,
        func: Callable[Concatenate[Callable[[], None], ProtectedPending[None], P], T | CoroutineType[Any, Any, T]],
        /,
        *,
        caller: Caller | None = None,
    ) -> Self:
        """Set the `func` and optionally specify the `caller` that runs it.

        This should only be called once.

        Args:
            func: The coroutine function to run as a task.
            caller: Specify the caller to run the function.

        Raises:
            RuntimeError: If called more than once.
        """
        if hasattr(self, "_caller_ref"):
            msg = "`func` can only be set once!"
            raise RuntimeError(msg)
        self._caller_ref: ReferenceType[Caller] = weakref.ref(caller or Caller())
        self.caller.stopping.add_done_callback(self.stop)
        self._start_token = ""
        self._func = func
        return self

    def _info(self, name: str) -> str:
        return f"{self.__class__}.{name}"

    def _wait_checks(self):
        if not self.stopping.done() and (hasattr(self, "_start_token") or not hasattr(self, "_caller_ref")):
            msg = "`start` must be called before entering the context! Tip: just add `.start()`."
            raise RuntimeError(msg)

    def __await__(self) -> Generator[Any, None, T]:
        self._wait_checks()
        return self.stopped.__await__()

    @contextlib.asynccontextmanager
    async def __asynccontextmanager__(self) -> AsyncGenerator[Self]:
        self._wait_checks()
        if hasattr(self, "_context_token"):
            msg = "The async context can only be used once!"
            raise RuntimeError(msg)
        if Caller.get_existing() is not self.caller:
            msg = "Async context can only be used by the same caller."
            raise RuntimeError(msg)
        self._context_token = ""

        def stop(_):
            caller.call_direct(scope.cancel, "The task has stopped!")

        async with self.caller as caller:
            try:
                with anyio.CancelScope() as scope:
                    self.stopped.add_done_callback(stop)
                    await self.started.wait(result=False)
                    yield self
                if scope.cancel_called and not self.stopping.done():
                    msg = f"The task stopped early {self._func!r}"
                    raise RuntimeError(msg)
            finally:
                self.stopped.remove_done_callback(stop)
                await self.stop().wait(shield=True)

    def start(self, *args: P.args, **kwargs: P.kwargs) -> Self:
        """Schedule `func` (that was set via [set_task_function][]) to run as a task.

        Args:
            *args: Arguments to use with the function.
            **kwargs: Keyword arguments to use with the function.

        Returns:
            Self: This makes it convenient to chain the call in an async context or await.
        """
        if not hasattr(self, "_func"):
            msg = "The `func` has not been set. Tip: Use the method `set_task_function`."
            raise RuntimeError(msg)
        with contextlib.suppress(AttributeError):
            del self._start_token
            ref = weakref.ref(self)

            def started() -> None:
                if self := ref():
                    if not self.started.done():
                        self.caller.log.debug("Task started %r", self._func)
                        self.started.set_result(None)
                    del self

            def done(pen: Pending) -> None:
                self = ref()
                assert self
                self.caller.log.debug("Task stopped %r", self._func)
                self.caller.stopping.remove_done_callback(self.stop)
                with self.caller._inst_lock:  # pyright: ignore[reportPrivateUsage]
                    self.caller._pen_stop.remove(pen)  # pyright: ignore[reportPrivateUsage]

                if pen.cancelled():
                    self.stopped.cancel(f"The Task was cancelled! {pen}")
                elif e := pen.exception():
                    self.stopped.set_exception(e)
                else:
                    self.stopped.set_result(pen.result())

                try:
                    self.stopped.set_exception(e) if (e := pen.exception()) else self.stopped.set_result(pen.result())
                except Exception as e:
                    self.stopped.set_exception(e)
                pen.metadata.clear()
                del pen, self

            if not self.stopping.done():
                args_ = (started, self.stopping, *args)
                pen = self.caller.schedule_call(self._func, args_, kwargs, None, PendingManager)
                pen.add_done_callback(done)
                pen.set_canceller(self.stop)
                with self.caller._inst_lock:  # pyright: ignore[reportPrivateUsage]
                    self.caller._pen_stop.append(pen)  # pyright: ignore[reportPrivateUsage]
                # Set the canceller again to override `Caller._scheduler`.
                self.started.add_done_callback(lambda _: pen.set_canceller(self.stop))
        return self

    def stop(self, _=None, /) -> ProtectedPending[T]:
        """Stop the task.

        This sets [stopping][] where `func` should respond to the change and exit normally.

        Returns:
            ProtectedPending: Resolves with the result of the function set using [StartStopTask.set_task_function][].
        """
        self.started.cancel("Stopped early!")
        self.stopping.set_result(None)
        with contextlib.suppress(AttributeError):
            if hasattr(self, "_caller_ref"):
                del self._start_token
            self.stopped.cancel("Stopped early!")
        return self.stopped

started `class-attribute` `instance-attribute` ¶

started: Fixed[Self, ProtectedPending[None]] = Fixed(
    lambda c: ProtectedPending(info=c["owner"]._info(c["name"]))
)

A ProtectedPending that is set once func indicates it has started.

Referenced by:

Reference caller StartStopTask

stopping `class-attribute` `instance-attribute` ¶

stopping: Fixed[Self, ProtectedPending[None]] = Fixed(
    lambda c: ProtectedPending(info=c["owner"]._info(c["name"]))
)

A ProtectedPending that is set when the method stop is called.

Referenced by:

Reference caller StartStopTask stop

stopped `class-attribute` `instance-attribute` ¶

stopped: Fixed[Self, ProtectedPending[T]] = Fixed(
    lambda c: ProtectedPending(info=c["owner"]._info(c["name"]))
)

A ProtectedPending that is set when the task has stopped.

Referenced by:

Reference caller StartStopTask

caller `property` ¶

caller: Caller

The caller where the task is running.

set_task_function ¶

set_task_function(
    func: Callable[
        Concatenate[Callable[[], None], ProtectedPending[None], P],
        T | CoroutineType[Any, Any, T],
    ],
    /,
    *,
    caller: Caller | None = None,
) -> Self

Set the func and optionally specify the caller that runs it.

This should only be called once.

Parameters:

func ¶
(Callable[Concatenate[Callable[[], None], ProtectedPending[None], P], T | CoroutineType[Any, Any, T]]) –

The coroutine function to run as a task.
caller ¶
(Caller | None, default: None ) –

Specify the caller to run the function.

Raises:

RuntimeError –

If called more than once.

Referenced by:

Reference caller StartStopTask
- start
- stop

Source code in src/async_kernel/caller.py

def set_task_function(
    self,
    func: Callable[Concatenate[Callable[[], None], ProtectedPending[None], P], T | CoroutineType[Any, Any, T]],
    /,
    *,
    caller: Caller | None = None,
) -> Self:
    """Set the `func` and optionally specify the `caller` that runs it.

    This should only be called once.

    Args:
        func: The coroutine function to run as a task.
        caller: Specify the caller to run the function.

    Raises:
        RuntimeError: If called more than once.
    """
    if hasattr(self, "_caller_ref"):
        msg = "`func` can only be set once!"
        raise RuntimeError(msg)
    self._caller_ref: ReferenceType[Caller] = weakref.ref(caller or Caller())
    self.caller.stopping.add_done_callback(self.stop)
    self._start_token = ""
    self._func = func
    return self

start ¶

start(*args: args, **kwargs: kwargs) -> Self

Schedule func (that was set via set_task_function) to run as a task.

Parameters:

*args ¶
(args, default: () ) –

Arguments to use with the function.
**kwargs ¶
(kwargs, default: {} ) –

Keyword arguments to use with the function.

Returns:

Self ( Self ) –

This makes it convenient to chain the call in an async context or await.

Referenced by:

Reference caller

Source code in src/async_kernel/caller.py

def start(self, *args: P.args, **kwargs: P.kwargs) -> Self:
    """Schedule `func` (that was set via [set_task_function][]) to run as a task.

    Args:
        *args: Arguments to use with the function.
        **kwargs: Keyword arguments to use with the function.

    Returns:
        Self: This makes it convenient to chain the call in an async context or await.
    """
    if not hasattr(self, "_func"):
        msg = "The `func` has not been set. Tip: Use the method `set_task_function`."
        raise RuntimeError(msg)
    with contextlib.suppress(AttributeError):
        del self._start_token
        ref = weakref.ref(self)

        def started() -> None:
            if self := ref():
                if not self.started.done():
                    self.caller.log.debug("Task started %r", self._func)
                    self.started.set_result(None)
                del self

        def done(pen: Pending) -> None:
            self = ref()
            assert self
            self.caller.log.debug("Task stopped %r", self._func)
            self.caller.stopping.remove_done_callback(self.stop)
            with self.caller._inst_lock:  # pyright: ignore[reportPrivateUsage]
                self.caller._pen_stop.remove(pen)  # pyright: ignore[reportPrivateUsage]

            if pen.cancelled():
                self.stopped.cancel(f"The Task was cancelled! {pen}")
            elif e := pen.exception():
                self.stopped.set_exception(e)
            else:
                self.stopped.set_result(pen.result())

            try:
                self.stopped.set_exception(e) if (e := pen.exception()) else self.stopped.set_result(pen.result())
            except Exception as e:
                self.stopped.set_exception(e)
            pen.metadata.clear()
            del pen, self

        if not self.stopping.done():
            args_ = (started, self.stopping, *args)
            pen = self.caller.schedule_call(self._func, args_, kwargs, None, PendingManager)
            pen.add_done_callback(done)
            pen.set_canceller(self.stop)
            with self.caller._inst_lock:  # pyright: ignore[reportPrivateUsage]
                self.caller._pen_stop.append(pen)  # pyright: ignore[reportPrivateUsage]
            # Set the canceller again to override `Caller._scheduler`.
            self.started.add_done_callback(lambda _: pen.set_canceller(self.stop))
    return self

stop ¶

stop(_=None) -> ProtectedPending[T]

Stop the task.

This sets stopping where func should respond to the change and exit normally.

Returns:

ProtectedPending ( ProtectedPending[T] ) –

Resolves with the result of the function set using StartStopTask.set_task_function.

Referenced by:

Reference caller
- StartStopTask
- caller

Source code in src/async_kernel/caller.py

def stop(self, _=None, /) -> ProtectedPending[T]:
    """Stop the task.

    This sets [stopping][] where `func` should respond to the change and exit normally.

    Returns:
        ProtectedPending: Resolves with the result of the function set using [StartStopTask.set_task_function][].
    """
    self.started.cancel("Stopped early!")
    self.stopping.set_result(None)
    with contextlib.suppress(AttributeError):
        if hasattr(self, "_caller_ref"):
            del self._start_token
        self.stopped.cancel("Stopped early!")
    return self.stopped

caller

Caller ¶

MAX_IDLE_POOL_INSTANCES class-attribute instance-attribute ¶

IDLE_WORKER_SHUTDOWN_DURATION class-attribute instance-attribute ¶

started class-attribute instance-attribute ¶

stopping class-attribute instance-attribute ¶

stopped class-attribute instance-attribute ¶

name property ¶

id property ¶

backend property ¶

backend_options property ¶

host property ¶

host_options property ¶

protected property writable ¶

children property ¶

thread property ¶

parent property ¶

__new__ ¶

modifier ¶

**kwargs ¶

id_current classmethod ¶

get_existing classmethod ¶

caller_id ¶

current_pending classmethod ¶

all_callers classmethod ¶

stop ¶

force ¶

get ¶

**kwargs ¶

schedule_call ¶

func ¶

args ¶

kwargs ¶

context ¶

trackers ¶

backend ¶

**metadata ¶

call_later ¶

delay ¶

func ¶

*args ¶

**kwargs ¶

call_soon ¶

func ¶

*args ¶

**kwargs ¶

call_using_backend ¶

backend ¶

func ¶

*args ¶

**kwargs ¶

call_direct ¶

func ¶

*args ¶

**kwargs ¶

to_thread ¶

func ¶

*args ¶

**kwargs ¶

queue_get ¶

queue_call ¶

func ¶

*args ¶

**kwargs ¶

queue_close ¶

func ¶

create_start_stop_task ¶

as_completed async ¶

items ¶

max_concurrent ¶

cancel_unfinished ¶

wait async ¶

items ¶

shield ¶

timeout ¶

return_when ¶

create_pending_group ¶

shield ¶

mode ¶

timeout ¶

MAX_IDLE_POOL_INSTANCES `class-attribute` `instance-attribute` ¶

IDLE_WORKER_SHUTDOWN_DURATION `class-attribute` `instance-attribute` ¶

started `class-attribute` `instance-attribute` ¶

stopping `class-attribute` `instance-attribute` ¶

stopped `class-attribute` `instance-attribute` ¶

name `property` ¶

id `property` ¶

backend `property` ¶

backend_options `property` ¶

host `property` ¶

host_options `property` ¶

protected `property` `writable` ¶

children `property` ¶

thread `property` ¶

parent `property` ¶

new ¶

`modifier` ¶

`kwargs`** ¶

id_current `classmethod` ¶

get_existing `classmethod` ¶

`caller_id` ¶

current_pending `classmethod` ¶

all_callers `classmethod` ¶

`force` ¶

`kwargs`** ¶

`func` ¶

`args` ¶

`kwargs` ¶

`context` ¶

`trackers` ¶

`backend` ¶

`metadata`** ¶

`delay` ¶

`func` ¶

**`*args`** ¶

`kwargs`** ¶

`func` ¶

**`*args`** ¶

`kwargs`** ¶

`backend` ¶

`func` ¶

**`*args`** ¶

`kwargs`** ¶

`func` ¶

**`*args`** ¶

`kwargs`** ¶

`func` ¶

**`*args`** ¶

`kwargs`** ¶

`func` ¶

**`*args`** ¶

`kwargs`** ¶

`func` ¶

as_completed `async` ¶

`items` ¶

`max_concurrent` ¶

`cancel_unfinished` ¶

wait `async` ¶

`items` ¶

`shield` ¶

`timeout` ¶

`return_when` ¶

`shield` ¶

`mode` ¶

`timeout` ¶

started `class-attribute` `instance-attribute` ¶

stopping `class-attribute` `instance-attribute` ¶

stopped `class-attribute` `instance-attribute` ¶

caller `property` ¶

`func` ¶

`caller` ¶

**`*args`** ¶

`kwargs`** ¶