使用多处理工作池

Question

我编写了以下代码,以使我的第二个CPU核心工作.代码基本上做的是首先在目录层次结构中找到所需的"海"文件,然后执行外部脚本集来处理这些二进制"海"文件,以生成50到100个文本和二进制文件.由于问题的标题以平行的方式提出,以提高处理速度.

这个问题起源于我们在IPython用户列表中长期讨论的标题为" 无法启动ipcluster ".从我对IPython并行处理功能的实验开始.

问题是我无法正常运行此代码.如果包含"sea"文件的文件夹仅包含"sea"文件,则脚本将在不完全执行外部脚本运行的情况下完成其执行.(假设我有30-50个外部脚本要运行,但是我的多处理启用脚本只有在执行这些外部脚本链中的第一个脚本后才会耗尽.)有趣的是,如果我在已处理的文件夹(即"海"文件)上运行此脚本预先处理并且输出文件已经在该文件夹中)然后它运行,但是这次我的速度提升到大约2.4到2.7X,相对于线性处理时序.由于笔记本电脑中只有Core 2 Duo 2.5 Ghz CPU,因此不是很理想.虽然我有一个CUDA驱动的GPU,但它与我目前的并行计算难度无关:)

您认为这个问题的来源是什么？

感谢您的所有意见和建议.

#!/usr/bin/env python

from multiprocessing import Pool
from subprocess import call
import os


def find_sea_files():

   file_list, path_list = [], []
   init = os.getcwd()

   for root, dirs, files in os.walk('.'):
      dirs.sort()
      for file in files:
          if file.endswith('.sea'):
              file_list.append(file)
              os.chdir(root)
              path_list.append(os.getcwd())
              os.chdir(init)

   return file_list, path_list


def process_all(pf):
   os.chdir(pf[0])
   call(['postprocessing_saudi', pf[1]])


if __name__ == '__main__':
   pool = Pool(processes=2)              # start 2 worker processes
   files, paths = find_sea_files()
   pathfile = [[paths[i],files[i]] for i in range(len(files))]
   pool.map(process_all, pathfile)

Answer 1

我首先要对工作流程的情况有一个更好的感觉.如果需要,多处理模块会为其子进程记录日志.由于您已经简化了代码以缩小问题范围,我只需要使用一些打印语句进行调试,就像这样(或者您可以使用PrettyPrint pf数组):

def process_all(pf):
   print "PID: ", os.getpid()
   print "Script Dir: ", pf[0]
   print "Script: ", pf[1]
   os.chdir(pf[0])
   call(['postprocessing_saudi', pf[1]])


if __name__ == '__main__':
   pool = Pool(processes=2)
   files, paths = find_sea_files()
   pathfile = [[paths[i],files[i]] for i in range(len(files))]
   pool.map(process_all, pathfile, 1) # Ensure the chunk size is 1
   pool.close()
   pool.join()

我用2.6.4完成的Python版本.