用于数值计算的多处理python函数

Question

希望能够在这里获得一些帮助，并行化我的python代码，我一直在苦苦挣扎了一段时间，并以我尝试的方式提出了几个错误，目前正在运行的代码将会涉及2-3小时完成，代码如下：

import numpy as np 
from scipy.constants import Boltzmann, elementary_charge as kb, e 
import multiprocessing 
from functools import partial 
Tc = 9.2 
x = [] 
g= [] 
def Delta(T): 
''' 
Delta(T) takes a temperature as an input and calculates a 
temperature dependent variable based on Tc which is defined as a 
global parameter 
''' 
    d0 = (pi/1.78)*kb*Tc 
    D0 = d0*(np.sqrt(1-(T**2/Tc**2))) 
    return D0 

def element_in_sum(T, n, phi): 
    D = Delta(T) 
    matsubara_frequency = (np.pi * kb * T) * (2*n + 1) 
    factor_d = np.sqrt((D**2 * cos(phi/2)**2) + matsubara_frequency**2) 
    element = ((2 * D * np.cos(phi/2))/ factor_d) * np.arctan((D * np.sin(phi/2))/factor_d) 
    return element 

def sum_elements(T, M, phi): 
''' 
    sum_elements(T,M,phi) is the most computationally heavy part 
    of the calculations, the larger the M value the more accurate the 
    results are. 
    T: temperature 
    M: number of steps for matrix calculation the larger the more accurate the calculation 
phi: The phase of the system can be between 0- pi 
''' 
    X = list(np.arange(0,M,1)) 
    Y = [element_in_sum(T, n, phi) for n in X] 
    return sum(Y) 

def KO_1(M, T, phi): 
    Iko1Rn = (2 * np.pi * kb * T /e) * sum_elements(T, M, phi) 
    return Iko1Rn 

def main(): 
    for j in range(1, 92): 
     T = 0.1*j 
     for i in range(1, 314): 
      phi = 0.01*i 
      pool = multiprocessing.Pool() 
      result = pool.apply_async(KO_1,args=(26000, T, phi,)) 
      g.append(result) 
      pool.close() 
      pool.join()  
    A = max(g); 
    x.append(A) 
    del g[:]  

我的做法是试图发送KO1功能成多池，但我要么得到一个Pickling错误或too many files open，任何帮助是极大的赞赏，如果多是错误的做法我很乐意的任何指南。

Answer 1

我还没有测试过你的代码，但你可以做几件事情来改善它。首先，不要不必要地创建数组。当它只能使用一个生成器时，sum_elements会创建三个类似数组的对象。首先，np.arange创建一个numpy数组，然后list函数创建一个列表对象，然后列表理解创建另一个列表。该功能是它应该做的4倍。

正确的方法来实现它（在python3）将是：

def sum_elements(T, M, phi): 
    return sum(element_in_sum(T, n, phi) for n in range(0, M, 1)) 

如果使用python2，与xrange更换range。 这篇技巧可能会帮助你编写任何python脚本。

此外，尝试更好地利用多处理。看来你需要做的是创建一个multiprocessing.Pool对象一次，并使用pool.map函数。

主要功能应该是这样的：我为了用一个元组

def job(args): 
    i, j = args 
    T = 0.1*j 
    phi = 0.01*i 
    return K0_1(26000, T, phi) 

def main():   
    pool = multiprocessing.Pool(processes=4) # You can change this number 
    x = [max(pool.imap(job, ((i, j) for i in range(1, 314)) for j in range(1, 92)] 

通知到多个参数传递给工作。

Answer 2

这不是问题的答案，但如果我可以，我会建议如何使用简单的numpy数组操作来加速代码。看看下面的代码：

import numpy as np 
from scipy.constants import Boltzmann, elementary_charge as kb, e 
import time 
Tc = 9.2 
RAM = 4*1024**2 # 4GB 

def Delta(T): 
    ''' 
    Delta(T) takes a temperature as an input and calculates a 
    temperature dependent variable based on Tc which is defined as a 
    global parameter 
    ''' 
    d0 = (np.pi/1.78)*kb*Tc 
    D0 = d0*(np.sqrt(1-(T**2/Tc**2))) 
    return D0 

def element_in_sum(T, n, phi): 
    D = Delta(T) 
    matsubara_frequency = (np.pi * kb * T) * (2*n + 1) 
    factor_d = np.sqrt((D**2 * np.cos(phi/2)**2) + matsubara_frequency**2) 
    element = ((2 * D * np.cos(phi/2))/ factor_d) * np.arctan((D * np.sin(phi/2))/factor_d) 
    return element 


def KO_1(M, T, phi): 
    X = np.arange(M)[:,np.newaxis,np.newaxis] 
    sizeX = int((float(RAM)/sum(T.shape))/sum(phi.shape)/8) #8byte 
    i0 = 0 
    Iko1Rn = 0. * T * phi 
    while (i0+sizeX) <= M: 
     print "X = %i"%i0 
     indices = slice(i0, i0+sizeX) 
     Iko1Rn += (2 * np.pi * kb * T /e) * element_in_sum(T, X[indices], phi).sum(0) 
     i0 += sizeX 
    return Iko1Rn 

def main(): 
    T = np.arange(0.1,9.2,0.1)[:,np.newaxis] 
    phi = np.linspace(0,np.pi, 361) 
    M = 26000 
    result = KO_1(M, T, phi) 
    return result, result.max() 

T0 = time.time() 
r, rmax = main() 
print time.time() - T0 

它在我的电脑上运行了超过20秒。必须小心不要使用太多的内存，这就是为什么仍然存在一个循环，其构造有点复杂，只能使用X片段。如果存在足够的内存，则不是必需的。

还应该注意，这只是加速的第一步。仍然可以使用例如只是及时编译或cython。