AnyMath 文档
Notebook

使用SharedArray的分布式计算

导言

SharedArray是一种特殊类型的数组，它允许同一设备上的多个进程直接访问内存中的相同数据，而不必在进程之间复制它。

与传递给另一个进程时复制的常规数组不同。, SharedArray 使用系统共享内存。 这意味着所有参与的工作流都会看到同一块RAM，因此一个进程所做的任何更改都会立即对其他人可见。

本文提出了一个最小但说明性的例子：通过两个独立的过程对随机数矩阵的元素进行分布式求和。 此模型问题说明了基本操作。:addprocs, @spawnat, wait, SharedArray. 它还展示了从顺序计算方案切换到并行计算方案的潜在性能优势。

加入图书馆

连接必要的库:

  • Distributed-用于组织多处理;
  • TickTock-测量执行时间。
In [ ]: 
for lib in ["Distributed", "TickTock"]
    try
        eval(Meta.parse("using $lib"))
    catch ex
        import Pkg
        Pkg.add(lib)
        eval(Meta.parse("using $lib"))
    end
end

准备两个工作流程

让我们使用以下命令运行两个额外的工作流 addprocs(2) 并通过以下方式显示活动工作人员列表，确保他们成功注册 workers(). 我们将得到一个向量，其标识符将进一步用于寻址任务。

In [ ]: 
addprocs(2)
workers()
Out[0]:
2-element Vector{Int64}:
 2
 3

创建SharedArray对象

在共享内存中创建两个数据结构:

*第一个 – SA，一个用随机数填充的二维6x2共享数组；它起到源数据的作用。
*第二个 – SR，用零初始化的两个元素的公共向量;它被设计为累积由不同过程计算的结果。

使用 SharedArray 无需在进程之间复制数据。

In [ ]: 
using SharedArrays
SA = convert(SharedArray{Float64, 2}, rand(6,2))
Out[0]:
6×2 SharedMatrix{Float64}:
 0.783205  0.327205
 0.432006  0.914733
 0.390884  0.326474
 0.305342  0.319739
 0.242857  0.0980725
 0.59598   0.954856
In [ ]: 
SR = SharedArray{Float64,1}(2)
Out[0]:
2-element SharedVector{Float64}:
 0.0
 0.0

分布式计算测试

让我们来看看分布式计算的基本机制：使用宏 @spawnat 我们将向进程2发送一个任务来计算第一列元素的总和。 SA 并将结果写入 SR[1].

In [ ]: 
@spawnat 2 SR[1] = sum(SA[:,1])
Out[0]:
Future(2, 1, 18, ReentrantLock(), nothing)
In [ ]: 
display(SR)
Out[0]:
2-element SharedVector{Float64}:
 0.0
 0.0

我们将向流程3发送第二列的类似任务。 让我们注意一个重要的细节。: @spawnat 返回一个对象 Future 并且它不会阻止主进程，计算是异步执行的。 因此，立即检查 SR 它还可以显示零值—这是预期的行为，说明了分布式任务的异步性质。

In [ ]: 
@spawnat 3 SR[2] = sum(SA[:,2])
Out[0]:
Future(3, 1, 19, ReentrantLock(), nothing)

分布式和单进程计算的性能比较

现在让我们用迭代次数N=10⁷进行一个更困难的实验。 首先，我们将在主进程中按顺序执行计算，然后分布在工作流2和3上，等待它们的完成。 wait.([f1, f2]). 在这两种情况下，我们将使用 tick() 和 tock(). 一个比较将显示分布式版本更快：在这样的工作量下，并行组织的开销被同时在不同内核上运行两个周期的好处反复超过。

In [ ]: 
N = 10^7
Out[0]:
1000000
In [ ]: 
# 一个过程
SR[1:2] .= 0.0
display(SR)
tick()
SR[1] = begin x = 0.0; for i in 1:N x += sum(SA[:,1]); end; x; end;
SR[2] = begin x = 0.0; for i in 1:N x += sum(SA[:,2]); end; x; end;
display(SR)
tock()

2-element SharedVector{Float64}:
 0.0
 0.0
[ Info:  started timer at: 2026-10-15T05:17:04.778
2-element SharedVector{Float64}:
 2.750272925045633e6
 2.941078833171551e6
[ Info:          6.015979337s: 6 seconds, 15 milliseconds
In [ ]: 
# 并行计算
SR[1:2] .= 0.0
display(SR)
tick()
f1 = @spawnat 2 SR[1] = begin x = 0.0; for i in 1:N x += sum(SA[:,1]); end; x; end;
f2 = @spawnat 3 SR[2] = begin x = 0.0; for i in 1:N x += sum(SA[:,2]); end; x; end;
@time wait.([f1, f2])
display(SR)
tock()

2-element SharedVector{Float64}:
 0.0
 0.0
[ Info:  started timer at: 2026-10-15T05:17:02.136
  1.802403 seconds (200.93 k allocations: 10.613 MiB, 10.99% compilation time)
2-element SharedVector{Float64}:
 2.750272925045633e6
 2.941078833171551e6
[ Info:          2.172321257s: 2 seconds, 172 milliseconds

清理资源

我们将使用命令完成工作流程 rmprocs(workers()) 并确保只有主进程保留在系统中。 专责小组 procs() 返回向量 1. 正确释放资源是使用分布式系统的重要部分，因为不完整的工作程序工作流会占用内存和处理器时间。

In [ ]: 
rmprocs(workers())
procs()
Out[0]:
1-element Vector{Int64}:
 1

结论

此示例涵盖了在AnyMath中使用分布式计算的所有主要阶段：启动工作流，创建共享数组，异步分配任务以及评估加速。 模块的捆绑 Distributed 和喜欢 SharedArray 在没有显式消息传递的情况下提供并行数据处理，返回的作业为 @spawnat，需要通过同步 wait 计算的顺序很重要。 即使在一个微不足道的求和问题上，也实现了几乎三倍的加速（≈2.17s与≈6.02s）。

所提出的方案是一系列任务的基础：大型数据阵列的并行聚合、蒙特卡洛方法中的独立模型运行、机器学习中的超参数枚举和集成训练、多变量工程分析，以及科学计算中求解偏微分方程和基因组分析。

因此，考虑的方法形成了一个基本的工具包，直接推广到更复杂的场景，其中需要有效使用多核系统和减少计数时间。