caller

Classes:

Caller –

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

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

Caller ¶

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: Pyodide does not support threads, It is a context-varible local class instead.

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.

Caller can run in any thread where a backend ('asyncio' or 'trio') is running. But in-order to start, it must be Created inside that thread. After which, methods can be called from any thread.

An async context is provided for lifecycle management. The async-context of a caller can be entered multiple times from any thread. Once entered, the caller is marked as 'protected' and a count of re-entry is maintained. When the first entered context exits it will wait until all other contexts have exited. When the count returns to zero the caller will stop, after which, the caller can not be restarted.

If the first entry of the async context is obtained inside a pending (a coroutine manage by the caller), the caller will also be force stopped.

Children callers (obtained using caller.get(...)) are always force stopped when the parent caller is stopped.

High level methods

Caller.call_soon: Schedule a function call in the caller's thread.
Caller.call_later: Schedule a function call in the caller's thread after a delay.
Caller.to_thread: Schedule a function call using a worker caller (separate thread).
Caller.call_using_backend: Schedule a function call using the backend in the caller's thread.
Caller.as_completed: An async iterator 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.
Caller.create_start_stop_task: Run a coroutine function in a Task.

Class methods

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

Referenced by:

Reference

Returned by:

Reference

Used by:

Reference

Methods:

__new__ –

Create or retrieve a caller.
id_current –

The immutable id of 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.
stop –

Stop the caller cancelling all incomplete tasks.
get –

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

A low-level function to schedule execution of funcin 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 –

A low-level function to schedule execution of func in caller's scheduler.
to_thread –

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

Returns the pending associated with the queue_call for func.
queue_call –

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

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

Wrap the coroutine function func with a StartStopTask.
as_completed –

An async iterator to yield a pending for each awaitable in items as they complete (are done).
wait –

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

Create a new PendingGroup.

Attributes:

MAX_IDLE_POOL_INSTANCES –

The number of 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.
started –

A pending that is set once the caller has started.
stopping –

A pending that is set done the first time stop is called.
stopped –

A pending that is done when the caller is stopped.
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.
running (bool) –

Returns True when the caller is available to run requests.
children (set[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:
    """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: Pyodide does not support threads, It is a context-varible local class instead.

    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.

    Caller can run in any thread where a backend ('asyncio' or 'trio') is running.  But
    in-order to start, it must be Created inside that thread. After which, methods can
    be called from any thread.

    An async context is provided for lifecycle management. The async-context of a caller
    can be entered multiple times from any thread.  Once entered, the caller is marked as
    'protected' and a count of re-entry is maintained. When the first entered context exits
    it will wait until all other contexts have exited. When the count returns to zero the
    caller will stop, after which, the caller can not be restarted.

    If the first entry of the async context is obtained inside a pending (a coroutine manage
    by the caller), the caller will also be force stopped.


    Children callers (obtained using `caller.get(...)`) are always force stopped when the
    parent caller is stopped.


    **High level methods**

    - [Caller.call_soon][]: Schedule a function call in the caller's thread.
    - [Caller.call_later][]: Schedule a function call in the caller's thread after a delay.
    - [Caller.to_thread][]: Schedule a function call using a worker caller (separate thread).
    - [Caller.call_using_backend][]: Schedule a function call using the backend in the caller's thread.
    - [Caller.as_completed][]: An async iterator 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.
    - [Caller.create_start_stop_task][]: Run a coroutine function in a Task.

    **Class methods**

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

    MAX_IDLE_POOL_INSTANCES = 10
    "The number of `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 = int(threading.main_thread().ident)  # pyright: ignore[reportArgumentType]

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

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

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

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

    started = Fixed(ProtectedPending)
    """A pending that is set once the caller has started."""

    stopping = Fixed(ProtectedPending)
    """A pending that is set done the first time stop is called."""

    stopped = Fixed(ProtectedPending)
    """A pending that is done when the caller is stopped."""

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

    log: logging.LoggerAdapter

    @property
    def name(self) -> str:
        """The name of the 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

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

    @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) -> set[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 changing state to stopped.
        """
        return {c for c in self._children.copy() if not c.stopping.done()}

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

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

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

    def _get_info(self) -> dict[str, Any]:
        if self._state.value < CallerState.running.value:
            return {"name": self._name, "backend": str(self._backend), "host": self._host}
        return {
            "name": self._name,
            "backend": str(self._backend),
            "host": self._host,
            "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() == getattr(self, "_caller_id", None) else "⚫"
        protected = "🔐 " if self.protected else " "
        n = len(self._children)
        children = "" if not n else (" 1 child" if n == 1 else f" {n} children")
        if self._wait_exit and self._enter_count:
            info = f"{info} waiting context {self._enter_count} remain"
        return f"<Caller {current}{self._state_reprs[self._state]}{protected}{info}{children}>"

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

        Args:
            modifier: Specifies the caller instance to retrieve.

                - "CurrentThread": The caller for the current thread.
                - "MainThread": The Caller associated with the main thread.
                - Advanced:
                    - "NewThread": Create a caller with a new thread.
                        [Caller.get][] and [Caller.to_thread][] are recommended for normal usage.

            **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.
                - 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 "MainThread":
                    caller_id, thread = cls.CALLER_MAIN_THREAD_ID, threading.main_thread()
                case "NewThread":
                    caller_id, thread = None, None
                case _:
                    caller_id, thread = cls.id_current(), threading.current_thread()

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

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

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

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

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

            # Set the scheduler to start in either the current thread or a new thread.
            # It is not possible to wait for it to start without using microthreads like greenlets.
            inst._start_sync(caller_id, thread, kwargs.get("no_debug", False))
            assert inst._caller_id is not None
            assert inst._caller_id not in cls._instances
            cls._instances[inst._caller_id] = inst
        return inst

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

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

    def _start_sync(self, caller_id: int | None, thread: threading.Thread | None, no_debug: bool = False) -> None:
        """Start synchronously.

        Args:
            caller_id: The id to use for the caller, which should match the thread in CPython.
            thread: The thread where the caller is running.
            no_debug: If debugpy should be disabled in the thread.
        """
        self._set_state(CallerState.start_sync)

        async def run_scheduler() -> None:
            if self._set_state(CallerState.starting) is not CallerState.start_sync:
                return
            if not self._name:
                self._name = self._thread.name
            if no_debug:
                utils.mark_thread_pydev_do_not_trace()
            try:
                async with task_factory() as create_task:
                    create_task(contextvars.Context(), self._scheduler, self._scheduler_queue)
                    self._set_state(CallerState.running)
                    await self.stopping
                    await self.wait(self._pen_stop, return_when="ALL_COMPLETED")
                    await self._children_countdown
                    await async_checkpoint(force=True)
                    self._scheduler_queue.stop()
                    await async_checkpoint(force=True)
            except anyio.get_cancelled_exc_class():
                # This may happen when the async event loop is shutting down.
                pass
            except BaseException as e:
                self.log.exception("Caller did not exit context nicely!", exc_info=e)
                raise
            finally:
                self._set_state(CallerState.stopping)
                self._set_state(CallerState.stopped)

        if caller_id and thread:
            assert thread is threading.current_thread()
            self._thread, self._caller_id = thread, caller_id

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

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

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

            def async_kernel_caller() -> None:
                self._thread, self._caller_id = threading.current_thread(), threading.get_ident()
                settings = RunSettings(
                    backend=self.backend,
                    host=self.host,
                    backend_options=self.backend_options,
                    host_options=self.host_options,
                )
                try:
                    async_kernel.event_loop.run(run_scheduler, (), settings)
                except Exception as e:
                    if not self.stopping.done():
                        self.started.set_exception(e)
                        self.stop(force=True)

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

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

    async def _scheduler(self, queue: SingleAsyncQueue) -> None:
        """A function that async iterates the queue and executes items as they arrive.

        It handles two types of items:
            - tuple: A tuple of `func`, `args`, `kwargs` is called directly in the scheduler.
            - Pending: A pending is started as an 'task'. The pending provides `func`, `args` and `kwargs`
                in it's metadata. The pending is set as `active_pending`in the context for the duration of execution.

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

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

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

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

    async def _idle_worker_cleanup(self) -> None:
        """Shutsdown idle thread workers."""
        while (t := self.IDLE_WORKER_SHUTDOWN_DURATION) and self._worker_pool:
            await anyio.sleep(t)
            for worker in self._worker_pool.copy():
                if (time.monotonic() - worker._idle_time) > t:
                    worker.stop(force=True)

    def _start_guest(self, backend: Backend, queue: SingleAsyncQueue) -> None:
        """Start a guest event loop."""
        assert self.get_existing() is self
        # Thread: caller

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

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

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

    @classmethod
    def get_existing(cls, caller_id: int | None = None, /) -> Self | None:
        """A [classmethod][] to get the caller instance 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 stop(self, *, force: bool = False) -> None:
        """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:
            self.log.debug("Non-force stop ignored for  %s", self)
            return
        self.stopping.set_result(None)
        if (parent := self.parent) and self in parent._worker_pool:
            with contextlib.suppress(IndexError):
                parent._worker_pool.remove(self)
        if (old_state := self._set_state(CallerState.stopping)) and old_state.value < CallerState.starting.value:
            self._set_state(CallerState.stopped)

    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.

        Returns:
            Self: The retrieved or newly created caller.

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

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

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

        Args:
            func: The function to be called.
            args: Arguments corresponding to in the call to  `func`.
            kwargs: Keyword arguments to use with in the call to `func`.
            context: A context to copy, if not provided the current context is copied.
            trackers: The tracker subclasses of active trackers which to add the pending.
            backend: The backend to use to execute func which may execute the code a backend running as a guest.
            **metadata: Additional metadata to store in the instance.

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

    def call_later(
        self,
        delay: float,
        func: Callable[P, T | CoroutineType[Any, Any, T]],
        /,
        *args: P.args,
        **kwargs: P.kwargs,
    ) -> Pending[T]:
        """Schedule func to be called in 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.
        """

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

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

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

        See Also:
            - [Caller.get][]
        """
        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:
        """A low-level function to schedule execution of `func` in caller's scheduler.

        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._scheduler_queue.append((func, args, kwargs))

    def to_thread(
        self,
        func: Callable[P, T | CoroutineType[Any, Any, T]],
        /,
        *args: P.args,
        **kwargs: P.kwargs,
    ) -> Pending[T]:
        """Call `func` in a worker thread using the same backend as the caller.

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

        Notes:
            - A pool of workers are maintained.
            - Structured concurrency tools should be used to creating task such as:
                - [Caller.create_pending_group][]
                - [asyncio.TaskGroup][]
                - [anyio.create_task_group][]
        """

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

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

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

        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.

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

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

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

                    async def queue_loop() -> None:
                        pen = self.current_pending()
                        assert pen is not None
                        try:
                            async for item in queue:
                                try:
                                    result = item[0](*item[1], **item[2])
                                    if iscoroutinelike(result):
                                        await result
                                    del result
                                except (anyio.get_cancelled_exc_class(), 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 [Caller.queue_call][async_kernel.caller.Caller.queue_call] execution queue associated with `func`.

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

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

        The returned `StartStopTask` will only be schedule after `start` method is called,
        and can only be started once, though it is safe to call `start` multiple times;
        subsequent calls are `noop`.

        When used as an async context, the method 'stop' will be called when the context
        exits and will wait for the protected task to  complete prior to stopping. `func`
        is expected to accept two positional arguments:
        1. started: A callable to indicate the task is started.
        2. stop: A protected pending, that should be awaited, or otherwise used to shutdown
            the task.

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


            task = caller.create_start_stop_task(func).start()
            await task.stop()
            # or
            async with caller.create_start_stop_task(func).start():
                pass
            ```
        """
        return StartStopTask().set_task_function(func, caller=self)

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

        How the pending was marked as done does not affect the iterator.

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

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

        def noop() -> None:
            pass

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

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

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

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

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

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

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

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

        [Pending][async_kernel.pending.Pending] created in the context that opt-in by including `PendingTracker`
        as a 'tracker', including all methods on [Caller][] that return pending are automatically registered.

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

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

        Usage:

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

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

MAX_IDLE_POOL_INSTANCES = 10

The number of 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 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).

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

started = Fixed(ProtectedPending)

A pending that is set once the caller has started.

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

stopping = Fixed(ProtectedPending)

A pending that is set done the first time stop is called.

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

stopped = Fixed(ProtectedPending)

A pending that is done when the caller is stopped.

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

protected: bool

Returns True if the caller is protected from stopping.

running `property` ¶

running: bool

Returns True when the caller is available to run requests.

children `property` ¶

children: set[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 changing state to stopped.

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"] = "CurrentThread",
    /,
    **kwargs: Unpack[CallerCreateOptions],
) -> Self

Create or retrieve a caller.

Parameters:

modifier ¶
(Literal['CurrentThread', 'MainThread', 'NewThread'], default: 'CurrentThread' ) –
Specifies the caller instance to retrieve.
- "CurrentThread": The caller for the current thread.
- "MainThread": The Caller associated with the main thread.
- Advanced:
  - "NewThread": Create a caller with a new thread. Caller.get and Caller.to_thread are recommended for normal usage.
**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. - 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"] = "CurrentThread",
    /,
    **kwargs: Unpack[CallerCreateOptions],
) -> Self:
    """Create or retrieve a caller.

    Args:
        modifier: Specifies the caller instance to retrieve.

            - "CurrentThread": The caller for the current thread.
            - "MainThread": The Caller associated with the main thread.
            - Advanced:
                - "NewThread": Create a caller with a new thread.
                    [Caller.get][] and [Caller.to_thread][] are recommended for normal usage.

        **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.
            - 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 "MainThread":
                caller_id, thread = cls.CALLER_MAIN_THREAD_ID, threading.main_thread()
            case "NewThread":
                caller_id, thread = None, None
            case _:
                caller_id, thread = cls.id_current(), threading.current_thread()

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

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

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

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

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

        # Set the scheduler to start in either the current thread or a new thread.
        # It is not possible to wait for it to start without using microthreads like greenlets.
        inst._start_sync(caller_id, thread, kwargs.get("no_debug", False))
        assert inst._caller_id is not None
        assert inst._caller_id not in cls._instances
        cls._instances[inst._caller_id] = inst
    return inst

id_current `classmethod` ¶

id_current() -> int

The immutable id of 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 immutable id of a caller for the current thread in CPython or context in Pyodide."""
    return cls._caller_token.get() if sys.platform == "emscripten" else threading.get_ident()

get_existing `classmethod` ¶

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

A classmethod to get the caller instance 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
- event_loop ZMQPoll event_handler

Source code in src/async_kernel/caller.py

@classmethod
def current_pending(cls) -> Pending[Any] | None:
    """A [classmethod][] that returns the 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]

stop ¶

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

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) -> None:
    """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:
        self.log.debug("Non-force stop ignored for  %s", self)
        return
    self.stopping.set_result(None)
    if (parent := self.parent) and self in parent._worker_pool:
        with contextlib.suppress(IndexError):
            parent._worker_pool.remove(self)
    if (old_state := self._set_state(CallerState.stopping)) and old_state.value < CallerState.starting.value:
        self._set_state(CallerState.stopped)

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.

Returns:

Self ( Self ) –

The retrieved or newly created caller.

Raises:

RuntimeError –

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

Notes

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

Referenced by:

Reference caller Caller

Source code in src/async_kernel/caller.py

def get(self, **kwargs: Unpack[CallerCreateOptions]) -> Self:
    """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.

    Returns:
        Self: The retrieved or newly created caller.

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

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

schedule_call ¶

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

A low-level function to schedule execution of funcin a task running in the caller's thread.

Parameters:

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

The function to be called.
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 ) –

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

The tracker subclasses of active trackers which to add the pending.
backend ¶
(NoValue | Backend, default: NoValue ) –

The backend to use to execute func which may execute the code a backend running as a guest.
**metadata ¶
(Any, default: {} ) –

Additional metadata to store in the instance.

Returns:

Pending ( Pending[T] ) –

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

Referenced by:

Reference caller Caller

Source code in src/async_kernel/caller.py

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

    Args:
        func: The function to be called.
        args: Arguments corresponding to in the call to  `func`.
        kwargs: Keyword arguments to use with in the call to `func`.
        context: A context to copy, if not provided the current context is copied.
        trackers: The tracker subclasses of active trackers which to add the pending.
        backend: The backend to use to execute func which may execute the code a backend running as a guest.
        **metadata: Additional metadata to store in the instance.

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

call_later ¶

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

Schedule func to be called in 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.
    """

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

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

call_soon ¶

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

Schedule func to be 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

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.

call_direct ¶

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

A low-level function to schedule execution of func in caller's scheduler.

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:
    """A low-level function to schedule execution of `func` in caller's scheduler.

    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._scheduler_queue.append((func, args, kwargs))

to_thread ¶

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

Call func in a worker thread using the same backend as the caller.

Parameters:

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

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

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

Keyword arguments to use with func.

Notes

A pool of workers are maintained.
Structured concurrency tools should be used to creating task such as:

Referenced by:

Reference
- caller Caller
  - MAX_IDLE_POOL_INSTANCES
  - __new__
- typing RunMode thread

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

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

    Notes:
        - A pool of workers are maintained.
        - Structured concurrency tools should be used to creating task such as:
            - [Caller.create_pending_group][]
            - [asyncio.TaskGroup][]
            - [anyio.create_task_group][]
    """

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

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

queue_get ¶

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

Returns the pending associated with the queue_call for func.

Notes

queue_close is the preferred means to shutdown the queue.

Referenced by:

Reference caller Caller

Source code in src/async_kernel/caller.py

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

    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.

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

Parameters:

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

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

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

Keyword arguments to use with func.

Notes

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

Referenced by:

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

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

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

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

                async def queue_loop() -> None:
                    pen = self.current_pending()
                    assert pen is not None
                    try:
                        async for item in queue:
                            try:
                                result = item[0](*item[1], **item[2])
                                if iscoroutinelike(result):
                                    await result
                                del result
                            except (anyio.get_cancelled_exc_class(), 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 Caller.queue_call execution queue associated with func.

Parameters:

func ¶
(Callable | int) –

The queue of the function to close.

Referenced by:

Reference caller Caller queue_call

Source code in src/async_kernel/caller.py

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

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

create_start_stop_task ¶

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

Wrap the coroutine function func with a StartStopTask.

The returned StartStopTask will only be schedule after start method is called, and can only be started once, though it is safe to call start multiple times; subsequent calls are noop.

When used as an async context, the method 'stop' will be called when the context exits and will wait for the protected task to complete prior to stopping. func is expected to accept two positional arguments: 1. started: A callable to indicate the task is started. 2. stop: A protected pending, that should be awaited, or otherwise used to shutdown the task.

Usage

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


task = caller.create_start_stop_task(func).start()
await task.stop()
# or
async with caller.create_start_stop_task(func).start():
    pass

Referenced by:

Reference caller Caller

Source code in src/async_kernel/caller.py

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

    The returned `StartStopTask` will only be schedule after `start` method is called,
    and can only be started once, though it is safe to call `start` multiple times;
    subsequent calls are `noop`.

    When used as an async context, the method 'stop' will be called when the context
    exits and will wait for the protected task to  complete prior to stopping. `func`
    is expected to accept two positional arguments:
    1. started: A callable to indicate the task is started.
    2. stop: A protected pending, that should be awaited, or otherwise used to shutdown
        the task.

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


        task = caller.create_start_stop_task(func).start()
        await task.stop()
        # or
        async with caller.create_start_stop_task(func).start():
            pass
        ```
    """
    return StartStopTask().set_task_function(func, caller=self)

as_completed `async` ¶

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

An async iterator to yield a pending for each awaitable in items as they complete (are done).

How the pending was marked as done does not affect the iterator.

Parameters:

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

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

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

Cancel any pending when exiting.

Tip

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

Referenced by:

Reference caller Caller

Source code in src/async_kernel/caller.py

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

    How the pending was marked as done does not affect the iterator.

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

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

    def noop() -> None:
        pass

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

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

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

wait `async` ¶

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

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

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

Parameters:

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

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

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

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

The same options as available for asyncio.wait.

Example

done, pending = await asyncio.wait(items)

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

Referenced by:

Reference caller Caller

Source code in src/async_kernel/caller.py

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

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

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

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

create_pending_group ¶

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

Create a new PendingGroup.

Pending created in the context that opt-in by including PendingTracker as a 'tracker', including all methods on Caller that return pending are automatically registered.

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

Parameters:

shield ¶
(bool, default: False ) –

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

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

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

Usage:

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

Referenced by:

Reference caller Caller to_thread

Source code in src/async_kernel/caller.py

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

    [Pending][async_kernel.pending.Pending] created in the context that opt-in by including `PendingTracker`
    as a 'tracker', including all methods on [Caller][] that return pending are automatically registered.

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

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

    Usage:

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

StartStopTask ¶

Bases: AsyncContextManagerMixin, Generic[P, T]

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

The stages are:

set_task_function: Set the task function and caller to associate it with.
start: Start the task.
started: The task indicates it is started at a position where it is considered to be running.
stopping: stop has been called which signals to the task that it should commence it's shutdown.
stopped: The task is finished.

For convenience, this class is awaitable and provides an async context manager. Both of which are only available after start has been called. The async context can only be entered once, and will initiate stop the protected task when the context exits.

Usage

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


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

Returned by:

Reference caller Caller create_start_stop_task

Subclassed by:

Reference
- interface
  - Interface
  - Interface
- messaging BaseMessage

Methods:

set_task_function –

Set the task function for this instance.
start –

Start the task function.
stop –

Stop the Task.

Attributes:

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

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

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

A ProtectedPending that is set when the shielded call of func is finished.
caller (Caller) –

The caller where the task is running.

Source code in src/async_kernel/caller.py

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

    The stages are:

    1. set_task_function: Set the task function and caller to associate it with.
    2. start: Start the task.
    3. started: The task indicates it is started at a position where it is considered to be running.
    4. stopping: `stop` has been called which signals to the task that it should commence it's shutdown.
    5. stopped: The task is finished.

    For convenience, this class is awaitable and provides an async context manager.
    Both of which are only available after `start` has been called. The async context can only
    be entered once, and will initiate stop the protected task when the context exits.

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


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

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

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

    stopped: Fixed[Self, ProtectedPending[T]] = Fixed(lambda c: ProtectedPending(info=c["owner"]._info(c["name"])))
    """A `ProtectedPending` that is set when the shielded call of `func` is finished."""

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

    def set_task_function(
        self,
        func: Callable[Concatenate[Callable[[], None], ProtectedPending[None], P], T | CoroutineType[Any, Any, T]],
        /,
        *,
        caller: Caller | None = None,
    ) -> Self:
        """Set the task function for this instance.

        Args:
            func: The coroutine function to run as a task.
            caller: Specify the caller to run the function.
        """
        if hasattr(self, "_caller_ref"):
            msg = "`func` can only be set once!"
            raise RuntimeError(msg)
        self._caller_ref: ReferenceType[Caller] = weakref.ref(caller or Caller())
        self.caller.stopping.add_done_callback(self._caller_stopping)
        self._start_token = ""
        self._func = func
        return self

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

    def _caller_stopping(self, _) -> None:
        self.stop()

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

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

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

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

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

    def start(self, *args: P.args, **kwargs: P.kwargs) -> Self:
        """Start the task function.

        Args:
            *args: Arguments to pass to the task coroutine function.
            **kwargs: Keyword arguments to pass to the task coroutine function.

        Returns:
            Self: Returns the instance to make it convenient to chain function calls.
        """
        if not hasattr(self, "_func"):
            msg = "The task function has not been set. Tip: Use the method `set_task_function`."
            raise RuntimeError(msg)
        with contextlib.suppress(AttributeError):
            del self._start_token
            ref = weakref.ref(self)

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

            def done(pen: Pending) -> None:
                self = ref()
                assert self
                self.caller.log.debug("Task stopped %r", self._func)
                self.caller.stopping.remove_done_callback(self._caller_stopping)

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

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

            if not self.stopping.done():
                self.caller.call_soon(self._func, started, self.stopping, *args, **kwargs).add_done_callback(done)
        return self

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

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

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

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

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

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

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

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

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

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

A ProtectedPending that is set when the shielded call of func is finished.

caller `property` ¶

caller: Caller

The caller where the task is running.

set_task_function ¶

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

Set the task function for this instance.

Parameters:

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

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

Specify the caller to run the function.

Source code in src/async_kernel/caller.py

def set_task_function(
    self,
    func: Callable[Concatenate[Callable[[], None], ProtectedPending[None], P], T | CoroutineType[Any, Any, T]],
    /,
    *,
    caller: Caller | None = None,
) -> Self:
    """Set the task function for this instance.

    Args:
        func: The coroutine function to run as a task.
        caller: Specify the caller to run the function.
    """
    if hasattr(self, "_caller_ref"):
        msg = "`func` can only be set once!"
        raise RuntimeError(msg)
    self._caller_ref: ReferenceType[Caller] = weakref.ref(caller or Caller())
    self.caller.stopping.add_done_callback(self._caller_stopping)
    self._start_token = ""
    self._func = func
    return self

start ¶

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

Start the task function.

Parameters:

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

Arguments to pass to the task coroutine function.
**kwargs ¶
(kwargs, default: {} ) –

Keyword arguments to pass to the task coroutine function.

Returns:

Self ( Self ) –

Returns the instance to make it convenient to chain function calls.

Source code in src/async_kernel/caller.py

def start(self, *args: P.args, **kwargs: P.kwargs) -> Self:
    """Start the task function.

    Args:
        *args: Arguments to pass to the task coroutine function.
        **kwargs: Keyword arguments to pass to the task coroutine function.

    Returns:
        Self: Returns the instance to make it convenient to chain function calls.
    """
    if not hasattr(self, "_func"):
        msg = "The task function has not been set. Tip: Use the method `set_task_function`."
        raise RuntimeError(msg)
    with contextlib.suppress(AttributeError):
        del self._start_token
        ref = weakref.ref(self)

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

        def done(pen: Pending) -> None:
            self = ref()
            assert self
            self.caller.log.debug("Task stopped %r", self._func)
            self.caller.stopping.remove_done_callback(self._caller_stopping)

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

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

        if not self.stopping.done():
            self.caller.call_soon(self._func, started, self.stopping, *args, **kwargs).add_done_callback(done)
    return self

stop ¶

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

Stop the Task.

Returns:

ProtectedPending ( ProtectedPending[T] ) –

Resolves with the result of the function.

Source code in src/async_kernel/caller.py

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

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

caller

Caller ¶

MAX_IDLE_POOL_INSTANCES class-attribute instance-attribute ¶

IDLE_WORKER_SHUTDOWN_DURATION class-attribute instance-attribute ¶

started class-attribute instance-attribute ¶

stopping class-attribute instance-attribute ¶

stopped class-attribute instance-attribute ¶

name property ¶

id property ¶

backend property ¶

backend_options property ¶

host property ¶

host_options property ¶

protected property writable ¶

running property ¶

children property ¶

thread property ¶

parent property ¶

__new__ ¶

modifier ¶

**kwargs ¶

id_current classmethod ¶

get_existing classmethod ¶

caller_id ¶

current_pending classmethod ¶

all_callers classmethod ¶

running_only ¶

stop ¶

force ¶

get ¶

**kwargs ¶

schedule_call ¶

func ¶

args ¶

kwargs ¶

context ¶

trackers ¶

backend ¶

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

create_start_stop_task ¶

as_completed async ¶

items ¶

max_concurrent ¶

cancel_unfinished ¶

wait async ¶

items ¶

shield ¶

timeout ¶

return_when ¶

create_pending_group ¶

shield ¶

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

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

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

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

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

name `property` ¶

id `property` ¶

backend `property` ¶

backend_options `property` ¶

host `property` ¶

host_options `property` ¶

protected `property` `writable` ¶

running `property` ¶

children `property` ¶

thread `property` ¶

parent `property` ¶

new ¶

`modifier` ¶

`kwargs`** ¶

id_current `classmethod` ¶

get_existing `classmethod` ¶

`caller_id` ¶

current_pending `classmethod` ¶

all_callers `classmethod` ¶

`running_only` ¶

`force` ¶

`kwargs`** ¶

`func` ¶

`args` ¶

`kwargs` ¶

`context` ¶

`trackers` ¶

`backend` ¶

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

`shield` ¶

`timeout` ¶

`return_when` ¶

`shield` ¶

`mode` ¶

`timeout` ¶

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

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

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

caller `property` ¶

`func` ¶

`caller` ¶

**`*args`** ¶

`kwargs`** ¶