Skip to content

Async runtime

Design rationale: ADR 0005.

Running coroutines

Shared background event loop, and dispatch of coroutines connected to signals.

redsun runs one background asyncio event loop per process. Device I/O and coroutines connected to psygnal signals run there, off the emitting GUI thread.

run_coro is for general use: synchronous code, such as a presenter method or a Qt slot, runs a coroutine on the loop with it and gets the result. The rest of the module is set up by the container at startup and torn down at shutdown. Components must not build their own loop or install a backend.

run_coro 

run_coro(
    coro: Coroutine[Any, Any, R],
    return_future: Literal[False] = ...,
) -> R

run_coro(
    coro: Coroutine[Any, Any, R],
    return_future: Literal[True] = ...,
) -> Future[R]

run_coro(
    coro: Coroutine[Any, Any, R],
    return_future: bool = False,
) -> R | Future[R]

Run a coroutine in the background event loop and return its result.

Parameters:

Name Type Description Default
coro Coroutine

The coroutine to run.

 required
return_future bool

Return the Future instead of waiting for the result.

 False

Returns:

Type Description
R

The result of the coroutine.
Source code in src/redsun/aio.py 
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
def run_coro(
    coro: Coroutine[Any, Any, R], return_future: bool = False
) -> R | Future[R]:
    """Run a coroutine in the background event loop and return its result.

    Parameters
    ----------
    coro : collections.abc.Coroutine
        The coroutine to run.
    return_future : bool, optional
        Return the `Future` instead of waiting for the result.

    Returns
    -------
    R
        The result of the coroutine.
    """
    future = asyncio.run_coroutine_threadsafe(coro, get_shared_loop())
    return future if return_future else future.result()

Internal machinery

Not part of the public API

The container sets these up at startup and tears them down at shutdown. They are documented so the runtime can be inspected, not for components to call: a second backend, or a loop beside the shared one, breaks signal dispatch for the whole process. Reach the shared loop with run_coro.

Return the background event loop, starting it on its own thread on first use.

Source code in src/redsun/aio.py 
67
68
69
70
71
72
73
74
75
76
@cache
def get_shared_loop() -> asyncio.AbstractEventLoop:
    """Return the background event loop, starting it on its own thread on first use."""
    loop = asyncio.new_event_loop()
    thread = Thread(target=loop.run_forever, daemon=True)
    thread.start()
    # bluesky's RunEngine looks up the thread of a loop that is already running,
    # and a loop it did not start itself is missing from that registry
    _ensure_event_loop_running.loop_to_thread[loop] = thread  # type: ignore[attr-defined]
    return loop

Install the culsans backend as psygnal's async backend.

Call it before connecting a coroutine to a signal. A second call returns the installed backend; tear it down with psygnal's clear_async_backend.

Returns:

Type Description
CulsansAsyncioBackend

The active backend.

Raises:

Type Description
RuntimeError

If a different async backend is already active.
Source code in src/redsun/aio.py 
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
def set_async_backend() -> CulsansAsyncioBackend:
    """Install the ``culsans`` backend as ``psygnal``'s async backend.

    Call it before connecting a coroutine to a signal. A second call returns the
    installed backend; tear it down with ``psygnal``'s ``clear_async_backend``.

    Returns
    -------
    CulsansAsyncioBackend
        The active backend.

    Raises
    ------
    RuntimeError
        If a different async backend is already active.
    """
    current = get_async_backend()
    if isinstance(current, CulsansAsyncioBackend):
        return current
    if current is not None:
        raise RuntimeError(f"Async backend already set to: {current._backend}")

    backend = CulsansAsyncioBackend()

    # psygnal resolves the active backend through its own module global, so
    # binding a name here is not enough for `get_async_backend()` to find it
    psygnal._async._ASYNC_BACKEND = backend
    return backend

Bases: _AsyncBackend, Loggable

psygnal async backend draining a culsans queue on the shared loop.

Queued callbacks run as tasks on the loop from get_shared_loop, so signals emitted on any thread are delivered.

Source code in src/redsun/aio.py 
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
class CulsansAsyncioBackend(_AsyncBackend, Loggable):
    """``psygnal`` async backend draining a ``culsans`` queue on the shared loop.

    Queued callbacks run as tasks on the loop from `get_shared_loop`, so signals
    emitted on any thread are delivered.
    """

    def __init__(self) -> None:
        super().__init__("culsans")
        self._queue: Queue[QueueItem] = Queue()
        self._running = AwaitableEvent()
        self._draining = False
        self._tasks: set[asyncio.Task[None]] = set()

        # the queue holds callbacks from here on, so work queued before the
        # loop thread picks the drain up is still delivered; marking the
        # backend running only once the drain executes would expose a window
        # in which callers see it as inert when it is not
        self._running.set()
        self._run_task = asyncio.run_coroutine_threadsafe(self.run(), get_shared_loop())

    @property
    def running(self) -> AwaitableEvent:
        """Return the event set while the backend accepts callbacks."""
        return self._running

    def put(self, item: QueueItem) -> None:
        """Queue a callback for dispatch on the shared loop."""
        self._queue.put_nowait(item)

    def close(self) -> None:
        """Shut the queue down; the drain cancels pending callbacks."""
        self._queue.shutdown()

    async def run(self) -> None:
        """Drain the queue until it is shut down or the drain is cancelled."""
        if self._draining:
            return
        self._draining = True
        try:
            loop = get_shared_loop()
            while True:
                item = await self._queue.async_get()
                task = loop.create_task(self.call_back(item))
                self._tasks.add(task)
                task.add_done_callback(self._tasks.discard)
                task.add_done_callback(self._log_slot_exception)
        except asyncio.CancelledError:
            self.logger.debug("Dispatch cancelled")
        except QueueShutDown:
            self.logger.debug("Dispatch queue shut down")
        except Exception as e:
            self.logger.error(f"Dispatch stopped: {e}", exc_info=e)
        finally:
            self._draining = False
            self._running.clear()
            for task in self._tasks:
                task.cancel()

    def _log_slot_exception(self, task: asyncio.Task[None]) -> None:
        """Report an exception raised by a slot, which nothing else awaits."""
        if task.cancelled():
            return
        if (exc := task.exception()) is not None:
            self.logger.error(f"Exception in async slot: {exc}", exc_info=exc)

    @property
    def name(self) -> str:
        """Name of the backend, for logging and debugging."""
        return f"psygnal-{self._backend}"

running property 

running: AwaitableEvent

Return the event set while the backend accepts callbacks.

name property 

name: str

Name of the backend, for logging and debugging.

put 

put(item: QueueItem) -> None

Queue a callback for dispatch on the shared loop.

Source code in src/redsun/aio.py 
105
106
107
def put(self, item: QueueItem) -> None:
    """Queue a callback for dispatch on the shared loop."""
    self._queue.put_nowait(item)

close 

close() -> None

Shut the queue down; the drain cancels pending callbacks.

Source code in src/redsun/aio.py 
109
110
111
def close(self) -> None:
    """Shut the queue down; the drain cancels pending callbacks."""
    self._queue.shutdown()

run async 

run() -> None

Drain the queue until it is shut down or the drain is cancelled.

Source code in src/redsun/aio.py 
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
async def run(self) -> None:
    """Drain the queue until it is shut down or the drain is cancelled."""
    if self._draining:
        return
    self._draining = True
    try:
        loop = get_shared_loop()
        while True:
            item = await self._queue.async_get()
            task = loop.create_task(self.call_back(item))
            self._tasks.add(task)
            task.add_done_callback(self._tasks.discard)
            task.add_done_callback(self._log_slot_exception)
    except asyncio.CancelledError:
        self.logger.debug("Dispatch cancelled")
    except QueueShutDown:
        self.logger.debug("Dispatch queue shut down")
    except Exception as e:
        self.logger.error(f"Dispatch stopped: {e}", exc_info=e)
    finally:
        self._draining = False
        self._running.clear()
        for task in self._tasks:
            task.cancel()

Resettable event whose wait is a coroutine.

Wraps aiologic.REvent, so the event can be set and cleared from any thread and awaited from a coroutine.

Source code in src/redsun/aio.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
class AwaitableEvent:
    """Resettable event whose ``wait`` is a coroutine.

    Wraps `aiologic.REvent`, so the event can be set and cleared from any thread
    and awaited from a coroutine.
    """

    def __init__(self) -> None:
        self._event = aiol.REvent()

    def is_set(self) -> bool:
        """Return ``True`` if the event is set."""
        return self._event.is_set()

    def set(self) -> None:
        """Set the event, waking every waiter."""
        self._event.set()

    def clear(self) -> None:
        """Unset the event."""
        self._event.clear()

    async def wait(self) -> None:
        """Wait until the event is set."""
        await self._event

is_set 

is_set() -> bool

Return True if the event is set.

Source code in src/redsun/aio.py 
47
48
49
def is_set(self) -> bool:
    """Return ``True`` if the event is set."""
    return self._event.is_set()

set 

set() -> None

Set the event, waking every waiter.

Source code in src/redsun/aio.py 
51
52
53
def set(self) -> None:
    """Set the event, waking every waiter."""
    self._event.set()

clear 

clear() -> None

Unset the event.

Source code in src/redsun/aio.py 
55
56
57
def clear(self) -> None:
    """Unset the event."""
    self._event.clear()

wait async 

wait() -> None

Wait until the event is set.

Source code in src/redsun/aio.py 
59
60
61
async def wait(self) -> None:
    """Wait until the event is set."""
    await self._event