嵌入式环境中的 Python 子进程块答案

【问题标题】：Python subprocess blocks in embedded environment嵌入式环境中的 Python 子进程块
【发布时间】：2019-05-30 14:13:35
【问题描述】：

我的脚本将大量文件作为输入，使用subprocess.call() 调用外部程序。它可以在常规操作系统控制台上完美运行，但在使用其嵌入式 Python 2.7.x 在嵌入式环境中运行时处理 10 个左右文件后挂起。

我参考了各种类似的问题，但没有找到适合我的问题：

Python: subprocess.Popen and subprocess.call hang

Python subprocess hangs

Python subprocess call hangs

还有这个深入的讨论：https://thraxil.org/users/anders/posts/2008/03/13/Subprocess-Hanging-PIPE-is-your-enemy/

它们都暗示了易受攻击的缓冲 PIPE，并建议为 stdout 和 stderr 使用类似文件的对象。所以我还添加了一个临时文本文件，打开它，并将其提供给subprocess.call() 的stdout 和stderr。它也没有工作。

我的旧代码相当简单：

# script1.py

folder = "/path/to/my/folder"
for root, dirs, files in os.walk(folder):
    for file in files:
        path = join(root, file)
        try:
            cmd = ['dir', path, '1>&2']
            _logger.debug(' '.join(cmd))
            completed = subprocess.call(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
        except subprocess.CalledProcessError as err:
            _logger.debug(err)
        else:
            _logger.debug('returncode: {}'.format(completed))
print('all done!!')

主脚本：

try:
    cmd = ['python', 'script1.py', '1>&2']
    ue.log(' '.join(cmd))
    completed = subprocess.call(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
except subprocess.CalledProcessError as err:
    _logger.error(err)

我已经混合并匹配了几个标准输出/标准错误解决方案。包括添加：

with open(join(_script_dir, 'tmp.txt'), 'w') as tmp:
    #
    # old code
    #
     completed = subprocess.call(cmd, shell=True, stdout=tmp, stderr=tmp)

都没有用。

【问题讨论】：

标签： python subprocess

【解决方案1】：

我最终丢弃了所有管道和 stdout/stderr，并使用 Python 3.7.2 的默认 subprocess.Popen() 参数，即，None 用于 stdout/stderr，并删除 1>&2。因此，当程序处理大量输入并可能通过狭窄和隐藏的管道产生巨大的输出波时，基本上不要费心检索另一个程序的实时输出。我转而只使用为每个单独的脚本编写的日志文件。

比起盲目地尝试各种 UNIX 技巧来解决 Windows 上的管道问题，我对这个解决方案更满意。

附言，我上面链接中的一些答案建议shell=False，我发现嵌入式环境会从每个subprocess.call() 中生成大量浮动控制台窗口，这很烦人。

【讨论】：

【解决方案2】：

有一个解决吊管问题的方法是使用from multiprocessing import Manager 你设置了一个任务队列：

mgr = Manager()

task_queue = mgr.Queue()

将其作为参数传递给进程：

gmat_args.append([gmat_arg, task_queue])
...
pool = Pool(processes=nrunp, maxtasksperchild=20)
...
results = pool.map(run_gmat, gmat_args, chunksize=ninstances)

进程写入池：

def run_gmat(args):
    q = args[1]

    scriptname = os.path.basename(args[0])

    proc = sp.Popen(['gmat', '-m', '-ns', '-x', '-r', str(args[0])])

    (outs, errors) = proc.communicate(timeout=cpto)

    outs = outs.decode('UTF-8')

    q.put(filter_outs(outs, scriptname))

然后回到主进程，读取队列并记录它：

while 1:

    qout = task_queue.get(cpto)   

    logging.info(qout)

    if task_queue.qsize() < 1:

        break

我不能说这是完美的，我正在运行 4000 多个作业，并且出现 2 - 3 次超时，这似乎与文件 I/O 相关（作业在完成时会写入大量报告文件）。为了防止挂起，我捕获了超时异常，将最终的标准输出、标准错误记录到队列中并终止作业。我丢失了报告文件，但我可以在队列中的日志中看到超时，因此只需重新运行这 2-3 个作业即可。

【讨论】：