Caller: A backend agnostic executor (notebook)¶

This notebook uses trio in some cells.

Caller is a thread-local object that coordinates the execution of functions and coroutines in the event loop of an asyncio or trio backend.

Caller instances are created on the following basis:

• CPython: per-thread
• Pyodide: per-context (until threading is supported in Pyodide)

A Caller instance can be started with one gui as a host to one backend started as a guest.

A second backend can also be started in the same thread/caller with or without a host gui event loop. Potentially the same thread could have up to:

• One host gui event loop (tk, qt, ...)
• Two asynchronous backends (asyncio, trio)

It should be noted that gui's generally assume/require that their mainloop runs in the main thread.

The magic command %callers prints a list of the current Caller instances.

In [1]:

%callers

%callers

      Name      Running Protected     Thread             Caller       
───────────────────────────────────────────────────────────────────────────
MainThread         ✓       🔐       MainThread   139786796679296 ← current
Control            ✓       🔐        Control     139786740659904

There are two caller instances associated with the kernel interface:

• Shell: The caller for the shell channel (normally the MainThread).
• Control: The caller for the control channel (a child of the shell).

Calling Caller() with no arguments will return the caller for the current thread (or context in Pyodide).

A runtime error will occur if there is no event loop (backend) running in the current thread (unlikely in Pyodide which provides a running event loop).

In [2]:

from async_kernel import Caller

caller = Caller()

from async_kernel import Caller caller = Caller()

Modifiers¶

There are three modifiers that can change the instance that is returned when calling Caller.

• CurrentThread (default): Access the caller that belongs to the current thread (or context with Pyodide).
• MainThread: Access main thread for the Caller.
• NewThread: Create a new thread.

In [3]:

assert caller is Caller()

assert caller is Caller("CurrentThread")

assert caller is Caller("MainThread")

assert caller is Caller() assert caller is Caller("CurrentThread") assert caller is Caller("MainThread")

In [4]:

caller1 = Caller("NewThread", name="My caller1")

assert caller1 is not caller

# We can start using it immediately.
await caller1.call_soon(display, "hello world!", caller1)

# We should stop the caller when we don't need it.
caller1.stop()
del caller1

caller1 = Caller("NewThread", name="My caller1") assert caller1 is not caller # We can start using it immediately. await caller1.call_soon(display, "hello world!", caller1) # We should stop the caller when we don't need it. caller1.stop() del caller1

'hello world!'

<Caller ⚪🏃 running name='My caller1' backend='asyncio' host=None thread='My caller1' id=139786455078592>

Caller creation options¶

The caller can be customised by passing keyword arguments when creating a new instance. The main options are:

• backend: The backend to use ('trio' or 'asyncio').
• name: A name for the caller.

In [5]:

import trio

trio_caller = Caller("NewThread", backend="trio")
await trio_caller.call_soon(trio.sleep, 0)

trio_caller.stop()
del trio_caller

import trio trio_caller = Caller("NewThread", backend="trio") await trio_caller.call_soon(trio.sleep, 0) trio_caller.stop() del trio_caller

Caller.get¶

Caller.get is a method to obtain a child instance of a Caller by name. Passing the same name on the same caller will return the same child caller on subsequent calls. The same options for creation of a new Caller can be specified. Calling caller.get() with no name will always return a new instance of a Caller. All children of a caller will stop when the parent stops.

In [6]:

trio_caller = caller.get(name="trio", backend="trio")
await trio_caller.call_soon(trio.sleep, 0)

# Callers returned from 'get' are children
assert trio_caller in caller.children

# The parent is accessible from the child
assert trio_caller.parent is caller

# The same caller can be accessed by using the name
assert caller.get(name="trio") is trio_caller

child1 = trio_caller.get()
child2 = trio_caller.get()

assert trio_caller.children == {child1, child2}
trio_caller.stop()
await trio_caller.stopped

# Stopping the parent will stop the children
assert child1.stopped
assert child2.stopped

trio_caller = caller.get(name="trio", backend="trio") await trio_caller.call_soon(trio.sleep, 0) # Callers returned from 'get' are children assert trio_caller in caller.children # The parent is accessible from the child assert trio_caller.parent is caller # The same caller can be accessed by using the name assert caller.get(name="trio") is trio_caller child1 = trio_caller.get() child2 = trio_caller.get() assert trio_caller.children == {child1, child2} trio_caller.stop() await trio_caller.stopped # Stopping the parent will stop the children assert child1.stopped assert child2.stopped

Caller methods that return Pending¶

Pending is a thread-safe representation of a future result. It was designed to provide thread-safe functionality similar asyncio.Future and was named Pending to avoid confusion about differences in functionality to that of asyncio.Future and concurrent.futures.Future.

The following functions return a Pending object:

• Caller.call_soon: Call a function in a new 'task'.
• Caller.call_later: Call a function in a new 'task' executing the func after a delay.
• Caller.schedule_call: An advanced function to schedule execution exposing the full functionality (used by call_soon, call_later, ...).
• Caller.call_using_backend: Call in the caller's thread ensuring a specific backend is used.

In [7]:

# Standard functions
assert await caller.call_soon(lambda: 1 + 1) == 2
assert await caller.call_later(0, lambda: 1 + 1) == 2
assert await caller.schedule_call(lambda: 1 + 1, (), {}) == 2
assert await caller.to_thread(lambda: 1 + 1) == 2
assert await caller.call_using_backend("asyncio", lambda: 1 + 1) == 2

# Standard functions assert await caller.call_soon(lambda: 1 + 1) == 2 assert await caller.call_later(0, lambda: 1 + 1) == 2 assert await caller.schedule_call(lambda: 1 + 1, (), {}) == 2 assert await caller.to_thread(lambda: 1 + 1) == 2 assert await caller.call_using_backend("asyncio", lambda: 1 + 1) == 2

In [8]:

# Coroutine functions
async def add(a, b):
    return a + b


assert await caller.call_soon(add, 1, 1) == 2
assert await caller.call_later(0, add, 1, 1) == 2
assert await caller.schedule_call(add, (1, 1), {}) == 2
assert await caller.to_thread(add, 1, 1) == 2
assert await caller.call_using_backend("asyncio", add, 1, 1) == 2

# It is also possible (though discouraged) to wrap a coroutine with a lambda
assert await caller.call_soon(lambda: add(1, 1)) == 2

# Coroutine functions async def add(a, b): return a + b assert await caller.call_soon(add, 1, 1) == 2 assert await caller.call_later(0, add, 1, 1) == 2 assert await caller.schedule_call(add, (1, 1), {}) == 2 assert await caller.to_thread(add, 1, 1) == 2 assert await caller.call_using_backend("asyncio", add, 1, 1) == 2 # It is also possible (though discouraged) to wrap a coroutine with a lambda assert await caller.call_soon(lambda: add(1, 1)) == 2

Caller.call_using_backend¶

This method calls functions and coroutines using a specific asynchronous backend in the same thread.

When the backend does not match the callers backend, it is run with the backend running as a guest.

In [9]:

import asyncio
import threading

import sniffio
import trio

await Caller().call_using_backend("asyncio", asyncio.sleep, 0)
await Caller().call_using_backend("trio", trio.sleep, 0)

for backend in ["asyncio", "trio"]:
    print("backend:", backend)
    print("thread:", await Caller().call_using_backend(backend, threading.current_thread))  # pyright: ignore[reportArgumentType]
    print("sniffio:", await Caller().call_using_backend(backend, sniffio.current_async_library), "\n")  # pyright: ignore[reportArgumentType]

import asyncio import threading import sniffio import trio await Caller().call_using_backend("asyncio", asyncio.sleep, 0) await Caller().call_using_backend("trio", trio.sleep, 0) for backend in ["asyncio", "trio"]: print("backend:", backend) print("thread:", await Caller().call_using_backend(backend, threading.current_thread)) # pyright: ignore[reportArgumentType] print("sniffio:", await Caller().call_using_backend(backend, sniffio.current_async_library), "\n") # pyright: ignore[reportArgumentType]

backend: asyncio

thread: <_MainThread(MainThread, started 139786796679296)>

sniffio: asyncio 

backend: trio

thread: <_MainThread(MainThread, started 139786796679296)>

sniffio: trio 

Low-level methods¶

Caller.queue_call¶

This method is used extensively by the kernel for message handling. Each time the queue_call is called with the same function, the function and call arguments are appended to a queue. On the first call a task is started in the backend to process the queue as items are added. The task remains alive until the function is deleted or the queue is stopped. Any exceptions that occur are logged.

In [10]:

from aiologic import CountdownEvent

N = 10
done = CountdownEvent(N)


def f(n):
    float(n)
    done.down()
    print(done.value, "Pending id:", id(Caller.current_pending()))


for i in range(N):
    caller.queue_call(f, i)

await done

assert caller.queue_get(f)
caller.queue_close(f)
assert caller.queue_get(f) is None

from aiologic import CountdownEvent N = 10 done = CountdownEvent(N) def f(n): float(n) done.down() print(done.value, "Pending id:", id(Caller.current_pending())) for i in range(N): caller.queue_call(f, i) await done assert caller.queue_get(f) caller.queue_close(f) assert caller.queue_get(f) is None

9 Pending id: 139786437103408

8 Pending id: 139786437103408

7 Pending id: 139786437103408

6 Pending id: 139786437103408

5 Pending id: 139786437103408

4 Pending id: 139786437103408

3 Pending id: 139786437103408

2 Pending id: 139786437103408

1 Pending id: 139786437103408

0 Pending id: 139786437103408

Caller.call_direct¶

This method will call the function/ coroutine function directly in the caller's scheduler. It is used by the Kernel extensively to execute coroutine functions that are known to run quickly.

Do not use this for long running coroutines as it will block the scheduler.

In [11]:

from aiologic import Event

event = Event()
caller.call_direct(event.set)
await event

from aiologic import Event event = Event() caller.call_direct(event.set) await event

Out[11]:

True

Caller.as_completed¶

This method

This example requires ipywidgets!

In [12]:

suppress-error

import random
import time

import ipywidgets as ipw

outputs = {}
stop = ipw.RadioButtons(value=None, options=["Stop"])
box = ipw.VBox([stop])
display(box)


def my_func(n):
    caller = Caller()
    if not (out := outputs.get(caller)):
        outputs[caller] = out = ipw.HTML(description=str(caller), style={"description_width": "initial"})
        box.children = [*box.children, out]
    sleep_time = random.random() / 4
    out.value = f"{n=:04d} sleeping {sleep_time * 1000:03.0f} ms"
    time.sleep(sleep_time)
    return n


async def run_forever():
    n = 0
    while not stop.value:
        n += 1
        yield Caller().to_thread(my_func, n)


async for fut in Caller().as_completed(run_forever()):
    result = await fut
    print(f"Finished: {result}", end="\r")

import random import time import ipywidgets as ipw outputs = {} stop = ipw.RadioButtons(value=None, options=["Stop"]) box = ipw.VBox([stop]) display(box) def my_func(n): caller = Caller() if not (out := outputs.get(caller)): outputs[caller] = out = ipw.HTML(description=str(caller), style={"description_width": "initial"}) box.children = [*box.children, out] sleep_time = random.random() / 4 out.value = f"{n=:04d} sleeping {sleep_time * 1000:03.0f} ms" time.sleep(sleep_time) return n async def run_forever(): n = 0 while not stop.value: n += 1 yield Caller().to_thread(my_func, n) async for fut in Caller().as_completed(run_forever()): result = await fut print(f"Finished: {result}", end="\r")

VBox(children=(RadioButtons(options=('Stop',), value=None),))

Finished: 5

Finished: 6

Finished: 7

Finished: 9

Finished: 1

Finished: 10

Finished: 2

Finished: 4

Finished: 12

Finished: 8

Finished: 3

Finished: 11

Finished: 13

Finished: 15

Finished: 19

Finished: 14

Finished: 22

Finished: 20

Finished: 16

Finished: 17

Finished: 29

Finished: 23

Finished: 18

Finished: 21

Finished: 31

Finished: 25

Finished: 24

Finished: 32

Finished: 36

Finished: 37

Finished: 26

Finished: 35

Finished: 30

Finished: 40

Finished: 27

Finished: 38

Finished: 33

Finished: 28

Finished: 44

Finished: 34

Finished: 47

Finished: 49

Finished: 46

Finished: 39

Finished: 51

Finished: 53

Finished: 50

Finished: 42

Finished: 57

Finished: 43

Finished: 41

Finished: 52

Finished: 55

Finished: 62

Finished: 45

Finished: 58

Finished: 48

Finished: 59

Finished: 54

Finished: 64

Finished: 56

Finished: 68

Finished: 67

Caller.wait¶

This method was inspired by asyncio.wait. A collection of awaitables or Pending can be passed which will wait until the condition is satisfied. Awaitables will be converted to pending by using caller.call_soon.

In [13]:

done, pending = await caller.wait(
    [asyncio.sleep(0), caller.to_thread(asyncio.sleep, 0), caller.call_using_backend("trio", trio.sleep, 0)],
    return_when="ALL_COMPLETED",
)

done, pending = await caller.wait( [asyncio.sleep(0), caller.to_thread(asyncio.sleep, 0), caller.call_using_backend("trio", trio.sleep, 0)], return_when="ALL_COMPLETED", )

Caller.create_pending_group¶

A PendingGroup is an asynchronous context manager for Pending. All pending that 'opt-in' when created are added to the pending group (Caller methods that return Pending 'opt-in').

The PendingGroup has different modes:

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

In [14]:

import anyio

async with caller.create_pending_group() as pg:
    caller.call_soon(anyio.sleep, 0.01)
    caller.call_later(0, anyio.sleep, 0)
    caller.to_thread(anyio.sleep, 0.01)
    caller.call_using_backend("asyncio", asyncio.sleep, 0.01)
    caller.call_using_backend("trio", trio.sleep, 0.01)
    print("In context pending=", len(pg.pending))

print("Out of context pending=", len(pg.pending))

import anyio async with caller.create_pending_group() as pg: caller.call_soon(anyio.sleep, 0.01) caller.call_later(0, anyio.sleep, 0) caller.to_thread(anyio.sleep, 0.01) caller.call_using_backend("asyncio", asyncio.sleep, 0.01) caller.call_using_backend("trio", trio.sleep, 0.01) print("In context pending=", len(pg.pending)) print("Out of context pending=", len(pg.pending))

In context pending= 5

Out of context pending= 0