cancelled asyncio tasks result in "Task was destroyed but it is pending"

Question

I am trying to write a job scheduler that can efficiently schedule M jobs across N cores. As soon as one job completes, a new one should be immediately started. Also, we should support timeout so that no task takes longer than a certain amount of time. This is what I came up with the for the main loop:

import asyncio
import sys

max_concurrency = 4

async def _sleep_asynchronously(time):
    index, seconds = time
    await asyncio.sleep(seconds)
    return (index, seconds)

def select_invocations(waiting, num_running):
    count = max_concurrency - num_running
    selected = waiting[:count]
    waiting = waiting[count:]
    return selected, waiting

async def _run_everything_asynchronously():
    tasks = []
    timeouts = [ 4, 3, 1, 2, 0.5, 7, 0.25, 3, 2, 1, 4.5, 5]
    timeouts = list(enumerate(timeouts))

    pending, waiting = select_invocations(tasks, 0)
    running = {_sleep_asynchronously(timeout) for timeout in timeouts}

    while len(running) > 0:
        try:
            done, running = await asyncio.wait(running, timeout=0.5, return_when=asyncio.FIRST_COMPLETED)
            if not done:
                for r in running:
                    r.cancel()
                    await r
            else:
                for d in done:
                    index, timeout = await d
                    print("Index {} finished waiting for {} seconds".format(index, timeout))

        except asyncio.CancelledError as e:
            running.clear()

        if len(waiting) > 0:
            pending, waiting = select_invocations(tasks, len(running))
            running = {_sleep_asynchronously(timeout) for timeout in timeouts}

if 'win32' in sys.platform:
    asyncio.set_event_loop_policy(asyncio.WindowsProactorEventLoopPolicy())

loop = asyncio.get_event_loop()
rc = loop.run_until_complete(_run_everything_asynchronously())
loop.close()

sys.exit(0)

If I run this, this is my output:

Index 6 finished waiting for 0.25 seconds
Index 4 finished waiting for 0.5 seconds
Index 9 finished waiting for 1 seconds
Index 2 finished waiting for 1 seconds
Task was destroyed but it is pending!
task: <Task pending coro=<_sleep_asynchronously() done, defined at D:\src\asynciotest.py:6> wait_for=<Future pending cb=[<TaskWakeupMethWrapper object at 0x000001E74D13BDF8>()]>>
Task was destroyed but it is pending!
task: <Task pending coro=<_sleep_asynchronously() done, defined at D:\src\asynciotest.py:6> wait_for=<Future pending cb=[<TaskWakeupMethWrapper object at 0x000001E74D352438>()]>>
Task was destroyed but it is pending!
task: <Task pending coro=<_sleep_asynchronously() done, defined at D:\src\asynciotest.py:6> wait_for=<Future pending cb=[<TaskWakeupMethWrapper object at 0x000001E74D37F9D8>()]>>
Task was destroyed but it is pending!
task: <Task pending coro=<_sleep_asynchronously() done, defined at D:\src\asynciotest.py:6> wait_for=<Future pending cb=[<TaskWakeupMethWrapper object at 0x000001E74D37FBE8>()]>>
Task was destroyed but it is pending!
task: <Task pending coro=<_sleep_asynchronously() done, defined at D:\src\asynciotest.py:6> wait_for=<Future pending cb=[<TaskWakeupMethWrapper object at 0x000001E74D3A15B8>()]>>
Task was destroyed but it is pending!
task: <Task pending coro=<_sleep_asynchronously() done, defined at D:\src\asynciotest.py:6> wait_for=<Future pending cb=[<TaskWakeupMethWrapper object at 0x000001E74D3E7498>()]>>
Task was destroyed but it is pending!
task: <Task pending coro=<_sleep_asynchronously() done, defined at D:\src\asynciotest.py:6> wait_for=<Future pending cb=[<TaskWakeupMethWrapper object at 0x000001E74D3E74C8>()]>>

What am I missing here? How do I properly clean up the tasks which were cancelled?

Answer 1

When you cancel a task and await on it, unless caught, it will throw a CancelledError.

Your current code takes the first CancelledError and clears the running list, without cancelling the rest:

except asyncio.CancelledError as e:
    running.clear()

Instead of awaiting, just cancel all of them:

for r in running:
    r.cancel()