[FIXED] Python asyncio in a thread for migrating existing codebase

Issue

We have a rather big project that is doing a lot of networking (API calls, Websocket messages) and that also has a lot of internal jobs running in intervals in threads. Our current architecture involves spawning a lot of threads and the app is not working very well when the system is under a big load, so we've decided to give asyncio a try.

I know that the best way would be to migrate the whole codebase to async code, but that is not realistic in the very near future because of the size of the codebase and the limited development resources. However, we would like to start migrating parts of our codebase to use asyncio event loop and hopefully, we will be able to convert the whole project at some point.

The problem we have encountered so far is that the whole codebase has sync code and in order to add non-blocking asyncio code inside, the code needs to be run in different thread since you can't really run async and sync code in the same thread.

In order to combine async and sync code, I came up with this approach of running the asyncio code in a separate thread that is created on app start. Other parts of the code add jobs to this loop simply by calling add_asyncio_task.

import threading
import asyncio
_tasks = []

def threaded_loop(loop):
    asyncio.set_event_loop(loop)
    global _tasks
    while True:
        if len(_tasks) > 0:
            # create a copy of needed tasks
            needed_tasks = _tasks.copy()
            # flush current tasks so that next tasks can be easily added
            _tasks = []
            # run tasks
            task_group = asyncio.gather(*needed_tasks)
            loop.run_until_complete(task_group)


def add_asyncio_task(task):
    _tasks.append(task)

def start_asyncio_loop():
    loop = asyncio.get_event_loop()
    t = threading.Thread(target=threaded_loop, args=(loop,))
    t.start()

and somewhere in app.py:

start_asyncio_loop()

and anywhere else in the code:

add_asyncio_task(some_coroutine)

Since I am new to asyncio, I am wondering if this is a good approach in our situation or if this approach is considered an anti-pattern and has some problems that will hit us later down the road? Or maybe asyncio already has some solution for this and I'm just trying to invent the wheel here?

Thanks for your inputs!

Solution

The approach is fine in general. You have some issues though:

(1) Almost all asyncio objects are not thread safe

(2) Your code is not thread safe on its own. What if a task appears after needed_tasks = _tasks.copy() but before _tasks = []? You need a lock here. Btw making a copy is pointless. Simple needed_tasks = _tasks will do.

(3) Some asyncio constructs are thread safe. Use them:

import threading
import asyncio

# asyncio.get_event_loop() creates a new loop per thread. Keep
# a single reference to the main loop. You can even try
#   _loop = asyncio.new_event_loop()
_loop = asyncio.get_event_loop()

def get_app_loop():
    return _loop

def asyncio_thread():
    loop = get_app_loop()
    asyncio.set_event_loop(loop)
    loop.run_forever()

def add_asyncio_task(task):
    asyncio.run_coroutine_threadsafe(task, get_app_loop())

def start_asyncio_loop():
    t = threading.Thread(target=asyncio_thread)
    t.start()

Answered By - freakish