parallel-processing - Julia 使用 Distributed 和 SharedArrays 和 @sync/@async 进行多次分配

Question

我试图了解如何使用 Distributed 包与 SharedArrays 一起使用 julia 执行并行操作。举个例子，我采用了一个简单的蒙特卡洛平均法

using Distributed 
using SharedArrays
using Statistics

const NWorkers = 2
const Ns = Int(1e6)


function parallelRun()

    addprocs(NWorkers)
    procsID = workers()

    A = SharedArray{Float64,1}(Ns)
    println("starting loop")

    for i=1:2:Ns
        #parallel block
        @sync for p=1:NWorkers
                @async A[i+p-1] = remotecall_fetch(rand,procsID[p]);
        end
    end

    println(mean(A))
end


function singleRun()
    A = zeros(Ns)
    for i=1:Ns
        A[i] = rand()
    end
 
    println(mean(A))
end

但是，如果我@time 这两个函数我都会得到

julia> @time singleRun()
0.49965531193003165
  0.009762 seconds (17 allocations: 7.630 MiB)
julia> @time parallelRun()
0.4994892300029917
 46.319737 seconds (66.99 M allocations: 2.665 GiB, 1.01% gc time)

特别是在并行版本中有更多的分配，这使得代码更慢。
我错过了什么吗？
顺便说一下，我使用 @sync 和 @async 的原因(即使在这个框架中不需要，因为每个样本都可以按随机顺序计算)只是因为我想应用相同的策略来数值求解抛物线偏微分方程在线的东西

    for t=1:time_steps

        #parallel block
        @sync for p=1:NWorkers
                @async remotecall(make_step_PDE,procsID[p],p);
        end
    end

由 p 索引的每个 worker 应该在我的方程的一组不相交的索引上工作。
提前致谢

Answer 1

您的代码存在以下问题:

您正在为 i 的每个值生成一个远程任务。 a 这只是昂贵的，最终需要很长时间。基本上经验法则是使用 @distributed用于跨工作器进行负载平衡的宏，这只会均匀地共享工作。

从不放 addprocs在您的工作函数中，因为每次运行它时，每次添加新进程时 - 产生一个新的 Julia 进程也需要大量时间，这包括在您的测量中。实际上，这意味着您要运行 addprocs在执行初始化的脚本的某些部分，或者通过启动 julia 添加进程。处理 -p或 --machine-file参数

最后，始终运行 @time总是两次 - 在第一次测量中 @time还测量编译时间，分布式环境中的编译比单个进程中的编译时间长得多。

你的函数应该或多或少像这样

using Distributed, SharedArrays
addprocs(4)
@everywhere using Distributed, SharedArrays
function parallelRun(Ns)
    A = SharedArray{Float64,1}(Ns)
    @sync @distributed for i=1:Ns
         A[i] = rand();
    end
    println(mean(A))
end

您还可以考虑在工作人员之间完全拆分数据。在某些情况下，这不太容易出错，并且允许您分布在许多节点上:

using Distributed, DistributedArrays
addprocs(4)
@everywhere using Distributed, DistributedArrays
function parallelRun2(Ns)
    d = dzeros(Ns) #creates an array distributed evenly on all workers
    @sync @distributed for i in 1:Ns
        p = localpart(d)
        p[((i-1) % Int(Ns/nworkers())+1] = rand()
    end
    println(mean(d))
end