如何在 asyncio.create_subprocess_exec 中创建我自己的管道答案

【问题标题】：How to create my own pipe in asyncio.create_subprocess_exec如何在 asyncio.create_subprocess_exec 中创建我自己的管道
【发布时间】：2022-08-19 01:07:15
【问题描述】：

我有一个程序，我必须从网络中提取文件（p4 print 从版本控制服务器中提取文件并打印到标准输出）。因为网络和 IO 是最大的瓶颈，我正在尝试使用 asyncio。我尝试使用标准的 asyncio.subprocess.PIPE，但因为我有多个子进程，我不断遇到死锁。我想尝试的解决方案是创建一个新文件并将标准输出写入那里。

这是我得到的一些错误

尝试 2：错误 \"OSError: [Errno 9] Bad file descriptor\"

async def _subprocess_wrapper(self, path):
    async with self.sem:
        _, write = os.pipe()
        proc = await asyncio.create_subprocess_exec(
            \'p4\', \'print\', \'-q\', path,
            stdout=write,
            stderr=write
        )
        status = await proc.wait()
        file = os.fdopen(write, \'r\')
        txt  = file.read()
        os.close(write)
        os.close(_)
        return status, txt

尝试 3：错误 \"AttributeError: \'NoneType\' 对象没有属性 \'read\'\"

async def _subprocess_wrapper(self, path):
    async with self.sem:
        _, write = os.pipe()
        proc = await asyncio.create_subprocess_exec(
            \'p4\', \'print\', \'-q\', path,
            stdout=write,
            stderr=write
        )
        status = await proc.wait()
        if status != 0:
            txt = await proc.stderr.read()
        else:
            txt = await proc.stdout.read()
        os.close(write)
        os.close(_)
        return status, txt.decode()

任何帮助，将不胜感激

我强烈建议返回标准 asyncio 并尝试识别和修复死锁。在 asyncio 中，所有 I/O 都必须是非阻塞的或由事件循环管理。该循环使用 select（或 poll）来识别准备好读/写的文件描述符（网络套接字、管道），并在这些 FD 和缓冲区之间传输数据。执行await 的I/O 应用程序代码与这些数据缓冲区交互，而不是与描述符交互。像您尝试做的那样使用直接管道 I/O 根本不适合 asyncio。

标签： python subprocess python-asyncio

【解决方案1】：

根据文档，我放弃了尝试使用自己的管道并更改了我的 wait() 进行通信...
[wait] 在使用 stdout=PIPE 或 stderr=PIPE 时可能会死锁，并且子进程生成的输出过多以至于它阻塞等待 OS 管道缓冲区接受更多数据。使用管道时使用communicate()方法来避免这种情况

我的工作代码

async def _subprocess_wrapper(self, path):
    async with self.sem:
        proc = await asyncio.create_subprocess_exec(
            'p4', 'print', '-q', path,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE
        )
        stdout, stderr = await proc.communicate()
        txt = stdout if proc.returncode == 0 else stderr
        return proc.returncode, txt.decode()

如果有人知道是否有更好的方法来制作这种规模，我将不胜感激

【讨论】：