1
\$\begingroup\$

A threading.Thread subclass intended for running an asyncio loop that has the ability to be stopped gracefully from another thread.

I spent a lot of time doing trial and error until it finally worked, but probably it still needs a lot of refinement. Warning to asyncio noobs like me: asyncio is very thread-unsafe in general. Don't assume otherwise.

Example:

import time


# This is a placeholder for an arbitrary async entry point
async def my_async_main():
    ...


async_thread = StoppableAsyncioThread(target=my_async_main)
time.sleep(5)
async_thread.stop()
async_thread.join()

Code:

import asyncio
import logging
import threading
from typing import ClassVar, Optional


class StoppableAsyncioThread(threading.Thread):
    """
    Stoppable asyncio thread.

    The target callable must be a coroutine function that can be executed with
    ``asyncio.run(target())``.
    """

    logger: ClassVar[logging.Logger] = logging.getLogger("threading")

    def __init__(self, *args, **kwargs) -> None:
        super().__init__(*args, **kwargs)
        # These are threading.Thread internals; equivalent to passing in the
        # corresponding keyword arguments to __init__. A bit lazy, but it guarantees
        # that the arguments and their defaults get handled correctly because
        # we delegate everything (*args, **kwargs) to threading.Thread.__init__
        self._args = (self._target_wrapper(self._target),)
        self._kwargs = {"debug": self.logger.getEffectiveLevel() <= logging.DEBUG}
        self._target = asyncio.run

        self.loop: Optional[asyncio.AbstractEventLoop] = None
        self._target_task = None

    def stop(self):
        if self.loop is None:
            raise RuntimeError("This asyncio thread is not running")
        self.logger.debug("Trying to stop %s", self)
        # cancelling the root task will be enough for the loop to close, because
        # we started it with asyncio.run, which only waits until the given task is done;
        # asyncio takes care of cancelling any other tasks
        self.loop.call_soon_threadsafe(self._target_task.cancel)

    async def _target_wrapper(self, target):
        self._target_task = asyncio.current_task()
        self.loop = asyncio.get_running_loop()
        try:
            await target()
        except asyncio.CancelledError:
            return
\$\endgroup\$
2
  • \$\begingroup\$ Could you also add the my_server function code? :) \$\endgroup\$ Commented Dec 23, 2020 at 7:33
  • \$\begingroup\$ Oh that's just a dummy example, it's not part of the code. I'd like to focus on the class itself, which should be able to support any async entry point passed as target. \$\endgroup\$
    – Anakhand
    Commented Dec 23, 2020 at 11:45

0

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.