Caller: Multi-thread and asynchronous backend function caller (notebook)¶

This notebook uses trio in some cells.

Caller is flexible interface for scheduling work across threads and asynchronous backends.

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 a host (gui) event loop where the backend runs a guest.

A backend can also run as a guest of another backend in the same thread with:

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]:

Copied!

%callers
%callers

        Caller ID              Parent       Backend     Name   Children  Protected
──────────────────────────────────────────────────────────────────────────────────────
current → 140399132094592                   asyncio  MainThread       1    🔐    
          140399088121536 140399132094592   asyncio  Control               🔐

The kernel interface uses a minimum of two callers:

Shell: The caller for the shell channel (normally the name is 'MainThread'). This caller performs execute requests and comm message handing.
Control: A child caller of the shell's caller that is named 'Control'. Control handles other message requests.

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]:

Copied!

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) (requires a running backend).
MainThread: The MainThread Caller (requires a running caller in the MainThread or running backend when called from the MainThread).
NewThread: Create new caller with a backend running in new thread.

In [3]:

Copied!

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")

NewThread¶

Create a new caller with a backend running in a new thread. This can be useful when you are in thread that does not have a running backend.

In [4]:

Copied!





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()
await caller1.stopped
del caller1

# We can also use an async context for lifecycle management
async with Caller("NewThread", name="My caller2") as caller_2:
    await caller_2.call_soon(lambda: print(Caller()))
assert caller_2.stopped.done()
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()
await caller1.stopped
del caller1

# We can also use an async context for lifecycle management
async with Caller("NewThread", name="My caller2") as caller_2:
    await caller_2.call_soon(lambda: print(Caller()))
assert caller_2.stopped.done()

'hello world!'

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

<Caller ⚪🏃 running🔐 name='My caller2' backend='asyncio' host=None id=140398823462592 parent=None>

Caller creation options¶

A new caller instance 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]:

Copied!

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

All children of a caller will stop when the parent stops.

In [6]:

Copied!





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

---------------------------------------------------------------------------
ValueError                                Traceback (most recent call last)
File ~/work/async-kernel/async-kernel/src/async_kernel/shell/ipshell.py:585, in IPShell.run_code(self, code_obj, result, async_)
    583         await eval(code_obj, self.user_global_ns, self.user_ns)
    584     else:
--> 585         exec(code_obj, self.user_global_ns, self.user_ns)
    586 except self.custom_exceptions:  # pragma: no cover
    587     etype, value, tb = sys.exc_info()

Cell In[6], line 13
      9 
     10 # The same caller can be accessed by using the name
     11 assert caller.get(name="trio") is trio_caller
     12 
---> 13 child1 = trio_caller.get()
     14 child2 = trio_caller.get()
     15 
     16 assert trio_caller.children == {child1, child2}

File ~/work/async-kernel/async-kernel/src/async_kernel/caller.py:745, in Caller.get(self, **kwargs)
    743 if name is None:
    744     msg = "Name must be specified"
--> 745     raise ValueError(msg)
    746 if "backend" not in kwargs:
    747     kwargs["backend"] = self._backend

ValueError: Name must be specified

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]:

Copied!





# 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]:

Copied!





# 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]:

Copied!





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 140399132094592)>

sniffio: asyncio

backend: trio

thread: <_MainThread(MainThread, started 140399132094592)>

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]:

Copied!

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: 140399062611936

8 Pending id: 140399062611936

7 Pending id: 140399062611936

6 Pending id: 140399062611936

5 Pending id: 140399062611936

4 Pending id: 140399062611936

3 Pending id: 140399062611936

2 Pending id: 140399062611936

1 Pending id: 140399062611936

0 Pending id: 140399062611936

Caller.call_direct¶

Call the function directly in the caller's scheduler.

In [11]:

Copied!

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

Collection based methods¶

These methods are provided to manage a collection of awaitables.

Caller.as_completed: An async iterator to yield Pending as they complete.
Caller.wait: Wait for an iterable of awaitables to complete blocking until the return_when condition is satisfied.
Caller.create_pending_group: Create a Pending context which automatically registers Pending that are created in the context.

Caller.as_completed¶

An async iterator that yields Pending as they complete. It accepts either an iterable or iterator with awaitable items. Non Pending awaitables are wrapped in Pending using call_soon.

This example requires ipywidgets!

In [12]:

suppress-error

Copied!





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: 2

Finished: 6

Finished: 5

Finished: 7

Finished: 12

Finished: 9

Finished: 14

Finished: 11

Finished: 4

Finished: 16

Finished: 1

Finished: 18

Finished: 17

Finished: 15

Finished: 21

Finished: 13

Finished: 20

Finished: 8

Finished: 10

Finished: 23

Finished: 3

Finished: 25

Finished: 24

Finished: 29

Finished: 22

Finished: 19

Finished: 35

Finished: 31

Finished: 27

Finished: 33

Finished: 32

Finished: 26

Finished: 28

Finished: 37

Finished: 40

Finished: 36

Finished: 34

Finished: 30

Finished: 46

Finished: 49

Finished: 42

Finished: 41

Finished: 47

Finished: 48

Finished: 39

Finished: 38

Finished: 44

Finished: 43

Finished: 56

Finished: 51

Finished: 50

Finished: 45

Finished: 52

Finished: 62

Finished: 64

Finished: 59

Finished: 54

Finished: 55

Finished: 57

Finished: 58

Finished: 61

Finished: 53

Finished: 72

Finished: 63

Finished: 68

Finished: 70

Finished: 74

Finished: 67

Finished: 60

Finished: 73

Finished: 80

Finished: 65

Finished: 77

Finished: 69

Finished: 66

Finished: 78

Finished: 81

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 are converted to Pending with caller.call_soon.

In [13]:

Copied!





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 including that return Pending do).

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]:

Copied!





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

In [15]:

Copied!

%callers
%callers

        Caller ID              Parent       Backend          Name       Children  Protected
───────────────────────────────────────────────────────────────────────────────────────────────
current → 140399132094592                   asyncio  MainThread               12    🔐    
          140399088121536 140399132094592   asyncio  Control                        🔐    
          140398823462592 140399132094592     trio   trio                                
          140398840243904 140399132094592   asyncio  async-kernel worker                 
          140398504699584 140399132094592   asyncio  async-kernel worker                 
          140398487918272 140399132094592   asyncio  async-kernel worker                 
          140398253045440 140399132094592   asyncio  async-kernel worker                 
          140398236264128 140399132094592   asyncio  async-kernel worker                 
          140398219482816 140399132094592   asyncio  async-kernel worker                 
          140397984609984 140399132094592   asyncio  async-kernel worker                 
          140397967828672 140399132094592   asyncio  async-kernel worker                 
          140397951047360 140399132094592   asyncio  async-kernel worker                 
          140397716174528 140399132094592   asyncio  async-kernel worker

Backends (anyio, asyncio, trio)¶

Running code in Caller is compatible with the asynchronous backends as follows:

flowchart TD
    Caller <---> asyncio
    Caller <--> anyio
    Caller <---> trio
    anyio <--> asyncio
    anyio <--> trio

Low-level primitives¶

It is recommended to use aiologic primitives when thread-safety is required. Async-library specific primitives (including Anyio) will work provided the use is limited to the backend in which it is created.