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 for the whole process. Device I/O and any coroutine connected to a psygnal signal execute there, off the GUI thread that emits.

Only run_coro is meant for general use: it is how synchronous code - a presenter method, a Qt slot - runs a coroutine on that loop and gets its result. Everything else in this module is application plumbing, set up by the application container during startup and torn down on shutdown. Components should not build a loop or install a backend of their own.

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

If True, return the Future object instead of waiting for the result.

 False

Returns:

Type Description
R

The result of the coroutine.
Source code in src/redsun/aio.py 
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
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
        If ``True``, return the `Future` object instead of waiting for the result.

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

Internal machinery

Not part of the public API

The symbols below are wired up by the application container at startup and torn down on shutdown. They are documented so that the runtime's behaviour is inspectable, not so that components call them: installing a second backend, or building a loop alongside the shared one, breaks signal dispatch for the whole process. Use run_coro to reach the shared loop.

Return the background event loop.

Returns:

Type Description
AbstractEventLoop

The shared event loop.
Source code in src/redsun/aio.py 
100
101
102
103
104
105
106
107
108
def get_shared_loop() -> asyncio.AbstractEventLoop:
    """Return the background event loop.

    Returns
    -------
    asyncio.AbstractEventLoop
        The shared event loop.
    """
    return _loop_factory()

Install the culsans backend as psygnal's active async backend.

Must be called before connecting a coroutine to a signal. Calling it again returns the backend installed by the first call; tear it down with psygnal's own 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 
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
def set_async_backend() -> CulsansAsyncioBackend:
    """Install the culsans backend as psygnal's active async backend.

    Must be called before connecting a coroutine to a signal. Calling it
    again returns the backend installed by the first call; tear it down with
    psygnal's own ``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 are dispatched as tasks on the loop returned by get_shared_loop, so signals emitted from any thread are delivered.

Source code in src/redsun/aio.py 
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
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
class CulsansAsyncioBackend(_AsyncBackend, Loggable):
    """Psygnal async backend draining a culsans queue on the shared loop.

    Queued callbacks are dispatched as tasks on the loop returned by
    `get_shared_loop`, so signals emitted from 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 indicating whether 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 outstanding 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 purposes."""
        return f"psygnal-{self._backend}"

running property 

running: AwaitableEvent

Return the event indicating whether the backend accepts callbacks.

name property 

name: str

Name of the backend, for logging and debugging purposes.

put 

put(item: QueueItem) -> None

Queue a callback for dispatch on the shared loop.

Source code in src/redsun/aio.py 
137
138
139
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 outstanding callbacks.

Source code in src/redsun/aio.py 
141
142
143
def close(self) -> None:
    """Shut the queue down; the drain cancels outstanding 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 
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
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 that the event can be set and cleared from any thread while still being 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 that the event can be set and cleared from any
    thread while still being 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