asyncio try/except/finally handling skipped if `asyncio.run()` catches KeyboardInterrupt - asyncio.start_server() keeps running
Port, board and/or hardware
probably all, at least posix and esp32
MicroPython version
MicroPython v1.24.0-preview.276.g1897fe622 on 2024-09-02; linux [GCC 11.4.0] version
also tried with
v1.23.0-preview.57.ge111793d8 on 2024-08-30; linux [GCC 11.4.0] version
Reproduction
demo.py:
import asyncio
async def async_func():
try:
while True:
print("step")
await asyncio.sleep(1)
except asyncio.CancelledError:
print("async_foo:CancelledError")
except KeyboardInterrupt:
print("async_foo:KeyboardInterrupt")
finally:
print("async_foo:finally")
try:
asyncio.run(async_func())
except KeyboardInterrupt:
print("run:KeyboardInterrupt")
finally:
print("run:finally")
Running the provided snippet on either micropython built for Linux or on an ESP32 via mpremote and aborting the script using CTRL-C terminates the program, but the finally block inside async_func() won't be called:
$ path/to/micropython demo.py
step
...
step
^Crun:KeyboardInterrupt
run:finally
Expected behaviour
CPython executing the same snippet gives
$ python demo.py
step
step
^Casync_foo:CancelledError
async_foo:finally
run:finally
indicating that the finally block is being executed.
Observed behaviour
It looks like MicroPythons asyncio.run() catches KeyboardInterrupt and instantly kills the event loop, skipping all code that might handle exceptions or doing unconditional stuff in finally.
While this might be desired behavior on IoT devices in general, the task created with asyncio.start_server wont be terminated, resulting in OSError: [Errno 112] EADDRINUSE error, if you started an asynchronous Micropython web server, terminated it, and tried to restart it.
See https://stackoverflow.com/questions/78886486/in-micropython-how-can-i-guarantee-asynchronous-tasks-being-shut-down
Additional Information
StackOverflow: In Micropython, how can I guarantee asynchronous tasks being shut down?](https://stackoverflow.com/questions/78886486/in-micropython-how-can-i-guarantee-asynchronous-tasks-being-shut-down)
Code of Conduct
Yes, I agree
uasyncio: current_task behaviour differs from CPython
Background: I am looking for a way detect whether a function has been called in a uasyncio context. The asyncio_running function works in CPython, but in MP it fails if a uasyncio script has previously run. In the absence of a fix, a non-hacky workround would be appreciated :)
Repro:
try:
import uasyncio as asyncio
except:
import asyncio
def asyncio_running():
try:
_ = asyncio.current_task()
except:
return False
return True
def prun(txt):
print(f"{txt}: asyncio {'' if asyncio_running() else 'not'} running")
prun("Start")
async def foo():
prun("In task")
await asyncio.sleep(1)
try:
asyncio.run(foo())
finally:
_ = asyncio.new_event_loop()
prun("At end")
On CPython 3.8.10:
>>> import rats25
Start: asyncio not running
In task: asyncio running
At end: asyncio not running
On MP 1.20 (tested on Unix and RP2):
>>> import rats25
Start: asyncio not running
In task: asyncio running
At end: asyncio running