Python asyncio : how to close a event loop after Task.cancel safely among threads?

Question

QUESTION

in my own project, i start a event loop in a thread, then cancel the task and close the loop in another thread safely. but i failed.

after reading task-object, i still cannot understand how to wait a task really cancelled after Task.cancel

Python Version:3.7.1

OS: windows

below is my debug process.

import threading
import asyncio
import time

async def visit_sth():
    print("start sleep")
    await asyncio.sleep(3)
    print("end sleep")


class Person(object):

    def __init__(self):
        self.loop = asyncio.new_event_loop()

    def visit(self):

        asyncio.set_event_loop(self.loop)
        try:
            self.loop.run_until_complete(visit_sth())
        except Exception as err:
            print(err)

    def pause(self):

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

        self.loop.stop()
        self.loop.close()

P = Person()
t1 = threading.Thread(target=P.visit)
t2 = threading.Thread(target=P.pause)

t1.start()
time.sleep(0.5)
t2.start()

t1.join()
t2.join()

errors below

start sleep
Exception in thread Thread-2:
Traceback (most recent call last):
  File "C:\Python3701\lib\threading.py", line 917, in _bootstrap_inner
    self.run()
  File "C:\Python3701\lib\threading.py", line 865, in run
    self._target(*self._args, **self._kwargs)
  File "c:\Users\zhouxin\Documents\jupyterlab\learning_asyncio\starkoverflow.py", line 31, in pause
    self.loop.close()
  File "C:\Python3701\lib\asyncio\selector_events.py", line 94, in close
    raise RuntimeError("Cannot close a running event loop")
RuntimeError: Cannot close a running event loop

after cancel, the event loop is still running.

also, the doc task-object said Task.cancel() does not guarantee that the Task will be cancelled

so i turn to stackoverflow, and read Kill tasks instead of awaiting them, and change pause method like

def pause(self):

    tasks = asyncio.all_tasks(loop=self.loop)
    for t in tasks:
        t.cancel()
        with suppress(asyncio.CancelledError):
            self.loop.run_until_complete(t)

    self.loop.stop()
    self.loop.close()

another error happended

start sleep
Exception in thread Thread-2:
Traceback (most recent call last):
  File "C:\Python3701\lib\threading.py", line 917, in _bootstrap_inner
    self.run()
  File "C:\Python3701\lib\threading.py", line 865, in run
    self._target(*self._args, **self._kwargs)
  File "c:\Users\zhouxin\Documents\jupyterlab\learning_asyncio\starkoverflow.py", line 31, in pause
    self.loop.run_until_complete(t)
  File "C:\Python3701\lib\asyncio\base_events.py", line 560, in run_until_complete
    self.run_forever()
  File "C:\Python3701\lib\asyncio\base_events.py", line 515, in run_forever
    raise RuntimeError('This event loop is already running')
RuntimeError: This event loop is already running

this way didnot work.

and now, i am really confused about how to wait a task cancelled then close the loop after Task.cancel.

score 0 · Answer 1 · answered Dec 18 '18 at 22:00

There are several issues with your code:

pause interacts with the event loop from outside the thread that runs the event loop. This is forbidden and must be replaced with calls to run_coroutine_threadsafe and call_soon_threadsafe.
The code creates a new event loop per business object. This is undesirable because the strong suite of asyncio is that it allows many coroutines inside a single event loop.

The recommended pattern is to have a single event loop, and submit tasks to it using run_coroutine_threadsafe. Instead of stopping the whole loop when you want to cancel a task, you just cancel that specific task. For example:

import threading
import asyncio
import time

async def visit_sth():
    print("start sleep")
    await asyncio.sleep(3)
    print("end sleep")


class Person:
    def visit(self):
        # returns a concurrent.futures.Future
        self._visit_fut = asyncio.run_coroutine_threadsafe(visit_sth(), loop)
        # result() waits for the future to be resolved and returns the
        # result
        try:
            return self._visit_fut.result()
        except Exception as err:
            print(type(err), err)

    def pause(self):
        self._visit_fut.cancel()

loop = asyncio.new_event_loop()
threading.Thread(target=loop.run_forever).start()

P = Person()
t1 = threading.Thread(target=P.visit)
t2 = threading.Thread(target=P.pause)

t1.start()
time.sleep(0.5)
t2.start()

t1.join()
t2.join()
loop.call_soon_threadsafe(loop.stop)

Python asyncio : how to close a event loop after Task.cancel safely among threads?

1 Answers1