[FIXED] 'Task was destroyed but it is pending' when running multiple tasks inside one asyncio loop

Issue

Inside a microservice I am running two separate asyncio tasks; one for incoming grpc calls and one for reading a redis queue. The main looks kinda like this:

if __name__ == "__main__":

    try:
        asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
        loop = asyncio.get_event_loop()
        loop.create_task(mainRedisLoop(redis_uri, consumer_name)) # create redis loop task

        loop.create_task(runGrpcServer()) # create grpc loop task

        loop.run_forever() # start event loop
    except Exception as exception:
        logger.error(exception, exc_info=True)
        sys.exit(-1)

Then what happens is that when a specific grpc call comes, the mainRedisLoop() task is somehow destroyed and throws the error:

ERROR  [asyncio:135] - Task was destroyed but it is pending!
task: <Task pending name='Task-1' coro=<mainRedisLoop() running at /path/to/app.py:177> wait_for=<Future pending cb=[<TaskWakeupMethWrapper object at 0x7f9762423190>()]>>
ERROR  [asyncio:135] - Task was destroyed but it is pending!
task: <Task pending name='Task-4' coro=<RedisConnection._read_data() running at /path/to/venv/lib/python3.8/site-packages/aioredis/connection.py:186> wait_for=<Future pending cb=[<TaskWakeupMethWrapper object at 0x7f976369da90>()]> cb=[RedisConnection.__init__.<locals>.<lambda>() at /path/to/venv/lib/python3.8/site-packages/aioredis/connection.py:168]>

Could it be some of the return statements in the logic of processing grpc requests from my side are causing the task to suddenly close?

Here is also the mainRedisLoop function:

async def mainRedisLoop(redis_uri, consumer_name):
    logger.info('begin mainRedisLoop function for redis')
    redis = await aioredis.create_redis(redis_uri, loop=loop)
    stream = 'some-stream-name'
    group = 'some-group-name'
    exists = await redis.exists(stream)
    if not exists:
        logger.info(
            f'stream {stream} does not exist. Trying to recreate.')
    try:
        await redis.xgroup_create(stream, group, latest_id='$', mkstream=True)
    except aioredis.errors.ReplyError as e:
        logger.info(f"Consumer group {group} already exists. {e}")
    await process_message(redis, loop, group, consumer_name, [stream]) # <<Error happens in the message processing
    redis.close()
    await redis.wait_closed()

I have researched the other questions in Stack regarding 'Task was destroyed but it is pending' and they mostly explain how to gracefully close tasks before ending the loop. My case is that I do not want the loop/tasks to end, rather continue in parallel with the grpc server expecting calls from other microservices or reading redis queues.

Solution

From the documentation for asyncio.create_task()

Important: Save a reference to the result of this function, to avoid a task disappearing mid-execution. The event loop only keeps weak references to tasks. A task that isn’t referenced elsewhere may get garbage collected at any time, even before it’s done. For reliable “fire-and-forget” background tasks, gather them in a collection:

background_tasks = set()

for i in range(10):
    task = asyncio.create_task(some_coro(param=i))

    # Add task to the set. This creates a strong reference.
    background_tasks.add(task)

    # To prevent keeping references to finished tasks forever,
    # make each task remove its own reference from the set after
    # completion:
    task.add_done_callback(background_tasks.discard)

Besides that: From my experience you should also consider calling process_message as a task instead of a coroutine depending on how long it takes. I personally had the problem that a running task was destroyed even when keeping a reference in a set. Inside the task was a long running coroutine. Running this coroutine as a task fixed it.

Answered By - Nik Schuetz