Unable to catch RuntimeError raised due to closing asyncio event loop

Question

I am writing a python script that uses signal_handler to capture SIGINT, SIGTSTP signals. The signal handler closes the event loop, which raises a RuntimeError. I'm trying to catch this error and display something else instead. My code is as follows:

async def signal_handler(self):
    try : 
        #perform some cleanup by executing some coroutines
        self.loop.stop()
    except RuntimeError as err:
        print('SIGINT or SIGTSTP raised')
        print("cleaning and exiting")
        sys.exit(0)

The output is as follows:

^CTraceback (most recent call last):
  File "pipeline.py", line 69, in <module>
    pipeline(sys.argv[1], (lambda : sys.argv[2] if len(sys.argv)==3 else os.getcwd())())
  File "pipeline.py", line 9, in __init__
    self.main(link)
  File "pipeline.py", line 52, in main
    asyncio.run(video.get_file(obj.metadata['name']+"_v."+obj.default_video_stream[1], obj.default_video_stream[0]))
  File "/usr/lib/python3.8/asyncio/runners.py", line 44, in run
    return loop.run_until_complete(main)
  File "/usr/lib/python3.8/asyncio/base_events.py", line 614, in run_until_complete
    raise RuntimeError('Event loop stopped before Future completed.')
RuntimeError: Event loop stopped before Future completed.

From the output I can infer that

self.loop.stop()

raised the RuntimeError.

How can I solve this issue?

Answer 1

MRE

Let's start with a Minimal Reproducible Example so that we know we are on the same page:

import asyncio
import signal
import sys


loop = asyncio.new_event_loop()

async def main():
    while True:
        print('Hey')
        await asyncio.sleep(0.5)

async def _signal_handler():
    try:
        loop.stop()
    except RuntimeError as err:
        print('SIGINT or SIGTSTP raised')
        print("cleaning and exiting")
        sys.exit(1)

def signal_handler(*args):
    loop.create_task(_signal_handler())

signal.signal(signal.SIGINT, signal_handler)

loop.run_until_complete(main())

This will print the following when SIGINT is received:

Hey
Hey
Hey
^CTraceback (most recent call last):
  File "73094030.py", line 26, in <module>
    loop.run_until_complete(main())
  File "/usr/lib/python3.8/asyncio/base_events.py", line 614, in run_until_complete
    raise RuntimeError('Event loop stopped before Future completed.')
RuntimeError: Event loop stopped before Future completed.

Problem

The error will be raised in main(). It happens because the loop is forced to exit before asyncio.sleep() task is finished.

Solution

To solve this, we should cancel the task before exiting. Let's replace

loop.stop()

with

tasks = asyncio.all_tasks(loop)
for task in tasks:
    task.cancel()

This still raises an exception:

Hey
Hey
Hey
^CTraceback (most recent call last):
  File "73094030.py", line 28, in <module>
    loop.run_until_complete(main())
  File "/usr/lib/python3.8/asyncio/base_events.py", line 616, in run_until_complete
    return future.result()
asyncio.exceptions.CancelledError

But we have proceeded to change RuntimeError to CancelledError which is more descriptive. It also allows the running functions to run their finally blocks, freeing resources. The event loop will stop automatically after all the tasks finish.

Of course we now except to get a CancelledException. So let's add a try/except block to main():

async def main():
    try:
        while True:
            print('Hey')
            await asyncio.sleep(0.5)
    except asyncio.CancelledError:
        print('\nFunction stopped manually.')

Now we get no clutter:

Hey
Hey
Hey
^C
Function stopped manually.

The final code looks like this:

import asyncio
import signal
import sys


loop = asyncio.new_event_loop()

async def main():
    try:
        while True:
            print('Hey')
            await asyncio.sleep(0.5)
    except asyncio.CancelledError:
        print('\nFunction stopped manually.')


async def _signal_handler():
    try:
        tasks = asyncio.all_tasks(loop)
        for task in tasks:
            task.cancel()
    except RuntimeError as err:
        print('SIGINT or SIGTSTP raised')
        print("cleaning and exiting")
        sys.exit(1)

def signal_handler(*args):
    loop.create_task(_signal_handler())

signal.signal(signal.SIGINT, signal_handler)

loop.run_until_complete(main())