Asyncio and coroutine synchronization support?

[OnceUponATimeInAmerica]

Thanks for wrapt.

I was wondering if you are going to add support for synchronizing coroutines (using @synchronized decorator) and accepting asyncio Locks to wrapt? Currently it does not accept asyncio locks.
Also because ayncio model is mostly single-threaded but highly re-entrant, rlocks (which are more suitable for parallel scenarios) would not work (I think).

[GrahamDumpleton]

Can you explain better what you think it needs to do and what would need to change in the implementation? Is it just a matter of it it using asyncio.Lock instead threading.RLock at:

• https://github.com/GrahamDumpleton/wrapt/blob/develop/src/wrapt/decorators.py#L516

I wasn't even aware that asyncio had locks so know nothing about them and how they are used.

[OnceUponATimeInAmerica]

Thanks for the reply.
I have to warn I am just a user.

To begin, asycio locks are awaited (their acquire method is a coroutine and invoked by async with ...).

Also (I assume) they would not be reentrant as it would basically allow the single thread servicing the event loop (and executing most/all the coroutines in the program, perhaps even concurrently) to go basically anywhere without any restrictions.

And because wrapper of a coroutine itself needs to be a coroutine (a chain of await ...) the syntax has to accommodate this. If we want to use the same decorator for both synchronous functions and also coroutines, the decorator function (which itself, of course, is an ordinary synchronous function) has to detect the type (of the wrapped routine) and return the appropriate wrapper. Somewhat similar to this or this.

Also if we couple this to the present feature in wrapt that the same synchronized decorator can be applied to both instance methods and also to static methods/class methods or even standalone functions (basically, it automatically decides the granularity for synchronization) the situation can get somewhat complex. Although the former problem (routine's type: sync vs. async) is orthogonal to the latter problem (instance vs class method).

Hope this was somewhat helpful.

[mentalisttraceur]

Quick note on wrapping either sync-or-async, since I've done this in compose: the only universal way to do this, last I checked, is to wait until you had the return value - because an async callable might be implemented as a normal sync function returning an "awaitable" value.

[pbryan]

I think adapting wrapt.synchronized to handle to asyncio coroutines is doable. As @mentalisttraceur mentions, the wrapper will need to perform runtime introspection on the wrapped function to determine if it is a synchronous or coroutine function. Also, if the semaphore is passed, it too would need to be identified as having synchronous or coroutine methods. @GrahamDumpleton If you see wrapt.synchronized as having a solid future, I'd be willing to take a crack at adding asyncio support to it, if you want.

[GrahamDumpleton]

Even if you can come up with a separate asynchronized decorator just to see if it is possible for async alone that would certainly be educational and help work out what to do and whether one could come up with a single decorator that does both. I have not played enough with Python async libraries to be sure that anything I worked out would be correct. So any help/guidance would be much appreciated.

[pbryan]

OK, will work on it. Just paste here for now, or propose in a PR?

[GrahamDumpleton]

Just follow up here initially with any conclusions you come to. My feeling right now is that an asynchronized decorator is only going to cover a subset of what synchronized can be used for. The asynchronized decorator might even need to use the synchronized decorator, or more basic standard mutex locks, internally to gate updates to stuff.

[mentalisttraceur]

Correction:

As @mentalisttraceur mentions, the wrapper will need to perform runtime introspection on the wrapped function to determine if it is a synchronous or coroutine function.

... on the return value of the wrapped function, to determine if it is an awaitable or not.

My point was that inspecting the function (as with inspect.iscoroutinefunction or asyncio.iscoroutinefunction) is an incomplete solution. It will miss cases such as

1. a callable class instance whose __call__ is async (to also detect those ahead-of-time, you gotta do inspect.iscoroutinefunction(f) or inspect.iscoroutinefunction(f.__call__));
2. a function not declared async but which still needs an await because it returned an "awaitable" value (this is impossible to detect ahead-of-time, but can be detected at call time by calling the function synchronously and then doing an inspect.isawaitable on the result).

That last method will work 100% of the time in all cases, but it can be trickier (or in some cases impossible) to write the logic in a way that can only decide how to act after calling the possibly-async function.

Take a look at how I've got it implemented in acompose, and then once that makes sense, also scroll down to the sacompose variant (whose __call__ only sometimes returns an awaitable).

The downside of just using the isawaitable method for all cases is that we end up with a wrapper that will always be false for the iscoroutinefunction check - it loses the ahead-of-time check in cases where it could be possible.

So the ideal wrapper combines these technique. Here's an untested/incomplete sketch of what that might look like:

def wrapper(function):
    if inspect.iscoroutinefunction(function) or inspect.iscoroutinefunction(function.__call__):
        async def _wrapper(*args, **kwargs):
            async with lock:
                return function(*args, **kwargs)
    else:
        def _wrapper(*args, **kwargs):
            result = function(*args, **kwargs)
            if inspect.isawaitable(result):
                async def _finish():
                    async with lock:
                        return await result
                return _finish()  # `await` intentionally omitted
            else:
                return result
    return _wrapper

[GrahamDumpleton]

FWIW, old example which tries to deal with sync or async functions.

#150 (comment)

Based on comments above though, that too probably needs more work to be complete, but important in that example is that it still uses @wrapt.decorator so introspection works on the resulting wrapped function.

[mentalisttraceur]

!!! 3.12 added inspect.markcoroutinefunction, arguably wiping out all my objections.

I think this tips the balance strongly in favor of: if a function returns an awaitable object but doesn't pass inspect.iscoroutinefunction, that's a bug in that callable, not something you need to handle.

If I had to re-design acompose and sacompose today, in a world where every Python implementation has a way to make any callable pass inspect.iscoroutinefunction, I would probably never bother with all that at-call-time inspect.isawaitable stuff - if inspect.iscoroutinefunction(function) or inspect.iscoroutinefunction(function.__call__): would be good enough.

Personally, I think it's fine for wrapt to have async detection which is basically perfect on Python 3.12+ and only fails for an obscure niche edge case on older Pythons.

(In fact, it might be possible to backport inspect.markcoroutinefunction using ctypes or a C extension to older Python versions, at least for CPython.)

[x42005e1f]

In aiologic 0.15.0, there is aiologic.synchronized(), which solves this issue. It can be used with any common type of lock, is well typed, and also relies on wrapt.

import asyncio
import threading
import time

from concurrent.futures import ThreadPoolExecutor

from aiologic import synchronized

UNITS: int = 10


def sync_main() -> None:
    @synchronized(threading.Lock())
    def sleep1() -> None:
        time.sleep(1 / UNITS)

    start = time.perf_counter()

    with ThreadPoolExecutor(UNITS) as executor:
        for _ in range(UNITS):
            executor.submit(sleep1)

    print("sync:", time.perf_counter() - start)  # ~one second


async def async_main() -> None:
    @synchronized(asyncio.Lock())
    async def sleep1() -> None:
        await asyncio.sleep(1 / UNITS)

    start = time.perf_counter()

    async with asyncio.TaskGroup() as tg:
        for _ in range(UNITS):
            tg.create_task(sleep1())

    print("async:", time.perf_counter() - start)  # ~one second


if __name__ == "__main__":
    sync_main()
    asyncio.run(async_main())

By default, aiologic.synchronized() uses aiologic.RLock, which can work with almost all popular concurrency libraries at the same time, with any number of threads, making the decorator as versatile as possible.

import time

from concurrent.futures import ThreadPoolExecutor
from functools import partial

import anyio

from aiologic import synchronized

SYNC_THREADS: int = 10
ASYNCIO_THREADS: int = 5
TRIO_THREADS: int = 5
ASYNC_TASKS: int = 20
ASYNC_THREADS: int = ASYNCIO_THREADS + TRIO_THREADS


class SynchronizedClass:
    @synchronized
    def sync_sleep5(self) -> None:
        time.sleep(5 / (SYNC_THREADS + ASYNC_THREADS*ASYNC_TASKS))

    @synchronized
    async def async_sleep5(self) -> None:
        await anyio.sleep(5 / (SYNC_THREADS + ASYNC_THREADS*ASYNC_TASKS))


def main() -> None:
    obj = SynchronizedClass()

    async def sleep5_with_tasks() -> None:
        async with anyio.create_task_group() as tg:
            for _ in range(ASYNC_TASKS):
                tg.start_soon(obj.async_sleep5)

    start = time.perf_counter()

    with ThreadPoolExecutor(SYNC_THREADS + ASYNC_THREADS) as executor:
        for _ in range(SYNC_THREADS):
            executor.submit(obj.sync_sleep5)

        for _ in range(ASYNCIO_THREADS):
            executor.submit(partial(
                anyio.run,
                sleep5_with_tasks,
                backend="asyncio",
            ))

        for _ in range(TRIO_THREADS):
            executor.submit(partial(
                anyio.run,
                sleep5_with_tasks,
                backend="trio",
            ))

    print(time.perf_counter() - start)  # ~five seconds


if __name__ == "__main__":
    main()

The documentation is still being written, but all features are already working as intended.

[x42005e1f]

I will also give my opinion on the separation of regular and asynchronous functions.

1. Using inspect.isawaitable() can lead to false positives. Suppose we pass asyncio.create_task(). The task object is awaitable, and instead of synchronizing the task creation, we will also synchronize its execution, and the return type changes: a coroutine object instead of a task object! It seems to me that it is easy to imagine user functions that schedule some action and return a future-like object for optional waiting for its completion. It would be slightly more accurate to use inspect.iscoroutine(), but it does not support generator-based coroutines (see types.coroutine()/asyncio.coroutine()) and user coroutines (implementing collections.abc.Coroutine).
2. Similar to inspect.iscoroutine(), inspect.iscoroutinefunction() is designed only for "native" coroutine functions. It will not detect anything that does not return an object that passes the inspect.iscoroutine() check, unless a fake coroutine function is marked with inspect.markcoroutinefunction(), but the latter violates the assumptions of type checkers. It also cannot detect unmarked instances of classes with async def __call__(), since such checking was rejected in the original PR as not directly relevant and was not proposed later. Furthermore, even support for functools.partialmethod objects was only added in Python 3.13, and no one who wants to support all non-EOL versions of Python will be able to rely on it for the next three years (since the end of support for Python 3.12 is October 2028).
3. inspect.markcoroutinefunction() is also far from ideal. At the moment, it is subject to inspect.iscoroutinefunction() does not detect marked partial objects python/cpython#142418. It is unknown when it will be fixed (as I have seen, some PRs may not be merged for several years, even when they are ready), and it is quite possible that the fix will not be backported to older versions of Python. Also, note that inspect.markcoroutinefunction() should not be used for callable objects that do not always return an object of types.CoroutineType, otherwise you will have type safety issues (a relatively common problem; please do not do this, otherwise inspect.iscoroutinefunction() loses its meaning!).

Because of this, I recently added iscoroutinelike(), iscoroutinefactory(), and markcoroutinefactory() to my library, which support all versions of Python starting with 3.8. iscoroutinelike() can detect both generator-based coroutines and user-defined coroutines. iscoroutinefactory() has all the improvements of inspect.iscoroutinefunction() and also detects classes with async def __call__() and much more complex cases (see the documentation). markcoroutinefactory() uses the standard asyncio marker for non-native coroutine factories on Python <3.12 (I may consider more precise version ranges, but for now I chose this version because django/asgiref relies on it), and the standard inspect marker for native coroutine factories on Python ≥3.12.

I think it would be more universal to also try to analyze the return type of the function in the annotations. I have already implemented the return type transformation in generator() and coroutine() (functions for transforming generator/coroutine factories into the corresponding object functions, in particular for the ability to implement __await__() via async def with zero cost), so it is quite possible that I will implement this as well.