7

我正在编写一些代码，这些代码对大量（数万到数十万个数值积分）问题进行了一些相当繁重的数值工作。幸运的是，这些集成是令人尴尬的并行，因此很容易使用 Pool.map() 将工作拆分到多个内核。

现在，我有一个具有以下基本工作流程的程序：

#!/usr/bin/env python
from multiprocessing import Pool
from scipy import *
from my_parser import parse_numpy_array
from my_project import heavy_computation

#X is a global multidimensional numpy array
X = parse_numpy_array("input.dat")
param_1 = 0.0168
param_2 = 1.505

def do_work(arg):
  return heavy_computation(X, param_1, param_2, arg)

if __name__=='__main__':
  pool = Pool()
  arglist = linspace(0.0,1.0,100)
  results = Pool.map(do_work,arglist)
  #save results in a .npy file for analysis
  save("Results", [X,results])

由于 X、param_1 和 param_2 是硬编码的，并且对于池中的每个进程都以完全相同的方式初始化，所以这一切都很好。现在我的代码已经工作了，我想让它让用户在运行时输入文件名、param_1 和 param_2，而不是硬编码。

需要注意的一点是，X、param_1 和 param_2 在工作完成时不会被修改。由于我不修改它们，我可以在程序开始时做这样的事情：

import sys
X = parse_numpy_array(sys.argv[1])
param_1 = float(sys.argv[2])
param_2 = float(sys.argv[3])

这样就可以解决问题，但是由于此代码的大多数用户都在 Windows 机器上运行代码，所以我宁愿不走命令行参数的路线。

我真正想做的是这样的：

X, param_1, param_2 = None, None, None

def init(x,p1, p2)
  X = x
  param_1 = p1
  param_2 = p2

if __name__=='__main__':
  filename = raw_input("Filename> ")
  param_1 = float(raw_input("Parameter 1: "))
  param_2 = float(raw_input("Parameter 2: "))
  X = parse_numpy_array(filename)
  pool = Pool(initializer = init, initargs = (X, param_1, param_2,))
  arglist = linspace(0.0,1.0,100)
  results = Pool.map(do_work,arglist)
  #save results in a .npy file for analysis
  save("Results", [X,results])

但是，当 pool.map 调用发生时，这当然会失败并且 X/param_1/param_2 都是 None 。我对多处理很陌生，所以我不确定为什么对初始化程序的调用失败。有没有办法做我想做的事？有没有更好的方法来解决这个问题？我也看过使用共享数据，但根据我对文档的理解，它只适用于不包括 numpy 数组的 ctypes。对此的任何帮助将不胜感激。

4

2 回答 2

5

我有一个类似的问题。如果您只想阅读我的解决方案，请跳过一些行:) 我必须：

  • 在运行在它的不同部分的线程之间共享一个 numpy.array 并且......
  • 向 Pool.map 传递一个具有多个参数的函数。

我注意到：

  • numpy.array 的数据被正确读取，但是......
  • numpy.array 上未永久更改的更改
  • Pool.map 在处理 lambda 函数时遇到问题，或者在我看来是这样（如果您不清楚这一点，请忽略它）

我的解决方案是：

  • 使目标函数仅作为参数列表
  • 使目标函数返回修改后的数据，而不是直接尝试在 numpy.array 上写入

我知道您的 do_work 函数已经返回了计算数据，因此您只需修改 to_work 以接受一个列表（包含 X、param_1、param_2 和 arg）作为参数，并在传递之前以这种格式将输入打包到目标函数它到 Pool.map。

这是一个示例实现：

def do_work2(args):
    X,param_1,param_2,arg = args
    return heavy_computation(X, param_1, param_2, arg)

现在您必须在调用 do_work 函数之前将输入打包。你的主要变成：

if __name__=='__main__':
   filename = raw_input("Filename> ")
   param_1 = float(raw_input("Parameter 1: "))
   param_2 = float(raw_input("Parameter 2: "))
   X = parse_numpy_array(filename)
   # now you pack the input arguments
   arglist = [[X,param1,param2,n] for n in linspace(0.0,1.0,100)]
   # consider that you're not making 100 copies of X here. You're just passing a reference to it
   results = Pool.map(do_work2,arglist)
   #save results in a .npy file for analysis
   save("Results", [X,results])
于 2012-09-23T06:28:29.067 回答
-2

为了使您的最后一个想法可行，我认为您可以在 if 语句中修改它们之前使用 global 关键字简单地制作X、param_1和全局变量。param_2所以添加以下内容：

global X
global param_1
global param_2

直接在if __name__ == '__main__'.

于 2012-08-15T02:55:38.390 回答