DjangoBB LoFi version

Найти - Пользователи

Полная версия: Как правильно ловить concurrent.futures._base.TimeoutError при использовании asyncio.wait_for и asyncio.Semaphore?

Начало » Python для экспертов » Как правильно ловить concurrent.futures._base.TimeoutError при использовании asyncio.wait_for и asyncio.Semaphore?

stratum

Май 20, 2020 11:15:47

Внимание! Я дублирую свой вопрос со stackoverflow, потому как ответа так и не дождался.

Сразу предупреждаю, я новичек в asyncio, хоть и представляю как работает сие чудо, но очень слабо представляю, что в этой библиотеке под капотом.

Вот мой код:

 import asyncio
semaphore = asyncio.Semaphore(50)
async def work(value):
    async with semaphore:
        print(value)
        await asyncio.sleep(10)
async def main():
    tasks = []
    for i in range(0, 10000):
        tasks.append(asyncio.wait_for(work(i), timeout=3))
    await asyncio.gather(*tasks)
loop = asyncio.get_event_loop()
future = asyncio.ensure_future(main())
loop.run_until_complete(future)

Мне необходимо, что-бы корутина work выполнялась не более 3 секунд, и не более 50-и штук в одно время. По истечению 3-ех секунд, выполнение корутины завершалось, и новые 50 тасков поступали в работу. Но в моем случае, через 3 секунды вылетает:

 Traceback (most recent call last):
  File "C:/Users/root/PycharmProjects/LogParser/ssh/async/asyn_test.py", line 19, in <module>
    loop.run_until_complete(future)
  File "C:\Code\Python3\lib\asyncio\base_events.py", line 579, in run_until_complete
    return future.result()
  File "C:/Users/root/PycharmProjects/LogParser/ssh/async/asyn_test.py", line 15, in main
    await asyncio.gather(*tasks)
  File "C:\Code\Python3\lib\asyncio\tasks.py", line 449, in wait_for
    raise futures.TimeoutError()
concurrent.futures._base.TimeoutError

И как-бы я не пробовал ловить это исключение, сколько-бы еще не оставалось тасков, работа завершается. А мне необходимо - что бы работа продолжалась Подскажите, как мне правильно это реализовать?

Python 3.7, asyncio 3.4.3

stratum

Май 20, 2020 14:29:32

Разобрался!

 import asyncio
semaphore = asyncio.Semaphore(50)
async def work(value):
    print(value)
    await asyncio.sleep(10)
async def work_with_timeout(value):
    async with semaphore:
        try:
            return await asyncio.wait_for(work(value), timeout=3)
        except asyncio.TimeoutError:
            return None
async def main():
    tasks = []
    for i in range(0, 10000):
        tasks.append(work_with_timeout(i))
    await asyncio.gather(*tasks)