使用外部命令多处理数千个文件

Question

我想从 Python 为大约 8000 个文件启动一个外部命令。每个文件都独立于其他文件进行处理。唯一的限制是在处理完所有文件后继续执行。我有 4 个物理核心，每个都有 2 个逻辑核心（multiprocessing.cpu_count() 返回 8）。我的想法是使用一个由四个并行独立进程组成的池，这些进程将在 8 个内核中的 4 个上运行。这样我的机器应该可以同时使用。

这是我一直在做的事情：

import multiprocessing
import subprocess
import os
from multiprocessing.pool import ThreadPool


def process_files(input_dir, output_dir, option):
    pool = ThreadPool(multiprocessing.cpu_count()/2)
    for filename in os.listdir(input_dir):  # about 8000 files
        f_in = os.path.join(input_dir, filename)
        f_out = os.path.join(output_dir, filename)
        cmd = ['molconvert', option, f_in, '-o', f_out]
        pool.apply_async(subprocess.Popen, (cmd,))
    pool.close()
    pool.join()


def main():
    process_files('dir1', 'dir2', 'mol:H')
    do_some_stuff('dir2')
    process_files('dir2', 'dir3', 'mol:a')
    do_more_stuff('dir3')

对一批 100 个文件进行顺序处理需要 120 秒。上面概述的多处理版本（函数process_files）批处理只需要 20 秒。但是，当我对整组 8000 个文件运行 process_files 时，我的电脑会挂起并且一小时后没有解冻。

我的问题是：

1) 我认为ThreadPool 应该初始化一个进程池（确切地说，这里是multiprocessing.cpu_count()/2 进程）。但是，我的计算机挂断了 8000 个文件而不是 100 个文件，这表明可能没有考虑到池的大小。要么，要么我做错了什么。你能解释一下吗？

2) 当每个独立进程都必须启动外部命令时，这是在 Python 下启动独立进程的正确方法，并且所有资源都不会被处理占用吗？

Answer 1

我认为您的基本问题是subprocess.Popen 的使用。该方法在返回之前不等待命令完成。由于函数会立即返回（即使命令仍在运行），所以就您的 ThreadPool 而言，该函数已完成，并且它可以产生另一个......这意味着您最终会产生 8000 个左右的进程。

使用subprocess.check_call 可能会更好：

Run command with arguments.  Wait for command to complete.  If
the exit code was zero then return, otherwise raise
CalledProcessError.  The CalledProcessError object will have the
return code in the returncode attribute.

所以：

def process_files(input_dir, output_dir, option):
    pool = ThreadPool(multiprocessing.cpu_count()/2)
    for filename in os.listdir(input_dir):  # about 8000 files
        f_in = os.path.join(input_dir, filename)
        f_out = os.path.join(output_dir, filename)
        cmd = ['molconvert', option, f_in, '-o', f_out]
        pool.apply_async(subprocess.check_call, (cmd,))
    pool.close()
    pool.join()

如果您真的不关心退出代码，那么您可能需要subprocess.call，它不会在进程退出代码非零的情况下引发异常。

Answer 2

如果您使用的是 Python 3，我会考虑使用 concurrent.futures.ThreadPoolExecutor 的 map 方法。

或者，您可以自己管理子流程列表。

下面的例子定义了一个函数来启动ffmpeg 来将视频文件转换为 Theora/Vorbis 格式。它为每个启动的子进程返回一个 Popen 对象。

def startencoder(iname, oname, offs=None):
    args = ['ffmpeg']
    if offs is not None and offs > 0:
        args += ['-ss', str(offs)]
    args += ['-i', iname, '-c:v', 'libtheora', '-q:v', '6', '-c:a',
            'libvorbis', '-q:a', '3', '-sn', oname]
    with open(os.devnull, 'w') as bb:
        p = subprocess.Popen(args, stdout=bb, stderr=bb)
    return p

在主程序中，代表正在运行的子进程的Popen对象列表是这样维护的。

outbase = tempname()
ogvlist = []
procs = []
maxprocs = cpu_count()
for n, ifile in enumerate(argv):
    # Wait while the list of processes is full.
    while len(procs) == maxprocs:
        manageprocs(procs)
    # Add a new process
    ogvname = outbase + '-{:03d}.ogv'.format(n + 1)
    procs.append(startencoder(ifile, ogvname, offset))
    ogvlist.append(ogvname)
# All jobs have been submitted, wail for them to finish.
while len(procs) > 0:
    manageprocs(procs)

因此，只有在运行的子进程少于核心时才会启动新进程。多次使用的代码被分离到manageprocs函数中。

def manageprocs(proclist):
    for pr in proclist:
        if pr.poll() is not None:
            proclist.remove(pr)
    sleep(0.5)

对sleep 的调用用于防止程序在循环中旋转。