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

Question

我想从Python中为大约8000个文件启动一个外部命令。每个文件都独立于其他文件进行处理。唯一的限制是在处理完所有文件后继续执行。我有4个物理核，每个物理核有2个逻辑核（multiprocessing.cpu\u 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秒。上述多处理版本（函数处理\u文件）只需20秒即可完成批处理。但是，当我对整套8000个文件运行process_files时，我的电脑会挂起，一小时后不会解除冻结。

我的问题是：

1） 我认为ThreadPool应该初始化一个进程池（确切地说是multiprocessing.cpu\u count（）/2进程池）。然而，我的电脑挂起了8000个文件，而不是100个，这表明可能没有考虑池的大小。要么这样，要么我做错了什么。你能解释一下吗？

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

Answer 1

如果您使用Python 3，我会考虑使用<代码> map < /COD>方法>代码>并发。期货ThreadPoolExecutor。

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

以下示例定义了一个函数，用于启动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可防止程序在循环中旋转。

Answer 2

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

使用子流程可能会更幸运。检查呼叫：

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()

如果您真的不关心退出代码，那么您可能需要子流程。调用，如果流程中存在非零退出代码，则不会引发异常。