Libraries

Note

This is the first section on advanced topics that cover the inner workings of aiologic as well as some complex scenarios. For the reader’s convenience, they are in a frequently asked questions (FAQ) format, making it easier to find information.

There are several goals that aiologic aims to achieve. One of them is to provide a convenient “glue” for different multitasking worlds, and thus different concurrency libraries. When you work with concurrent.futures.ThreadPoolExecutor, you are actually working with threading. When you work with aiohttp.ClientSession, you are actually working with asyncio. The huge number of asynchronous libraries available for your use rely on a very small set of “root” concurrency libraries.

However, the concurrency libraries are usually unaware of each other, making different asynchronous libraries incompatible with each other. Acting as a bridge between different worlds, aiologic should be aware of the largest number of concurrency libraries to cover the maximum number of possible combinations.

Which libraries does aiologic support?

Currently, aiologic supports the following concurrency libraries:

• asyncio, curio, trio, and anyio (coroutine-based)

• eventlet, and gevent (greenlet-based)

• threading (thread-based)

But uses a slightly different division in its work:

• coroutine-based libraries are “async” libraries

• greenlet-based and thread-based libraries are “green” libraries

This division is based on the fact that only coroutine-based libraries use asynchronous functions (PEP 492), while the code for thread-based libraries cannot be distinguished from the code for greenlet-based libraries by function signatures.

Why is AnyIO mentioned?

While anyio runs on top of either asyncio or trio, it implements trio-like structured concurrency with a different cancellation semantics than asyncio. Therefore, aiologic has to support both semantics for asyncio at the same time by using anyio features when they are available.

Why is Twisted not supported?

While twisted has some features of newer concurrency libraries and even supports asyncio via a separate reactor, it lacks the concept of a current execution unit, commonly called a current task.

The problem can be illustrated most clearly with the example of aiologic.RLock: how to handle lock acquisitions in different callbacks? Acquire reentrantly, assuming the entire reactor is one execution unit? Or acquire non-reentrantly, like DeferredLock, assuming that the different callbacks are different execution units? But then how to distinguish them from each other?

Since there is no clear way to implement support for twisted that would allow aiologic primitives to work as native twisted primitives, it is currently not supported at all. Instead, you can use aiologic with twisted via its asyncio support.

Why is multiprocessing not supported?

Unlike threads, which share common memory, each process manages its own memory. This is a different situation than the one in which aiologic primitives operate. In order to support multiprocessing, new approaches are needed.

There are several ways to enable interactions between processes, such as:

1. Allocate shared memory and rely on operations on it.

2. Use standard communication mechanisms such as pipes or sockets.

The first way is closest to the spirit of aiologic, but has limited scalability and cannot be implemented via effectively atomic operations in pure Python. The second way would require implementing an architecture without a server process to preserve the spirit of aiologic, which is a challenge that requires a separate research.

Thus, there is no way to easily implement multiprocessing support that would not conflict with aiologic’s ideas. Nevertheless, it is still an open question, and the situation may change in the future.

Why are the APIs separated?

Different approaches to providing both synchronous and asynchronous APIs coexist in the wild. In contrast to API separation, which implies using different functions for different types of libraries, there is an approach of providing functionality for different worlds via the same functions behaving differently in different contexts (returning awaitable objects in an asynchronous context and plain values in others). This improves both compatibility and usability, but has some non-trivial drawbacks:

1. Such dual nature is bad for static type checking. Let’s imagine a function, such as acquire(), that returns bool in a synchronous context. If it were to return Awaitable[bool] in an asynchronous context, its actual return type would be a union of both types. This raises a question about whether to allow the await expression for objects of that type. If allowed, a type checker might miss a situation where an object is not actually an awaitable object, resulting in a type error. If disallowed, the await expression can be used only after runtime type checking (or after type casting), which is inconvenient for the user.

2. One of the simplest ways to implement such behavior is to check that there is a running event loop of an asynchronous library in the current thread. This makes the behavior runtime dependent, which in turn can lead to undefined behavior. Suppose it is implemented by some library that is used by some other, non-asynchronous library that does not inherit this behavior. Then accidental use of the second library in an asynchronous context will activate the asynchronous API that is not expected by the library, and this in turn will lead to errors, perhaps even hard-to-detect ones (e.g. in case of wait()-like methods).

3. A more correct way is to check that functions are used from a coroutine, which is particularly implemented by curio. This makes the behavior less runtime dependent, but still has some nuances. First, it prevents calling the asynchronous version of a function from synchronous functions, making it incompatible with wrappers (such as functools.partial(), but in pure Python for curio implementation due to special handling of list comprehensions and generator expressions). Second, this check negatively affects performance, especially on PyPy, where the difference can be a hundredfold or more.

The API separation has drawbacks too, but they are related to the development process rather than runtime specifics.

Why are the release methods not asynchronous?

Once we have dealt with the previous question, we are bound to come to the next, equally difficult question in software design: should we represent the entire asynchronous API as coroutine functions? It refers to one of the most important ideas from software architecture, called “separation of concerns”, and is a strict form of API separation. It is the curio way, and it directly affects release methods, such as release().

Let’s assume that we decided to make release methods as coroutine functions. Then we will be able to use any asynchronous API in such methods to create any complex logic, which opens the way to implementing primitives like eventlet.semaphore.CappedSemaphore, whose release methods can be blocking. This is a strong advantage, but let’s move on to the disadvantages:

1. More complex cancellation handling. Because release methods are used after acquire methods, they must always be fully executed, which would require shielding release method calls from cancellation. Otherwise, cancelling a release method call (e.g. due to its blocking nature or a checkpoint) will result in the primitive no longer being able to be acquired.

2. A cascade effect. Coroutine functions can only be used from coroutines — the async/await syntax says so. As a consequence, in addition to splitting event.set() into event.green_set() and event.async_set(), any code that used event.set() in an asynchronous environment must also become asynchronous. This problem has long been a source of friction in anyio, resulting in its dropping curio support.

3. Just lower performance. With checkpoints enabled, any code that uses several primitives in a row will switch context several times in a row even if there is no workload between context switches. For example, if acquire() comes right after release().

Due to these disadvantages, aiologic does not use async/await syntax where it can be done without it, which is the same approach as asyncio, trio, and other libraries other than curio. However, while aiologic does not follow the curio way, it fully supports curio, although not with interfaces native to its architectural model.

How does aiologic import libraries?

aiologic is strong in that it efficiently supports the wide variety of concurrency libraries without depending on any of them. The techniques it uses for this purpose resemble lazy imports in action, but differ from them in flexibility and higher efficiency.

The first of them may have different names, but let us call it “global rebinding”. When you call an aiologic function that needs to access a third-party library’s API, on the first call it does all the necessary imports and replaces itself at the module level with the actual implementation. This eliminates the need to import optional libraries that are not currently required.

Here is how, for example, a function to check that there is a running twisted reactor in the current thread can be implemented:

def _twisted_running() -> bool:
    global _twisted_running

    from twisted.python.threadable import isInIOThread

    _twisted_running = isInIOThread  # global rebinding

    return _twisted_running()

One serious disadvantage of this example is that if you do not use twisted, you have to install it anyway, because otherwise ImportError will be raised. Well, there are several ways to solve this problem.

The naive one is to suppress the exception and return a default value instead. Despite its simplicity, it is very, very slow because each time it is called, it runs the complex import system that can make a lot of file system calls.

def _twisted_running() -> bool:
    global _twisted_running

    try:
        import twisted.internet
    except ImportError:  # reactor is not in use
        return False

    from twisted.python.threadable import isInIOThread

    _twisted_running = isInIOThread  # global rebinding

    return _twisted_running()

The other is to check sys.modules for the imported module. It is much better than the naive way and gives performance almost comparable to a direct return of a default value, but only almost, which may matter when calling several functions at a time.

def _twisted_running() -> bool:
    global _twisted_running

    if "twisted.internet" not in sys.modules:  # reactor is not in use
        return False

    from twisted.python.threadable import isInIOThread

    _twisted_running = isInIOThread  # global rebinding

    return _twisted_running()

Thus, we come to the second technique. That is post import hooks (PEP 369) by wrapt. These allow to use a very fast dummy function before importing a third-party library on the side, and then replace it with a function that does the imports on the first call according to the first technique. This combination of techniques bypasses the need to import when it is known that the library has not yet started to be used, and also preserves proper exception propagation.

With the second technique, the example can be improved as follows:

from wrapt import when_imported

def _twisted_running() -> bool:
    return False

@when_imported("twisted.internet")
def _(_):  # post import hook
    global _twisted_running

    def _twisted_running():  # global rebinding
        global _twisted_running

        from twisted.python.threadable import isInIOThread

        _twisted_running = isInIOThread  # global rebinding

        return _twisted_running()

The result is the following semantics: libraries are imported when functions are called, and only when the libraries are actually required. Combined with aiologic’s architecture, this gives us one interesting effect: primitives can work without needing to import any optional libraries at all. For example, if a lock is only used in one task at a time, that is, non-blocking.

In addition to these techniques, aiologic also uses functools.update_wrapper() to copy original function information, such as annotations and docstrings, into replacement functions, but removes the __wrapped__ attribute to eliminate memory leaks when a function is called by many threads at a time.

Note

As a careful reader may notice, functions built on any of the techniques should not be imported directly, since they are only replaced at the level of their module. In fact, aiologic uses such functions indirectly, to build more general abstractions based on them.

Hint

One consequence is that the fewer libraries you import, the faster aiologic runs.

How does aiologic detect libraries?

The example given during the previous question showed one of the cases of checking that a library is “running”, and it is for a reason. Using similar functions, aiologic implements detection of the currently running library — the one whose API should be used to create a waiter or get the current task.

aiologic.lowlevel.current_green_library(*, failsafe: Literal[False] = False) → str

aiologic.lowlevel.current_green_library(*, failsafe: Literal[True]) → str | None

Detect which green library is currently running.

Parameters:

failsafe – Unless set to True, the function will raise an exception when there is no current green library. Otherwise the function returns None in that case.

Returns:

A string like "gevent" or None.

Raises:

GreenLibraryNotFoundError – if the current green library was not recognized.

Example

def green_sleep(seconds: float) -> None:
    match library := aiologic.lowlevel.current_green_library():
        case "threading":
            time.sleep(seconds)
        case "eventlet":
            eventlet.sleep(seconds)
        case "gevent":
            gevent.sleep(seconds)
        case _:
            msg = f"unsupported green library {library!r}"
            raise RuntimeError(msg)

aiologic.lowlevel.current_async_library(*, failsafe: Literal[False] = False) → str

aiologic.lowlevel.current_async_library(*, failsafe: Literal[True]) → str | None

Detect which async library is currently running.

Parameters:

failsafe – Unless set to True, the function will raise an exception when there is no current async library. Otherwise the function returns None in that case.

Returns:

A string like "trio" or None.

Raises:

AsyncLibraryNotFoundError – if the current async library was not recognized.

Example

async def async_sleep(seconds: float) -> None:
    match library := aiologic.lowlevel.current_async_library():
        case "asyncio":
            await asyncio.sleep(seconds)
        case "curio":
            await curio.sleep(seconds)
        case "trio":
            await trio.sleep(seconds)
        case _:
            msg = f"unsupported async library {library!r}"
            raise RuntimeError(msg)

exception aiologic.lowlevel.GreenLibraryNotFoundError

Bases: RuntimeError

Exception raised by the aiologic.lowlevel.current_green_library() function if the current green library was not recognized.

exception aiologic.lowlevel.AsyncLibraryNotFoundError

Bases: RuntimeError

Exception raised by the aiologic.lowlevel.current_async_library() function if the current async library was not recognized.

When are libraries counted as running?

Libraries are counted as “currently running” in the current thread by the following semantics:

• threading: always running

• eventlet: running if there is a hub

• gevent: running if there is a hub

• asyncio: running if asyncio.get_running_loop() returns a non-None object

• curio: running if curio.meta.curio_running() returns True

• trio: running if trio.lowlevel.in_trio_run() returns True

It is much clearer in action. The green libraries except threading are counted as currently running if at least one of their functions that touches the current hub has been called, such as eventlet.sleep() or gevent.spawn().

>>> aiologic.lowlevel.current_green_library()
'threading'
>>> eventlet.spawn(eventlet.sleep)
<GreenThread object at 0x7ff1db690200 (otid=0x(nil)) pending>
>>> aiologic.lowlevel.current_green_library()
'eventlet'
>>> gevent.spawn(gevent.sleep)
<Greenlet at 0x7ff1db5191c0: sleep>
>>> aiologic.lowlevel.current_green_library()
'gevent'

The async libraries are counted as currently running if the function was called in an active library run (asyncio.run() / curio.run() / trio.run() / anyio.run()).

>>> async def test() -> str:
...     return aiologic.lowlevel.current_async_library()
...
>>> asyncio.run(test())
'asyncio'
>>> curio.run(test)
'curio'
>>> trio.run(test)
'trio'
>>> anyio.run(test, backend="asyncio")
'asyncio'
>>> anyio.run(test, backend="trio")
'trio'

In case there are several running libraries, priority is given to the one that is lower in the list (newer). This was already shown in the green libraries example, where both eventlet and gevent hubs coexisted at the same time. A similar situation for async libraries is nested run() calls.

Note

Due to the fact that threading is always counted as running, current_green_library() will always be able to recognize the current green library. As a consequence, the failsafe parameter has no effect, and GreenLibraryNotFoundError is actually useless. Nevertheless, they are still present for consistency.

How to manually specify a library as running?

There are situations where you need to explicitly specify the currently running library. For example, if you are migrating from eventlet to gevent, so you have eventlet and gevent hubs coexisting in the same thread, and you want to switch between them. For these situations, aiologic provides thread-local objects with which you can explicitly set the name of any supported library for the current thread.

aiologic.lowlevel.current_green_library_tlocal: threading.local

Thread-local data to control the return value of aiologic.lowlevel.current_green_library().

current_green_library_tlocal.name: str | None = None

Unless set to a non-None object, the function detects the current green library with its own algorithms. Otherwise the function returns exactly the set object.

Example

library = aiologic.lowlevel.current_green_library_tlocal.name

aiologic.lowlevel.current_green_library_tlocal.name = "somelet"

try:
    ...  # aiologic.lowlevel.current_green_library() == "somelet"
finally:
    aiologic.lowlevel.current_green_library_tlocal.name = library

aiologic.lowlevel.current_async_library_tlocal: threading.local

Thread-local data to control the return value of aiologic.lowlevel.current_async_library().

current_async_library_tlocal.name: str | None = None

Unless set to a non-None object, the function detects the current async library with its own algorithms. Otherwise the function returns exactly the set object.

Example

library = aiologic.lowlevel.current_async_library_tlocal.name

aiologic.lowlevel.current_async_library_tlocal.name = "someio"

try:
    ...  # aiologic.lowlevel.current_async_library() == "someio"
finally:
    aiologic.lowlevel.current_async_library_tlocal.name = library

It is worth noting that aiologic.lowlevel.current_async_library_tlocal.name is the same as sniffio.thread_local.name. So you can specify the current async library via either aiologic or sniffio. This is how trio and trio-asyncio detection is implemented.

Tip

If the performance of library detection is critical for you, you could manually specify a library even when it can be recognized automatically. This way the function will not use its own, slower, detection algorithms.

Is runtime installation supported?

When it comes to user scripts, there are different approaches to dependency management. One of them is to install missing dependencies straight within a script. But its ambiguous weakness is that it requires a special approach to optional dependency management on the side of the packages being installed, depending on where it is applied.

A good example would be the httpx package. It optionally supports decoding for “zstd” compressed responses via the zstandard package, using the following module-level code:

# Zstandard support is optional
try:
    import zstandard
except ImportError:  # pragma: no cover
    zstandard = None  # type: ignore

If a user script installs zstandard after importing httpx, the support will not be enabled.

The problem is taken into account in the aiologic design. You can install it and its optional dependencies both before and after importing, and even in parallel during use, which is achieved by using the techniques described above.

Note

Of course, this is not the main reason why aiologic does not use the module-level approach. If it tried to import all supported optional dependencies at initialization time, this could lead to either conflicts (such as between eventlet and trio) or incompatibility with their unsupported versions, even if they are not needed by the user.