python - 如何为标准输入/标准输出创建异步流读取器/写入器？

Question

我需要编写两个程序，它们将作为父进程及其子进程运行。父进程生成子进程，然后它们通过连接到子进程的标准输入和标准输出的一对管道进行通信。通信是点对点的，这就是我需要 asyncio 的原因。一个简单的读取/回复循环是行不通的。

我写了父级。没问题，因为 asyncio 在 create_subprocess_exec() 中提供了我需要的一切。

但是我不知道如何在 child 中创建类似的流读取器/写入器。我没想到会有任何问题。因为管道已经创建并且文件描述符 0 和 1 已准备好在子进程启动时使用。无需打开任何连接，无需生成任何进程。

这是我不成功的尝试:

import asyncio
import sys

_DEFAULT_LIMIT = 64 * 1024

async def connect_stdin_stdout(limit=_DEFAULT_LIMIT, loop=None):
    if loop is None:
        loop = asyncio.get_event_loop()
    reader = asyncio.StreamReader(limit=limit, loop=loop)
    protocol = asyncio.StreamReaderProtocol(reader, loop=loop)
    r_transport, _ = await loop.connect_read_pipe(lambda: protocol, sys.stdin)
    w_transport, _ = await loop.connect_write_pipe(lambda: protocol, sys.stdout)
    writer = asyncio.StreamWriter(w_transport, protocol, reader, loop)
    return reader, writer

问题是我有两个运输工具，而我应该有一个。该函数失败，因为它尝试两次设置协议(protocol)的传输:

await loop.connect_read_pipe(lambda: protocol, sys.stdin)
await loop.connect_write_pipe(lambda: protocol, sys.stdout)
# !!!! assert self._transport is None, 'Transport already set'

我试图将一个虚拟协议(protocol)传递给第一行，但是这一行也不正确，因为需要两种传输方式，而不仅仅是一种传输方式:

writer = asyncio.StreamWriter(w_transport, protocol, reader, loop)

我想我需要以某种方式将两个单向传输组合成一个双向传输。还是我的方法完全错误？你能给我一些建议吗？

---

更新:经过一些测试，这似乎有效(但对我来说不太好):

async def connect_stdin_stdout(limit=_DEFAULT_LIMIT, loop=None):
    if loop is None:
        loop = asyncio.get_event_loop()
    reader = asyncio.StreamReader(limit=limit, loop=loop)
    protocol = asyncio.StreamReaderProtocol(reader, loop=loop)
    dummy = asyncio.Protocol()
    await loop.connect_read_pipe(lambda: protocol, sys.stdin) # sets read_transport
    w_transport, _ = await loop.connect_write_pipe(lambda: dummy, sys.stdout)
    writer = asyncio.StreamWriter(w_transport, protocol, reader, loop)
return reader, writer

Answer 1

你的第一个版本失败了，因为你在编写端使用了错误的协议(protocol)； StreamReaderProtocol 实现了钩子(Hook)来对传入的连接和数据使用react，这是编写方不需要也不应该处理的事情。

loop.connect_write_pipe() 协程使用您传入的协议(protocol)工厂并返回生成的协议(protocol)实例。您确实希望在流写入器中使用相同的协议(protocol)对象，而不是用于读取器的协议(protocol)。

接下来，您不想要将stdin 读取器传递给stdout 流编写器!该类假定读取器和写入器连接到同一个文件描述符，但实际情况并非如此。

在recent past我构建了以下内容来处理子进程的 stdio； stdio() 函数基于 Nathan Hoad gist on the subject ，加上 Windows 的回退，其中 support for treating stdio as pipes is limited .

您确实希望编写器正确处理背压，因此我的版本使用(未记录的)asyncio.streams.FlowControlMixin 类作为协议(protocol)；您真的只需要这些:

import asyncio
import os
import sys

async def stdio(limit=asyncio.streams._DEFAULT_LIMIT, loop=None):
    if loop is None:
        loop = asyncio.get_event_loop()

    if sys.platform == 'win32':
        return _win32_stdio(loop)

    reader = asyncio.StreamReader(limit=limit, loop=loop)
    await loop.connect_read_pipe(
        lambda: asyncio.StreamReaderProtocol(reader, loop=loop), sys.stdin)

    writer_transport, writer_protocol = await loop.connect_write_pipe(
        lambda: asyncio.streams.FlowControlMixin(loop=loop),
        os.fdopen(sys.stdout.fileno(), 'wb'))
    writer = asyncio.streams.StreamWriter(
        writer_transport, writer_protocol, None, loop)

    return reader, writer

def _win32_stdio(loop):
    # no support for asyncio stdio yet on Windows, see https://bugs.python.org/issue26832
    # use an executor to read from stdio and write to stdout
    # note: if nothing ever drains the writer explicitly, no flushing ever takes place!
    class Win32StdinReader:
        def __init__(self):
            self.stdin = sys.stdin.buffer 
        async def readline():
            # a single call to sys.stdin.readline() is thread-safe
            return await loop.run_in_executor(None, self.stdin.readline)

    class Win32StdoutWriter:
        def __init__(self):
            self.buffer = []
            self.stdout = sys.stdout.buffer
        def write(self, data):
            self.buffer.append(data)
        async def drain(self):
            data, self.buffer = self.buffer, []
            # a single call to sys.stdout.writelines() is thread-safe
            return await loop.run_in_executor(None, sys.stdout.writelines, data)

    return Win32StdinReader(), Win32StdoutWriter()

虽然可能有点过时，但我找到了this 2016 blog post by by Nathaniel J. Smith on asyncio and curio 非常有助于理解 asyncio、协议(protocol)、传输和背压以及所有这些如何相互作用并结合在一起。那篇文章还说明了为什么目前为 stdio 创建读取器和写入器对象如此冗长和繁琐。