python - Multiprocessing.Pool.map 返回错误值

Question

我想计算这个for loop：

import numpy as np
import cmath
import multiprocessing as mp
from functools import partial


def main():

    nodes = 53
    f_nom = 50
    U = [1.000005327453146, 1.0000057162412561, 1.0000070448192537, 1.0000069227537496, 1.0000069227537496, 1.0000041605235332, 1.0000035222393044, 1.000004263922246, 1.0000047513736152, 1.0000047550593771, 1.0000049581713173, 1.000004263922246, 1.0000039401443763, 1.0000026446387185, 1.0000014323088564, 1.0000026446387185, 1.0000036287590721, 1.0000026446387185, 1.0000039401443768, 1.0000014323088589, 1.0000039401443768, 1.0000014323088553, 1.0000014323088553, 1.0000030231390993, 1.0000014323088615, 1.0000014323088615, 1.0000014323088606, 1.0000014323088604, 1.0000014323088597, 1.0000014323088606, 1.0000014323088604, 1.0000014323088597, 1.0000014323088593, 1.000001432308859, 1.000001432308858, 1.000001590333777, 1.0000020451347542, 1.000001432308861, 1.0000014323088602, 1.000001432308863, 1.00000143230886, 1.000001432308859, 1.000001432308863, 1.000001432308863, 1.000001432308859, 1.000001432308863, 1.00000143230886, 1.0000014323088615, 1.000001432308859, 1.0000014323088615, 1.0000057853629756, 1.0000059478472048, 1.0000000056539258, 1.0]
    theta = [-7.355656043185493e-06, -3.464594992534604e-06, -1.0583594144863342e-05, -9.626585410799617e-06, -9.626585410799618e-06, -5.6945086783273145e-06, -2.674752704682853e-06, -5.841699103904964e-06, -6.535596440474819e-06, -3.118569281977461e-06, -6.829976913326792e-06, -5.841699103904964e-06, -2.825199109428268e-06, -2.3588148383740883e-06, -1.1617097845584577e-06, -2.3588148383740883e-06, -4.937528901600708e-06, -2.3588148383740883e-06, -2.8251991094452968e-06, -1.1617097837371578e-06, -2.8251991094452976e-06, -1.1617097849706262e-06, -1.1617097849706262e-06, -4.075412032614385e-06, -1.1617097820641042e-06, -1.1617097820641045e-06, -1.16170978206629e-06, -1.161709782064495e-06, -1.1617097822154165e-06, -1.16170978206629e-06, -1.161709782064495e-06, -1.1617097822154165e-06, -1.1617097834417375e-06, -1.1617097836565141e-06, -1.1617097839483619e-06, -2.0357655204310186e-06, -1.766840164692082e-06, -1.1617097820892918e-06, -1.1617097821105786e-06, -1.1617097821630116e-06, -1.1617097822743808e-06, -1.1617097836888828e-06, -1.161709782163012e-06, -1.1617097821630119e-06, -1.1617097836179758e-06, -1.1617097821630119e-06, -1.1617097822743808e-06, -1.1617097820641042e-06, -1.161709783617976e-06, -1.1617097820641047e-06, -8.007497618793153e-06, -8.238796094878206e-06, -1.972012101284986e-08, 0.0]
    U_start = [396.98922, 397.590444, 396.6735, 396.663384, 396.625768, 397.134352, 397.85536, 397.132184, 397.123008, 397.680784, 397.074752, 397.131104, 397.78544, 397.975768, 397.642204, 398.124984, 397.186264, 397.87854, 397.766736, 397.505908, 397.706156, 397.640524, 397.643368, 397.294984, 397.816152, 397.855528, 397.792792, 397.785664, 397.736036, 397.802636, 397.783008, 397.746864, 397.641188, 397.56948, 397.635652, 397.676168, 397.972024, 397.835444, 397.76222, 397.440376, 397.740988, 397.405996, 397.40994, 397.429992, 397.336748, 397.466372, 397.744532, 397.846052, 397.311572, 397.728728, 396.841804, 396.767284, 397.679944, 400.]
    t = np.linspace(0, 0.1, int(0.1 * 20000), dtype=np.float64)
    wave_line = np.zeros((3, nodes, len(t)))
    wave_line_mp = np.zeros((3, nodes, len(t)))
    looprange = 5

    for i in range(looprange):
        harm_order = i + 2
        for n in range(nodes):
            U_line = cmath.polar(cmath.rect(1.0, 0.0) - cmath.rect(U[n], theta[n]))
            wave_line[0, n] += np.sqrt(2) / np.sqrt(3) * U_line[0] * U_start[n] * np.sin((t * 2 * np.pi * f_nom + U_line[1]) * harm_order)
            wave_line[1, n] += np.sqrt(2) / np.sqrt(3) * U_line[0] * U_start[n] * np.sin((t * 2 * np.pi * f_nom + U_line[1] - 2 / 3 * np.pi) * harm_order)
            wave_line[2, n] += np.sqrt(2) / np.sqrt(3) * U_line[0] * U_start[n] * np.sin((t * 2 * np.pi * f_nom + U_line[1] + 2 / 3 * np.pi) * harm_order)

        print(i, wave_line[0][0][0])
        wave_line[:] = 0


if __name__ == '__main__':
    main()

输出 1：

0 -0.0027971102683270466
1 0.000897146546025977
2 0.0017446246602729246
3 -0.0029438493445009956
4 0.0017089460869939772

在一个multiprocessing.Pool.map函数中：

import numpy as np
import cmath
import multiprocessing as mp
from functools import partial


def main():

    nodes = 53
    f_nom = 50
    U = [1.000005327453146, 1.0000057162412561, 1.0000070448192537, 1.0000069227537496, 1.0000069227537496, 1.0000041605235332, 1.0000035222393044, 1.000004263922246, 1.0000047513736152, 1.0000047550593771, 1.0000049581713173, 1.000004263922246, 1.0000039401443763, 1.0000026446387185, 1.0000014323088564, 1.0000026446387185, 1.0000036287590721, 1.0000026446387185, 1.0000039401443768, 1.0000014323088589, 1.0000039401443768, 1.0000014323088553, 1.0000014323088553, 1.0000030231390993, 1.0000014323088615, 1.0000014323088615, 1.0000014323088606, 1.0000014323088604, 1.0000014323088597, 1.0000014323088606, 1.0000014323088604, 1.0000014323088597, 1.0000014323088593, 1.000001432308859, 1.000001432308858, 1.000001590333777, 1.0000020451347542, 1.000001432308861, 1.0000014323088602, 1.000001432308863, 1.00000143230886, 1.000001432308859, 1.000001432308863, 1.000001432308863, 1.000001432308859, 1.000001432308863, 1.00000143230886, 1.0000014323088615, 1.000001432308859, 1.0000014323088615, 1.0000057853629756, 1.0000059478472048, 1.0000000056539258, 1.0]
    theta = [-7.355656043185493e-06, -3.464594992534604e-06, -1.0583594144863342e-05, -9.626585410799617e-06, -9.626585410799618e-06, -5.6945086783273145e-06, -2.674752704682853e-06, -5.841699103904964e-06, -6.535596440474819e-06, -3.118569281977461e-06, -6.829976913326792e-06, -5.841699103904964e-06, -2.825199109428268e-06, -2.3588148383740883e-06, -1.1617097845584577e-06, -2.3588148383740883e-06, -4.937528901600708e-06, -2.3588148383740883e-06, -2.8251991094452968e-06, -1.1617097837371578e-06, -2.8251991094452976e-06, -1.1617097849706262e-06, -1.1617097849706262e-06, -4.075412032614385e-06, -1.1617097820641042e-06, -1.1617097820641045e-06, -1.16170978206629e-06, -1.161709782064495e-06, -1.1617097822154165e-06, -1.16170978206629e-06, -1.161709782064495e-06, -1.1617097822154165e-06, -1.1617097834417375e-06, -1.1617097836565141e-06, -1.1617097839483619e-06, -2.0357655204310186e-06, -1.766840164692082e-06, -1.1617097820892918e-06, -1.1617097821105786e-06, -1.1617097821630116e-06, -1.1617097822743808e-06, -1.1617097836888828e-06, -1.161709782163012e-06, -1.1617097821630119e-06, -1.1617097836179758e-06, -1.1617097821630119e-06, -1.1617097822743808e-06, -1.1617097820641042e-06, -1.161709783617976e-06, -1.1617097820641047e-06, -8.007497618793153e-06, -8.238796094878206e-06, -1.972012101284986e-08, 0.0]
    U_start = [396.98922, 397.590444, 396.6735, 396.663384, 396.625768, 397.134352, 397.85536, 397.132184, 397.123008, 397.680784, 397.074752, 397.131104, 397.78544, 397.975768, 397.642204, 398.124984, 397.186264, 397.87854, 397.766736, 397.505908, 397.706156, 397.640524, 397.643368, 397.294984, 397.816152, 397.855528, 397.792792, 397.785664, 397.736036, 397.802636, 397.783008, 397.746864, 397.641188, 397.56948, 397.635652, 397.676168, 397.972024, 397.835444, 397.76222, 397.440376, 397.740988, 397.405996, 397.40994, 397.429992, 397.336748, 397.466372, 397.744532, 397.846052, 397.311572, 397.728728, 396.841804, 396.767284, 397.679944, 400.]
    t = np.linspace(0, 0.1, int(0.1 * 20000), dtype=np.float64)
    wave_line = np.zeros((3, nodes, len(t)))
    wave_line_mp = np.zeros((3, nodes, len(t)))
    looprange = 5

    func_part = partial(worker_function, nodes=nodes, f_nom=f_nom, U=U, theta=theta, U_start=U_start, t=t, wave_line_mp=wave_line_mp)
    wave_line_results = mp.Pool(1).map(func_part, range(looprange))

    print('')
    for idx, wave_line_mp in wave_line_results:
        print(idx, wave_line_mp[0][0][0])


def worker_function(i=None, nodes=None, f_nom=None, U=None, theta=None, U_start=None, t=None, wave_line_mp=None):
    harm_order = i + 2
    wave_line_mp[:] = 0
    for n in range(nodes):
        U_line = cmath.polar(cmath.rect(1.0, 0.0) - cmath.rect(U[n], theta[n]))
        wave_line_mp[0, n] += np.sqrt(2) / np.sqrt(3) * U_line[0] * U_start[n] * np.sin((t * 2 * np.pi * f_nom + U_line[1]) * harm_order)
        wave_line_mp[1, n] += np.sqrt(2) / np.sqrt(3) * U_line[0] * U_start[n] * np.sin((t * 2 * np.pi * f_nom + U_line[1] - 2 / 3 * np.pi) * harm_order)
        wave_line_mp[2, n] += np.sqrt(2) / np.sqrt(3) * U_line[0] * U_start[n] * np.sin((t * 2 * np.pi * f_nom + U_line[1] + 2 / 3 * np.pi) * harm_order)

    print(i, wave_line_mp[0][0][0])

    return i, wave_line_mp

if __name__ == '__main__':
    main()

输出 2：

0 -0.0027971102683270466
1 0.000897146546025977
2 0.0017446246602729246
3 -0.0029438493445009956
4 0.0017089460869939772

0 0.000897146546025977
1 0.000897146546025977
2 -0.0029438493445009956
3 -0.0029438493445009956
4 0.0017089460869939772

如您所见，其中的输出 (2.1)worker_function是正确的（类似于上述 for 循环的输出 (1)），而返回的wave_line_results输出 (2.2) 则不正确。似乎存在值重叠，可能是Pool函数的调度问题？

标签： pythonmultiprocessing

Answer 1

似乎您正在将（可变）数组传递给wave_line_mp函数然后更改它，并且您应该在它的单独副本上进行操作才能这样做。一个简单的单行修复方法是wave_line_mp = wave_line_mp.copy()在 的开头插入worker_function，但更好的解决方案可能是将其从函数的参数列表中完全删除（当然也从调用它的位置删除），然后移动该行：

    wave_line_mp = np.zeros((3, nodes, len(t)))

从里面main到开始worker_function。（这是有效的，因为变量nodes和t被传递给 内部的同名变量worker_function。）您还可以删除该行wave_line_mp[:] = 0，这就是多余的。

这样做，您将获得所需的输出：

0 -0.00279711026833
1 0.000897146546026
2 0.00174462466027
3 -0.0029438493445
4 0.00170894608699

0 -0.00279711026833
1 0.000897146546026
2 0.00174462466027
3 -0.0029438493445
4 0.00170894608699