使用外部命令多處理數千個文件

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

一個連續的治療需要120秒爲一個批次的100個文件。上面概述的多處理版本（功能process_files）僅需20秒。但是，當我在整個8000個文件集上運行process_files時，我的PC掛起並在一小時後不凍結。

我的問題是：

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，我會考慮使用map方法concurrent.futures.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的調用用於防止程序在循環中旋轉。