caller

Classes:

PendingGroup –

An asynchronous context manager for tracking Pending that are created in it's context.
Caller –

A thread-local class that facilitates inter-thread function and coroutine scheduling in asynchronous backends (asyncio or trio).

PendingGroup ¶

Bases: PendingTracker, AsyncContextManagerMixin

An asynchronous context manager for tracking Pending that are created in it's context.

Usage

Enter the async context and create new pending.

async with PendingGroup() as pg:
    assert pg.caller.to_thread(lambda: None) in pg.pending

Referenced by:

Reference
- caller Caller create_pending_group
- pending Pending

Returned by:

Reference caller Caller create_pending_group

Methods:

__init__ –

An async context to capture all pending (that opt in) created in the context.
cancel –

Cancel the pending group (internally synchronised).
cancelled –

Returns: If the pending group is marked as cancelled.

Attributes:

cancellation_timeout –

The maximum time to wait for cancelled pending to be done.
caller (Fixed[Self, Caller]) –

The caller where the pending group was instantiated.

Source code in src/async_kernel/pending.py

@final
class PendingGroup(PendingTracker, anyio.AsyncContextManagerMixin):
    """
    An asynchronous context manager for tracking `Pending` that are created in it's context.

    Usage:
        Enter the async context and create new pending.

        ```python
        async with PendingGroup() as pg:
            assert pg.caller.to_thread(lambda: None) in pg.pending
        ```
    """

    _parent_id: None | str = None
    _cancel_scope: anyio.CancelScope
    _cancelled: str | None = None
    _leaving_context: bool = False
    _failed: Fixed[Self, list[Pending]] = Fixed(list)
    cancellation_timeout = 10
    "The maximum time to wait for cancelled pending to be done."

    caller: Fixed[Self, Caller] = Fixed(lambda _: async_kernel.Caller())
    "The caller where the pending group was instantiated."

    def __init__(self, *, shield: bool = False, mode: int = 0) -> None:
        """
        An async context to capture all pending (that opt in) created in the context.

        The pending group will only exit once all pending in the group are `done`.

        Pending can be added to and removed from the group manually.

        Args:
            shield: Passed to the cancel scope.
            mode: The mode.
                - 0: Ignore cancellation of pending.
                - 1: Cancel if any pending is cancelled - raise PendingCancelled on exit.
                - 2: Cancel if any pending is cancelled - exit quietly.
        """
        assert mode in [0, 1, 2]
        self._mode = mode
        self._shield = shield
        self.caller  # noqa: B018
        super().__init__()

    @override
    def __repr__(self) -> str:
        info = " ⛔ cancelled" if self._cancelled else ""
        return f"<PendingGroup at {id(self)}{info} | {len(self.pending)} pending | mode:{self._mode}>"

    @override
    def _activate(self) -> Token[str | None]:
        self._parent_id = None if (parent_id := self._id_contextvar.get()) == self.id else parent_id
        return super()._activate()

    @contextlib.asynccontextmanager
    async def __asynccontextmanager__(self) -> AsyncGenerator[Self]:
        if self._leaving_context:
            msg = f"Re-entry of {self.__class__} is not supported!"
            raise InvalidStateError(msg)
        self._cancel_scope = anyio.CancelScope(shield=self._shield)
        self._all_done = create_async_event()
        token = self._activate()
        try:
            with self._cancel_scope:
                try:
                    yield self
                    self._leaving_context = True
                    if self._pending:
                        await self._all_done
                except (anyio.get_cancelled_exc_class(), Exception) as e:
                    self.cancel(f"An error occurred: {e!r}")
                    raise
            if self._cancelled is not None:
                if exceptions := [e for pen in self._failed if isinstance(e := pen.exception(), Exception)]:
                    msg = f"One or more exceptions occurred in this context! {list(map(str, exceptions))}"
                    raise ExceptionGroup(msg, exceptions)
                if self._mode in [0, 1]:
                    raise PendingCancelled(self._cancelled)
        finally:
            self._leaving_context = True
            self._deactivate(token)
            self._instances.pop(self.id, None)
            if self._pending:
                if self._all_done or self._all_done.cancelled():
                    self._all_done = create_async_event()
                if self._pending and not self._all_done:
                    with anyio.CancelScope(shield=True), anyio.move_on_after(self.cancellation_timeout):
                        await self._all_done

    @override
    def add(self, pen: Pending) -> None:
        if pen not in self._pending:
            self._pending.add(pen)
            pen.add_done_callback(self._on_pending_done)
        if (id_ := self._parent_id) and (parent := self._instances.get(id_)):
            parent.add(pen)

    @override
    def _on_pending_done(self, pen: Pending) -> None:
        try:
            self._pending.remove(pen)
            if pen.cancelled():
                if self._mode in [1, 2]:
                    self.cancel(f"A monitored pending was cancelled {pen=}")
            elif pen.exception():
                self._failed.append(pen)
                self.cancel(f"Exception in member: {pen}")
        except KeyError:
            pass
        if self._leaving_context and not self._pending:
            self._all_done.set()

    def cancel(self, msg: str | None = None) -> bool:
        "Cancel the pending group (internally synchronised)."
        self._cancelled = "\n".join(((self._cancelled or ""), msg or ""))
        if not self._cancel_scope.cancel_called:
            self.caller.call_direct(self._cancel_scope.cancel, msg)
            for pen_ in self.pending:
                pen_.cancel(msg)
        return self.cancelled()

    def cancelled(self) -> bool:
        """Returns: If the pending group is marked as cancelled."""
        return self._cancelled is not None

cancellation_timeout `class-attribute` `instance-attribute` ¶

cancellation_timeout = 10

The maximum time to wait for cancelled pending to be done.

caller `class-attribute` `instance-attribute` ¶

caller: Fixed[Self, Caller] = Fixed(lambda _: Caller())

The caller where the pending group was instantiated.

init ¶

__init__(*, shield: bool = False, mode: int = 0) -> None

The pending group will only exit once all pending in the group are done.

Pending can be added to and removed from the group manually.

Parameters:

shield ¶
(bool, default: False ) –

Passed to the cancel scope.
mode ¶
(int, default: 0 ) –

The mode. - 0: Ignore cancellation of pending. - 1: Cancel if any pending is cancelled - raise PendingCancelled on exit. - 2: Cancel if any pending is cancelled - exit quietly.

Source code in src/async_kernel/pending.py

def __init__(self, *, shield: bool = False, mode: int = 0) -> None:
    """
    An async context to capture all pending (that opt in) created in the context.

    The pending group will only exit once all pending in the group are `done`.

    Pending can be added to and removed from the group manually.

    Args:
        shield: Passed to the cancel scope.
        mode: The mode.
            - 0: Ignore cancellation of pending.
            - 1: Cancel if any pending is cancelled - raise PendingCancelled on exit.
            - 2: Cancel if any pending is cancelled - exit quietly.
    """
    assert mode in [0, 1, 2]
    self._mode = mode
    self._shield = shield
    self.caller  # noqa: B018
    super().__init__()

cancel ¶

cancel(msg: str | None = None) -> bool

Cancel the pending group (internally synchronised).

Source code in src/async_kernel/pending.py

def cancel(self, msg: str | None = None) -> bool:
    "Cancel the pending group (internally synchronised)."
    self._cancelled = "\n".join(((self._cancelled or ""), msg or ""))
    if not self._cancel_scope.cancel_called:
        self.caller.call_direct(self._cancel_scope.cancel, msg)
        for pen_ in self.pending:
            pen_.cancel(msg)
    return self.cancelled()

cancelled ¶

cancelled() -> bool

Returns: If the pending group is marked as cancelled.

Source code in src/async_kernel/pending.py

def cancelled(self) -> bool:
    """Returns: If the pending group is marked as cancelled."""
    return self._cancelled is not None

Caller ¶

Bases: AsyncContextManagerMixin

A thread-local class that facilitates inter-thread function and coroutine scheduling in asynchronous backends (asyncio or trio).

CPython: there is only one caller instance per thread.
Pyodide: Multiple caller instances can exist in the same thread, but is limited to one instance per context.

Multi-eventloop management is supported including:

zero or one host gui event loop.
one or two backends.

Code execution is always done within the context of an asynchronous backend.

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 to access pending as they complete.
Caller.wait: A method to wait for pending to complete with a timeout.
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.

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

Returned by:

Reference

Used by:

Reference
- caller PendingGroup caller
- pending PendingGroup caller

Methods:

__new__ –

Create or retrieve a Caller instance.
start_sync –

Start synchronously.
stop –

Stop the caller cancelling all incomplete tasks.
__asynccontextmanager__ –

The asynchronous context for caller.
id_current –

The id that is used for a caller for the current thread in CPython or context in Pyodide.
get_existing –

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

A classmethod that returns the current pending when called from inside a function scheduled by Caller.
all_callers –

A classmethod to get a list of the callers.
get –

Retrieves an existing child caller by name and backend, or creates a new one if not found.
schedule_call –

Schedule func to be called inside a task running in the caller's thread.
call_later –

Schedule func to be called in caller's event loop copying the current context.
call_soon –

Schedule func to be executed.
call_using_backend –

Schedule func to be executed using the specified backend.
call_direct –

Schedule func to be called in caller's event loop directly.
to_thread –

Call func in a worker thread using the same backend as the current instance.
queue_get –

Returns Pending instance for func where the queue is running.
queue_call –

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

Close the execution queue associated with func.
as_completed –

An iterator to get result as they complete.
wait –

Wait for the results given by items to complete.
create_pending_group –

Create a new PendingGroup instance.

Attributes:

MAX_IDLE_POOL_INSTANCES –

The number of pool instances to leave idle (See also to_thread).
IDLE_WORKER_SHUTDOWN_DURATION –

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

An event that is set when the caller has stopped.
log (LoggerAdapter[Any]) –
iopub_sockets (dict[int, Socket]) –
iopub_url (ClassVar) –
name (str) –

The name of the thread when the caller was created.
id (int) –

The id for the caller.
backend (Backend) –

The backend used by caller.
backend_options (dict | None) –

Options used to create the backend.
host (Hosts | None) –

The name of the gui event loop if there is one.
host_options (dict | None) –

Options used to create the gui event loop.
protected (bool) –

Returns True if the caller is protected from stopping.
zmq_context (Context | None) –

A zmq socket, which if present indicates that an iopub socket is loaded.
running (bool) –

Returns True when the caller is available to run requests.
children (frozenset[Self]) –

A frozenset copy of the instances that were created by the caller.
thread (Thread) –

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

The parent caller if it exists.

Source code in src/async_kernel/caller.py

@final
class Caller(anyio.AsyncContextManagerMixin):
    """
    A thread-local class that facilitates inter-thread function and coroutine scheduling in asynchronous backends (asyncio or trio).

    - CPython: there is only one caller instance per thread.
    - Pyodide: Multiple caller instances can exist in the same thread, but is limited to one instance per context.

    Multi-eventloop management is supported including:

    - zero or one host gui event loop.
    - one or two backends.

    Code execution is always done within the context of an asynchronous backend.

    **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 to access pending as they complete.
    - [Caller.wait][]: A method to wait for pending to complete with a timeout.
    - [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.

    **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 `pool` instances to leave idle (See also [to_thread][async_kernel.caller.Caller.to_thread])."

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

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

    CALLER_MAIN_THREAD_ID: int = id(threading.main_thread())

    _caller_token = contextvars.ContextVar("caller_tokens", default=CALLER_MAIN_THREAD_ID)
    _instances: ClassVar[dict[int, Self]] = {}
    _lock: ClassVar = BinarySemaphore()

    _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
    _state_reprs: ClassVar[dict] = {
        CallerState.initial: "❗ not running",
        CallerState.start_sync: "starting sync",
        CallerState.running: "🏃 running",
        CallerState.stopping: "🏁 stopping",
        CallerState.stopped: "🏁 stopped",
    }
    _zmq_context: zmq.Context[Any] | None = None

    _parent_ref: weakref.ref[Self] | None = None

    # Fixed
    _child_lock = Fixed(BinarySemaphore)
    _children: Fixed[Self, set[Self]] = Fixed(set)
    _tasks: Fixed[Self, set[asyncio.Task]] = Fixed(set)
    _worker_pool: Fixed[Self, deque[Self]] = Fixed(deque)
    _queue_map: Fixed[Self, dict[int, Pending]] = Fixed(dict)
    _queue: Fixed[Self, SingleConsumerAsyncQueue[Pending | tuple[Callable, tuple, dict]]] = Fixed(
        lambda c: SingleConsumerAsyncQueue(c["owner"].backend, reject=c["owner"]._reject)
    )

    stopped = Fixed(Event)
    "An event that is set when the caller has stopped."

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

    log: logging.LoggerAdapter[Any]
    ""
    iopub_sockets: ClassVar[dict[int, zmq.Socket]] = {}
    ""
    iopub_url: ClassVar = "inproc://iopub"
    ""

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

    @property
    def id(self) -> int:
        "The id for the caller."
        return self._caller_id

    @property
    def backend(self) -> Backend:
        "The backend used by caller."
        return self._backend

    @property
    def backend_options(self) -> dict | None:
        "Options used to create the backend."
        return self._backend_options

    @property
    def host(self) -> Hosts | None:
        "The [name][async_kernel.typing.Hosts] of the gui event loop if there is one."
        return self._host

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

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

    @property
    def zmq_context(self) -> zmq.Context | None:
        "A zmq socket, which if present indicates that an iopub socket is loaded."
        return self._zmq_context

    @property
    def running(self) -> bool:
        "Returns `True` when the caller is available to run requests."
        return self._state is CallerState.running

    @property
    def children(self) -> frozenset[Self]:
        """A frozenset copy of the instances that were created by the caller.

        Notes:
            - When the parent is stopped, all children are stopped.
            - All children are stopped prior to the parent exiting its async context.
        """
        return frozenset(self._children)

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

    @property
    def parent(self) -> Self | None:
        "The parent caller if it exists."
        if (ref := self._parent_ref) and (inst := ref()) and not inst.stopped:
            return inst
        return None

    def _get_info(self) -> dict[str, Any]:
        return {
            "name": self._name,
            "backend": str(self._backend),
            "host": self._host,
            "thread": self._thread.name,
            "id": self._caller_id,
        }

    @override
    def __repr__(self) -> str:
        info = " ".join(f"{k}={v!r}" for k, v in self._get_info().items())
        current = "🟢" if self.id_current() == self._caller_id else "〇"  # noqa: RUF001
        protected = "🔐 " if self.protected else " "
        n = len(self._children)
        children = "" if not n else ("1 child" if n == 1 else f"{n} children")
        return f"<Caller {current}{protected}{info}{children}>"

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

        Args:
            modifier: Specifies how the Caller instance should be created or retrieved.

                - "CurrentThread": Automatically create or retrieve the instance.
                - "MainThread": Use the main thread for the Caller.
                - "NewThread": Create a new thread.
                - "manual": Manually create a new instance for the current thread.

            **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.
                - protected: Whether the Caller is protected.
                - zmq_context: ZeroMQ context.
                - log: Logger instance.

        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.
        """
        with cls._lock:
            name, backend = kwargs.get("name", ""), kwargs.get("backend")
            match modifier:
                case "CurrentThread" | "manual":
                    caller_id = cls.id_current()
                case "MainThread":
                    caller_id = cls.CALLER_MAIN_THREAD_ID
                case "NewThread":
                    caller_id = None

            # Locate existing
            if caller_id is not None and (caller := cls._instances.get(caller_id)):
                if modifier == "manual":
                    msg = f"An instance already exists for {caller_id=}"
                    raise RuntimeError(msg)
                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

            # create a new instance
            inst = super().__new__(cls)
            inst._name = name
            inst._backend = Backend(backend or current_async_library())
            inst._host = Hosts(loop) if (loop := kwargs.get("host")) else None
            inst._backend_options = kwargs.get("backend_options")
            inst._host_options = kwargs.get("host_options")
            inst._protected = kwargs.get("protected", False)
            inst._zmq_context = kwargs.get("zmq_context")
            inst.log = kwargs.get("log") or logging.LoggerAdapter(logging.getLogger())
            if (sys.platform == "emscripten") and (caller_id is None):
                caller_id = id(inst)
            if caller_id is not None:
                inst._caller_id = caller_id
                inst._thread = threading.current_thread()

            # finalize
            if modifier != "manual":
                inst.start_sync(no_debug=kwargs.get("no_debug", False))
            assert inst._caller_id
            assert inst._caller_id not in cls._instances
            cls._instances[inst._caller_id] = inst
        return inst

    def start_sync(self, *, no_debug: bool = False) -> None:
        """
        Start synchronously.

        Args:
            no_debug: If debugpy should be disabled in the thread.
        """

        assert self._state is CallerState.initial
        self._state = CallerState.start_sync

        async def run_caller_in_context() -> None:
            if self._state is CallerState.start_sync:
                if no_debug:
                    utils.mark_thread_pydev_do_not_trace()
                self._state = CallerState.initial
                async with self:
                    await self.stopped

        if getattr(self, "_caller_id", None) is not None:
            # An event loop for the current thread.

            if self.backend == Backend.asyncio:
                self._tasks.add(asyncio.create_task(run_caller_in_context()))
            else:
                # trio
                token = current_token()

                def to_thread():
                    utils.mark_thread_pydev_do_not_trace()
                    try:
                        anyio.from_thread.run(run_caller_in_context, token=token)
                    except (BaseExceptionGroup, BaseException) as e:
                        if not "shutdown" not in str(e):
                            raise

                threading.Thread(target=to_thread, daemon=False).start()
        else:
            name = self.name or "async_kernel_caller"
            settings = RunSettings(
                backend=self.backend,
                host=self.host,
                backend_options=self.backend_options,
                host_options=self.host_options,
            )
            args = [run_caller_in_context, (), settings]
            self._thread = threading.Thread(target=async_kernel.event_loop.run, args=args, name=name, daemon=True)
            self._caller_id = id(self._thread)
            self._thread.start()

    def stop(self, *, force: bool = False) -> CallerState:
        """
        Stop the caller cancelling all incomplete tasks.

        Args:
            force: If the caller is protected the call is a no-op unless force=True.
        """
        if (self._protected and not force) or self._state is CallerState.stopping:
            return self._state
        set_stop = self._state in [CallerState.initial, CallerState.start_sync]
        self._state = CallerState.stopping
        self._instances.pop(self._caller_id, None)
        if parent := self.parent:
            try:
                parent._worker_pool.remove(self)
            except ValueError:
                pass
        for child in self.children:
            child.stop(force=True)
        if queue := getattr(self, "_guest_queue", None):
            queue.stop()
        self._queue.stop()
        for func in tuple(self._queue_map):
            self.queue_close(func)
        if set_stop:
            self._state = CallerState.stopped
            self.stopped.set()
        return self._state

    @asynccontextmanager
    async def __asynccontextmanager__(self) -> AsyncGenerator[Self]:
        "The asynchronous context for caller."
        if (state := self._state) is not CallerState.initial:
            if state is CallerState.start_sync:
                msg = 'Already starting! Did you mean to use Caller("manual")?'
            else:
                msg = "Caller cannot be been re-entered!"
            raise RuntimeError(msg)
        if not self._name:
            self._name = threading.current_thread().name
        async with anyio.create_task_group() as tg:
            socket = None
            self._state = CallerState.running
            tg.start_soon(self._scheduler, self.backend, self._queue, tg)
            if self._zmq_context:
                socket = self._zmq_context.socket(1)  # zmq.SocketType.PUB
                socket.linger = 50
                socket.connect(self.iopub_url)
                self.iopub_sockets[self._caller_id] = socket
            try:
                yield self
            finally:
                if stop_guest := getattr(self, "_stop_guest", None):
                    with anyio.CancelScope(shield=True):
                        await stop_guest()
                self.stop(force=True)
                if socket:
                    self.iopub_sockets.pop(self._caller_id, None)
                    socket.close()
                with anyio.CancelScope(shield=True):
                    while self._children:
                        await self._children.pop().stopped
                    if parent := self.parent:
                        parent._children.discard(self)
                    self._state = CallerState.stopped
                    self.stopped.set()
                    await self._checkpoint()

    async def _scheduler(self, backend: Backend, queue: SingleConsumerAsyncQueue, tg: TaskGroup) -> None:
        """
        An asynchronous coroutine to schedule or execute functions as they arrive in the queue.

        It handles two different types of items in the queue:
            - tuple: A tuple of func, args, kwargs intended to be called directly in the scheduler.
            - Pending: A pending that is to be started as a 'task' with the backend wrapped with

        Args:
            backend: The backend where the scheduler is operating.
            tg: The task group used to manage concurrent tasks.
            task_status: Used to signal when the scheduler has started.
        """
        kwgs = {}
        asyncio_backend = backend == Backend.asyncio
        if asyncio_backend:
            loop = asyncio.get_running_loop()
            coro = asyncio.sleep(0)
            try:
                await loop.create_task(coro, eager_start=True)  # pyright: ignore[reportCallIssue]
                kwgs["eager_start"] = True
            except Exception:
                coro.close()
        try:
            async for item in queue:
                result = None
                if not isinstance(item, Pending):
                    try:
                        result = item[0](*item[1], **item[2])
                        if inspect.iscoroutine(result):
                            await result
                    except Exception as e:
                        self.log.exception("Direct call failed", exc_info=e)
                else:
                    if asyncio_backend:
                        task = loop.create_task(self._wrap_call(item), context=item.context, **kwgs)  # pyright: ignore[reportPossiblyUnboundVariable]
                        if not task.done():
                            self._tasks.add(task)
                            task.add_done_callback(self._tasks.discard)
                        del task
                    else:
                        if context := item.context:
                            context.run(tg.start_soon, self._wrap_call, item)
                        else:
                            tg.start_soon(self._wrap_call, item)
                del item, result
        finally:
            if asyncio_backend:
                for task in self._tasks:
                    task.cancel()
            tg.cancel_scope.cancel()

    async def _wrap_call(self, pen: Pending) -> None:
        """
        Asynchronously executes the function associated with the given pending, handling cancellation, delays, and exceptions.

        Args:
            pen: The [async_kernel.Pending][] object containing metadata about the function to execute, its arguments, and execution state.
        """
        if not pen.done():
            md = pen.metadata
            token_pending = self._pending_var.set(pen)
            token_ident = self._caller_token.set(self._caller_id)
            try:
                with anyio.CancelScope() as scope:
                    pen.set_canceller(lambda msg: self.call_direct(scope.cancel, msg))
                    # Call later.
                    if (delay := md.get("delay")) and ((delay := delay - time.monotonic() + md["start_time"]) > 0):
                        await anyio.sleep(delay)
                    # Call now.
                    try:
                        result = md["func"](*md["args"], **md["kwargs"])
                        if inspect.iscoroutine(result):
                            result = await result
                        pen.set_result(result)
                    # Cancelled.
                    except anyio.get_cancelled_exc_class():
                        if not pen.cancelled():
                            pen.cancel()
                        pen.set_result(None)
                    # Catch exceptions.
                    except Exception as e:
                        pen.set_exception(e)
            except Exception as e:
                if not pen.done():
                    pen.set_exception(e)
            finally:
                self._pending_var.reset(token_pending)
                self._caller_token.reset(token_ident)

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

    @classmethod
    def _start_idle_worker_cleanup_thead(cls) -> None:
        "A single thread to shutdown idle workers that have not been used for an extended duration."
        if cls.IDLE_WORKER_SHUTDOWN_DURATION > 0 and not hasattr(cls, "_thread_cleanup_idle_workers"):

            def _cleanup_workers():
                utils.mark_thread_pydev_do_not_trace()
                n = 0
                cutoff = time.monotonic()
                time.sleep(cls.IDLE_WORKER_SHUTDOWN_DURATION)
                for caller in tuple(cls._instances.values()):
                    for worker in frozenset(caller._worker_pool):
                        n += 1
                        if worker._idle_time < cutoff:
                            with contextlib.suppress(IndexError):
                                caller._worker_pool.remove(worker)
                                worker.stop()
                if n:
                    _cleanup_workers()
                else:
                    del cls._thread_cleanup_idle_workers

            cls._thread_cleanup_idle_workers = threading.Thread(target=_cleanup_workers, daemon=True)
            cls._thread_cleanup_idle_workers.start()

    @classmethod
    def id_current(cls) -> int:
        "The id that is used for a caller for the current thread in CPython or context in Pyodide."
        if sys.platform == "emscripten":
            return cls._caller_token.get()
        return id(threading.current_thread())

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

        Args:
            caller_id: The id of the caller which in CPython is also the the id of the thread in which it is running.
        """
        caller_id = cls.id_current() if caller_id is None else caller_id
        with cls._lock:
            return cls._instances.get(caller_id)

    @classmethod
    def current_pending(cls) -> Pending[Any] | None:
        """A [classmethod][] that returns the current pending when called from inside a function scheduled by Caller."""
        return cls._pending_var.get()

    @classmethod
    def all_callers(cls, running_only: bool = True) -> list[Caller]:
        """
        A [classmethod][] to get a list of the callers.

        Args:
            running_only: Restrict the list to callers that are active (running in an async context).
        """
        return [caller for caller in Caller._instances.values() if caller.running or not running_only]

    def get(self, **kwargs: Unpack[CallerCreateOptions]) -> Self:
        """
        Retrieves an existing child caller by name and backend, or creates a new one if not found.

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

        Returns:
            Self: The retrieved or newly created caller instance.

        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 this instance.
            - If 'backend' or 'zmq_context' are not specified they are copied from this instance.
        """

        with self._child_lock:
            if name := kwargs.get("name"):
                for caller in self.children:
                    if caller.name == name:
                        if (backend := kwargs.get("backend")) and caller.backend != backend:
                            msg = f"Backend mismatch! {backend=} {caller.backend=}"
                            raise RuntimeError(msg)
                        if (host := kwargs.get("host")) and caller.host != host:
                            msg = f"Host mismatch! {host=} {caller.host=}"
                            raise RuntimeError(msg)
                        return caller
            if "backend" not in kwargs:
                kwargs["backend"] = self._backend
                kwargs["backend_options"] = self.backend_options
            if "zmq_context" not in kwargs and self._zmq_context:
                kwargs["zmq_context"] = self._zmq_context
            caller = self.__class__("NewThread", **kwargs)
            self._children.add(caller)
            caller._parent_ref = weakref.ref(self)
            return caller

    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,  # pyright: ignore[reportInvalidTypeForm]
        /,
        **metadata: Any,
    ) -> Pending[T]:
        """
        Schedule `func` to be called inside a task running in the caller's thread.

        The methods [call_soon][Caller.call_soon] and [call_later][Caller.call_later]
        use this method in the background,  they should be used in preference to this method since they provide type hinting for the arguments.

        Args:
            func: The function to be called. If it returns a coroutine, it will be awaited and its result will be returned.
            args: Arguments corresponding to in the call to  `func`.
            kwargs: Keyword arguments to use with in the call to `func`.
            context: The context to use, if not provided the current context is used.
            trackers: The tracker subclasses of active trackers which to add the pending.
            **metadata: Additional metadata to store in the instance.
        """
        pen = Pending(context, trackers, func=func, args=args, kwargs=kwargs, caller=self, **metadata)
        if backend is NoValue or (backend := Backend(backend)) is self.backend:
            self._queue.append(pen)
        else:
            if not hasattr(self, "_guest_queue"):
                self._guest_queue = SingleConsumerAsyncQueue(backend, reject=self._reject)
                self.schedule_call(
                    self._guest_backend_loop, (), {"backend": backend, "queue": self._guest_queue}, None, ()
                )
            self._guest_queue.append(pen)
        return pen

    async def _guest_backend_loop(self, backend: Backend, queue: SingleConsumerAsyncQueue) -> None:
        async def _guest_scheduler() -> None:
            async with anyio.create_task_group() as tg:
                tg.start_soon(self._scheduler, Backend(backend), queue, tg)

        if self._state is CallerState.running:
            # Prefer callbacks from the host.
            host = Host.current(self.thread)
            run_soon_threadsafe_queue = SingleConsumerAsyncQueue(self.backend)
            start_guest_run = get_start_guest_run(backend)
            start_guest_run(
                _guest_scheduler,
                done_callback=lambda _: run_soon_threadsafe_queue.stop(),
                run_sync_soon_threadsafe=host.run_sync_soon_threadsafe if host else run_soon_threadsafe_queue.append,
                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,
            )
            if pen := self.current_pending():
                self._stop_guest = lambda: [queue.stop(), pen.wait(result=False)][1]
            with anyio.CancelScope(shield=True):
                async for func in run_soon_threadsafe_queue:
                    try:
                        func()
                    except Exception:
                        pass

    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 caller's event loop copying the current context.

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

        Info:
            All call arguments are packed into the instance's metadata.
        """
        return self.schedule_call(func, args, kwargs, delay=delay, start_time=time.monotonic())

    def call_soon(
        self,
        func: Callable[P, T | CoroutineType[Any, Any, T]],
        /,
        *args: P.args,
        **kwargs: P.kwargs,
    ) -> Pending[T]:
        """
        Schedule func to be executed.

        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.

        This methods enables coroutines written for a specific function to be run irresective
        of the callers backend.

        - `backend == caller.backend` - `func` is executed via [Caller.call_soon][].
        - `backend != caller.backend` - `func` is executed with a backend running as a guest.

        Args:
            backend: The backend in which `func` must be executed.
            func: The function.
            *args: Arguments to use with `func`.
            **kwargs: Keyword arguments to use with `func`.

        Notes:

            - **Only use this to execute coroutines that require a specific backend to run in the caller's thread.**
            - Where possible use a separate caller/thread with [Caller.get][] instead.
        """
        return self.schedule_call(func, args, kwargs, None, PendingTracker, Backend(backend))

    def call_direct(
        self,
        func: Callable[P, T | CoroutineType[Any, Any, T]],
        /,
        *args: P.args,
        **kwargs: P.kwargs,
    ) -> None:
        """
        Schedule `func` to be called in caller's event loop directly.

        Use this for short-running function calls only.

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

        Warning:

            **Use this method for lightweight calls only!**
        """
        self._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 worker thread using the same backend as the current instance.

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

        Notes:
            - A minimum number of caller instances are retained for this method.
            - Async code run inside func should use taskgroups for creating task.
        """

        def _to_thread_on_done(_) -> None:
            if not caller.stopped and self.running:
                if len(self._worker_pool) < self.MAX_IDLE_POOL_INSTANCES:
                    caller._idle_time = time.monotonic()
                    self._worker_pool.append(caller)
                    self._start_idle_worker_cleanup_thead()
                else:
                    caller.stop()

        try:
            caller = self._worker_pool.popleft()
        except IndexError:
            caller = self.get()
        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 `Pending` instance for `func` where the queue 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 function to queue the execution of `func` in a queue unique to it and the caller instance.

        This sets up a long-lived task to provide a fast pathway for repetitive calls to a function
        where call order is also respected.

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

        Notes:
            - The queue runs inside a pending 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 registered with PendingManager subclassed
                trackers and **not** PendingGroup.
        """
        if not (pen_ := self._queue_map.get(key := hash(func))):
            queue = SingleConsumerAsyncQueue[tuple[Callable, tuple, dict]](self.backend)
            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
                try:
                    async for item in queue:
                        try:
                            await item[0](*item[1], **item[2])
                        except TypeError:
                            continue
                        except (anyio.get_cancelled_exc_class(), Exception) as e:
                            if pen.cancelled():
                                raise
                            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 execution queue associated with `func`.

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

    async def _checkpoint(self) -> None:
        "Yield to the event loop."
        if not self._use_safe_checkpoint:
            try:
                if self.backend is Backend.trio:
                    await trio_checkpoint()
                else:
                    await asyncio_checkpoint()
            except Exception:
                self._use_safe_checkpoint = True
            else:
                return
        await async_checkpoint(force=True)

    async def as_completed(
        self,
        items: Iterable[Awaitable[T]] | AsyncGenerator[Awaitable[T]],
        *,
        max_concurrent: NoValue | int = NoValue,  # pyright: ignore[reportInvalidTypeForm]
        cancel_unfinished: bool = True,
    ) -> AsyncGenerator[Pending[T], Any]:
        """
        An iterator to get result as they complete.

        Args:
            items: Either a container with existing results or generator of Pendings.
            max_concurrent: The maximum number of concurrent results to monitor at a time.
                This is useful when `items` is a generator utilising [Caller.to_thread][].
                By default this will limit to `Caller.MAX_IDLE_POOL_INSTANCES`.
            cancel_unfinished: Cancel any `pending` when exiting.

        Tip:
            1. Pass a generator if you wish to limit the number result jobs when calling to_thread/to_task etc.
            2. Pass a container with all results when the limiter is not relevant.
        """
        resume = noop
        queue: SingleConsumerAsyncQueue[Pending[T]] = SingleConsumerAsyncQueue(self.backend)
        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 queue.stopped and (pen := await anext(gen, None) if is_async else next(gen, None)) is not None:
                if pen is pen_current:
                    queue.stop()
                    msg = "Waiting for the pending in which it is running would result in deadlock!"
                    raise RuntimeError(msg)
                if not isinstance(pen, Pending):
                    pen = cast("Pending[T]", self.call_soon(await_for, pen))
                pen.add_done_callback(queue.append)
                if not pen.done():
                    unfinished.add(pen)
                    if max_concurrent_ and len(unfinished) == max_concurrent_:
                        event = create_async_event()
                        resume = event.set
                        if len(unfinished) == max_concurrent_:
                            await event
                        resume = noop
            if not queue.queue and not unfinished:
                queue.stop()

        pen_ = self.call_soon(scheduler)
        try:
            async for pen in queue:
                unfinished.discard(pen)
                yield pen
                if pen_.done() and not unfinished and not queue.queue:
                    break
                else:
                    if max_concurrent_ and len(unfinished) < max_concurrent_:
                        resume()
            pen_.result()
        finally:
            queue.stop()
            pen_.cancel()
            for pen in unfinished:
                pen.remove_done_callback(queue.append)
                if cancel_unfinished:
                    pen.cancel("Cancelled by as_completed")
            with anyio.CancelScope():
                await pen_.wait(result=False)

    async def wait(
        self,
        items: Iterable[Awaitable[T]],
        *,
        timeout: float | None = None,
        return_when: Literal["FIRST_COMPLETED", "FIRST_EXCEPTION", "ALL_COMPLETED"] = "ALL_COMPLETED",
    ) -> tuple[set[Pending[T]], set[Pending[T]]]:
        """
        Wait for the results given by items to complete.

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

        Args:
            items: An iterable of results to wait for.
            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:
                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:
            with anyio.move_on_after(timeout):
                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] = 0) -> PendingGroup:
        """
        Create a new [PendingGroup][async_kernel.pending.PendingGroup] instance.

        The pending group registers all pending created in its context that opt in (via trackers). On exiting
        the context, it will await all remaining pending to complete. The exit and cancellation behaviour of
        the instance is a function of `mode`.

        If any pending result in exception, the pending group and all registered pending are cancelled.
        If the pending group context is cancelled or results in exception, all pending in the group are
        also cancelled.

        Args:
            shield: Shield the pending group from external cancellation.
            mode: The mode.
                - 0: Ignore cancellation of pending.
                - 1: Cancel if any pending is cancelled - raise PendingCancelled on exit.
                - 2: Cancel if any pending is cancelled - exit quietly.

        Usage:

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

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

MAX_IDLE_POOL_INSTANCES = 10

The number of pool instances to leave idle (See also to_thread).

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

IDLE_WORKER_SHUTDOWN_DURATION = 0 if 'pytest' in modules else 60

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

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

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

stopped = Fixed(Event)

An event that is set when the caller has stopped.

log `instance-attribute` ¶

log: LoggerAdapter[Any]

iopub_sockets `class-attribute` ¶

iopub_sockets: dict[int, Socket] = {}

iopub_url `class-attribute` `instance-attribute` ¶

iopub_url: ClassVar = 'inproc://iopub'

name `property` ¶

name: str

The name of the thread when the caller was created.

id `property` ¶

id: int

The id for the caller.

backend `property` ¶

backend: Backend

The backend used by caller.

backend_options `property` ¶

backend_options: dict | None

Options used to create the backend.

host `property` ¶

host: Hosts | None

The name of the gui event loop if there is one.

host_options `property` ¶

host_options: dict | None

Options used to create the gui event loop.

protected `property` ¶

protected: bool

Returns True if the caller is protected from stopping.

zmq_context `property` ¶

zmq_context: Context | None

A zmq socket, which if present indicates that an iopub socket is loaded.

running `property` ¶

running: bool

Returns True when the caller is available to run requests.

children `property` ¶

children: frozenset[Self]

A frozenset copy of the instances that were created by the caller.

Notes

When the parent is stopped, all children are stopped.
All children are stopped prior to the parent exiting its async context.

thread `property` ¶

thread: Thread

The thread where the caller is running.

parent `property` ¶

parent: Self | None

The parent caller if it exists.

new ¶

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

Create or retrieve a Caller instance.

Parameters:

modifier ¶
(Literal['CurrentThread', 'MainThread', 'NewThread', 'manual'], default: 'CurrentThread' ) –
Specifies how the Caller instance should be created or retrieved.
- "CurrentThread": Automatically create or retrieve the instance.
- "MainThread": Use the main thread for the Caller.
- "NewThread": Create a new thread.
- "manual": Manually create a new instance for the current thread.
**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. - protected: Whether the Caller is protected. - zmq_context: ZeroMQ context. - log: Logger instance.

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", "manual"] = "CurrentThread",
    /,
    **kwargs: Unpack[CallerCreateOptions],
) -> Self:
    """
    Create or retrieve a Caller instance.

    Args:
        modifier: Specifies how the Caller instance should be created or retrieved.

            - "CurrentThread": Automatically create or retrieve the instance.
            - "MainThread": Use the main thread for the Caller.
            - "NewThread": Create a new thread.
            - "manual": Manually create a new instance for the current thread.

        **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.
            - protected: Whether the Caller is protected.
            - zmq_context: ZeroMQ context.
            - log: Logger instance.

    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.
    """
    with cls._lock:
        name, backend = kwargs.get("name", ""), kwargs.get("backend")
        match modifier:
            case "CurrentThread" | "manual":
                caller_id = cls.id_current()
            case "MainThread":
                caller_id = cls.CALLER_MAIN_THREAD_ID
            case "NewThread":
                caller_id = None

        # Locate existing
        if caller_id is not None and (caller := cls._instances.get(caller_id)):
            if modifier == "manual":
                msg = f"An instance already exists for {caller_id=}"
                raise RuntimeError(msg)
            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

        # create a new instance
        inst = super().__new__(cls)
        inst._name = name
        inst._backend = Backend(backend or current_async_library())
        inst._host = Hosts(loop) if (loop := kwargs.get("host")) else None
        inst._backend_options = kwargs.get("backend_options")
        inst._host_options = kwargs.get("host_options")
        inst._protected = kwargs.get("protected", False)
        inst._zmq_context = kwargs.get("zmq_context")
        inst.log = kwargs.get("log") or logging.LoggerAdapter(logging.getLogger())
        if (sys.platform == "emscripten") and (caller_id is None):
            caller_id = id(inst)
        if caller_id is not None:
            inst._caller_id = caller_id
            inst._thread = threading.current_thread()

        # finalize
        if modifier != "manual":
            inst.start_sync(no_debug=kwargs.get("no_debug", False))
        assert inst._caller_id
        assert inst._caller_id not in cls._instances
        cls._instances[inst._caller_id] = inst
    return inst

start_sync ¶

start_sync(*, no_debug: bool = False) -> None

Start synchronously.

Parameters:

no_debug ¶
(bool, default: False ) –

If debugpy should be disabled in the thread.

Source code in src/async_kernel/caller.py

def start_sync(self, *, no_debug: bool = False) -> None:
    """
    Start synchronously.

    Args:
        no_debug: If debugpy should be disabled in the thread.
    """

    assert self._state is CallerState.initial
    self._state = CallerState.start_sync

    async def run_caller_in_context() -> None:
        if self._state is CallerState.start_sync:
            if no_debug:
                utils.mark_thread_pydev_do_not_trace()
            self._state = CallerState.initial
            async with self:
                await self.stopped

    if getattr(self, "_caller_id", None) is not None:
        # An event loop for the current thread.

        if self.backend == Backend.asyncio:
            self._tasks.add(asyncio.create_task(run_caller_in_context()))
        else:
            # trio
            token = current_token()

            def to_thread():
                utils.mark_thread_pydev_do_not_trace()
                try:
                    anyio.from_thread.run(run_caller_in_context, token=token)
                except (BaseExceptionGroup, BaseException) as e:
                    if not "shutdown" not in str(e):
                        raise

            threading.Thread(target=to_thread, daemon=False).start()
    else:
        name = self.name or "async_kernel_caller"
        settings = RunSettings(
            backend=self.backend,
            host=self.host,
            backend_options=self.backend_options,
            host_options=self.host_options,
        )
        args = [run_caller_in_context, (), settings]
        self._thread = threading.Thread(target=async_kernel.event_loop.run, args=args, name=name, daemon=True)
        self._caller_id = id(self._thread)
        self._thread.start()

stop ¶

stop(*, force: bool = False) -> CallerState

Stop the caller cancelling all incomplete tasks.

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) -> CallerState:
    """
    Stop the caller cancelling all incomplete tasks.

    Args:
        force: If the caller is protected the call is a no-op unless force=True.
    """
    if (self._protected and not force) or self._state is CallerState.stopping:
        return self._state
    set_stop = self._state in [CallerState.initial, CallerState.start_sync]
    self._state = CallerState.stopping
    self._instances.pop(self._caller_id, None)
    if parent := self.parent:
        try:
            parent._worker_pool.remove(self)
        except ValueError:
            pass
    for child in self.children:
        child.stop(force=True)
    if queue := getattr(self, "_guest_queue", None):
        queue.stop()
    self._queue.stop()
    for func in tuple(self._queue_map):
        self.queue_close(func)
    if set_stop:
        self._state = CallerState.stopped
        self.stopped.set()
    return self._state

asynccontextmanager `async` ¶

__asynccontextmanager__() -> AsyncGenerator[Self]

The asynchronous context for caller.

Source code in src/async_kernel/caller.py

@asynccontextmanager
async def __asynccontextmanager__(self) -> AsyncGenerator[Self]:
    "The asynchronous context for caller."
    if (state := self._state) is not CallerState.initial:
        if state is CallerState.start_sync:
            msg = 'Already starting! Did you mean to use Caller("manual")?'
        else:
            msg = "Caller cannot be been re-entered!"
        raise RuntimeError(msg)
    if not self._name:
        self._name = threading.current_thread().name
    async with anyio.create_task_group() as tg:
        socket = None
        self._state = CallerState.running
        tg.start_soon(self._scheduler, self.backend, self._queue, tg)
        if self._zmq_context:
            socket = self._zmq_context.socket(1)  # zmq.SocketType.PUB
            socket.linger = 50
            socket.connect(self.iopub_url)
            self.iopub_sockets[self._caller_id] = socket
        try:
            yield self
        finally:
            if stop_guest := getattr(self, "_stop_guest", None):
                with anyio.CancelScope(shield=True):
                    await stop_guest()
            self.stop(force=True)
            if socket:
                self.iopub_sockets.pop(self._caller_id, None)
                socket.close()
            with anyio.CancelScope(shield=True):
                while self._children:
                    await self._children.pop().stopped
                if parent := self.parent:
                    parent._children.discard(self)
                self._state = CallerState.stopped
                self.stopped.set()
                await self._checkpoint()

id_current `classmethod` ¶

id_current() -> int

The id that is used for a caller for the current thread in CPython or context in Pyodide.

Referenced by:

Reference caller Caller

Source code in src/async_kernel/caller.py

@classmethod
def id_current(cls) -> int:
    "The id that is used for a caller for the current thread in CPython or context in Pyodide."
    if sys.platform == "emscripten":
        return cls._caller_token.get()
    return id(threading.current_thread())

get_existing `classmethod` ¶

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

A classmethod to get the caller instance from the corresponding thread if it exists.

Parameters:

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

The id of the caller which in CPython is also the the id of the thread in which it is running.

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 from the corresponding thread if it exists.

    Args:
        caller_id: The id of the caller which in CPython is also the the id of the thread in which it is running.
    """
    caller_id = cls.id_current() if caller_id is None else caller_id
    with cls._lock:
        return cls._instances.get(caller_id)

current_pending `classmethod` ¶

current_pending() -> Pending[Any] | None

A classmethod that returns the current pending when called from inside a function scheduled by Caller.

Referenced by:

Reference caller Caller

Source code in src/async_kernel/caller.py

@classmethod
def current_pending(cls) -> Pending[Any] | None:
    """A [classmethod][] that returns the current pending when called from inside a function scheduled by Caller."""
    return cls._pending_var.get()

all_callers `classmethod` ¶

all_callers(running_only: bool = True) -> list[Caller]

A classmethod to get a list of the callers.

Parameters:

running_only ¶
(bool, default: True ) –

Restrict the list to callers that are active (running in an async context).

Referenced by:

Reference caller Caller

Source code in src/async_kernel/caller.py

@classmethod
def all_callers(cls, running_only: bool = True) -> list[Caller]:
    """
    A [classmethod][] to get a list of the callers.

    Args:
        running_only: Restrict the list to callers that are active (running in an async context).
    """
    return [caller for caller in Caller._instances.values() if caller.running or not running_only]

get ¶

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

Retrieves an existing child caller by name and backend, or creates a new one if not found.

Parameters:

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

Options for creating or retrieving a caller instance.

Returns:

Self ( Self ) –

The retrieved or newly created caller instance.

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 this instance.
If 'backend' or 'zmq_context' are not specified they are copied from this instance.

Referenced by:

Reference caller Caller

Source code in src/async_kernel/caller.py

def get(self, **kwargs: Unpack[CallerCreateOptions]) -> Self:
    """
    Retrieves an existing child caller by name and backend, or creates a new one if not found.

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

    Returns:
        Self: The retrieved or newly created caller instance.

    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 this instance.
        - If 'backend' or 'zmq_context' are not specified they are copied from this instance.
    """

    with self._child_lock:
        if name := kwargs.get("name"):
            for caller in self.children:
                if caller.name == name:
                    if (backend := kwargs.get("backend")) and caller.backend != backend:
                        msg = f"Backend mismatch! {backend=} {caller.backend=}"
                        raise RuntimeError(msg)
                    if (host := kwargs.get("host")) and caller.host != host:
                        msg = f"Host mismatch! {host=} {caller.host=}"
                        raise RuntimeError(msg)
                    return caller
        if "backend" not in kwargs:
            kwargs["backend"] = self._backend
            kwargs["backend_options"] = self.backend_options
        if "zmq_context" not in kwargs and self._zmq_context:
            kwargs["zmq_context"] = self._zmq_context
        caller = self.__class__("NewThread", **kwargs)
        self._children.add(caller)
        caller._parent_ref = weakref.ref(self)
        return caller

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]

Schedule func to be called inside a task running in the caller's thread.

The methods call_soon and call_later use this method in the background, they should be used in preference to this method since they provide type hinting for the arguments.

Parameters:

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

The function to be called. If it returns a coroutine, it will be awaited and its result will be returned.
args ¶
(tuple) –

Arguments corresponding to in the call to func.
kwargs ¶
(dict) –

Keyword arguments to use with in the call to func.
context ¶
(Context | None, default: None ) –

The context to use, if not provided the current context is used.
trackers ¶
(type[PendingTracker] | tuple[type[PendingTracker], ...], default: PendingTracker ) –

The tracker subclasses of active trackers which to add the pending.
**metadata ¶
(Any, default: {} ) –

Additional metadata to store in the instance.

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,  # pyright: ignore[reportInvalidTypeForm]
    /,
    **metadata: Any,
) -> Pending[T]:
    """
    Schedule `func` to be called inside a task running in the caller's thread.

    The methods [call_soon][Caller.call_soon] and [call_later][Caller.call_later]
    use this method in the background,  they should be used in preference to this method since they provide type hinting for the arguments.

    Args:
        func: The function to be called. If it returns a coroutine, it will be awaited and its result will be returned.
        args: Arguments corresponding to in the call to  `func`.
        kwargs: Keyword arguments to use with in the call to `func`.
        context: The context to use, if not provided the current context is used.
        trackers: The tracker subclasses of active trackers which to add the pending.
        **metadata: Additional metadata to store in the instance.
    """
    pen = Pending(context, trackers, func=func, args=args, kwargs=kwargs, caller=self, **metadata)
    if backend is NoValue or (backend := Backend(backend)) is self.backend:
        self._queue.append(pen)
    else:
        if not hasattr(self, "_guest_queue"):
            self._guest_queue = SingleConsumerAsyncQueue(backend, reject=self._reject)
            self.schedule_call(
                self._guest_backend_loop, (), {"backend": backend, "queue": self._guest_queue}, None, ()
            )
        self._guest_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 caller's event loop copying the current context.

Parameters:

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

The function.
delay ¶
(float) –

The minimum delay to add between submission and execution.
*args ¶
(args, default: () ) –

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

Keyword arguments to use with func.

Info

All call arguments are packed into the instance's metadata.

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 caller's event loop copying the current context.

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

    Info:
        All call arguments are packed into the instance's metadata.
    """
    return self.schedule_call(func, args, kwargs, delay=delay, start_time=time.monotonic())

call_soon ¶

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

Schedule func to be executed.

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
- caller Caller
- typing RunMode task

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 executed.

    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.

This methods enables coroutines written for a specific function to be run irresective of the callers backend.

backend == caller.backend - func is executed via Caller.call_soon.
backend != caller.backend - func is executed with a backend running as a guest.

Parameters:

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

The backend in which func must be executed.
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:

- **Only use this to execute coroutines that require a specific backend to run in the caller's thread.**
- Where possible use a separate caller/thread with [Caller.get][] instead.

Referenced by:

Usage Usage Features
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.

    This methods enables coroutines written for a specific function to be run irresective
    of the callers backend.

    - `backend == caller.backend` - `func` is executed via [Caller.call_soon][].
    - `backend != caller.backend` - `func` is executed with a backend running as a guest.

    Args:
        backend: The backend in which `func` must be executed.
        func: The function.
        *args: Arguments to use with `func`.
        **kwargs: Keyword arguments to use with `func`.

    Notes:

        - **Only use this to execute coroutines that require a specific backend to run in the caller's thread.**
        - Where possible use a separate caller/thread with [Caller.get][] instead.
    """
    return self.schedule_call(func, args, kwargs, None, PendingTracker, Backend(backend))

call_direct ¶

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

Schedule func to be called in caller's event loop directly.

Use this for short-running function calls only.

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.

Warning:

**Use this method for lightweight calls only!**

Referenced by:

Reference caller Caller

Source code in src/async_kernel/caller.py

def call_direct(
    self,
    func: Callable[P, T | CoroutineType[Any, Any, T]],
    /,
    *args: P.args,
    **kwargs: P.kwargs,
) -> None:
    """
    Schedule `func` to be called in caller's event loop directly.

    Use this for short-running function calls only.

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

    Warning:

        **Use this method for lightweight calls only!**
    """
    self._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 worker thread using the same backend as the current instance.

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 minimum number of caller instances are retained for this method.
Async code run inside func should use taskgroups for creating task.

Referenced by:

Reference
- caller Caller
- typing RunMode

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 worker thread using the same backend as the current instance.

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

    Notes:
        - A minimum number of caller instances are retained for this method.
        - Async code run inside func should use taskgroups for creating task.
    """

    def _to_thread_on_done(_) -> None:
        if not caller.stopped and self.running:
            if len(self._worker_pool) < self.MAX_IDLE_POOL_INSTANCES:
                caller._idle_time = time.monotonic()
                self._worker_pool.append(caller)
                self._start_idle_worker_cleanup_thead()
            else:
                caller.stop()

    try:
        caller = self._worker_pool.popleft()
    except IndexError:
        caller = self.get()
    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 Pending instance for func where the queue 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 `Pending` instance for `func` where the queue 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 function to queue the execution of func in a queue unique to it and the caller instance.

This sets up a long-lived task to provide a fast pathway for repetitive calls to a function where call order is also respected.

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 pending 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 registered with PendingManager subclassed trackers and not PendingGroup.

Referenced by:

Reference
- caller Caller
- typing RunMode queue
Thread Safety of Built-in Types async_kernel.caller.Caller Guarantees

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 function to queue the execution of `func` in a queue unique to it and the caller instance.

    This sets up a long-lived task to provide a fast pathway for repetitive calls to a function
    where call order is also respected.

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

    Notes:
        - The queue runs inside a pending 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 registered with PendingManager subclassed
            trackers and **not** PendingGroup.
    """
    if not (pen_ := self._queue_map.get(key := hash(func))):
        queue = SingleConsumerAsyncQueue[tuple[Callable, tuple, dict]](self.backend)
        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
            try:
                async for item in queue:
                    try:
                        await item[0](*item[1], **item[2])
                    except TypeError:
                        continue
                    except (anyio.get_cancelled_exc_class(), Exception) as e:
                        if pen.cancelled():
                            raise
                        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 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 execution queue associated with `func`.

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

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 iterator to get result as they complete.

Parameters:

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

Either a container with existing results or generator of Pendings.
max_concurrent ¶
(NoValue | int, default: NoValue ) –

The maximum number of concurrent results to monitor at a time. This is useful when items is a generator utilising Caller.to_thread. By default this will limit to Caller.MAX_IDLE_POOL_INSTANCES.
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.

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,  # pyright: ignore[reportInvalidTypeForm]
    cancel_unfinished: bool = True,
) -> AsyncGenerator[Pending[T], Any]:
    """
    An iterator to get result as they complete.

    Args:
        items: Either a container with existing results or generator of Pendings.
        max_concurrent: The maximum number of concurrent results to monitor at a time.
            This is useful when `items` is a generator utilising [Caller.to_thread][].
            By default this will limit to `Caller.MAX_IDLE_POOL_INSTANCES`.
        cancel_unfinished: Cancel any `pending` when exiting.

    Tip:
        1. Pass a generator if you wish to limit the number result jobs when calling to_thread/to_task etc.
        2. Pass a container with all results when the limiter is not relevant.
    """
    resume = noop
    queue: SingleConsumerAsyncQueue[Pending[T]] = SingleConsumerAsyncQueue(self.backend)
    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 queue.stopped and (pen := await anext(gen, None) if is_async else next(gen, None)) is not None:
            if pen is pen_current:
                queue.stop()
                msg = "Waiting for the pending in which it is running would result in deadlock!"
                raise RuntimeError(msg)
            if not isinstance(pen, Pending):
                pen = cast("Pending[T]", self.call_soon(await_for, pen))
            pen.add_done_callback(queue.append)
            if not pen.done():
                unfinished.add(pen)
                if max_concurrent_ and len(unfinished) == max_concurrent_:
                    event = create_async_event()
                    resume = event.set
                    if len(unfinished) == max_concurrent_:
                        await event
                    resume = noop
        if not queue.queue and not unfinished:
            queue.stop()

    pen_ = self.call_soon(scheduler)
    try:
        async for pen in queue:
            unfinished.discard(pen)
            yield pen
            if pen_.done() and not unfinished and not queue.queue:
                break
            else:
                if max_concurrent_ and len(unfinished) < max_concurrent_:
                    resume()
        pen_.result()
    finally:
        queue.stop()
        pen_.cancel()
        for pen in unfinished:
            pen.remove_done_callback(queue.append)
            if cancel_unfinished:
                pen.cancel("Cancelled by as_completed")
        with anyio.CancelScope():
            await pen_.wait(result=False)

wait `async` ¶

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

Wait for the results given by items to complete.

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

Parameters:

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

An iterable of results to wait for.
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]],
    *,
    timeout: float | None = None,
    return_when: Literal["FIRST_COMPLETED", "FIRST_EXCEPTION", "ALL_COMPLETED"] = "ALL_COMPLETED",
) -> tuple[set[Pending[T]], set[Pending[T]]]:
    """
    Wait for the results given by items to complete.

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

    Args:
        items: An iterable of results to wait for.
        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:
            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:
        with anyio.move_on_after(timeout):
            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] = 0) -> PendingGroup

Create a new PendingGroup instance.

The pending group registers all pending created in its context that opt in (via trackers). On exiting the context, it will await all remaining pending to complete. The exit and cancellation behaviour of the instance is a function of mode.

If any pending result in exception, the pending group and all registered pending are cancelled. If the pending group context is cancelled or results in exception, all pending in the group are also cancelled.

Parameters:

shield ¶
(bool, default: False ) –

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

The mode. - 0: Ignore cancellation of pending. - 1: Cancel if any pending is cancelled - raise PendingCancelled on exit. - 2: Cancel if any pending is cancelled - exit quietly.

Usage:

```python
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] = 0) -> PendingGroup:
    """
    Create a new [PendingGroup][async_kernel.pending.PendingGroup] instance.

    The pending group registers all pending created in its context that opt in (via trackers). On exiting
    the context, it will await all remaining pending to complete. The exit and cancellation behaviour of
    the instance is a function of `mode`.

    If any pending result in exception, the pending group and all registered pending are cancelled.
    If the pending group context is cancelled or results in exception, all pending in the group are
    also cancelled.

    Args:
        shield: Shield the pending group from external cancellation.
        mode: The mode.
            - 0: Ignore cancellation of pending.
            - 1: Cancel if any pending is cancelled - raise PendingCancelled on exit.
            - 2: Cancel if any pending is cancelled - exit quietly.

    Usage:

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

caller

PendingGroup ¶

cancellation_timeout class-attribute instance-attribute ¶

caller class-attribute instance-attribute ¶

__init__ ¶

shield ¶

mode ¶

cancel ¶

cancelled ¶

Caller ¶

MAX_IDLE_POOL_INSTANCES class-attribute instance-attribute ¶

IDLE_WORKER_SHUTDOWN_DURATION class-attribute instance-attribute ¶

stopped class-attribute instance-attribute ¶

log instance-attribute ¶

iopub_sockets class-attribute ¶

iopub_url class-attribute instance-attribute ¶

name property ¶

id property ¶

backend property ¶

backend_options property ¶

host property ¶

host_options property ¶

protected property ¶

zmq_context property ¶

running property ¶

children property ¶

thread property ¶

parent property ¶

__new__ ¶

modifier ¶

**kwargs ¶

start_sync ¶

no_debug ¶

stop ¶

force ¶

__asynccontextmanager__ async ¶

id_current classmethod ¶

get_existing classmethod ¶

caller_id ¶

current_pending classmethod ¶

all_callers classmethod ¶

running_only ¶

get ¶

**kwargs ¶

schedule_call ¶

func ¶

args ¶

kwargs ¶

context ¶

trackers ¶

**metadata ¶

call_later ¶

func ¶

delay ¶

*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 ¶

cancellation_timeout `class-attribute` `instance-attribute` ¶

caller `class-attribute` `instance-attribute` ¶

init ¶

`shield` ¶

`mode` ¶

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

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

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

log `instance-attribute` ¶

iopub_sockets `class-attribute` ¶

iopub_url `class-attribute` `instance-attribute` ¶

name `property` ¶

id `property` ¶

backend `property` ¶

backend_options `property` ¶

host `property` ¶

host_options `property` ¶

protected `property` ¶

zmq_context `property` ¶

running `property` ¶

children `property` ¶

thread `property` ¶

parent `property` ¶

new ¶

`modifier` ¶

`kwargs`** ¶

`no_debug` ¶

`force` ¶

asynccontextmanager `async` ¶

id_current `classmethod` ¶

get_existing `classmethod` ¶

`caller_id` ¶

current_pending `classmethod` ¶

all_callers `classmethod` ¶

`running_only` ¶

`kwargs`** ¶

`func` ¶

`args` ¶

`kwargs` ¶

`context` ¶

`trackers` ¶

`metadata`** ¶

`func` ¶

`delay` ¶

**`*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` ¶

`timeout` ¶

`return_when` ¶

`shield` ¶

`mode` ¶