Why?

Cooperative (coroutines, greenlets) and preemptive (threads) multitasking are not usually used together. Typically, you have an application that uses only threads (classic application) or only coroutines/greenlets (asynchronous application). But sometimes so different styles need to coexist.

# cooperative multitasking (deterministic execution order)

async def foo():
    print("foo (in)")
    await asyncio.sleep(0)  # switch to bar()
    print("foo (out)")

async def bar():
    print("bar (in)")
    await asyncio.sleep(0)  # switch to foo()
    print("bar (out)")

async with asyncio.TaskGroup() as tg:
    tg.create_task(foo())
    tg.create_task(bar())

# preemptive multitasking (non-deterministic execution order)

def foo():
    print("foo (in)")
    time.sleep(0)  # maybe switch to the main thread
    time.sleep(0)  # maybe switch to bar()
    print("foo (out)")

def bar():
    print("bar (in)")
    time.sleep(0)  # maybe switch to the main thread
    time.sleep(0)  # maybe switch to foo()
    print("bar (out)")

with ThreadPoolExecutor(2) as executor:
    executor.submit(foo)
    executor.submit(bar)

The main problem is notification when some event occurs, since both synchronization and communication depend on it. Cooperative-only (async-only) and preemptive-only (sync-only) worlds already have suitable primitives, but when they collide, things get much more complicated. Here are some of those situations (assuming that the primary multitasking style is cooperative):

  • Using a library that manages threads itself (e.g. a web app).

  • Reusing the same worker thread for different asynchronous operations (e.g. to access a serial port).

  • Requirement to guarantee even distribution of CPU resources between different groups of tasks (e.g. a chatbot working in multiple chats).

  • Interaction of two or more frameworks that cannot be run in the same event loop (e.g. a GUI framework with any other framework).

  • Parallelization of code whose synchronous part cannot be easily delegated to a thread pool (e.g. a CPU-bound network application that needs low response times).

  • Simultaneous use of incompatible concurrency libraries in different threads (e.g. due to legacy code).

  • Accelerating asynchronous applications in a nogil world.

These situations have one thing in common: you may need a way to interact between threads, at least one of which may run an event loop. However, you cannot use primitives from the threading module because they block the event loop. You also cannot use primitives from the asyncio module because they are not thread-safe/thread-aware.

Known solutions (only for some special cases) use one of the following ideas:

  • Delegate waiting to a thread pool (executor), e.g. via run_in_executor().

  • Delegate calling to an event loop, e.g. via call_soon_threadsafe().

  • Perform polling via timeouts and non-blocking calls.

All these ideas have disadvantages. Polling consumes a lot of CPU resources, actually blocks the event loop for a short time, and has poor responsiveness. The call_soon_threadsafe() approach does not actually do any real work until the event loop scheduler handles a callback. The run_in_executor() approach requires a worker thread per call and has issues with cancellation and timeouts:

import asyncio
import threading

from concurrent.futures import ThreadPoolExecutor

executor = ThreadPoolExecutor(8)
semaphore = threading.Semaphore(0)


async def main() -> None:
    loop = asyncio.get_running_loop()

    for _ in range(8):
        future = loop.run_in_executor(executor, semaphore.acquire)

        try:
            await asyncio.wait_for(future, 0)
        except asyncio.TimeoutError:
            pass


print("active threads:", threading.active_count())  # 1
asyncio.run(main())
print("active threads:", threading.active_count())  # 9 — wow, thread leak!

# program will hang until you press Control-C

Note

You can learn a little more about the various solutions and their limitations in the related Stack Overflow answer. Also, a comment on a PR for omnilib/aiomultiprocess describes a problem that is virtually impossible to solve using threads and/or polling.

However, aiologic has none of these disadvantages. Using its approach based on low-level events, it gives you much more than you can get with alternatives. That’s why it’s there.

Relevance

Despite all of aiologic’s features and the complexity of the problems it solves, it would not make sense if no one used it. With this subsection I want to show that aiologic is not only an interesting puzzle for its author (because it really is), but also a tool that can solve real problems. I think Stack Overflow’s related questions are good enough for this purpose.

The first related questions found on Stack Overflow predate the introduction of the asyncio module in Python 3.4. These are questions about mixing greenlets and threads:

But none of these questions address the problem of interaction between threads. Such a question was only asked in 2014, a few weeks before the Python 3.4 release, and another question was asked the following year:

And although they have such nice titles, the actual problems are unrelated: in the first case, importing the logging module before monkey patching is enough to solve the problem, while in the second, interaction with asyncio is only required if the asker considers an implicit thread-safety issue.

The really related questions started to be asked in 2015 after the Python 3.5 release. There were two questions that year, the first about a serial device and the second about a serial port:

So back in 2015, the real need for thread-safe primitives was visible. Curiously, the first Janus (thread-safe asyncio-aware queue) release, version 0.1.0, was published on June 11, 2015 — before October 1, 2015, when the related question was asked.

Since then, more and more questions have appeared. Here are just some of them:

And outside of the asyncio ecosystem, too:

Until now, there has been no universal library for all of these questions. Now there is.