asyncio: exception handling while waiting for a long task run by executor to finish

Question

I have the following code snippet being run by python 3.10.5:

import time
import asyncio

async def main():
    loop = asyncio.get_running_loop()
    loop.run_in_executor(None, blocking)
    print(f"{time.ctime()} Hello!")
    await asyncio.sleep(1.0)
    print(f"{time.ctime()} Goodbye!")

def blocking():
    time.sleep(5.0)
    print(f"{time.ctime()} Hello from thread!")


try:
    asyncio.run(main())
except KeyboardInterrupt:
    print("Cancelled.")

When I let it run it exits gracefully due to shutdown_default_executor() method added in python 3.9 which allows to solve the problem of task running in executor that outlasts the main event loop by wrapping this task in the coroutine. So I have the following output:

Sun Sep 11 19:04:25 2022 Hello!
Sun Sep 11 19:04:26 2022 Goodbye!
Sun Sep 11 19:04:30 2022 Hello from thread!

Next when I am pressing Ctrl-C after the first line of output, I am getting:

Sun Sep 11 19:04:42 2022 Hello!
^CSun Sep 11 19:04:47 2022 Hello from thread!
Cancelled.

So it is still able to handle the situation. But when I do it after Goodbye! line (when main coroutine is already finished and I am waiting for task in the executor to finish) I am getting:

Sun Sep 11 19:04:49 2022 Hello!
Sun Sep 11 19:04:50 2022 Goodbye!
^CCancelled.
Sun Sep 11 19:04:54 2022 Hello from thread!
exception calling callback for <Future at 0x7f58475183d0 state=finished returned NoneType>
Traceback (most recent call last):
  File "/usr/lib/python3.10/concurrent/futures/_base.py", line 330, in _invoke_callbacks
    callback(self)
  File "/usr/lib/python3.10/asyncio/futures.py", line 398, in _call_set_state
    dest_loop.call_soon_threadsafe(_set_state, destination, source)
  File "/usr/lib/python3.10/asyncio/base_events.py", line 795, in call_soon_threadsafe
    self._check_closed()
  File "/usr/lib/python3.10/asyncio/base_events.py", line 515, in _check_closed
    raise RuntimeError('Event loop is closed')
RuntimeError: Event loop is closed
Exception in thread Thread-1 (_do_shutdown):
Traceback (most recent call last):
  File "/usr/lib/python3.10/asyncio/base_events.py", line 576, in _do_shutdown
    self.call_soon_threadsafe(future.set_result, None)
  File "/usr/lib/python3.10/asyncio/base_events.py", line 795, in call_soon_threadsafe
    self._check_closed()
  File "/usr/lib/python3.10/asyncio/base_events.py", line 515, in _check_closed
    raise RuntimeError('Event loop is closed')
RuntimeError: Event loop is closed

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/usr/lib/python3.10/threading.py", line 1016, in _bootstrap_inner
    self.run()
  File "/usr/lib/python3.10/threading.py", line 953, in run
    self._target(*self._args, **self._kwargs)
  File "/usr/lib/python3.10/asyncio/base_events.py", line 578, in _do_shutdown
    self.call_soon_threadsafe(future.set_exception, ex)
  File "/usr/lib/python3.10/asyncio/base_events.py", line 795, in call_soon_threadsafe
    self._check_closed()
  File "/usr/lib/python3.10/asyncio/base_events.py", line 515, in _check_closed
    raise RuntimeError('Event loop is closed')
RuntimeError: Event loop is closed

The question is why am I getting runtime error here but manage to avoid it when I was hitting Ctrl-C after Hello! line (second case)? How do I handle this Runtime Error gracefully?

Answer 1

I think I've figured it out, but please, be cautious - I'm quite new to asyncio.

I base it on a python 3.10 asyncio.run code:

def run(main, *, debug=None):

    # More code here
    # ...

    try:
        events.set_event_loop(loop)
        if debug is not None:
            loop.set_debug(debug)
        return loop.run_until_complete(main)
    finally:
        try:
            _cancel_all_tasks(loop)
            loop.run_until_complete(loop.shutdown_asyncgens())
            loop.run_until_complete(loop.shutdown_default_executor())
        finally:
            events.set_event_loop(None)
            loop.close()

The behaviour you describe is explained by those nested try-finally blocks.

If KeyboardInterrupt happens during loop.run_until_complete(main) - in your case before Goodbye! - the inner try - finally is executed which properly handles blocking with loop.shutdown_default_executor().

If, on the other hand, the exception happens after Goodbye!, the code that is currently executing is loop.shutdown_default_executor() and since further clean up closes the loop without awaiting anything, the Future containing blocking throws RuntimeError('Event loop is closed').

The Future still seems to be awaited after that... unless you hit Ctrl-C again. Then the

Exception ignored in: <module 'threading' from '/usr/lib/python3.10/threading.py'>
Exception in thread Thread-1 (_do_shutdown):
Traceback (most recent call last):
Traceback (most recent call last):
  File "/usr/lib/python3.10/threading.py", line 1537, in _shutdown
  File "/usr/lib/python3.10/asyncio/base_events.py", line 576, in _do_shutdown
    self.call_soon_threadsafe(future.set_result, None)
  File "/usr/lib/python3.10/asyncio/base_events.py", line 795, in call_soon_threadsafe
    atexit_call()
  File "/usr/lib/python3.10/concurrent/futures/thread.py", line 31, in _python_exit
    self._check_closed()
  File "/usr/lib/python3.10/asyncio/base_events.py", line 515, in _check_closed
    t.join()
  File "/usr/lib/python3.10/threading.py", line 1096, in join
    self._wait_for_tstate_lock()
    raise RuntimeError('Event loop is closed')
  File "/usr/lib/python3.10/threading.py", line 1116, in _wait_for_tstate_lock
RuntimeError: Event loop is closed

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/usr/lib/python3.10/threading.py", line 1016, in _bootstrap_inner
    if lock.acquire(block, timeout):
KeyboardInterrupt:

exception is thrown.

So it would seem there is one more layer - this time at the thread level - waiting for the executor :D

Anyway, I think this is expected - we should allow user to kill the program without awaiting (which could take forever).

Back to your question - how to handle it gracefully. I don't think you can. The error is thrown inside Future logic and what we see on the stderr is just a log of an unconsumed result. But just because we can't handle it doesn't mean user have to see it.

I don't know if it's a good practice, but you could redirect stderr to null device. Or maybe figure out which logger does that an patch him? Code that worked for me:

from contextlib import redirect_stderr
from os import devnull

fnull = open(devnull, 'w')
r = redirect_stderr(fnull)
r.__enter__()