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

Question

Сразу предупреждаю, я новичек в 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

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

UPD: Python 3.7, asyncio 3.4.3

UPD: Разобрался Разобрался!

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)

Answer 1

semaphore = asyncio.Semaphore(50)

async def work(value):
    async with semaphore:
        print(value)
        await asyncio.sleep(10)

async def work_timed(i):
    task = work(i)
    try:
        await asyncio.wait_for(task, timeout=3)
    except futures.TimeoutError:
        task.cancel()
    return await task


async def main():
    tasks = []
    for i in range(0, 10000):
        tasks.append(work_timed(i))

    while tasks:
        done, tasks = await asyncio.wait(tasks, return_when=FIRST_EXCEPTION)
        for task in done:
            try:
                result = await task
            except asyncio.CancelledError:
                print('timeout')


loop = asyncio.get_event_loop()
future = asyncio.ensure_future(main())
loop.run_until_complete(future)

Я бы решил примерно так. Добавил обработку таймаута внутрь пачки тасков, потому что wait_for не завершает таск.

БЛОГ НА HUSL

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

Ответы (1 шт):