Julia pmap 速度 - 并行处理 - 动态规划

Question

我正在尝试加快 Julia (v0.6.0) 中动态编程问题的矩阵填充速度，但使用 pmap 似乎无法获得太多额外的速度。这与我差不多一年前发布的这个问题有关：Filling a matrix using parallel processing in Julia。那时我能够在一些很大的帮助下加速串行处理，现在我正在尝试从 Julia 的并行处理工具中获得额外的速度。

对于串行处理案例，我使用了一个 3 维矩阵（本质上是一组大小相等的矩阵，由第一维索引）并在第一维上进行迭代。不过，我想尝试一下pmap，以更有效地迭代矩阵集。

这里是代码设置。要将pmap 与下面的v_iter 函数一起使用，我将三维矩阵转换为字典对象，字典键等于第一维中的索引值（下面代码中的v_dict，gcc等于第一维大小）。 v_iter 函数将其他字典对象（下面的E_opt_dict 和gridpoint_m_dict）作为附加输入：

function v_iter(a,b,c)
   diff_v = 1
   while diff_v>convcrit
     diff_v = -Inf

     #These lines efficiently multiply the value function by the Markov transition matrix, using the A_mul_B function
     exp_v       = zeros(Float64,gkpc,1)
     A_mul_B!(exp_v,a[1:gkpc,:],Zprob[1,:])
     for j=2:gz
       temp=Array{Float64}(gkpc,1)
       A_mul_B!(temp,a[(j-1)*gkpc+1:(j-1)*gkpc+gkpc,:],Zprob[j,:])
       exp_v=hcat(exp_v,temp)
     end    

     #This tries to find the optimal value of v
     for h=1:gm
       for j=1:gz
         oldv = a[h,j]
         newv = (1-tau)*b[h,j]+beta*exp_v[c[h,j],j]
         a[h,j] = newv
         diff_v = max(diff_v, oldv-newv, newv-oldv)
       end
     end
   end
end

gz =  9  
gp =  13  
gk =  17  
gcc =  5  
gm    = gk * gp * gcc * gz
gkpc  = gk * gp * gcc
gkp = gk*gp
beta  = ((1+0.015)^(-1))
tau        = 0.35
Zprob = [0.43 0.38 0.15 0.03 0.00 0.00 0.00 0.00 0.00; 0.05 0.47 0.35 0.11 0.02 0.00 0.00 0.00 0.00; 0.01 0.10 0.50 0.30 0.08 0.01 0.00 0.00 0.00; 0.00 0.02 0.15 0.51 0.26 0.06 0.01  0.00 0.00; 0.00 0.00 0.03 0.21 0.52 0.21 0.03 0.00 0.00 ; 0.00 0.00  0.01  0.06 0.26 0.51 0.15 0.02 0.00 ; 0.00 0.00 0.00 0.01 0.08 0.30 0.50 0.10 0.01 ; 0.00 0.00 0.00 0.00 0.02 0.11 0.35 0.47 0.05; 0.00 0.00 0.00 0.00 0.00 0.03 0.15 0.38 0.43]
convcrit = 0.001   # chosen convergence criterion

E_opt                  = Array{Float64}(gcc,gm,gz)    
fill!(E_opt,10.0)

gridpoint_m   = Array{Int64}(gcc,gm,gz)
fill!(gridpoint_m,fld(gkp,2)) 

v_dict=Dict(i => zeros(Float64,gm,gz) for i=1:gcc)
E_opt_dict=Dict(i => E_opt[i,:,:] for i=1:gcc)
gridpoint_m_dict=Dict(i => gridpoint_m[i,:,:] for i=1:gcc) 

对于并行处理，我执行了以下两个命令：

wp = CachingPool(workers())
addprocs(3)
pmap(wp,v_iter,values(v_dict),values(E_opt_dict),values(gridpoint_m_dict))

...产生了这种表现：

135.626417 seconds (3.29 G allocations: 57.152 GiB, 3.74% gc time)

然后我尝试改为串行处理：

for i=1:gcc
    v_iter(v_dict[i],E_opt_dict[i],gridpoint_m_dict[i])
end

...并获得了更好的性能。

128.263852 seconds (3.29 G allocations: 57.101 GiB, 4.53% gc time)

这也给了我与在原始 3 维对象上运行 v_iter 大致相同的性能：

v=zeros(Float64,gcc,gm,gz)
for i=1:gcc
    v_iter(v[i,:,:],E_opt[i,:,:],gridpoint_m[i,:,:])
end

我知道并行处理涉及设置时间，但是当我增加gcc 的值时，串行和并行的处理时间仍然大致相同。这似乎是并行处理的一个很好的候选，因为不需要在工作人员之间进行消息传递！但我似乎无法让它有效地工作。

Answer 1

在添加工作进程之前创建CachingPool。因此，您的缓存池传递给pmap 告诉它只使用一个工作人员。 您可以通过运行wp.workers 来简单地检查它，您将看到类似Set([1]) 的内容。 因此它应该是： addprocs(3) wp = CachingPool(workers()) 您还可以考虑运行 Julia -p 命令行参数，例如julia -p 3 然后你可以跳过addprocs(3) 命令。

除此之外，您的 for 和 pmap 循环并不相同。 Julia Dict 对象是一个哈希图，与其他语言类似，它不提供元素顺序之类的东西。因此，在您的 for 循环中，您可以保证获得相同的匹配 i-th 元素，而使用 values 值的顺序不需要与原始顺序匹配（并且您可以为这三个变量中的每一个设置不同的顺序在pmap 循环中）。 由于您的Dicts 的键只是从1 到gcc 的数字，因此您应该简单地使用数组。您可以使用与 Python 非常相似的生成器。例如，而不是 v_dict=Dict(i => zeros(Float64,gm,gz) for i=1:gcc) 采用 v_dict_a = [zeros(Float64,gm,gz) for i=1:gcc]

希望对您有所帮助。

Answer 2

根据@Przemyslaw Szufeul 的有用建议，我已将正确执行并行处理的代码放在下面。运行一次后，我在运行时间上取得了实质性的提升： 77.728264 seconds (181.20 k allocations: 12.548 MiB)

除了重新排序 wp 命令和使用 Przemyslaw 推荐的生成器之外，我还将 v_iter 重铸为匿名函数，以避免在代码周围散布@everywhere 以将函数和数据提供给工人。

我还在v_iter 函数中添加了return a，并将v_a 设置为等于pmap 的输出，因为您不能通过引用传递到远程对象。

addprocs(3)
v_iter = function(a,b,c)
   diff_v = 1
   while diff_v>convcrit
     diff_v = -Inf

     #These lines efficiently multiply the value function by the Markov transition matrix, using the A_mul_B function
     exp_v       = zeros(Float64,gkpc,1)
     A_mul_B!(exp_v,a[1:gkpc,:],Zprob[1,:])
     for j=2:gz
       temp=Array{Float64}(gkpc,1)
       A_mul_B!(temp,a[(j-1)*gkpc+1:(j-1)*gkpc+gkpc,:],Zprob[j,:])
       exp_v=hcat(exp_v,temp)
     end    

     #This tries to find the optimal value of v
     for h=1:gm
       for j=1:gz
         oldv = a[h,j]
         newv = (1-tau)*b[h,j]+beta*exp_v[c[h,j],j]
         a[h,j] = newv
         diff_v = max(diff_v, oldv-newv, newv-oldv)
       end
     end
   end
  return a
end

gz =  9  
gp =  13  
gk =  17  
gcc =  5  
gm    = gk * gp * gcc * gz
gkpc  = gk * gp * gcc
gkp   =gk*gp
beta  = ((1+0.015)^(-1))
tau        = 0.35
Zprob = [0.43 0.38 0.15 0.03 0.00 0.00 0.00 0.00 0.00; 0.05 0.47 0.35 0.11 0.02 0.00 0.00 0.00 0.00; 0.01 0.10 0.50 0.30 0.08 0.01 0.00 0.00 0.00; 0.00 0.02 0.15 0.51 0.26 0.06 0.01  0.00 0.00; 0.00 0.00 0.03 0.21 0.52 0.21 0.03 0.00 0.00 ; 0.00 0.00  0.01  0.06 0.26 0.51 0.15 0.02 0.00 ; 0.00 0.00 0.00 0.01 0.08 0.30 0.50 0.10 0.01 ; 0.00 0.00 0.00 0.00 0.02 0.11 0.35 0.47 0.05; 0.00 0.00 0.00 0.00 0.00 0.03 0.15 0.38 0.43]
convcrit = 0.001   # chosen convergence criterion

E_opt                  = Array{Float64}(gcc,gm,gz)    
fill!(E_opt,10.0)

gridpoint_m   = Array{Int64}(gcc,gm,gz)
fill!(gridpoint_m,fld(gkp,2)) 

v_a=[zeros(Float64,gm,gz) for i=1:gcc]
E_opt_a=[E_opt[i,:,:] for i=1:gcc]
gridpoint_m_a=[gridpoint_m[i,:,:] for i=1:gcc]

wp = CachingPool(workers())
v_a = pmap(wp,v_iter,v_a,E_opt_a,gridpoint_m_a)