如何在Julia中运行简单的并行数组赋值操作

Question

我必须多次求解微分方程组，迭代一个参数。为此，我在参数列表上运行一个循环，并存储每个参数的解决方案(在时间值数组中计算)。所以我有一个2D数组，在其中存储解决方案(每一行都代表参数的值)。

现在，由于任何迭代都与另一个迭代无关，我想并行进行。

这是我的代码：

using DifferentialEquations
using SharedArrays
using DelimitedFiles
using Distributed
function tf(x,w)
return x*sin(w*x)
end
function sys!(dv,v,w,t)
dv[1] = w*v[1]
dv[2] = tf(v[1],w)
end
times = LinRange(0.1,2,25)
params = LinRange(0.1,1.2,100)
sols = SharedArray{Float64,2}((length(times),length(params)))
@distributed for i=1:length(params)
println(i)
init_val = [1.0,1.0]
tspan = (0.0,2.0)
prob = ODEProblem(sys!,init_val,tspan,params[i])
sol = solve(prob)
sols[:,i] .= sol(times)[2,:]
end
writedlm("output.txt",sols)

现在，当我在循环中不加前缀@distributed的情况下运行时，它可以完美地运行。

然而，当我运行这段代码时，println语句不起作用，尽管存储了文件"output.txt"，但它充满了零。

我正在从命令行以这种方式运行此代码

julia -p 4 trycode.jl

虽然存储了文件"output.txt"，但它不显示任何输出，只工作一分钟，什么也不做。就好像从未进入过循环。

我真的很感激关于如何设置这个简单的并行循环的一些帮助。

Answer 1

正如Bill所说，在Julia中有两种主要的并行方式：线程模型，它在Julia 1.3中引入，通过Threads.@threads宏进行共享内存并行，以及使用Distributed.@distributed宏进行分布式处理，它在不同的Julia进程之间进行并行。

线程肯定更接近于"automagic"并行加速，只需最少或不需要重写代码，而且通常是一个很好的选择，尽管必须注意确保运行的任何操作都是线程安全的，所以一定要检查结果是否相同。

由于您的问题最初是关于@distributed并行性的，所以让我也回答这个问题。如果你做@distributed并行，思考正在发生的事情的最简单的心理模型(我相信)是想象你在完全独立的Julia REPL中运行代码。

以下是适用于@distributed型号的代码版本：

using Distributed
addprocs(2)
using SharedArrays
using DelimitedFiles
@everywhere begin 
using DifferentialEquations
tf(x,w) = x*sin(w*x)
function sys!(dv,v,w,t)
dv[1] = w*v[1]
dv[2] = tf(v[1],w)
end
times = LinRange(0.1,2,25)
params = LinRange(0.1,1.2,100)
end
sols = SharedArray{Float64,2}((length(times),length(params)))
@sync @distributed for i=1:length(params)
println(i)
init_val = [1.0,1.0]
tspan = (0.0,2.0)
prob = ODEProblem(sys!,init_val,tspan,params[i])
sol = solve(prob)
sols[:,i] .= sol(times)[2,:]
end
sols

发生了什么变化？

• 我在脚本的开头添加了addprocs(2)。如果您在执行Julia时使用p -2(或任何您想要的进程数量)启动它，那么这是没有必要的，但我经常发现，当它直接在代码中显式设置并行环境时，更容易对代码进行推理。请注意，目前线程不可能这样做，即在启动Julia之前需要设置JULIA_NUM_THREADS环境变量，并且在启动并运行后无法更改线程数。

• 然后，我将代码的比特移动到@everywhere begin ... end块中。这本质上是同时在所有进程上运行块中包含的操作。回到运行单独Julia实例的心理模型，您必须查看@distributed循环中的内容，并确保所有函数和变量实际上都在所有进程上定义。因此，例如，为了确保每个进程都知道ODEProblem是什么，您需要对所有进程执行using DifferentialEquations。

• 最后，我将@sync添加到分布式循环中。@distributed的文档中引用了这一点。使用for循环运行@distributed宏会为分布式执行生成一个异步绿线程(Task)句柄，并向前移动到下一行。由于您希望等待执行真正完成，因此需要同步@sync。原始代码的问题是，在不等待绿色线程完成(同步)的情况下，它会吞下错误并立即返回，这就是sol数组为空的原因。如果您运行原始代码，并且只添加@sync，您就可以看到这一点，然后您将得到一个TaskFailedException: on worker 2 - UndefVarError: #sys! not defined，它告诉您的工作进程不知道您在主进程上定义的函数。在实践中，您几乎总是希望执行@sync，除非您计划并行运行许多这样的分布式循环。在分布式循环(循环的@distributed (func) for i in 1:1000形式)中使用聚合器函数时，也不需要@sync关键字

现在这里的最佳解决方案是什么？答案是我不知道。@threads是一个很好的选择，可以在不重写代码的情况下快速并行化线程安全操作，并且仍在积极开发和改进中，因此在未来可能会变得更好。分布式标准库中还有pmap，它为您提供了额外的选项，但这个答案已经足够长了！根据我的个人经验，没有什么能取代(1)思考问题和(2)基准执行。您需要考虑的是问题的运行时(包括总操作和要分发的每个单独操作)以及消息传递/内存访问需求。

好的一面是，虽然你可能需要花一些精力思考问题，Julia有很多很好的选择来充分利用每一种硬件情况，从一台有两个核心的破旧笔记本电脑(就像我正在打字的那台)到多节点超高性能集群(这使Julia成为为数不多的实现PB级性能的编程语言之一——尽管公平地说，这比我或Bill的答案要棘手一点：)

Answer 2

您能从线程化for而不是@distributed for中获益吗？本作品(Julia 1.4)：

using DifferentialEquations
using SharedArrays
using DelimitedFiles
using Distributed
function tf(x,w)
return x*sin(w*x)
end
function sys!(dv,v,w,t)
dv[1] = w*v[1]
dv[2] = tf(v[1],w)
end
times = LinRange(0.1,2,25)
params = LinRange(0.1,1.2,100)
sols = SharedArray{Float64,2}((length(times),length(params)))
@Threads.threads for i=1:length(params)
println(i)
init_val = [1.0,1.0]
tspan = (0.0,2.0)
prob = ODEProblem(sys!,init_val,tspan,params[i])
sol = solve(prob)
sols[:,i] .= sol(times)[2,:]
end
writedlm("output.txt",sols)