时间戳

首页 > 解决方案 > 即使在循环停止后，run_until_complete 也会失败

python - 即使在循环停止后，run_until_complete 也会失败

问题描述

我正在尝试编写一个 SIGTERM 处理程序，它将具有我的run_forever() -loop

这是我写的一个学习演示：

import asyncio
import signal
import logging

logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s [%(name)s]: %(message)s', datefmt='%H:%M:%S')
_log = logging.getLogger(__name__)


class Looper:
    def __init__(self, loop):
        self._loop = loop
        self._shutdown = False
        signal.signal(signal.SIGINT, self._exit)
        signal.signal(signal.SIGTERM, self._exit)

    def _exit(self, sig, frame):
        name = signal.Signals(sig).name
        _log.info(f"Received shutdown-signal: {sig} ({name})")
        self._shutdown = True
        self._loop.stop() # << Stopping the event loop here.
        _log.info(f"Loop stop initiated.")
        pending = asyncio.all_tasks(loop=self._loop)
        _log.info(f"Collected {len(pending)} tasks that have been stopped.")
        if pending:
            _log.info("Attempting to gather pending tasks: " + str(pending))
            gatherer_set = asyncio.gather(*pending, loop=self._loop)
            # self._loop.run_until_complete(gatherer_set) # << "RuntimeError: This event loop is already running"
        _log.info("Shutting down for good.")

    async def thumper(self, id, t):
        print(f"{id}: Winding up...")
        while not self._shutdown:
            await asyncio.sleep(t)
            print(f'{id}: Thump!')
        print(f'{id}: Thud.')


loop = asyncio.get_event_loop()
lp = Looper(loop)
loop.create_task(lp.thumper('North Hall', 2))
loop.create_task(lp.thumper('South Hall', 3))
loop.run_forever()
_log.info("Done.")

在 Windows 10 和 Debian 10 上，上面的脚本都会对 SIGINT 做出反应并产生输出

North Hall: Winding up...
South Hall: Winding up...
North Hall: Thump!
South Hall: Thump!
North Hall: Thump!
South Hall: Thump!
North Hall: Thump!
09:55:53 INFO [__main__]: Received shutdown-signal: 2 (SIGINT)
09:55:53 INFO [__main__]: Loop stop initiated.
09:55:53 INFO [__main__]: Collected 2 tasks that have been stopped.
09:55:53 INFO [__main__]: Attempting to gather pending tasks: {<Task pending coro=<Looper.thumper() running at amazing_grace.py:42> wait_for=<Future pending cb=[<TaskWakeupMethWrapper object at 0x02F91BF0>()]>>, <Task pending coro=<Looper.thumper() running at amazing_grace.py:42> wait_for=<Future pending cb=[<TaskWakeupMethWrapper object at 0x02F91C10>()]>>}
09:55:53 INFO [__main__]: Shutting down for good.
09:55:53 INFO [__main__]: Done.

遗憾的是，表示thumper(..)演示调用实际上已经结束的“Thud.”行不会显示。我想，这是因为“聚集”只是给我带来了一组未实现的未来。但是，如果我敢激活run_until_complete() - 行，即使它位于self._loop.stop()之后，输出结束如下：

[...]
10:24:25 INFO [__main__]: Collected 2 tasks that have been stopped.
10:24:25 INFO [__main__]: Attempting to gather pending tasks: {<Task pending coro=<Looper.thumper() running at amazing_grace.py:41> wait_for=<Future pending cb=[<TaskWakeupMethWrapper object at 0x03E417D0>()]>>, <Task pending coro=<Looper.thumper() running at amazing_grace.py:41> wait_for=<Future pending cb=[<TaskWakeupMethWrapper object at 0x03E41BF0>()]>>}
Traceback (most recent call last):
  File "amazing_grace.py", line 50, in <module>
    loop.run_forever()
  File "C:\Python37\lib\asyncio\base_events.py", line 539, in run_forever
    self._run_once()
  File "C:\Python37\lib\asyncio\base_events.py", line 1739, in _run_once
    event_list = self._selector.select(timeout)
  File "C:\Python37\lib\selectors.py", line 323, in select
    r, w, _ = self._select(self._readers, self._writers, [], timeout)
  File "C:\Python37\lib\selectors.py", line 314, in _select
    r, w, x = select.select(r, w, w, timeout)
  File "amazing_grace.py", line 35, in _exit
    self._loop.run_until_complete(gatherer_set) # << "This event loop is already running"
  File "C:\Python37\lib\asyncio\base_events.py", line 571, in run_until_complete
    self.run_forever()
  File "C:\Python37\lib\asyncio\base_events.py", line 526, in run_forever
    raise RuntimeError('This event loop is already running')
RuntimeError: This event loop is already running

问题归结为

该程序应在Windows 10和Linux下的Python 3.7上运行。

几天后编辑

正如 zaquest 在他/她的回答中所说的那样，当只是分配一个信号处理程序并create_task在其中添加一个调用时，就会自找麻烦。正如我所观察到的，该例程可能会运行也可能不会运行（即使没有其他任务）。所以现在我添加了一个sys.platform检查，看看脚本是否在 UNIX () 下运行。如果是这样，我更喜欢更可靠loop.add_signal_handler地定义回调函数，这是我真正需要的。幸运的是 UNIX 是我的主要用例。主线：

self._loop.add_signal_handler(signal.signal(signal.SIGINT, self._exit, signal.SIGINT, None)

为什么要检查平台？：按照文档https://docs.python.org/3/library/asyncio-eventloop.html#unix-signals，loop.add_signal_handler() 在 Windows 上不可用，这不是真的惊讶地认为有问题的信号是 UNIX 术语。

标签： pythonpython-3.7python-asyncio

解决方案

Python 信号处理程序在主线程中执行，在循环运行的同一线程中。BaseEventLoop.stop()方法不会立即停止循环，而只是设置一个标志，这样当你的循环下次运行时，它只会执行已经安排好的回调，并且不再安排任何回调（参见run_forever）。但是，在您的信号处理程序返回之前，无法运行循环。这意味着您不能等到循环在信号处理程序中停止。相反，您可以安排另一个任务，该任务将等待您长时间运行的任务对更改做出反应，self._shutdown然后停止循环。

class Looper:
    ...

    def _exit(self, sig, frame):
        name = signal.Signals(sig).name
        _log.info("Received shutdown-signal: %s (%s)", sig, name)
        self._shutdown = True

        pending = asyncio.all_tasks(loop=self._loop)
        _log.info("Attempting to gather pending tasks: " + str(pending))
        if pending:
            self._loop.create_task(self._wait_for_stop(pending))

    async def _wait_for_stop(self, tasks):
        await asyncio.gather(*tasks)
        self._loop.stop()  # << Stopping the event loop here.
        _log.info("Loop stop initiated.")

    ...

还要提到的一件事是，文档说signal.signal()处理程序not allowed要与循环交互，但没有说明原因（请参阅）

推荐阅读