您好, 欢迎来到 !    登录 | 注册 | | 设为首页 | 收藏本站

子流程中有多个管道

子流程中有多个管道

模拟bash进程替换

#!/usr/bin/env python
from subprocess import check_call

check_call('someprogram <(someprocess) <(anotherprocess)',
           shell=True, executable='/bin/bash')

在Python中,您可以使用命名管道:

#!/usr/bin/env python
from subprocess import Popen

with named_pipes(n=2) as paths:
    someprogram = Popen(['someprogram'] + paths)
    processes = []
    for path, command in zip(paths, ['someprocess', 'anotherprocess']):
        with open(path, 'wb', 0) as pipe:
            processes.append(Popen(command, stdout=pipe, close_fds=True))
    for p in [someprogram] + processes:
        p.wait()

在哪里named_pipes(n)

import os
import shutil
import tempfile
from contextlib import contextmanager

@contextmanager
def named_pipes(n=1):
    dirname = tempfile.mkdtemp()
    try:
        paths = [os.path.join(dirname, 'named_pipe' + str(i)) for i in range(n)]
        for path in paths:
            os.mkfifo(path)
        yield paths
    finally:
        shutil.rmtree(dirname)

实现bash进程替换的另一种更可取的方式(无需在磁盘上创建命名条目)是使用@Dunes建议的/dev/fd/N文件名(如果可用)。在FreeBSD上,)为进程打开的所有文件描述符创建条目。要测试可用性,请运行:@L_419_4@fdescfs(5)/dev/fd/#

$ test -r /dev/fd/3 3</dev/null && echo /dev/fd is available

如果失败;尝试进行符号链接/dev/fdproc(5)就像在某些Linux上所做的那样:

$ ln -s /proc/self/fd /dev/fd

这是/dev/fd基于someprogram <(someprocess) <(anotherprocess)bash命令的实现:

#!/usr/bin/env python3
from contextlib import ExitStack
from subprocess import CalledProcessError, Popen, PIPE

def kill(process):
    if process.poll() is None: # still running
        process.kill()

with ExitStack() as stack: # for proper cleanup
    processes = []
    for command in [['someprocess'], ['anotherprocess']]:  # start child processes
        processes.append(stack.enter_context(Popen(command, stdout=PIPE)))
        stack.callback(kill, processes[-1]) # kill on someprogram exit

    fds = [p.stdout.fileno() for p in processes]
    someprogram = stack.enter_context(
        Popen(['someprogram'] + ['/dev/fd/%d' % fd for fd in fds], pass_fds=fds))
    for p in processes: # close pipes in the parent
        p.stdout.close()
# exit stack: wait for processes
if someprogram.returncode != 0: # errors shouldn't go unnoticed
   raise CalledProcessError(someprogram.returncode, someprogram.args)

注意:在我的Ubuntu机器上,该subprocess代码仅在Python 3.4+中有效,尽管pass_fds自Python 3.2起可用。

其他 2022/1/1 18:26:37 有432人围观

撰写回答


你尚未登录,登录后可以

和开发者交流问题的细节

关注并接收问题和回答的更新提醒

参与内容的编辑和改进,让解决方法与时俱进

请先登录

推荐问题


联系我
置顶