Asyncio threadsafe primitives

Question

I need Threadsafe primitives (locking, conditions Semaphore), do they exist in the asyncio ecosystem?

I created some code myself, but it feels a bit sluggish:

import asyncio
from threading import get_ident,Lock,Event

class AsyncThreadsafeEvent():
    def __init__(self,*args,**kwargs):
        super(AsyncThreadsafeEvent, self).__init__()
        self.waiters = {}
        self.event = Event()

    def set(self):
        self.event.set()
        events = self.waiters.values()
        for loop,event in events:
            loop.call_soon_threadsafe(event.set)


    async def wait(self):
        event = asyncio.Event()
        #not strictly necessary but could provide tiny speedup
        if self.event.is_set():
            return

        self.waiters[get_ident()] = (asyncio.get_event_loop(),event)

        #to ensure thread safty
        if self.event.is_set():
            return

        await event.wait()

Any help would be appreciated!

Answer 1

They do exist. You can use aiologic.Lock, aiologic.Semaphore, aiologic.Condition, and many others (I'm the creator of aiologic). All the primitives that aiologic provides are fully thread-safe. And of course, they are both async-aware and thread-aware: they can be used in both asynchronous and synchronous environments.

import asyncio

from aiologic import Condition

cond = Condition()


def work():
    with cond:
        print("thread")

        cond.notify()
        cond.wait()

        print("thread")


async def main():
    async with asyncio.TaskGroup() as tasks, cond:
        tasks.create_task(asyncio.to_thread(work))

        print("asyncio")

        await cond
        cond.notify()

        print("asyncio")


asyncio.run(main())

However, due to their dual nature, they have some differences in interface and semantics from those primitives you can find in the threading module or in the asyncio module. So feel free to create new discussions on GitHub. I'll be happy to answer your questions.