不确定如何取消多长时间

我正在开发一个与Web服务交互的CLI。 运行时，它将尝试与它建立通信，发送请求，接收并处理回复，然后终止。 我在我的代码和asyncio的各个部分使用协程来驱动它们。 我希望能够执行所有这些步骤，然后让所有协程完全终止（即以不会导致异步抱怨的方式）。 不幸的是，我发现asyncio比C＃等其他语言中的异步性更难以使用和理解。

我定义了一个类来处理通过websocket连接与Web服务的所有直接通信：

class CommunicationService(object):
    def __init__(self):
        self._ws = None
        self._listen_task = None
        self._loop = asyncio.get_event_loop()

    ...

    async def _listen_for_responses(self):
        logging.debug('Listening for responses')
        while True:
            message = await self._ws.recv()
            self.__received_response.on_next(message)

    def establish_communication(self, hostname: str, port: int) -> None:
        websocket_address = 'ws://{}:{}/'.format(hostname, port)

        self._ws = self._loop.run_until_complete(websockets.connect(websocket_address))
        self._listen_task = asyncio.ensure_future(self._listen_for_responses())

    def send_request(self, request: str) -> None:
        return asyncio.ensure_future(self._ws.send(request))

    def stop(self):
        if self._listen_task:
            self._loop.call_soon_threadsafe(self._listen_task.cancel)

        if self._ws and self._ws.open:
            self._ws.close()

这个类使用websockets和RxPY库。 建立通信时，此类的实例将创建一个无限期运行的任务，等待来自Web服务的响应并将其发布到RxPY主题上。

我在主CLI方法中运行CommunicationService.establish_communication ：

def cli(context, hostname, port):
    log_level = context.meta['click_log.core.logger']['level']
    _initialize_logging(log_level)

    # Create a new event loop for processing commands asynchronously on.
    loop = asyncio.new_event_loop()
    loop.set_debug(log_level == logging.DEBUG)
    asyncio.set_event_loop(loop)

    # Establish communication with TestCube Web Service.
    context.comms = CommunicationService()
    context.comms.establish_communication(hostname, port)
    ...

根据提供的CLI参数，这可能会调用一个子命令回调函数，我将其作为协程函数来实现。

然后我注册一个函数来处理被调用的子命令的结果，该命令将是None或者一个协程对象：

@cli.resultcallback()
@click.pass_context
def _handle_command_task(context, task: Coroutine, **_) -> None:
    if task:
        loop = asyncio.get_event_loop()
        result = loop.run_until_complete(task)
        context.comms.stop()
        loop.close()
        if result:
            print(result, end='')

我的程序可以正常工作，但是我得到以下输出（在INFO日志级别运行CLI时）：

$ testcube relays.0.enabled false
2016-09-06 12:33:51,157 [INFO    ] testcube.comms - Establishing connection to TestCube Web Service @ 127.0.0.1:36364
2016-09-06 12:33:51,219 [ERROR   ] asyncio - Task was destroyed but it is pending!
task: <Task pending coro=<CommunicationService._listen_for_responses() running at c:usersdavidfallahpycharmprojectstestcubetestcubecomms.py:34> wait_for=<Future pending cb=[Task._wakeup()]>>
2016-09-06 12:33:51,219 [ERROR   ] asyncio - Task was destroyed but it is pending!
task: <Task pending coro=<WebSocketCommonProtocol.run() running at c:usersdavidfallahappdatalocalprogramspythonpython35-32libsite-packageswebsocketsprotocol.py:413> wait_for=<Future pending cb=[Task._wakeup()]> cb=[_wait.<locals>._on_completion() at c:usersdavidfallahappdatalocalprogramspythonpython35-32libasynciotasks.py:414]>
2016-09-06 12:33:51,219 [ERROR   ] asyncio - Task was destroyed but it is pending!
task: <Task pending coro=<Queue.get() running at c:usersdavidfallahappdatalocalprogramspythonpython35-32libasyncioqueues.py:168> wait_for=<Future pending cb=[Task._wakeup()]> cb=[_wait.<locals>._on_completion() at c:usersdavidfallahappdatalocalprogramspythonpython35-32libasynciotasks.py:414]>
Exception ignored in: <generator object Queue.get at 0x03643600>
Traceback (most recent call last):
  File "c:usersdavidfallahappdatalocalprogramspythonpython35-32libasyncioqueues.py", line 170, in get
  File "c:usersdavidfallahappdatalocalprogramspythonpython35-32libasynciofutures.py", line 227, in cancel
  File "c:usersdavidfallahappdatalocalprogramspythonpython35-32libasynciofutures.py", line 242, in _schedule_callbacks
  File "c:usersdavidfallahappdatalocalprogramspythonpython35-32libasynciobase_events.py", line 497, in call_soon
  File "c:usersdavidfallahappdatalocalprogramspythonpython35-32libasynciobase_events.py", line 506, in _call_soon
  File "c:usersdavidfallahappdatalocalprogramspythonpython35-32libasynciobase_events.py", line 334, in _check_closed
RuntimeError: Event loop is closed

如果我更改CommunicationService.stop()以直接取消响应侦听任务（而不是调度）...

self._listen_task.cancel()
#self._loop.call_soon_threadsafe(self._listen_task.cancel)

我得到以下输出：

...
task: <Task pending coro=<CommunicationService._listen_for_responses() running at c:usersdavidfallahpycharmprojectstestcubetestcubecomms.py:34> wait_for=<Future cancelled>>
...

其中wait_for <Future cancelled> （与wait_for=<Future pending cb=[Task._wakeup()]>> ） wait_for=<Future pending cb=[Task._wakeup()]>> 。 我不明白我如何调用Task.cancel() ，它说future cancelled但任务仍在等待。 我是否需要对任务做一些特殊的事情，例如，在try...except asyncio.CancelledException...包装代码try...except asyncio.CancelledException... ？

如果它有用，则这是同一命令的DEBUG级输出：

$ testcube -v DEBUG relays.0.enabled false
2016-09-06 12:48:10,145 [DEBUG   ] asyncio - Using selector: SelectSelector
2016-09-06 12:48:10,147 [DEBUG   ] Rx - CurrentThreadScheduler.schedule(state=None)
2016-09-06 12:48:10,147 [INFO    ] testcube.comms - Establishing connection to TestCube Web Service @ 127.0.0.1:36364
2016-09-06 12:48:10,153 [DEBUG   ] asyncio - connect <socket.socket fd=608, family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=6> to ('127.0.0.1', 36364)
2016-09-06 12:48:10,156 [DEBUG   ] asyncio - poll took 0.000 ms: 1 events
2016-09-06 12:48:10,163 [DEBUG   ] asyncio - <socket.socket fd=608, family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=6, laddr=('127.0.0.1', 56647), raddr=('127.0.0.1', 36364)> connected to 127.0.0.1:36364: (<_SelectorSocketTransport fd=608 read=polling write=<idle, bufsize=0>>, <websockets.client.WebSocketClientProtocol object at 0x03623BF0>)
2016-09-06 12:48:10,198 [DEBUG   ] asyncio - poll took 31.000 ms: 1 events
2016-09-06 12:48:10,202 [DEBUG   ] root - Connected using websocket address: ws://127.0.0.1:36364/
2016-09-06 12:48:10,202 [DEBUG   ] Rx - CurrentThreadScheduler.schedule(state=None)
2016-09-06 12:48:10,203 [DEBUG   ] testcube.components.core - Using write handler
2016-09-06 12:48:10,203 [DEBUG   ] root - Listening for responses
2016-09-06 12:48:10,205 [DEBUG   ] testcube.comms - Sending request: {"op": "replace", "value": false, "path": "testcube.relays[0].enabled"}
2016-09-06 12:48:10,208 [DEBUG   ] websockets.protocol - client >> Frame(fin=True, opcode=1, data=b'{"op": "replace", "value": false, "path": "testcube.relays[0].enabled"}')
2016-09-06 12:48:10,209 [DEBUG   ] asyncio - Close <_WindowsSelectorEventLoop running=False closed=False debug=True>
2016-09-06 12:48:10,222 [ERROR   ] asyncio - Task was destroyed but it is pending!
source_traceback: Object created at (most recent call last):
  File "C:UsersdavidfallahAppDataLocalProgramsPythonPython35-32Scriptstestcube-script.py", line 9, in <module>
    load_entry_point('testcube', 'console_scripts', 'testcube')()
  File "c:usersdavidfallahpycharmprojectstestcubetestcubetestcube.py", line 198, in main
    cli(default_map=_get_default_settings())
  File "c:usersdavidfallahappdatalocalprogramspythonpython35-32libsite-packagesclickcore.py", line 716, in __call__
    return self.main(*args, **kwargs)
  File "c:usersdavidfallahappdatalocalprogramspythonpython35-32libsite-packagesclickcore.py", line 696, in main
    rv = self.invoke(ctx)
  File "c:usersdavidfallahappdatalocalprogramspythonpython35-32libsite-packagesclickcore.py", line 1057, in invoke
    Command.invoke(self, ctx)
  File "c:usersdavidfallahappdatalocalprogramspythonpython35-32libsite-packagesclickcore.py", line 889, in invoke
    return ctx.invoke(self.callback, **ctx.params)
  File "c:usersdavidfallahappdatalocalprogramspythonpython35-32libsite-packagesclickcore.py", line 534, in invoke
    return callback(*args, **kwargs)
  File "c:usersdavidfallahappdatalocalprogramspythonpython35-32libsite-packagesclickdecorators.py", line 17, in new_func
    return f(get_current_context(), *args, **kwargs)
  File "c:usersdavidfallahpycharmprojectstestcubetestcubetestcube.py", line 168, in cli
    context.comms.establish_communication(hostname, port)
  File "c:usersdavidfallahpycharmprojectstestcubetestcubecomms.py", line 48, in establish_communication
    self._listen_task = asyncio.ensure_future(self._listen_for_responses())
task: <Task pending coro=<CommunicationService._listen_for_responses() running at c:usersdavidfallahpycharmprojectstestcubetestcubecomms.py:34> wait_for=<Future cancelled created at c:usersdavidfallahappdatalocalprogramspythonpython35-32libasynciobase_events.py:252> created at c:usersdavidfallahpycharmprojectstestcubetestcubecomms.py:48>
2016-09-06 12:48:10,223 [ERROR   ] asyncio - Task was destroyed but it is pending!
source_traceback: Object created at (most recent call last):
  File "C:UsersdavidfallahAppDataLocalProgramsPythonPython35-32Scriptstestcube-script.py", line 9, in <module>
    load_entry_point('testcube', 'console_scripts', 'testcube')()
  File "c:usersdavidfallahpycharmprojectstestcubetestcubetestcube.py", line 198, in main
    cli(default_map=_get_default_settings())
  File "c:usersdavidfallahappdatalocalprogramspythonpython35-32libsite-packagesclickcore.py", line 716, in __call__
    return self.main(*args, **kwargs)
  File "c:usersdavidfallahappdatalocalprogramspythonpython35-32libsite-packagesclickcore.py", line 696, in main
    rv = self.invoke(ctx)
  File "c:usersdavidfallahappdatalocalprogramspythonpython35-32libsite-packagesclickcore.py", line 1057, in invoke
    Command.invoke(self, ctx)
  File "c:usersdavidfallahappdatalocalprogramspythonpython35-32libsite-packagesclickcore.py", line 889, in invoke
    return ctx.invoke(self.callback, **ctx.params)
  File "c:usersdavidfallahappdatalocalprogramspythonpython35-32libsite-packagesclickcore.py", line 534, in invoke
    return callback(*args, **kwargs)
  File "c:usersdavidfallahappdatalocalprogramspythonpython35-32libsite-packagesclickdecorators.py", line 17, in new_func
    return f(get_current_context(), *args, **kwargs)
  File "c:usersdavidfallahpycharmprojectstestcubetestcubetestcube.py", line 168, in cli
    context.comms.establish_communication(hostname, port)
  File "c:usersdavidfallahpycharmprojectstestcubetestcubecomms.py", line 47, in establish_communication
    self._ws = self._loop.run_until_complete(websockets.connect(websocket_address))
  File "c:usersdavidfallahappdatalocalprogramspythonpython35-32libasynciobase_events.py", line 375, in run_until_complete
    self.run_forever()
  File "c:usersdavidfallahappdatalocalprogramspythonpython35-32libasynciobase_events.py", line 345, in run_forever
    self._run_once()
  File "c:usersdavidfallahappdatalocalprogramspythonpython35-32libasynciobase_events.py", line 1304, in _run_once
    handle._run()
  File "c:usersdavidfallahappdatalocalprogramspythonpython35-32libasyncioevents.py", line 125, in _run
    self._callback(*self._args)
  File "c:usersdavidfallahappdatalocalprogramspythonpython35-32libasynciostreams.py", line 238, in connection_made
    self._stream_writer)
  File "c:usersdavidfallahappdatalocalprogramspythonpython35-32libsite-packageswebsocketsprotocol.py", line 633, in client_connected
    self.worker_task = asyncio_ensure_future(self.run(), loop=self.loop)
task: <Task pending coro=<WebSocketCommonProtocol.run() running at c:usersdavidfallahappdatalocalprogramspythonpython35-32libsite-packageswebsocketsprotocol.py:413> wait_for=<Future pending cb=[Task._wakeup()] created at c:usersdavidfallahappdatalocalprogramspythonpython35-32libasynciobase_events.py:252> cb=[_wait.<locals>._on_completion() at c:usersdavidfallahappdatalocalprogramspythonpython35-32libasynciotasks.py:414] created at c:usersdavidfallahappdatalocalprogramspythonpython35-32libsite-packageswebsocketsprotocol.py:633>
2016-09-06 12:48:10,223 [ERROR   ] asyncio - Task was destroyed but it is pending!
source_traceback: Object created at (most recent call last):
  File "C:UsersdavidfallahAppDataLocalProgramsPythonPython35-32Scriptstestcube-script.py", line 9, in <module>
    load_entry_point('testcube', 'console_scripts', 'testcube')()
  File "c:usersdavidfallahpycharmprojectstestcubetestcubetestcube.py", line 198, in main
    cli(default_map=_get_default_settings())
  File "c:usersdavidfallahappdatalocalprogramspythonpython35-32libsite-packagesclickcore.py", line 716, in __call__
    return self.main(*args, **kwargs)
  File "c:usersdavidfallahappdatalocalprogramspythonpython35-32libsite-packagesclickcore.py", line 696, in main
    rv = self.invoke(ctx)
  File "c:usersdavidfallahappdatalocalprogramspythonpython35-32libsite-packagesclickcore.py", line 1060, in invoke
    return _process_result(sub_ctx.command.invoke(sub_ctx))
  File "c:usersdavidfallahappdatalocalprogramspythonpython35-32libsite-packagesclickcore.py", line 1025, in _process_result
    **ctx.params)
  File "c:usersdavidfallahappdatalocalprogramspythonpython35-32libsite-packagesclickcore.py", line 534, in invoke
    return callback(*args, **kwargs)
  File "c:usersdavidfallahappdatalocalprogramspythonpython35-32libsite-packagesclickdecorators.py", line 17, in new_func
    return f(get_current_context(), *args, **kwargs)
  File "c:usersdavidfallahpycharmprojectstestcubetestcubetestcube.py", line 190, in _handle_command_task
    result = loop.run_until_complete(task)
  File "c:usersdavidfallahappdatalocalprogramspythonpython35-32libasynciobase_events.py", line 375, in run_until_complete
    self.run_forever()
  File "c:usersdavidfallahappdatalocalprogramspythonpython35-32libasynciobase_events.py", line 345, in run_forever
    self._run_once()
  File "c:usersdavidfallahappdatalocalprogramspythonpython35-32libasynciobase_events.py", line 1304, in _run_once
    handle._run()
  File "c:usersdavidfallahappdatalocalprogramspythonpython35-32libasyncioevents.py", line 125, in _run
    self._callback(*self._args)
  File "c:usersdavidfallahappdatalocalprogramspythonpython35-32libasynciotasks.py", line 239, in _step
    result = coro.send(None)
  File "c:usersdavidfallahpycharmprojectstestcubetestcubecomms.py", line 34, in _listen_for_responses
    message = await self._ws.recv()
  File "c:usersdavidfallahappdatalocalprogramspythonpython35-32libsite-packageswebsocketsprotocol.py", line 280, in recv
    self.messages.get(), loop=self.loop)
task: <Task pending coro=<Queue.get() running at c:usersdavidfallahappdatalocalprogramspythonpython35-32libasyncioqueues.py:168> wait_for=<Future pending cb=[Task._wakeup()] created at c:usersdavidfallahappdatalocalprogramspythonpython35-32libasynciobase_events.py:252> cb=[_wait.<locals>._on_completion() at c:usersdavidfallahappdatalocalprogramspythonpython35-32libasynciotasks.py:414] created at c:usersdavidfallahappdatalocalprogramspythonpython35-32libsite-packageswebsocketsprotocol.py:280>
Exception ignored in: <generator object Queue.get at 0x03641240>
Traceback (most recent call last):
  File "c:usersdavidfallahappdatalocalprogramspythonpython35-32libasyncioqueues.py", line 170, in get
  File "c:usersdavidfallahappdatalocalprogramspythonpython35-32libasynciofutures.py", line 227, in cancel
  File "c:usersdavidfallahappdatalocalprogramspythonpython35-32libasynciofutures.py", line 242, in _schedule_callbacks
  File "c:usersdavidfallahappdatalocalprogramspythonpython35-32libasynciobase_events.py", line 497, in call_soon
  File "c:usersdavidfallahappdatalocalprogramspythonpython35-32libasynciobase_events.py", line 506, in _call_soon
  File "c:usersdavidfallahappdatalocalprogramspythonpython35-32libasynciobase_events.py", line 334, in _check_closed
RuntimeError: Event loop is closed

---

我想通了 - 我需要定义CommunicationService.stop()如下：

def stop(self):
    if self._listen_task is None or self._ws is None:
        return

    self._listen_task.cancel()
    self._loop.run_until_complete(asyncio.wait([self._listen_task, self._ws.close()]))

    self._listen_task = None
    self._ws = None

作为可能最终解决相关问题的其他人的文档，完整的清理代码现在是：

@cli.resultcallback()
@click.pass_context
def _handle_command_task(context, task: Coroutine, **_) -> None:
    if task:
        loop = asyncio.get_event_loop()
        result = loop.run_until_complete(task)
        context.comms.stop()
        loop.close()
        if result:
            print(result, end='')

链接地址: http://www.djcxy.com/p/53211.html

上一篇: Unsure how to cancel long

下一篇: Asyncio coroutines