Мониторить каналы asyncio.create_subprocess_exec на наличие ошибок. ⇐ Python

[Anonymous]

Я пытаюсь передать несколько команд Linux и прервать выполнение в случае ошибки. При использовании Popen метод communication() ожидает завершения всех команд. Вот почему я сейчас пробую asyncio.
У меня есть следующий MWE, работающий должным образом, если нет ошибок:

Код: Выделить всё

async def forward(src, dest):
"""Read data from src and write it to dest."""
while True:
chunk = await src.read(4096)
if not chunk:
dest.write_eof()
await dest.drain()
break
dest.write(chunk)
await dest.drain()

async def stderr_watch(stream):
err = await stream.read()
if err.strip():
raise RuntimeError(f"stderr: {err.decode()}")

async def main():
p1 = await asyncio.create_subprocess_exec(
"find","/", "-name", "*.py",
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)

p2 = await asyncio.create_subprocess_exec(
"wc", "-l",
stdin=asyncio.subprocess.PIPE,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)

output = []
async def stream_reader(stream):
while True:
line = await stream.readline()
if not line:
break
output.append(line.decode())

try:
async with asyncio.TaskGroup() as tg:
t1 = tg.create_task(stderr_watch(p1.stderr))
t2 = tg.create_task(stderr_watch(p2.stderr))
t3 = tg.create_task(forward(p1.stdout, p2.stdin))
t4 = tg.create_task(stream_reader(p2.stdout))

except* Exception as eg:
for e in eg.exceptions:
print(e)
pass
else:
return "".join(output)

if __name__ == '__main__':
output = asyncio.run(main())
print(output)

Однако я получаю ошибку RuntimeError: цикл событий закрыт, если я создаю исключение в p2, например. путем передачи по конвейеру "wc", "-l", "abc",. Где моя ошибка?

Подробнее здесь: https://stackoverflow.com/questions/798 ... for-errors