1
Я разрабатываю CLI, который взаимодействует с веб-сервисом. При запуске он попытается установить связь с ним, отправлять запросы, получать и обрабатывать ответы, а затем прекращать работу. Я использую сопрограммы в различных частях моего кода и асинхронно, чтобы управлять ими. Я хотел бы иметь возможность выполнить все эти шаги, а затем завершить все сопрограммы в конечном итоге (т. Е. Таким образом, чтобы не асинхронно жаловаться). К сожалению, я нахожу asyncio намного сложнее в использовании и понимании, чем асинхронность на других языках, таких как C#.Неуверенный, как отменить долгосрочную задачу asyncio
Я определяю класс, который обрабатывает все прямые связи с веб-службой через WebSocket соединений:
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 библиотек. При установлении связи экземпляры этого класса создадут задачу с неопределенным сроком, которая будет ждать ответов от веб-службы и публиковать их по теме RxPY.
Я бегу CommunicationService.establish_communication в главном методе CLI:
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='')
Моя программа работает, но я получаю следующий результат (при запуске CLI уровень ведения журнала на INFO):
$ 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:\users\davidfallah\pycharmprojects\testcube\testcube\comms.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:\users\davidfallah\appdata\local\programs\python\python35-32\lib\site-packages\websockets\protocol.py:413> wait_for=<Future pending cb=[Task._wakeup()]> cb=[_wait.<locals>._on_completion() at c:\users\davidfallah\appdata\local\programs\python\python35-32\lib\asyncio\tasks.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:\users\davidfallah\appdata\local\programs\python\python35-32\lib\asyncio\queues.py:168> wait_for=<Future pending cb=[Task._wakeup()]> cb=[_wait.<locals>._on_completion() at c:\users\davidfallah\appdata\local\programs\python\python35-32\lib\asyncio\tasks.py:414]>
Exception ignored in: <generator object Queue.get at 0x03643600>
Traceback (most recent call last):
File "c:\users\davidfallah\appdata\local\programs\python\python35-32\lib\asyncio\queues.py", line 170, in get
File "c:\users\davidfallah\appdata\local\programs\python\python35-32\lib\asyncio\futures.py", line 227, in cancel
File "c:\users\davidfallah\appdata\local\programs\python\python35-32\lib\asyncio\futures.py", line 242, in _schedule_callbacks
File "c:\users\davidfallah\appdata\local\programs\python\python35-32\lib\asyncio\base_events.py", line 497, in call_soon
File "c:\users\davidfallah\appdata\local\programs\python\python35-32\lib\asyncio\base_events.py", line 506, in _call_soon
File "c:\users\davidfallah\appdata\local\programs\python\python35-32\lib\asyncio\base_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:\users\davidfallah\pycharmprojects\testcube\testcube\comms.py:34> wait_for=<Future cancelled>>
...
В которой wait_for является <Future cancelled> (в отличие от wait_for=<Future pending cb=[Task._wakeup()]>>). Я не понимаю, как я звоню Task.cancel(), и он говорит future cancelled, но задача все еще не выполнена. Нужно ли мне делать что-то особенное с заданием, например. оберните код в 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:\Users\davidfallah\AppData\Local\Programs\Python\Python35-32\Scripts\testcube-script.py", line 9, in <module>
load_entry_point('testcube', 'console_scripts', 'testcube')()
File "c:\users\davidfallah\pycharmprojects\testcube\testcube\testcube.py", line 198, in main
cli(default_map=_get_default_settings())
File "c:\users\davidfallah\appdata\local\programs\python\python35-32\lib\site-packages\click\core.py", line 716, in __call__
return self.main(*args, **kwargs)
File "c:\users\davidfallah\appdata\local\programs\python\python35-32\lib\site-packages\click\core.py", line 696, in main
rv = self.invoke(ctx)
File "c:\users\davidfallah\appdata\local\programs\python\python35-32\lib\site-packages\click\core.py", line 1057, in invoke
Command.invoke(self, ctx)
File "c:\users\davidfallah\appdata\local\programs\python\python35-32\lib\site-packages\click\core.py", line 889, in invoke
return ctx.invoke(self.callback, **ctx.params)
File "c:\users\davidfallah\appdata\local\programs\python\python35-32\lib\site-packages\click\core.py", line 534, in invoke
return callback(*args, **kwargs)
File "c:\users\davidfallah\appdata\local\programs\python\python35-32\lib\site-packages\click\decorators.py", line 17, in new_func
return f(get_current_context(), *args, **kwargs)
File "c:\users\davidfallah\pycharmprojects\testcube\testcube\testcube.py", line 168, in cli
context.comms.establish_communication(hostname, port)
File "c:\users\davidfallah\pycharmprojects\testcube\testcube\comms.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:\users\davidfallah\pycharmprojects\testcube\testcube\comms.py:34> wait_for=<Future cancelled created at c:\users\davidfallah\appdata\local\programs\python\python35-32\lib\asyncio\base_events.py:252> created at c:\users\davidfallah\pycharmprojects\testcube\testcube\comms.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:\Users\davidfallah\AppData\Local\Programs\Python\Python35-32\Scripts\testcube-script.py", line 9, in <module>
load_entry_point('testcube', 'console_scripts', 'testcube')()
File "c:\users\davidfallah\pycharmprojects\testcube\testcube\testcube.py", line 198, in main
cli(default_map=_get_default_settings())
File "c:\users\davidfallah\appdata\local\programs\python\python35-32\lib\site-packages\click\core.py", line 716, in __call__
return self.main(*args, **kwargs)
File "c:\users\davidfallah\appdata\local\programs\python\python35-32\lib\site-packages\click\core.py", line 696, in main
rv = self.invoke(ctx)
File "c:\users\davidfallah\appdata\local\programs\python\python35-32\lib\site-packages\click\core.py", line 1057, in invoke
Command.invoke(self, ctx)
File "c:\users\davidfallah\appdata\local\programs\python\python35-32\lib\site-packages\click\core.py", line 889, in invoke
return ctx.invoke(self.callback, **ctx.params)
File "c:\users\davidfallah\appdata\local\programs\python\python35-32\lib\site-packages\click\core.py", line 534, in invoke
return callback(*args, **kwargs)
File "c:\users\davidfallah\appdata\local\programs\python\python35-32\lib\site-packages\click\decorators.py", line 17, in new_func
return f(get_current_context(), *args, **kwargs)
File "c:\users\davidfallah\pycharmprojects\testcube\testcube\testcube.py", line 168, in cli
context.comms.establish_communication(hostname, port)
File "c:\users\davidfallah\pycharmprojects\testcube\testcube\comms.py", line 47, in establish_communication
self._ws = self._loop.run_until_complete(websockets.connect(websocket_address))
File "c:\users\davidfallah\appdata\local\programs\python\python35-32\lib\asyncio\base_events.py", line 375, in run_until_complete
self.run_forever()
File "c:\users\davidfallah\appdata\local\programs\python\python35-32\lib\asyncio\base_events.py", line 345, in run_forever
self._run_once()
File "c:\users\davidfallah\appdata\local\programs\python\python35-32\lib\asyncio\base_events.py", line 1304, in _run_once
handle._run()
File "c:\users\davidfallah\appdata\local\programs\python\python35-32\lib\asyncio\events.py", line 125, in _run
self._callback(*self._args)
File "c:\users\davidfallah\appdata\local\programs\python\python35-32\lib\asyncio\streams.py", line 238, in connection_made
self._stream_writer)
File "c:\users\davidfallah\appdata\local\programs\python\python35-32\lib\site-packages\websockets\protocol.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:\users\davidfallah\appdata\local\programs\python\python35-32\lib\site-packages\websockets\protocol.py:413> wait_for=<Future pending cb=[Task._wakeup()] created at c:\users\davidfallah\appdata\local\programs\python\python35-32\lib\asyncio\base_events.py:252> cb=[_wait.<locals>._on_completion() at c:\users\davidfallah\appdata\local\programs\python\python35-32\lib\asyncio\tasks.py:414] created at c:\users\davidfallah\appdata\local\programs\python\python35-32\lib\site-packages\websockets\protocol.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:\Users\davidfallah\AppData\Local\Programs\Python\Python35-32\Scripts\testcube-script.py", line 9, in <module>
load_entry_point('testcube', 'console_scripts', 'testcube')()
File "c:\users\davidfallah\pycharmprojects\testcube\testcube\testcube.py", line 198, in main
cli(default_map=_get_default_settings())
File "c:\users\davidfallah\appdata\local\programs\python\python35-32\lib\site-packages\click\core.py", line 716, in __call__
return self.main(*args, **kwargs)
File "c:\users\davidfallah\appdata\local\programs\python\python35-32\lib\site-packages\click\core.py", line 696, in main
rv = self.invoke(ctx)
File "c:\users\davidfallah\appdata\local\programs\python\python35-32\lib\site-packages\click\core.py", line 1060, in invoke
return _process_result(sub_ctx.command.invoke(sub_ctx))
File "c:\users\davidfallah\appdata\local\programs\python\python35-32\lib\site-packages\click\core.py", line 1025, in _process_result
**ctx.params)
File "c:\users\davidfallah\appdata\local\programs\python\python35-32\lib\site-packages\click\core.py", line 534, in invoke
return callback(*args, **kwargs)
File "c:\users\davidfallah\appdata\local\programs\python\python35-32\lib\site-packages\click\decorators.py", line 17, in new_func
return f(get_current_context(), *args, **kwargs)
File "c:\users\davidfallah\pycharmprojects\testcube\testcube\testcube.py", line 190, in _handle_command_task
result = loop.run_until_complete(task)
File "c:\users\davidfallah\appdata\local\programs\python\python35-32\lib\asyncio\base_events.py", line 375, in run_until_complete
self.run_forever()
File "c:\users\davidfallah\appdata\local\programs\python\python35-32\lib\asyncio\base_events.py", line 345, in run_forever
self._run_once()
File "c:\users\davidfallah\appdata\local\programs\python\python35-32\lib\asyncio\base_events.py", line 1304, in _run_once
handle._run()
File "c:\users\davidfallah\appdata\local\programs\python\python35-32\lib\asyncio\events.py", line 125, in _run
self._callback(*self._args)
File "c:\users\davidfallah\appdata\local\programs\python\python35-32\lib\asyncio\tasks.py", line 239, in _step
result = coro.send(None)
File "c:\users\davidfallah\pycharmprojects\testcube\testcube\comms.py", line 34, in _listen_for_responses
message = await self._ws.recv()
File "c:\users\davidfallah\appdata\local\programs\python\python35-32\lib\site-packages\websockets\protocol.py", line 280, in recv
self.messages.get(), loop=self.loop)
task: <Task pending coro=<Queue.get() running at c:\users\davidfallah\appdata\local\programs\python\python35-32\lib\asyncio\queues.py:168> wait_for=<Future pending cb=[Task._wakeup()] created at c:\users\davidfallah\appdata\local\programs\python\python35-32\lib\asyncio\base_events.py:252> cb=[_wait.<locals>._on_completion() at c:\users\davidfallah\appdata\local\programs\python\python35-32\lib\asyncio\tasks.py:414] created at c:\users\davidfallah\appdata\local\programs\python\python35-32\lib\site-packages\websockets\protocol.py:280>
Exception ignored in: <generator object Queue.get at 0x03641240>
Traceback (most recent call last):
File "c:\users\davidfallah\appdata\local\programs\python\python35-32\lib\asyncio\queues.py", line 170, in get
File "c:\users\davidfallah\appdata\local\programs\python\python35-32\lib\asyncio\futures.py", line 227, in cancel
File "c:\users\davidfallah\appdata\local\programs\python\python35-32\lib\asyncio\futures.py", line 242, in _schedule_callbacks
File "c:\users\davidfallah\appdata\local\programs\python\python35-32\lib\asyncio\base_events.py", line 497, in call_soon
File "c:\users\davidfallah\appdata\local\programs\python\python35-32\lib\asyncio\base_events.py", line 506, in _call_soon
File "c:\users\davidfallah\appdata\local\programs\python\python35-32\lib\asyncio\base_events.py", line 334, in _check_closed
RuntimeError: Event loop is closed