优化循环性能：并行化

Question

优化循环性能：并行化

multithreadingperformanceloopsparallel-processingjulia

3

我正在努力理解Julia的并行化选项。我将随机过程建模成马尔可夫链。由于链是独立的复制品，外层循环是独立的——这使得问题变得非常适合并行处理。

我尝试了@distributed和@threads两种解决方案，两者都可以正常运行，但速度并没有更快。

以下是我代码的简化版本（顺序执行）：

function dummy(steps = 10000, width = 100, chains = 4)
    out_N = zeros(steps, width, chains)
    initial = zeros(width)
    for c = 1:chains
        # print("c=$c\n")
        N = zeros(steps, width)
        state = copy(initial)
        N[1,:] = state
        for i = 1:steps
            state = state + rand(width)
            N[i,:] = state
        end
        out_N[:,:,c] = N
    end
    return out_N
end

如何正确地将此问题并行化以提高性能？

- Levasco

3

看起来你正在使用按行主序索引数组。Julia的数组是以列主序排列的，因此访问N[:, i]将更快。而且，如果避免使用向量化代码，性能可能会更好。最后，如果手动指定线程本地的随机数生成器，则多线程的rand通常最快。 - mbauman

谢谢你的建议，@Matt。我会尝试着去实践它们。向量化编程是我从我的第一门语言R中养成的习惯，我应该加以改进。 - Levasco

2个回答

1

你可以使用 @distributed 宏来并行运行进程。

@everywhere using Distributed, SharedArrays

addprocs(4)

@everywhere function inner_loop!(out_N, chain_number,steps,width)
    N = zeros(steps, width)
    state = zeros(width)
    for i = 1:steps
        state .+= rand(width)
        N[i,:] .= state
    end
    out_N[:,:,chain_number] .= N
    nothing
end

function dummy(steps = 10000, width = 100, chains = 4)
    out_N = SharedArray{Float64}((steps, width, chains); pids = collect(1:4))
    @sync for c = 1:chains
        # print("c=$c\n")
        @spawnat :any inner_loop!(out_N, c, steps,width)
    end
    sdata(out_N)
end

- Paul Dydyshko

1

嗨保罗，谢谢你的回答。你的代码确实非常快，但语义上是错误的。它生成的是随机矩阵，而不是马尔可夫链。 - Levasco

你在问题中的代码片段也产生随机矩阵，而非马尔可夫链。 - Paul Dydyshko

嗯...不是吗？它生成（虚拟的）MC，因为 N[i+1,:] 依赖于 N[i,:]。你的矩阵条目是完全独立的元素。检查输出。 - Levasco

我修改了代码以满足您的需求。现在它应该可以正常工作了。 - Paul Dydyshko

1

男女们，这两个代码都是完全错误的！第一个代码不会起作用，因为远程进程不知道 inner_loop!，结果也不会被存储！数组只会单向传递到远程进程。应该使用 SharedArray。宏 @async 会启动一个绿色线程，只能用于加速 IO 密集型循环（例如网络爬虫、文件传输等）。 - Przemyslaw Szufel

显示剩余6条评论

网页内容由stack overflow 提供, 点击上面的

可以查看英文原文，
原文链接

- Przemyslaw Szufel · Accepted Answer

以下是正确的操作方法（在撰写本答案时，其他答案不起作用-请参见我的评论）。

我将使用比问题中略微简单的例子（但非常相似）。

1. 非并行版本（基准方案）

using Random
const m = MersenneTwister(0);

function dothestuff!(out_N, N, ic, m)
    out_N[:, ic] .= rand(m, N)
end

function dummy_base(m=m, N=100_000,c=256)
    out_N = Array{Float64}(undef,N,c)
    for ic in 1:c
        dothestuff!(out_N, N, ic, m)
    end
    out_N 
end

测试：

julia> using BenchmarkTools; @btime dummy_base();
  106.512 ms (514 allocations: 390.64 MiB)

2. 使用线程并行化

#remember to run before starting Julia:
# set JULIA_NUM_THREADS=4
# OR (Linux)
# export JULIA_NUM_THREADS=4

using Random

const mt = MersenneTwister.(1:Threads.nthreads());
# required for older Julia versions, look still good in later versions :-)

function dothestuff!(out_N, N, ic, m)
    out_N[:, ic] .= rand(m, N)
end
function dummy_threads(mt=mt, N=100_000,c=256)
    out_N = Array{Float64}(undef,N,c)
    Threads.@threads for ic in 1:c
        dothestuff!(out_N, N, ic, mt[Threads.threadid()])
    end
    out_N 
end

让我们来测试性能：

julia> using BenchmarkTools; @btime dummy_threads();
  46.775 ms (535 allocations: 390.65 MiB)

3. 使用进程实现并行化（在单台机器上）

using Distributed

addprocs(4) 

using Random, SharedArrays
@everywhere using Random, SharedArrays, Distributed
@everywhere Random.seed!(myid())

@everywhere function dothestuff!(out_N, N, ic)
    out_N[:, ic] .= rand(N)
end
function dummy_distr(N=100_000,c=256)
    out_N = SharedArray{Float64}(N,c)
    @sync @distributed for ic in 1:c
        dothestuff!(out_N, N, ic)
    end
    out_N 
end

性能（注意，进程间通信需要一些时间，因此对于小型计算，线程通常更好）：

julia> using BenchmarkTools; @btime dummy_distr();
  62.584 ms (1073 allocations: 45.48 KiB)