将字节从 asyncio StreamReader 泵入文件描述符答案

【问题标题】：Pump bytes from asyncio StreamReader into a file descriptor将字节从 asyncio StreamReader 泵入文件描述符
【发布时间】：2016-12-21 09:06:18
【问题描述】：

我有一个 Python 函数（用 C++ 实现），它从文件描述符（在 C++ 端包装在 FILE* 中）读取，我需要从 asyncio.StreamReader 提供函数。具体来说，reader 就是一个 HTTP 响应的内容：aiohttp.ClientResponse.content。

我想我可能会 open a pipe，将读取端传递给 C++ 函数，并将 connect the write-end 传递给 asyncio 的事件循环。但是，如何通过适当的流控制和尽可能少的复制将数据从流读取器移动到管道？

缺少部分的代码骨架如下：

# obtain the StreamReader from aiohttp
content = aiohttp_client_response.content
# create a pipe
(pipe_read_fd, pipe_write_fd) = os.pipe()

# now I need a suitable protocol to manage the pipe transport
protocol = ?
(pipe_transport, __) = loop.connect_write_pipe(lambda: protocol, pipe_write_fd)

# the protocol should start reading from `content` and writing into the pipe
return pipe_read_fd

【问题讨论】：

标签： python python-asyncio aiohttp

【解决方案1】：

来自subprocess_attach_write_pipe asyncio 示例：

rfd, wfd = os.pipe()
pipe = open(wfd, 'wb', 0)
transport, _ = await loop.connect_write_pipe(asyncio.Protocol, pipe)
transport.write(b'data')

EDIT - 写流控制，见以下方法：

这是一个可能的FlowControl 实现，灵感来自StreamWriter.drain：

class FlowControl(asyncio.streams.FlowControlMixin):
    async def drain(self):
        await self._drain_helper()

用法：

transport, protocol = await loop.connect_write_pipe(FlowControl, pipe)
transport.write(b'data')
await protocol.drain()

【讨论】：

这显示了如何打开管道以使用 asyncio 进行写入，但没有显示如何正确地从 asyncio.StreamReader 复制到管道。特别是，如果从管道读取的一方太慢而无法跟上StreamReader，则简单地从读取器读取字节块并将它们提供给transport.write 可能会溢出缓冲区。
@JanŠpaček 查看我关于写入流控制的编辑，希望对您有所帮助。

【解决方案2】：

我通过使用ThreadPoolExecutor 并阻止对os.write 的调用解决了这个问题：

(read_fd, write_fd) = os.pipe()
task_1 = loop.create_task(pump_bytes_into_fd(write_fd))
task_2 = loop.run_in_executor(executor_1, parse_bytes_from_fd(read_fd))

async def pump_bytes_into_fd(write_fd):
    while True:
        chunk = await stream.read(CHUNK_SIZE)
        if chunk is None: break
        # process the chunk
        await loop.run_in_executor(executor_2, os.write, write_fd, chunk)

使用两个不同的执行器来阻止读取和写入以避免死锁是至关重要的。

【讨论】：