Python mixing asyncio and threading

From an architectural perspective, are there any concerns mixing asyncio and threading code?

Example,

import asyncio
from threading import Thread

class ThreadTask:
    def __init__(self, universe_reload_id) -> None:
        self.keep_running = True
        self.universe_reload_id = universe_reload_id


class SomeAlgo:

    def process_universe(self, universe, task):
        while task.keep_running:
            ...

    async def start(self):

        task = None

        universe_reload_id = 1
        for message in self.app_context.kafka_consumer:
            if task:
                task.keep_running = False

            pairs = json.loads(message.value.decode('utf-8'))

            task = ThreadTask(universe_reload_id=universe_reload_id)
            t = Thread(target=self.process_universe, args = (pairs, task))
            t.start() # here we'd call process_universe in non-blocking manner

async def _run_jobs():
    provider = SomeAlgo(param)
    await provider.start()

asyncio.get_event_loop().run_until_complete(_run_jobs())

I notice Python has an asyncio version of a semaphore:
Asyncio Semaphore in Python vs. Threading Semaphore in Python.

So in the above code, which semaphore should/can I use? The asyncio version or the plain version?

For the first question a Perspective :

Combining asyncio and threading can lead to complex code because they have different concurrency models. However, it's not inherently problematic if done carefully. Here are a few concerns you might want to keep in mind:

1. Thread-Safety: If both your async and threaded code are accessing shared data structures, you need to ensure that the access is thread-safe. This usually involves using locks or other synchronization primitives to prevent race conditions. In your case, task.keep_running is a shared variable that is read from a thread and written from the asyncio context. You should protect this variable with a lock or make it an atomic operation.

2. Error Handling: Error handling can get tricky when mixing async and threaded code. If an exception is thrown in your threaded code, it won't propagate to your async code and vice versa. You need to make sure that errors are properly caught and handled in both contexts.

3. Performance: Python's GIL (Global Interpreter Lock) can cause threads to run slower than expected because only one thread can execute Python bytecode at a time. However, this might not be a problem if your threaded code is I/O bound or if it's written in C and releases the GIL.

4. Stopping Threads: In your current setup, you stop a thread by setting task.keep_running to False. However, there's no guarantee when the thread will actually stop because it depends on when it next checks this variable. If you need more immediate stopping of threads, you may need to use a different approach, like thread.interrupt() (available in Python 3.7 and later).

5. Resource Usage: Each thread uses a certain amount of resources, and having a large number of threads can lead to high memory usage. Also, context-switching between threads can consume CPU. Depending on the number of threads you're planning to use and the resources available, this may or may not be a concern.

Overall, mixing asyncio and threading can be useful for running blocking code in a non-blocking manner, but it adds complexity to your program. Whenever possible, it's simpler to stick to one model of concurrency. If your threaded code is I/O bound, consider converting it to async code. If it's CPU-bound, consider using processes instead of threads to bypass the GIL.

For the second part question :

The Semaphore you should use depends on the context in which it's being used.

In the context of your provided code, where you're interacting with both asyncio and Thread, the choice of Semaphore would depend on where and what you're trying to limit.

• If you're trying to limit the concurrency of asynchronous tasks within your asyncio event loop, you should use asyncio.Semaphore.

• If you're trying to limit the concurrency of threads or synchronize across threads, you should use threading.Semaphore.

Remember, asyncio and threading are two different paradigms and the synchronization primitives (like Semaphore) are not interchangeable.

asyncio.Semaphore is designed to work with asyncio tasks and respects asyncio's event loop. It yields control back to the event loop when the semaphore is locked, allowing other tasks to run.

On the other hand, threading.Semaphore is a traditional blocking semaphore. When a thread waits for a semaphore to be released, it truly waits and does not let other Python threads run.

In your case, if you're planning to use a semaphore for the shared task.keep_running variable, you would likely use threading.Semaphore, as task.keep_running is being accessed from a thread. However, be aware that using a semaphore might be overkill if you're just protecting a single boolean variable. Using a threading.Lock or threading.Condition may be simpler and more efficient.

Cheers**!!