README.rst

aiologic

aiologic is an async-aware library for tasks synchronization and their communication in different threads and different event loops. For example, if there is interaction between classic synchronous (threaded) code and asynchronous one, between two asynchronous codes in different threads, or any other combination that you want. Let's take a look at the example:

from threading import Thread

import anyio

from aiologic import Lock

lock = Lock()


async def func(i, j):
    print(f"thread={i} task={j} started")

    async with lock:
        await anyio.sleep(1)

    print(f"thread={i} task={j} stopped")


async def main(i):
    async with anyio.create_task_group() as tasks:
        for j in range(2):
            tasks.start_soon(func, i, j)


for i in range(2):
    Thread(target=anyio.run, args=[main, i]).start()

It prints something like this:

thread=0 task=0 started
thread=1 task=0 started
thread=0 task=1 started
thread=1 task=1 started
thread=0 task=0 stopped
thread=1 task=0 stopped
thread=0 task=1 stopped
thread=1 task=1 stopped

As you can see, when using aiologic.Lock, tasks from different event loops are all able to acquire a lock. In the same case if you use anyio.Lock, it will raise a RuntimeError. And threading.Lock will cause a deadlock.

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 there are situations when these so different styles need to coexist:

• 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.

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, and in the case of a queue only works when there is only one consumer. 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():
    loop = asyncio.get_running_loop()

    for _ in range(8):
        try:
            await asyncio.wait_for(loop.run_in_executor(
                executor,
                semaphore.acquire,
            ), 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

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, and that's why you're here.

Features

• Python 3.8+ support
• CPython and PyPy support
• Pickling and weakrefing support
• Cancellation and timeouts support
• Optional Trio-style checkpoints:
  • enabled by default for Trio itself
  • disabled by default for all others
• Only one checkpoint per asynchronous call:
  • exactly one context switch if checkpoints are enabled
  • zero or one context switch if checkpoints are disabled
• Fairness wherever possible (with some caveats)
• Thread safety wherever possible
• Zero required dependencies
• Lock-free implementation

Synchronization primitives:

• Semaphores: counting and bounded
• Locks: primitive, ownable and reentrant
• Capacity limiters: simple and reentrant
• Condition variables
• Barriers: single-use and cyclic
• Events: one-time, reusable and countdown
• Resource guards

Communication primitives:

• Queues: FIFO, LIFO and priority

Supported concurrency libraries:

• asyncio and trio (coroutine-based)
• eventlet and gevent (greenlet-based)

All synchronization and communication primitives are implemented entirely on effectively atomic operations, which gives an incredible speedup on PyPy compared to alternatives from the threading module. All this works because of GIL, but per-object locks also ensure that the same operations are still atomic, so aiologic also works when running in a free-threaded mode.

Installation

Install from PyPI (recommended):

pip install aiologic

Or from GitHub:

pip install git+https://github.com/x42005e1f/aiologic.git

You can also use other package managers, such as uv.

Communication channels

GitHub Discussions: https://github.com/x42005e1f/aiologic/discussions

Feel free to post your questions and ideas here.

Derivatives

• x42005e1f/culsans - Janus-like sync-async queue. Unlike aiologic queues, provides API compatible interfaces.

Stars

License

The aiologic library is offered under the ISC License.