QUERY · ISSUE

asyncio: Calling .run() within a task does not behave as per CPython

openby peterhinchopened 2024-06-02updated 2026-03-25

bugextmod

Port, board and/or hardware

Unix build

MicroPython version

MicroPython v1.23.0-preview.72.g4a2e510a8.dirty on 2024-02-04; linux [GCC 11.4.0] version

Reproduction

import asyncio

async def foo(s):
    while True:
        print(f"Task {s}")
        await asyncio.sleep(1)

async def main():
    asyncio.create_task(foo(1))
    await asyncio.sleep(2)
    asyncio.run(foo(2))  # CPython throws a RuntimeError
    await asyncio.sleep(2)
    asyncio.create_task(foo(1))
    await asyncio.sleep(5)
    print("Done")  # never happens in MP

asyncio.run(main())

Expected behaviour

On CPython issuing run() in a running task throws a RuntimeError:

Python 3.10.12 (main, Nov 20 2023, 15:14:05) [GCC 11.4.0] on linux
Type "help", "copyright", "credits" or "license" for more information.
>>> import rats17
Task 1
Task 1
Task 1
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/home/adminpete/rats17.py", line 17, in <module>
    asyncio.run(main())
  File "/usr/lib/python3.10/asyncio/runners.py", line 44, in run
    return loop.run_until_complete(main)
  File "/usr/lib/python3.10/asyncio/base_events.py", line 649, in run_until_complete
    return future.result()
  File "/home/adminpete/rats17.py", line 11, in main
    asyncio.run(foo(2))  # CPython throws a RuntimeError
  File "/usr/lib/python3.10/asyncio/runners.py", line 33, in run
    raise RuntimeError(
RuntimeError: asyncio.run() cannot be called from a running event loop
>>>

Observed behaviour

The run() command appears to work, starting the second foo instance. However main never terminates.

MicroPython v1.23.0-preview.72.g4a2e510a8.dirty on 2024-02-04; linux [GCC 11.4.0] version
Use Ctrl-D to exit, Ctrl-E for paste mode
>>> import rats17
Task 1
Task 1
Task 1
Task 2
Task 1
Task 2
Task 1
Task 2
Task 1
Task 2
... continues forever

Additional Information

This arose from this issue.

A possible fix to asyncio.core.py is

def run(coro):
    if cur_task is None:
        return run_until_complete(create_task(coro))
    else:
        raise RuntimeError("asyncio.run() cannot be called from a running event loop")

Code of Conduct

Yes, I agree

CANDIDATE · ISSUE

uasyncio: current_task behaviour differs from CPython

closedby peterhinchopened 2023-05-17updated 2024-02-28

bug

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