Browse Source

extmod/uasyncio: Add Loop.new_event_loop method.

This commit adds Loop.new_event_loop() which is used to reset the singleton
event loop.  This functionality is put here instead of in Loop.close() to
make it possible to write code that is compatible with CPython.
pull/5890/head
Damien George 5 years ago
parent
commit
db137e70dc
  1. 7
      docs/library/uasyncio.rst
  2. 20
      extmod/uasyncio/core.py
  3. 36
      tests/extmod/uasyncio_new_event_loop.py
  4. 6
      tests/extmod/uasyncio_new_event_loop.py.exp

7
docs/library/uasyncio.rst

@ -244,6 +244,13 @@ Event Loop
Return the event loop used to schedule and run tasks. See `Loop`.
.. function:: new_event_loop()
Reset the event loop and return it.
Note: since MicroPython only has a single event loop this function just
resets the loop's state, it does not create a new one.
.. class:: Loop()
This represents the object which schedules and runs tasks. It cannot be

20
extmod/uasyncio/core.py

@ -132,13 +132,6 @@ class IOQueue:
################################################################################
# Main run loop
# TaskQueue of Task instances
_task_queue = TaskQueue()
# Task queue and poller for stream IO
_io_queue = IOQueue()
# Ensure the awaitable is a task
def _promote_to_task(aw):
return aw if isinstance(aw, Task) else create_task(aw)
@ -269,3 +262,16 @@ class Loop:
# The runq_len and waitq_len arguments are for legacy uasyncio compatibility
def get_event_loop(runq_len=0, waitq_len=0):
return Loop
def new_event_loop():
global _task_queue, _io_queue
# TaskQueue of Task instances
_task_queue = TaskQueue()
# Task queue and poller for stream IO
_io_queue = IOQueue()
return Loop
# Initialise default event loop
new_event_loop()

36
tests/extmod/uasyncio_new_event_loop.py

@ -0,0 +1,36 @@
# Test Loop.new_event_loop()
try:
import uasyncio as asyncio
except ImportError:
try:
import asyncio
except ImportError:
print("SKIP")
raise SystemExit
async def task():
for i in range(4):
print("task", i)
await asyncio.sleep(0)
await asyncio.sleep(0)
async def main():
print("start")
loop.create_task(task())
await asyncio.sleep(0)
print("stop")
loop.stop()
# Use default event loop to run some tasks
loop = asyncio.get_event_loop()
loop.create_task(main())
loop.run_forever()
# Create new event loop, old one should not keep running
loop = asyncio.new_event_loop()
loop.create_task(main())
loop.run_forever()

6
tests/extmod/uasyncio_new_event_loop.py.exp

@ -0,0 +1,6 @@
start
task 0
stop
start
task 0
stop
Loading…
Cancel
Save