Асинхронный потокобезопасный семафор Python
Я ищу поточно-ориентированную реализацию семафора, которую можно использовать в Python.
Стандартные библиотеки asyncio.Semaphore не являются потокобезопасными.
Стандартные библиотеки threading.Semaphore не имеют awaitable интерфейса.
Я использую sanic, который имеет несколько потоков (работников), а также асинхронный цикл в каждом потоке. Я хочу иметь возможность возвращать выполнение циклу событий на каждом из рабочих процессов всякий раз, когда он сталкивается с заблокированным семафором, пока он ожидает.
ОБНОВЛЕНИЕ: я имел в виду здесь процесс, а не потоки. Итак, должно быть так, что Sanic разделяется на процессы и multiprocessing.Semaphore. Я считаю, что данный ответ по-прежнему актуален для того, где я могу применить аналогичное решение.
Ответы 1
Если вы говорите, что вам нужен один экземпляр семафора, который будет использоваться во всех потоках, создайте экземпляр threading.Semaphore и асинхронную функцию acquire_semaphore, например:
import asyncio
import threading
from concurrent.futures import ThreadPoolExecutor
executor = ThreadPoolExecutor()
semaphore = threading.Semaphore()
async def acquire_semaphore():
"""Make a threading.Semaphore awaitable."""
def acquirer():
semaphore.acquire()
loop = asyncio.get_running_loop()
await loop.run_in_executor(executor, acquirer)
async def test():
"""Acquire and release a semaphore sharable across threads."""
await acquire_semaphore()
print('acquired')
await asyncio.sleep(1)
semaphore.release() # This never blocks
print('released')
def worker():
asyncio.run(test())
def main():
threads = [
threading.Thread(target=worker)
for _ in range(3)
]
for thread in threads:
thread.start()
for thread in threads:
thread.join()
if __name__ == '__main__':
main()
Распечатки:
acquired
released
acquired
released
acquired
released