如何使用 asyncio 从使用子进程协议的子进程读取并在任意点终止该子进程?

Question

使用这里的答案作为基础(使用SubprocessProtocol)，我只是尝试从子进程读取并在我选择的点停止读取(并终止子进程)(例如，我已经读取了足够的数据)。

请注意，我确实希望根据另一个讨论使用run_until_complete的好处。

我碰巧正在使用Windows，下面的例子使用的是Cygwin的cat。我使用的实际实用程序只是一个本机Windows控制台应用程序 - 但它将流式传输，直到手动关闭。

我可以很好地读取数据，但是我尝试停止读取并关闭子进程(例如，从pipe_data_received()中调用loop.stop())会导致异常(RuntimeError: Event loop is closed和ValueError: I/O operation on closed pipe)。我想立即优雅地终止子进程。

我不认为这是平台，而是我没有看到在哪里适当地中断事情以达到预期的效果。关于如何实现这一目标的任何想法？

我的Python 3.7+代码(根据示例修改)：

import asyncio
import os
external_program = "cat"  # Something that will output to stdio
external_option = "a"  # An arbitrarily large amount of data
saved_data = []
class SubprocessProtocol(asyncio.SubprocessProtocol):
def pipe_data_received(self, fd, data):
if fd == 1: # got stdout data (bytes)
data_len = len(data)
print(''.join(' {:02x}'.format(x) for x in data), flush=True)
saved_data.extend(data)
if len(saved_data) > 512:  # Stop once we've read this much data
loop.call_soon_threadsafe(loop.stop)
def connection_lost(self, exc):
print("Connection lost")
loop.stop() # end loop.run_forever()
print("START")
if os.name == 'nt':
# On Windows, the ProactorEventLoop is necessary to listen on pipes
loop = asyncio.ProactorEventLoop() # for subprocess' pipes on Windows
asyncio.set_event_loop(loop)
else:
loop = asyncio.get_event_loop()
try:
loop.run_until_complete(
loop.subprocess_exec(
SubprocessProtocol, 
external_program,
external_option,
)
)
loop.run_forever()
finally:
loop.close()
print("DONE")
loop.close()

Answer 1

不是异步专家，但这样的东西应该可以工作。

import time
import asyncio
import threading
class SubprocessProtocol(asyncio.SubprocessProtocol):
def __init__(self, loop):
self.transport = None
self.loop = loop
def pipe_data_received(self, fd, data):
print('data received')
def connection_lost(self, exc):
print("Connection lost")
def connection_made(self, transport):
print("Connection made")
self.transport = transport
# becasue calc won't call pipe_data_received method.
t = threading.Thread(target=self._endme)
t.setDaemon(True)
t.start()
def _endme(self):
time.sleep(2)
# You'd normally use these inside pipe_data_received, connection_lost methods
self.transport.close()
self.loop.stop()

def main():
loop = asyncio.ProactorEventLoop()
asyncio.set_event_loop(loop)
loop.run_until_complete(loop.subprocess_exec(
lambda: SubprocessProtocol(loop), 
'calc.exe'
))
loop.run_forever()
loop.close()
if __name__ == "__main__":
main()