并行计算

对于多线程和并行计算的新手来说，首先了解Jullia所提供的 不同层级并行是非常有用的。这里我们主要将其分成三类：

  1. Julia 协程（绿色线程）
  2. 多线程
  3. 多核心或分布式处理

我们首先考虑 Julia 任务 Task（也就是协程）以及其它依赖于 Julia 运行时库的模块，通过运行时库，我们无需手动与操作系统的调度进行交互就可以挂起和恢复计算，并且对 Task 间内部通信拥有完全控制。Julia 同样支持利用一些操作在 Task 间进行通信，比如 wait 以及 fetch。另外，通信和数据同步是通过管道 Channel 完成的，它也为 Task 间内部通信提供了渠道。

Julia also supports experimental multi-threading, where execution is forked and an anonymous function is run across all threads. Known as the fork-join approach, parallel threads execute independently, and must ultimately be joined in Julia’s main thread to allow serial execution to continue. Multi-threading is supported using the Base.Threads module that is still considered experimental, as Julia is not yet fully thread-safe. In particular segfaults seem to occur during I/O operations and task switching. As an up-to-date reference, keep an eye on the issue tracker. Multi-Threading should only be used if you take into consideration global variables, locks and atomics, all of which are explained later.

最后我们将介绍 Julia 的分布式和并行计算的实现方法。鉴于以科学计算为主要目的，Julia 底层实现上提供了跨多核或多机器对任务并行的接口。同时我们还将介绍一些有用的分布式编程的外部包，比如 MPI.jl 以及 DistributedArrays.jl。

协程

Julia 的并行编程平台采用协程任务 Tasks (aka Coroutines) 来进行多个计算之间的切换。为了表示轻量线程之间的执行顺序，必须提供一种通信的原语。Julia 提供了函数 Channel(func::Function, ctype=Any, csize=0, taskref=nothing)，根据 func 创建 task，然后将其绑定到一个新的大小为 csize、 类型为 ctype 的管道上，并调度 task。Channel 可以当作是一种 task 之间通信的方式，Channel{T}(sz::Int) 会创建一个类型为 T、 大小为 sz 的管道。无论何时发起一个通信操作，如 fetch 或 wait，当前 task 都会挂起，然后调度器会选择其它 task 去执行，在一个 task 等待的事件结束之后会重新恢复执行。

对于许多问题而言，并不需要直接考虑 task。不过，task 可以用来同时等待多个事件，从而实现动态调度。在动态调度的过程中，程序可以决定计算什么，或者根据其它任务执行结束的时间决定接下来在哪里执行计算。这对于不可预测或不平衡的计算量来说是必须的，因为我们只希望给那些已经完成了其当前任务的进程分配更多的任务。

管道

在 流程控制 中有关 Task 的部分，已经讨论了如何协调多个函数的执行。Channel 可以很方便地在多个运行中的 task 传递数据，特别是那些涉及 I/O 的操作。

典型的 I/O 操作包括读写文件、访问 web 服务、执行外部程序等。在所有这些场景中，如果其它 task 可以在读取文件（等待外部服务或程序执行完成）时继续执行，那么总的执行时间能够得到大大提升。

一个管道可以形象得看做是一个管子，一端可读，另一端可写：

  • 不同的 task 可以通过 put! 往同一个 channel 并发地写入。

  • 不同的 task 也可以通过 take! 从同一个 channel 并发地取数据

  • 举个例子：

    1. # Given Channels c1 and c2,
    2. c1 = Channel(32)
    3. c2 = Channel(32)
    5. # and a function `foo` which reads items from c1, processes the item read
    6. # and writes a result to c2,
    7. function foo()
    8.     while true
    9.         data = take!(c1)
    10.         [...]               # process data
    11.         put!(c2, result)    # write out result
    12.     end
    13. end
    15. # we can schedule `n` instances of `foo` to be active concurrently.
    16. for _ in 1:n
    17.     @async foo()
    18. end

  • Channel 可以通过 Channel{T}(sz) 构造，得到的 channel 只能存储类型 T 的数据。如果 T 没有指定，那么 channel 可以存任意类型。sz 表示该 channel 能够存储的最大元素个数。比如 Channel(32) 得到的 channel 最多可以存储32个元素。而 Channel{MyType}(64) 则可以最多存储64个 MyType 类型的数据。

  • 如果一个 Channel 是空的，读取的 task(即执行 take! 的 task)会被阻塞直到有新的数据准备好了。

  • 如果一个 Channel 是满的，那么写入的 task(即执行 put! 的 task)则会被阻塞，直到 Channel 有空余。

  • isready 可以用来检查一个 channel 中是否有已经准备好的元素，而等待一个元素准备好 则用 wait

  • 一个 Channel 一开始处于开启状态，也就是说可以被 take! 读取和 put! 写入。close 会关闭一个 Channel，对于一个已经关闭的 Channel，put! 会失败，例如：

    1. julia> c = Channel(2);
    3. julia> put!(c, 1) # `put!` on an open channel succeeds
    4. 1
    6. julia> close(c);
    8. julia> put!(c, 2) # `put!` on a closed channel throws an exception.
    9. ERROR: InvalidStateException("Channel is closed.",:closed)
    10. Stacktrace:
    11. [...]

  • take! 和 fetch (只读取，不会将元素从 channel 中删掉)仍然可以从一个已经关闭的 channel 中读数据，直到 channel 被取空了为止。继续上面的例子：

    1. julia> fetch(c) # Any number of `fetch` calls succeed.
    2. 1
    4. julia> fetch(c)
    5. 1
    7. julia> take!(c) # The first `take!` removes the value.
    8. 1
    10. julia> take!(c) # No more data available on a closed channel.
    11. ERROR: InvalidStateException("Channel is closed.",:closed)
    12. Stacktrace:
    13. [...]

Channel 可以在 for 循环中遍历，此时，循环会一直运行直到 Channel 中有数据，遍历过程中会取遍加入到 Channel 中的所有值。一旦 Channel关闭或者取空了，for 循环就会终止。

例如，下面的 for 循环会等待新的数据：

  1. julia> c = Channel{Int}(10);
  3. julia> foreach(i->put!(c, i), 1:3) # add a few entries
  5. julia> data = [i for i in c]

而下面的则会返回已经读取的数据：

  1. julia> c = Channel{Int}(10);
  3. julia> foreach(i->put!(c, i), 1:3); # add a few entries
  5. julia> close(c);                    # `for` loops can exit
  7. julia> data = [i for i in c]
  8. 3-element Array{Int64,1}:
  9.  1
  10.  2
  11.  3

考虑这样一个用 channel 做 task 之间通信的例子。首先，起 4 个 task 来处理一个 jobs channel 中的数据。jobs 中的每个任务通过 job_id 来表示，然后每个 task 模拟读取一个 job_id，然后随机等待一会儿，然后往一个 results channel 中写入一个元组，它分别包含 job_id 和执行的时间，最后将结果打印出来：

  1. julia> const jobs = Channel{Int}(32);
  3. julia> const results = Channel{Tuple}(32);
  5. julia> function do_work()
  6.            for job_id in jobs
  7.                exec_time = rand()
  8.                sleep(exec_time)                # 模拟执行实际外部工作所需的时间
  9.                put!(results, (job_id, exec_time))
  10.            end
  11.        end;
  13. julia> function make_jobs(n)
  14.            for i in 1:n
  15.                put!(jobs, i)
  16.            end
  17.        end;
  19. julia> n = 12;
  21. julia> @async make_jobs(n); # 用 "n" 个 jobs 填充 jobs channel
  23. julia> for i in 1:4 # 启动 4 个 tasks 来并行的处理请求
  24.            @async do_work()
  25.        end
  27. julia> @elapsed while n > 0 # 打印结果
  28.            job_id, exec_time = take!(results)
  29.            println("$job_id finished in $(round(exec_time; digits=2)) seconds")
  30.            global n = n - 1
  31.        end
  32. 4 finished in 0.22 seconds
  33. 3 finished in 0.45 seconds
  34. 1 finished in 0.5 seconds
  35. 7 finished in 0.14 seconds
  36. 2 finished in 0.78 seconds
  37. 5 finished in 0.9 seconds
  38. 9 finished in 0.36 seconds
  39. 6 finished in 0.87 seconds
  40. 8 finished in 0.79 seconds
  41. 10 finished in 0.64 seconds
  42. 12 finished in 0.5 seconds
  43. 11 finished in 0.97 seconds
  44. 0.029772311

当前版本的 Julia 会将所有 task 分发到一个操作系统的线程，因此，涉及 I/O 的操作会从并行执行中获利，而计算密集型的 task 则会顺序地在单独这个线程上执行。未来 Julia 将支持在多个线程上调度 task，从而让计算密集型 task 也能从并行计算中获利。

Multi-Threading (Experimental)

除了 task 之外，Julia 还原生支持多线程。本部分内容是实验性的，未来相关接口可能会改变。

设置

Julia 默认启动一个线程执行代码，这点可以通过 Threads.nthreads() 来确认：

  1. julia> Threads.nthreads()
  2. 1

Julia 启动时的线程数可以通过环境变量 JULIA_NUM_THREADS 设置，下面启动4个线程：

  1. export JULIA_NUM_THREADS=4

(上面的代码只能在 Linux 和 OSX 系统中运行，如果你在以上平台中使用的是 C shell，那么将 export 改成 set，如果你是在 Windows 上运行，那么将 export 改成 set 同时启动 Julia 时指定 julia.exe 的完整路径。)

现在确认下确实有4个线程：

  1. julia> Threads.nthreads()
  2. 4

不过我们现在是在 master 线程，用 Threads.threadid 确认下：

  1. julia> Threads.threadid()
  2. 1

@threads宏

下面用一个简单的例子测试我们原生的线程，首先创建一个全零的数组：

  1. julia> a = zeros(10)
  2. 10-element Array{Float64,1}:
  3.  0.0
  4.  0.0
  5.  0.0
  6.  0.0
  7.  0.0
  8.  0.0
  9.  0.0
  10.  0.0
  11.  0.0
  12.  0.0

现在用4个线程模拟操作这个数组，每个线程往对应的位置写入线程ID。

Julia 用 Threads.@threads 宏实现并行循环，该宏加在 for 循环前面，提示 Julia 循环部分是一个多线程的区域：

  1. julia> Threads.@threads for i = 1:10
  2.            a[i] = Threads.threadid()
  3.        end

每次迭代会分配到各个线程，然后每个线程往对应位置写入线程 ID：

  1. julia> a
  2. 10-element Array{Float64,1}:
  3.  1.0
  4.  1.0
  5.  1.0
  6.  2.0
  7.  2.0
  8.  2.0
  9.  3.0
  10.  3.0
  11.  4.0
  12.  4.0

注意 Threads.@threads 并没有一个像 @distributed 一样的可选的 reduction 参数。

原子操作

Julia 支持访问和修改值的原子操作，即以一种线程安全的方式来避免竞态条件。一个值（必须是基本类型的，primitive type）可以通过 Threads.Atomic 来包装起来从而支持原子操作。下面看个例子：

  1. julia> i = Threads.Atomic{Int}(0);
  3. julia> ids = zeros(4);
  5. julia> old_is = zeros(4);
  7. julia> Threads.@threads for id in 1:4
  8.            old_is[id] = Threads.atomic_add!(i, id)
  9.            ids[id] = id
  10.        end
  12. julia> old_is
  13. 4-element Array{Float64,1}:
  14.  0.0
  15.  1.0
  16.  7.0
  17.  3.0
  19. julia> ids
  20. 4-element Array{Float64,1}:
  21.  1.0
  22.  2.0
  23.  3.0
  24.  4.0

如果不加 Atomic 的话，那么会因为竞态条件而得到错误的结果，下面是一个没有避免竞态条件的例子：

  1. julia> using Base.Threads
  3. julia> nthreads()
  4. 4
  6. julia> acc = Ref(0)
  7. Base.RefValue{Int64}(0)
  9. julia> @threads for i in 1:1000
  10.           acc[] += 1
  11.        end
  13. julia> acc[]
  14. 926
  16. julia> acc = Atomic{Int64}(0)
  17. Atomic{Int64}(0)
  19. julia> @threads for i in 1:1000
  20.           atomic_add!(acc, 1)
  21.        end
  23. julia> acc[]
  24. 1000

并非所有的原始类型都能放在 Atomic 标签内封装起来，支持的类型有Int8, Int16, Int32, Int64, Int128, UInt8, UInt16, UInt32, UInt64, UInt128, Float16, Float32, 以及 Float64。此外，Int128 和 UInt128 在 AAarch32 和 ppc64le 上不支持。

副作用和可变的函数参数

在使用多线程时，要非常小心使用了不纯的函数，例如，用到了以!结尾的函数，通常这类函数会修改其参数，因而是不纯的。此外还有些函数没有以 ! 结尾，其实也是有副作用的，比如 findfirst(regex, str) 就会改变 regex 参数，或者是 rand() 会修改 Base.GLOBAL_RNG:

  1. julia> using Base.Threads
  3. julia> nthreads()
  4. 4
  6. julia> function f()
  7.            s = repeat(["123", "213", "231"], outer=1000)
  8.            x = similar(s, Int)
  9.            rx = r"1"
  10.            @threads for i in 1:3000
  11.                x[i] = findfirst(rx, s[i]).start
  12.            end
  13.            count(v -> v == 1, x)
  14.        end
  15. f (generic function with 1 method)
  17. julia> f() # the correct result is 1000
  18. 1017
  20. julia> function g()
  21.            a = zeros(1000)
  22.            @threads for i in 1:1000
  23.                a[i] = rand()
  24.            end
  25.            length(unique(a))
  26.        end
  27. g (generic function with 1 method)
  29. julia> Random.seed!(1); g() # the result for a single thread is 1000
  30. 781

此时，应该重新设计代码来避免可能的竞态条件或者是使用 同步机制。

例如，为了修正上面 findfirst 的例子，每个线程都要拷贝一份 rx。

  1. julia> function f_fix()
  2.              s = repeat(["123", "213", "231"], outer=1000)
  3.              x = similar(s, Int)
  4.              rx = [Regex("1") for i in 1:nthreads()]
  5.              @threads for i in 1:3000
  6.                  x[i] = findfirst(rx[threadid()], s[i]).start
  7.              end
  8.              count(v -> v == 1, x)
  9.          end
  10. f_fix (generic function with 1 method)
  12. julia> f_fix()
  13. 1000

现在使用 Regex("1") 而不是 r"1" 来保证 Julia 对每个 rx 向量的元素都创建了一个 Regex 的对象。

rand 的例子更复杂点，因为我们需要保证每个线程使用的是不重叠的随机数序列，这可以简单地通过 Future.randjump 函数保证：

  1. julia> using Random; import Future
  3. julia> function g_fix(r)
  4.            a = zeros(1000)
  5.            @threads for i in 1:1000
  6.                a[i] = rand(r[threadid()])
  7.            end
  8.            length(unique(a))
  9.        end
  10. g_fix (generic function with 1 method)
  12. julia>  r = let m = MersenneTwister(1)
  13.                 [m; accumulate(Future.randjump, fill(big(10)^20, nthreads()-1), init=m)]
  14.             end;
  16. julia> g_fix(r)
  17. 1000

这里将 r 向量发送到 g_fix，由于生成多个随机数是很昂贵的操作，因此我们不希望每次执行函数都重复该操作。

@threadcall （实验性功能）

所有的 I/O task，计时器，REPL 命令等都是通过一个事件循环复用的一个系统线程。有一个补丁版的 libuv(http://docs.libuv.org/en/v1.x/) 提供了该功能，从而在同一个系统线程上协调调度多个 task。I/O task 和计时器在等待某个事件发生时，会隐式地退出（yield），而显式地调用 yield 则会允许其它 task 被调度。

因此，一个执行 ccall 的 task 会阻止 Julia 的调度器执行其它 task，直到调用返回，这种情况对于所有外部库的调用都存在，例外的情况是，某些自定义的C代码调用返回到了 Julia中（此时有可能 yield ）或者 C 代码执行了 jl_yield()（C 中等价的 yield）。

注意，尽管 Julia 的代码默认是单线程的，但是 Julia 调用的库可能会用到其内部的多线程，例如，BLAS 会在一台机器上使用尽可能多的线程。

@threadcall 就是要解决 ccall 会卡住主线程的这个问题，它会在一个额外的线程中调度 C 函数的执行，有一个默认大小为4的线程库用来做这个事情，该线程库的大小可以通过环境变量 UV_THREADPOOL_SIZE 控制。在等待一个空闲线程，以及在函数执行过程中某个线程空闲下来时，（主线程的事件循环中）正在请求的 task 会 yield 到其它 task，注意，@threadcall 并不会返回，直到执行结束。从用户的角度来看，就是一个和其它 Julia API 一样会阻塞的模块。

非常关键的一点是，被调用的函数不会再调用回 Julia。

@threadcall 在 Julia 未来的版本中可能会被移除或改变。

多核心或分布式处理

作为 Julia 标准库之一，Distributed 库提供了一种分布式内存并行计算的实现。

大多数现代计算机都拥有不止一个 CPU，而且多台计算机可以组织在一起形成一个集群。借助多个 CPU 的计算能力，许多计算过程能够更快地完成，这其中影响性能的两个主要因素分别是：CPU 自身的速度以及它们访问内存的速度。显然，在一个集群中，一个 CPU 访问同一个节点的 RAM 速度是最快的，不过令人吃惊的是，在一台典型的多核笔记本电脑上，由于访问主存和缓存的速度存在差别，类似的现象也会存在。因此，一个良好的多进程环境应该能够管理好某一片内存区域“所属”的CPU。Julia提供的多进程环境是基于消息传递来实现的，可以做到同时让程序在多个进程的不同内存区域中运行。

Julia 的消息传递机制与一些其它的框架不太一样，比如 MPI [1]。在 Julia 中，进程之间的通信通常是单向的，这里单向的意思是说，在实现2个进程之间的操作时，只需要显式地管理一个进程即可。此外，这些操作并不像是“发送消息”，“接收消息”这类操作，而是一些高阶的操作，比如调用用户定义的函数。

Julia 中的分布式编程基于两个基本概念：远程引用(remote references)和远程调用(remote calls)。远程引用是一个对象，任意一个进程可以通过它访问存储在某个特定进程上的对象。远程调用指是某个进程发起的执行函数的请求，该函数会在另一个（也可能是同一个）进程中执行。

Remote references come in two flavors: Future and RemoteChannel.

A remote call returns a Future to its result. Remote calls return immediately; the process that made the call proceeds to its next operation while the remote call happens somewhere else. You can wait for a remote call to finish by calling wait on the returned Future, and you can obtain the full value of the result using fetch.

对于 RemoteChannel 而言，它可以被反复写入。例如，多个进程可以通过引用同一个远程 Channel 来协调相互之间的操作。

每个进程都有一个对应的 id，提供 Julia 交互环境的进程的 id 永远是1。我们把用来执行并行任务的进程称为 “worker”，假如总共只有一个进程，那么进程1就被认为是 worker，否则，除了进程1以外的进程都称作 worker。

一起试一下吧。执行 julia -p n 就可以在本地起 n 个进程。一般来说，将 n 设成与你机器上（物理的内核数）CPU 个数一致比较合适。需要注意 -p 参数会隐式地载入 Distributed 模块。

  1. $ ./julia -p 2
  3. julia> r = remotecall(rand, 2, 2, 2)
  4. Future(2, 1, 4, nothing)
  6. julia> s = @spawnat 2 1 .+ fetch(r)
  7. Future(2, 1, 5, nothing)
  9. julia> fetch(s)
  10. 2×2 Array{Float64,2}:
  11.  1.18526  1.50912
  12.  1.16296  1.60607

remotecall 的第一个参数是想要调用的函数，第二个参数是执行函数的进程 id，其余的参数会喂给将要被调用的函数。在 Julia 中进行并行编程时，一般不需要显示地指明具体在哪个进程上执行，不过 remotecall 是一个相对底层的接口用来提供细粒度的管理。

可以看到，第一行代码请求进程2构建一个随机矩阵，第二行代码对该矩阵执行加一操作。每次执行的结果存在对应的 Future 中，即 r 和 s。这里 @spawnat 宏会在第一个参数所指定的进程中执行后面第二个参数中的表达式。

有时候，你可能会希望立即获取远程计算的结果，比如，在接下来的操作中就需要读取远程调用的结果，这时候你可以使用 remotecall_fetch 函数，其效果相当于 fetch(remotecall(...))，不过更高效些。

  1. julia> remotecall_fetch(getindex, 2, r, 1, 1)
  2. 0.18526337335308085

回忆下，这里 getindex(r,1,1) 相当于 r[1,1]，因此，上面的调用相当于获取 r 的第一个元素。

To make things easier, the symbol :any can be passed to [@spawnat], which picks where to do the operation for you:

  1. julia> r = @spawnat :any rand(2,2)
  2. Future(2, 1, 4, nothing)
  4. julia> s = @spawnat :any 1 .+ fetch(r)
  5. Future(3, 1, 5, nothing)
  7. julia> fetch(s)
  8. 2×2 Array{Float64,2}:
  9.  1.38854  1.9098
  10.  1.20939  1.57158

Note that we used 1 .+ fetch(r) instead of 1 .+ r. This is because we do not know where the code will run, so in general a fetch might be required to move r to the process doing the addition. In this case, @spawnat is smart enough to perform the computation on the process that owns r, so the fetch will be a no-op (no work is done).

(It is worth noting that @spawnat is not built-in but defined in Julia as a macro. It is possible to define your own such constructs.)

An important thing to remember is that, once fetched, a Future will cache its value locally. Further fetch calls do not entail a network hop. Once all referencing Futures have fetched, the remote stored value is deleted.

@async is similar to @spawnat, but only runs tasks on the local process. We use it to create a “feeder” task for each process. Each task picks the next index that needs to be computed, then waits for its process to finish, then repeats until we run out of indices. Note that the feeder tasks do not begin to execute until the main task reaches the end of the @sync block, at which point it surrenders control and waits for all the local tasks to complete before returning from the function. As for v0.7 and beyond, the feeder tasks are able to share state via nextidx because they all run on the same process. Even if Tasks are scheduled cooperatively, locking may still be required in some contexts, as in asynchronous I/O. This means context switches only occur at well-defined points: in this case, when remotecall_fetch is called. This is the current state of implementation and it may change for future Julia versions, as it is intended to make it possible to run up to N Tasks on M Process, aka M:N Threading#Models). Then a lock acquiring\releasing model for nextidx will be needed, as it is not safe to let multiple processes read-write a resource at the same time.

Code Availability and Loading Packages

对于想要并行执行的代码，需要所有对所有线程都可见。例如，在 Julia 命令行中输入以下命令：

  1. julia> function rand2(dims...)
  2.            return 2*rand(dims...)
  3.        end
  5. julia> rand2(2,2)
  6. 2×2 Array{Float64,2}:
  7.  0.153756  0.368514
  8.  1.15119   0.918912
  10. julia> fetch(@spawnat :any rand2(2,2))
  11. ERROR: RemoteException(2, CapturedException(UndefVarError(Symbol("#rand2"))
  12. Stacktrace:
  13. [...]

进程1知道函数 rand2 的存在，但进程2并不知道。

大多数情况下，你会从文件或者库中加载代码，在此过程中你可以灵活地控制哪个进程加载哪部分代码。假设有这样一个文件，DummyModule.jl，其代码如下：

  1. module DummyModule
  3. export MyType, f
  5. mutable struct MyType
  6.     a::Int
  7. end
  9. f(x) = x^2+1
  11. println("loaded")
  13. end

为了在所有进程中引用 MyType，DummyModule.jl 需要在每个进程中载入。单独执行 include("DummyModule.jl") 只会在一个线程中将其载入。为了让每个线程都载入它，可以用 @everywhere 宏来实现(启动 Julia 的时候，执行 julia -p 2)。

  1. julia> @everywhere include("DummyModule.jl")
  2. loaded
  3.       From worker 3:    loaded
  4.       From worker 2:    loaded

和往常一样，这么做并不会将 DummyModule 引入到每个线程的命名空间中，除非显式地使用 using 或 import。此外，显式地将 DummyModule 引入一个线程中，并不会影响其它线程：

  1. julia> using .DummyModule
  3. julia> MyType(7)
  4. MyType(7)
  6. julia> fetch(@spawnat 2 MyType(7))
  7. ERROR: On worker 2:
  8. UndefVarError: MyType not defined
  9. ⋮
  11. julia> fetch(@spawnat 2 DummyModule.MyType(7))
  12. MyType(7)

不过，我们仍然可以在已经包含(include)过 DummyModule 的进程中，发送 MyType 类型的实例，尽管此时该进程的命名空间中并没有 MyType 变量:

  1. julia> put!(RemoteChannel(2), MyType(7))
  2. RemoteChannel{Channel{Any}}(2, 1, 13)

文件代码还可以在启动的时候，通过 -L 参数指定，从而提前在多个进程中载入，然后通过一个 driver.jl 文件控制执行逻辑:

  1. julia -p <n> -L file1.jl -L file2.jl driver.jl

上面执行 driver.jl 的进程 id 为1，就跟提供交互式命令行的 Julia 进程一样。

最后，如果 DummyModule.jl 不是一个单独的文件，而是一个包的话，那么 using DummyModule 只会在所有线程中载入 DummyModule.jl，也就是说 DummyModule 只会在 using 执行的线程中被引入命名空间。

启动和管理 worker 进程

Julia 自带两种集群管理模式：

  • 本地集群，前面通过启动时指定 -p 参数就是这种模式
  • 跨机器的集群，通过 --machine-file 指定。这种模式采用没有密码的 ssh 登陆并对应的机器上（与 host 相同的路径下）启动 Julia 的 worker 进程。

addprocs, rmprocs, workers 这些函数可以分别用来对集群中的进程进行增加，删除和修改。

  1. julia> using Distributed
  3. julia> addprocs(2)
  4. 2-element Array{Int64,1}:
  5.  2
  6.  3

在 master 主线程中，Distributed 模块必须显式地在调用 addprocs 之前载入，该模块会自动在其它进程中可见。

需要注意的时，worker 进程并不会执行 ~/.julia/config/startup.jl 启动脚本，也不会同步其它进程的全局状态（比如全局变量，新定义的方法，加载的模块等）。

其它类型的集群可以通过自己写一个 ClusterManager 来实现，下面 集群管理器 部分会介绍。

数据转移

分布式程序的性能瓶颈主要是由发送消息和数据转移造成的，减少发送消息和转移数据的数量对于获取高性能和可扩展性至关重要，因此，深入了解 Julia 分布式程序是如何转移数据的非常有必要。

fetch can be considered an explicit data movement operation, since it directly asks that an object be moved to the local machine. @spawnat (and a few related constructs) also moves data, but this is not as obvious, hence it can be called an implicit data movement operation. Consider these two approaches to constructing and squaring a random matrix:

方法一：

  1. julia> A = rand(1000,1000);
  3. julia> Bref = @spawnat :any A^2;
  5. [...]
  7. julia> fetch(Bref);

方法二：

  1. julia> Bref = @spawnat :any rand(1000,1000)^2;
  3. [...]
  5. julia> fetch(Bref);

The difference seems trivial, but in fact is quite significant due to the behavior of @spawnat. In the first method, a random matrix is constructed locally, then sent to another process where it is squared. In the second method, a random matrix is both constructed and squared on another process. Therefore the second method sends much less data than the first.

In this toy example, the two methods are easy to distinguish and choose from. However, in a real program designing data movement might require more thought and likely some measurement. For example, if the first process needs matrix A then the first method might be better. Or, if computing A is expensive and only the current process has it, then moving it to another process might be unavoidable. Or, if the current process has very little to do between the @spawnat and fetch(Bref), it might be better to eliminate the parallelism altogether. Or imagine rand(1000,1000) is replaced with a more expensive operation. Then it might make sense to add another @spawnat statement just for this step.

全局变量

Expressions executed remotely via @spawnat, or closures specified for remote execution using remotecall may refer to global variables. Global bindings under module Main are treated a little differently compared to global bindings in other modules. Consider the following code snippet:

  1. A = rand(10,10)
  2. remotecall_fetch(()->sum(A), 2)

这个例子中 sum 必须已经在远程的线程中定义了。注意这里 A 是当前线程中的一个全局变量，起初 worker 2 在其 Main 中并没有一个叫做 A 的变量。上面代码中，将闭包 ()->sum(A) 发送到 worker 2 之后，会在 worker 2 中定义一个变量 Main.A，而且，Main.A 即使在执行完 remotecall_fetch 之后，仍然会存在与 worker 2 中。远程调用中包含的全局（这里仅仅指 Main 模块中的）引用会按如下方式管理：

  • 在全局调用中引用的全局绑定会在将要执行该调用的 worker 中被创建。

  • 全局常量仍然在远端结点定义为常量。

  • 全局绑定会在下一次远程调用中引用到的时候，当其值发生改变时，再次发送给目标 worker。此外，集群并不会所有结点的全局绑定。例如：

    1. A = rand(10,10)
    2. remotecall_fetch(()->sum(A), 2) # worker 2
    3. A = rand(10,10)
    4. remotecall_fetch(()->sum(A), 3) # worker 3
    5. A = nothing

    可以看到，A 作为全局变量在 worker 2中有定义，而 B 是一个局部变量，因而最后在 worker 2 中并没有 B 的绑定。 执行以上代码之后，worker 2 和 worker 3中的 Main.A 的值是不同的，同时，节点1上的值则为 nothing。

也许你也注意到了，在 master 主节点上被赋值为 nothing 之后，全局变量的内存会被回收，但在 worker 节点上的全局变量并没有被回收掉。执行 clear 可以手动将远端结点上的特定全局变量置为 nothing，然后对应的内存会被周期性的垃圾回收机制回收。

因此，在远程调用中，需要非常小心地引用全局变量。事实上，应当尽量避免引用全局变量，如果必须引用，那么可以考虑用let代码块将全局变量局部化：

  1. julia> A = rand(10,10);
  3. julia> remotecall_fetch(()->A, 2);
  5. julia> B = rand(10,10);
  7. julia> let B = B
  8.            remotecall_fetch(()->B, 2)
  9.        end;
  11. julia> @fetchfrom 2 InteractiveUtils.varinfo()
  12. name           size summary
  13. ––––––––– ––––––––– ––––––––––––––––––––––
  14. A         800 bytes 10×10 Array{Float64,2}
  15. Base                Module
  16. Core                Module
  17. Main                Module

可以看到，A 作为全局变量在 worker 2中有定义，而 B 是一个局部变量，因而最后在 worker 2 中并没有 B 的绑定。

并行的Map和Loop

Fortunately, many useful parallel computations do not require data movement. A common example is a Monte Carlo simulation, where multiple processes can handle independent simulation trials simultaneously. We can use @spawnat to flip coins on two processes. First, write the following function in count_heads.jl:

  1. function count_heads(n)
  2.     c::Int = 0
  3.     for i = 1:n
  4.         c += rand(Bool)
  5.     end
  6.     c
  7. end

函数 count_heads 只是简单地将 n 个随机 0-1 值累加，下面在两个机器上进行试验，并将结果叠加：

  1. julia> @everywhere include_string(Main, $(read("count_heads.jl", String)), "count_heads.jl")
  3. julia> a = @spawnat :any count_heads(100000000)
  4. Future(2, 1, 6, nothing)
  6. julia> b = @spawnat :any count_heads(100000000)
  7. Future(3, 1, 7, nothing)
  9. julia> fetch(a)+fetch(b)
  10. 100001564

上面的例子展示了一种非常常见而且有用的并行编程模式，在一些进程中执行多次独立的迭代，然后将它们的结果通过某个函数合并到一起，这个合并操作通常称作聚合(reduction)，也就是一般意义上的张量降维(tensor-rank-reducing)，比如将一个向量降维成一个数，或者是将一个 tensor 降维到某一行或者某一列等。在代码中，通常具有 x = f(x, v[i]) 这种形式，其中 x 是一个叠加器，f 是一个聚合函数，而 v[i] 则是将要被聚合的值。一般来说，f 要求满足结合律，这样不管执行的顺序如何，都不会影响计算结果。

Notice that our use of this pattern with count_heads can be generalized. We used two explicit @spawnat statements, which limits the parallelism to two processes. To run on any number of processes, we can use a parallel for loop, running in distributed memory, which can be written in Julia using @distributed like this:

  1. nheads = @distributed (+) for i = 1:200000000
  2.     Int(rand(Bool))
  3. end

上面的写法将多次迭代分配到了不同的线程，然后通过一个聚合函数（这里是 (+)）合并计算结果，其中，每次迭代的结果作为 for 循环中的表达式的结果，最后整个循环的结果聚合后得到最终的结果。

注意，尽管这里 for 循环看起来跟串行的 for 循环差不多，实际表现完全不同。这里的迭代并没有特定的执行顺序，而且由于所有的迭代都在不同的进程中进行，其中变量的写入对全局来说不可见。所有并行的 for 循环中的变量都会复制并广播到每个进程。

比如，下面这段代码并不会像你想要的那样执行：

  1. a = zeros(100000)
  2. @distributed for i = 1:100000
  3.     a[i] = i
  4. end

这段代码并不会把 a 的所有元素初始化，因为每个进程都会有一份 a 的拷贝，因此类似的 for 循环一定要避免。幸运的是，共享数组 可以用来突破这种限制：

  1. using SharedArrays
  3. a = SharedArray{Float64}(10)
  4. @distributed for i = 1:10
  5.     a[i] = i
  6. end

当然，对于 for 循环外面的变量来说，如果是只读的话，使用起来完全没问题：

  1. a = randn(1000)
  2. @distributed (+) for i = 1:100000
  3.     f(a[rand(1:end)])
  4. end

这里每次迭代都会从共享给每个进程的向量 a 中随机选一个样本，然后用来计算 f。

As you could see, the reduction operator can be omitted if it is not needed. In that case, the loop executes asynchronously, i.e. it spawns independent tasks on all available workers and returns an array of Future immediately without waiting for completion. The caller can wait for the Future completions at a later point by calling fetch on them, or wait for completion at the end of the loop by prefixing it with @sync, like @sync @distributed for.

在一些不需要聚合函数的情况下，我们可能只是像对某个范围内的整数应用一个函数(或者，更一般地，某个序列中的所有元素)，这种操作称作并行的 map，在 Julia 中有一个对应的函数 pmap。例如，可以像下面这样计算一些随机大矩阵的奇异值：

  1. julia> M = Matrix{Float64}[rand(1000,1000) for i = 1:10];
  3. julia> pmap(svdvals, M);

Julia 中的 pmap 是被设计用来处理一些计算量比较复杂的函数的并行化的。与之对比的是，@distributed for 是用来处理一些每次迭代计算都很轻量的计算，比如简单地对两个数求和。pmap 和 @distributed for 都只会用到 worker 的进程。对于 @distributed for 而言，最后的聚合计算由发起者的进程完成。

远程引用和 AbstractChannel

远程引用通常指某种 AbstractChannel 的实现。

A concrete implementation of an AbstractChannel (like Channel), is required to implement put!, take!, fetch, isready and wait. The remote object referred to by a Future is stored in a Channel{Any}(1), i.e., a Channel of size 1 capable of holding objects of Any type.

RemoteChannel 可以被反复写入，可以指向任意大小和类型的 channel（或者是任意 AbstractChannel 的实现）。

RemoteChannel(f::Function, pid)() 构造器可以构造一些引用，而这些引用指向的 channel 可以容纳多个某种具体类型的数据。其中 f 是将要在 pid 上执行的函数，其返回值必须是 AbstractChannel 类型。

例如，RemoteChannel(()->Channel{Int}(10), pid) 会创建一个 channel，其类型是 Int，容量是 10，这个 channel 存在于 pid 进程中。

针对 RemoteChannel 的 put!, take!, fetch, isready 和 wait 方法会被重定向到其底层存储着 channel 的进程。

因此，RemoteChannel 可以用来引用用户自定义的 AbstractChannel 对象。在 Examples repository 中的 dictchannel.jl 文件中有一个简单的例子，其中使用了一个字典用于远端存储。

Channel 和 RemoteChannel

  • 一个 Channel 仅对局部的进程可见，worker 2 无法直接访问 worker 3 上的 Channel，反之亦如此。不过 RemoteChannel 可以跨 worker 获取和写入数据。
  • RemoteChannel 可以看作是对 Channel 的封装。
  • RemoteChannel 的 pid 就是其封装的 channel 所在的进程 id。
  • 任意拥有 RemoteChannel 引用的进程都可以对其进行读写，数据会自动发送到 RemoteChannel 底层 channel 的进程（或从中获取数据）
  • 序列化 Channel 会将其中的所有数据也都序列化，因此反序列化的时候也就可以得到一个原始数据的拷贝。
  • 不过，对 RemoteChannel 的序列化则只会序列化其底层指向的 channel 的 id，因此反序列化之后得到的对象仍然会指向之前存储的对象。

如上的通道示例可以修改为进程间通信，如下所示

首先，起 4 个 worker 进程处理同一个 remote channel jobs，其中的每个 job 都有一个对应的 job_id，然后每个 task 读取一个 job_id，然后模拟随机等待一段时间，然后往存储结果的 RemoteChannel 中写入一个 Tuple 对象，其中包含 job_id 和等待的时间。最后将结果打印出来。

  1. julia> addprocs(4); # add worker processes
  3. julia> const jobs = RemoteChannel(()->Channel{Int}(32));
  5. julia> const results = RemoteChannel(()->Channel{Tuple}(32));
  7. julia> @everywhere function do_work(jobs, results) # define work function everywhere
  8.            while true
  9.                job_id = take!(jobs)
  10.                exec_time = rand()
  11.                sleep(exec_time) # simulates elapsed time doing actual work
  12.                put!(results, (job_id, exec_time, myid()))
  13.            end
  14.        end
  16. julia> function make_jobs(n)
  17.            for i in 1:n
  18.                put!(jobs, i)
  19.            end
  20.        end;
  22. julia> n = 12;
  24. julia> @async make_jobs(n); # feed the jobs channel with "n" jobs
  26. julia> for p in workers() # start tasks on the workers to process requests in parallel
  27.            remote_do(do_work, p, jobs, results)
  28.        end
  30. julia> @elapsed while n > 0 # print out results
  31.            job_id, exec_time, where = take!(results)
  32.            println("$job_id finished in $(round(exec_time; digits=2)) seconds on worker $where")
  33.            global n = n - 1
  34.        end
  35. 1 finished in 0.18 seconds on worker 4
  36. 2 finished in 0.26 seconds on worker 5
  37. 6 finished in 0.12 seconds on worker 4
  38. 7 finished in 0.18 seconds on worker 4
  39. 5 finished in 0.35 seconds on worker 5
  40. 4 finished in 0.68 seconds on worker 2
  41. 3 finished in 0.73 seconds on worker 3
  42. 11 finished in 0.01 seconds on worker 3
  43. 12 finished in 0.02 seconds on worker 3
  44. 9 finished in 0.26 seconds on worker 5
  45. 8 finished in 0.57 seconds on worker 4
  46. 10 finished in 0.58 seconds on worker 2
  47. 0.055971741

远程调用和分布式垃圾回收

远程引用所指向的对象可以在其所有引用都被集群删除之后被释放掉。

The node where the value is stored keeps track of which of the workers have a reference to it. Every time a RemoteChannel or a (unfetched) Future is serialized to a worker, the node pointed to by the reference is notified. And every time a RemoteChannel or a (unfetched) Future is garbage collected locally, the node owning the value is again notified. This is implemented in an internal cluster aware serializer. Remote references are only valid in the context of a running cluster. Serializing and deserializing references to and from regular IO objects is not supported.

上面说到的通知都是通过发送”跟踪”信息来实现的，当一个引用被序列化的时候，就会发送”添加引用”的信息，而一个引用被本地的垃圾回收器回收的时候，就会发送一个”删除引用”的信息。

Since Futures are write-once and cached locally, the act of fetching a Future also updates reference tracking information on the node owning the value.

一旦指向某个值的引用都被删除了，对应的节点会将其释放。

With Futures, serializing an already fetched Future to a different node also sends the value since the original remote store may have collected the value by this time.

此外需要注意的是，本地的垃圾回收到底发生在什么时候取决于具体对象的大小以及当时系统的内存压力。

In case of remote references, the size of the local reference object is quite small, while the value stored on the remote node may be quite large. Since the local object may not be collected immediately, it is a good practice to explicitly call finalize on local instances of a RemoteChannel, or on unfetched Futures. Since calling fetch on a Future also removes its reference from the remote store, this is not required on fetched Futures. Explicitly calling finalize results in an immediate message sent to the remote node to go ahead and remove its reference to the value.

一旦执行了 finalize 之后，引用就不可用了。

Local invocations(@id man-distributed-local-invocations)

Data is necessarily copied over to the remote node for execution. This is the case for both remotecalls and when data is stored to aRemoteChannel / Future on a different node. As expected, this results in a copy of the serialized objects on the remote node. However, when the destination node is the local node, i.e. the calling process id is the same as the remote node id, it is executed as a local call. It is usually(not always) executed in a different task - but there is no serialization/deserialization of data. Consequently, the call refers to the same object instances as passed - no copies are created. This behavior is highlighted below:

  1. julia> using Distributed;
  3. julia> rc = RemoteChannel(()->Channel(3));   # RemoteChannel created on local node
  5. julia> v = [0];
  7. julia> for i in 1:3
  8.            v[1] = i                          # Reusing `v`
  9.            put!(rc, v)
  10.        end;
  12. julia> result = [take!(rc) for _ in 1:3];
  14. julia> println(result);
  15. Array{Int64,1}[[3], [3], [3]]
  17. julia> println("Num Unique objects : ", length(unique(map(objectid, result))));
  18. Num Unique objects : 1
  20. julia> addprocs(1);
  22. julia> rc = RemoteChannel(()->Channel(3), workers()[1]);   # RemoteChannel created on remote node
  24. julia> v = [0];
  26. julia> for i in 1:3
  27.            v[1] = i
  28.            put!(rc, v)
  29.        end;
  31. julia> result = [take!(rc) for _ in 1:3];
  33. julia> println(result);
  34. Array{Int64,1}[[1], [2], [3]]
  36. julia> println("Num Unique objects : ", length(unique(map(objectid, result))));
  37. Num Unique objects : 3

As can be seen, put! on a locally owned RemoteChannel with the same object v modifed between calls results in the same single object instance stored. As opposed to copies of v being created when the node owning rc is a different node.

It is to be noted that this is generally not an issue. It is something to be factored in only if the object is both being stored locally and modifed post the call. In such cases it may be appropriate to store a deepcopy of the object.

This is also true for remotecalls on the local node as seen in the following example:

  1. julia> using Distributed; addprocs(1);
  3. julia> v = [0];
  5. julia> v2 = remotecall_fetch(x->(x[1] = 1; x), myid(), v);     # Executed on local node
  7. julia> println("v=$v, v2=$v2, ", v === v2);
  8. v=[1], v2=[1], true
  10. julia> v = [0];
  12. julia> v2 = remotecall_fetch(x->(x[1] = 1; x), workers()[1], v); # Executed on remote node
  14. julia> println("v=$v, v2=$v2, ", v === v2);
  15. v=[0], v2=[1], false

As can be seen once again, a remote call onto the local node behaves just like a direct invocation. The call modifies local objects passed as arguments. In the remote invocation, it operates on a copy of the arguments.

To repeat, in general this is not an issue. If the local node is also being used as a compute node, and the arguments used post the call, this behavior needs to be factored in and if required deep copies of arguments must be passed to the call invoked on the local node. Calls on remote nodes will always operate on copies of arguments.

共享数组

共享数组使用系统共享内存将数组映射到多个进程上，尽管和 DArray 有点像，但其实际表现有很大不同。在 DArray 中，每个进程可以访问数据中的一块，但任意两个进程都不能共享同一块数据，而对于 SharedArray，每个进程都可以访问整个数组。如果你想在一台机器上，让一大块数据能够被多个进程访问到，那么 SharedArray 是个不错的选择。

共享数组由 SharedArray 提供，必须在所有相关的 worker 中都显式地加载。

对 SharedArray 索引（访问和复制）操作就跟普通的数组一样，由于底层的内存对本地的进程是可见的，索引的效率很高，因此大多数单进程上的算法对 SharedArray 来说都是适用的，除非某些算法必须使用 Array 类型（此时可以通过调用 sdata 来获取 SharedArray 数组）。对于其它类型的 AbstractArray 类型数组来说，sdata 仅仅会返回数组本身，因此，可以放心地使用 sdata 对任意类型的 Array 进行操作。

共享数组可以通过以下形式构造：

  1. SharedArray{T,N}(dims::NTuple; init=false, pids=Int[])

上面的代码会创建一个 N 维，类型为 T，大小为 dims 的共享数组，通过 pids 指定可见的进程。与分布式数组不同的是，只有通过 pids 指定的 worker 才可见。

如果提供了一个类型为 initfn(S::SharedArray) 的 init 函数，那么所有相关的 worker 都会调用它。你可以让每个 worker 都在共享数组不同的地方执行 init 函数，从而实现并行初始化。

下面是个例子：

  1. julia> using Distributed
  3. julia> addprocs(3)
  4. 3-element Array{Int64,1}:
  5.  2
  6.  3
  7.  4
  9. julia> @everywhere using SharedArrays
  11. julia> S = SharedArray{Int,2}((3,4), init = S -> S[localindices(S)] = myid())
  12. 3×4 SharedArray{Int64,2}:
  13.  2  2  3  4
  14.  2  3  3  4
  15.  2  3  4  4
  17. julia> S[3,2] = 7
  18. 7
  20. julia> S
  21. 3×4 SharedArray{Int64,2}:
  22.  2  2  3  4
  23.  2  3  3  4
  24.  2  7  4  4

SharedArrays.localindices 提供了一个以为的切片，可以很方便地用来将 task 分配到各个进程上。当然你可以按你想要的方式做区分：

  1. julia> S = SharedArray{Int,2}((3,4), init = S -> S[indexpids(S):length(procs(S)):length(S)] = myid())
  2. 3×4 SharedArray{Int64,2}:
  3.  2  2  2  2
  4.  3  3  3  3
  5.  4  4  4  4

由于所有的进程都能够访问底层的数据，因此一定要小心避免出现冲突：

  1. @sync begin
  2.     for p in procs(S)
  3.         @async begin
  4.             remotecall_wait(fill!, p, S, p)
  5.         end
  6.     end
  7. end

上面的代码会导致不确定的结果，因为每个进程都将整个数组赋值为其 pid，从而导致最后一个执行完成的进程会保留其 pid。

考虑更复杂的一种情况：

  1. q[i,j,t+1] = q[i,j,t] + u[i,j,t]

这个例子中，如果首先将任务用按照一维的索引作区分，那么就会出问题：如果 q[i,j,t] 位于分配给某个 worker 的最后一个位置，而 q[i,j,t+1] 位于下一个 worker 的开始位置，那么后面这个 worker 开始计算的时候，可能 q[i,j,t] 还没有准备好，这时候，更好的做法是，手动分区，比如可以定义一个函数，按照 (irange,jrange) 给每个 worker 分配任务。

  1. julia> @everywhere function myrange(q::SharedArray)
  2.            idx = indexpids(q)
  3.            if idx == 0 # This worker is not assigned a piece
  4.                return 1:0, 1:0
  5.            end
  6.            nchunks = length(procs(q))
  7.            splits = [round(Int, s) for s in range(0, stop=size(q,2), length=nchunks+1)]
  8.            1:size(q,1), splits[idx]+1:splits[idx+1]
  9.        end

然后定义计算内核：

  1. julia> @everywhere function advection_chunk!(q, u, irange, jrange, trange)
  2.            @show (irange, jrange, trange)  # display so we can see what's happening
  3.            for t in trange, j in jrange, i in irange
  4.                q[i,j,t+1] = q[i,j,t] + u[i,j,t]
  5.            end
  6.            q
  7.        end

然后定义一个 wrapper：

  1. julia> @everywhere advection_shared_chunk!(q, u) =
  2.            advection_chunk!(q, u, myrange(q)..., 1:size(q,3)-1)

接下来，比较三个不同的版本，第一个是单进程版本：

  1. julia> advection_serial!(q, u) = advection_chunk!(q, u, 1:size(q,1), 1:size(q,2), 1:size(q,3)-1);

然后是使用 @distributed:

  1. julia> function advection_parallel!(q, u)
  2.            for t = 1:size(q,3)-1
  3.                @sync @distributed for j = 1:size(q,2)
  4.                    for i = 1:size(q,1)
  5.                        q[i,j,t+1]= q[i,j,t] + u[i,j,t]
  6.                    end
  7.                end
  8.            end
  9.            q
  10.        end;

最后是使用分区：

  1. julia> function advection_shared!(q, u)
  2.            @sync begin
  3.                for p in procs(q)
  4.                    @async remotecall_wait(advection_shared_chunk!, p, q, u)
  5.                end
  6.            end
  7.            q
  8.        end;

如果创建好了 SharedArray 之后，计算这些函数的执行时间，那么可以得到以下结果（用 julia -p 4 启动）：

  1. julia> q = SharedArray{Float64,3}((500,500,500));
  3. julia> u = SharedArray{Float64,3}((500,500,500));

先执行一次以便 JIT 编译，然后用 @time 宏测试其第二次执行的时间：

  1. julia> @time advection_serial!(q, u);
  2. (irange,jrange,trange) = (1:500,1:500,1:499)
  3.  830.220 milliseconds (216 allocations: 13820 bytes)
  5. julia> @time advection_parallel!(q, u);
  6.    2.495 seconds      (3999 k allocations: 289 MB, 2.09% gc time)
  8. julia> @time advection_shared!(q,u);
  9.         From worker 2:       (irange,jrange,trange) = (1:500,1:125,1:499)
  10.         From worker 4:       (irange,jrange,trange) = (1:500,251:375,1:499)
  11.         From worker 3:       (irange,jrange,trange) = (1:500,126:250,1:499)
  12.         From worker 5:       (irange,jrange,trange) = (1:500,376:500,1:499)
  13.  238.119 milliseconds (2264 allocations: 169 KB)

这里 advection_shared! 最大的优势在于，最小程度地降低了 woker 之间的通信，从而让每个 worker 能针对被分配的部分持续地计算一段时间。

共享数组与分布式垃圾回收

和远程引用一样，共享数组也依赖于创建节点上的垃圾回收来释放所有参与的 worker 上的引用。因此，创建大量生命周期比较短的数组，并尽可能快地显式 finilize 这些对象，代码会更高效，这样与之对用的内存和文件句柄都会更快地释放。

集群管理器

Julia 通过集群管理器实现对多个进程（所构成的逻辑上的集群）的启动，管理以及网络通信。一个 ClusterManager 负责：

  • 在一个集群环境中启动 worker 进程
  • 管理每个 worker 生命周期内的事件
  • （可选），提供数据传输

一个 Julia 集群由以下特点：

  • 初始进程，称为 master，其 id 为 1
  • 只有 master 进程可以增加或删除 worker 进程
  • 所有进程之间都可以直接通信

worker 之间的连接（用的是内置的 TCP/IP 传输）按照以下方式进行：

  • master 进程对一个 ClusterManager 对象调用 addprocs
  • addprocs 调用对应的 launch 方法，然后在对应的机器上启动相应数量的 worker 进程
  • 每个 worker 监听一个端口，然后将其 host 和 port 信息传给 stdout
  • 集群管理器捕获 stdout 中每个 worker 的信息，并提供给 master 进程
  • master 进程解析信息并与相应的 worker 建立 TCP/IP 连接
  • 每个 worker 都会被通知集群中的其它 worker
  • 每个 worker 与 id 小于自己的 worker 连接
  • 这样，一个网络就建立了，从而，每个 worker 都可以与其它 worker 建立连接

尽管默认的传输层使用的是 TCPSocket，对于一个自定义的集群管理器来说，完全可以使用其它传输方式。

Julia 提供了两种内置的集群管理器：

LocalManager 用来在同一个 host 上启动多个 worker，从而利用多核/多处理器硬件。

因此，一个最小的集群管理器需要：

  • 是一个 ClusterManager 抽象类的一个子类
  • 实现 launch 接口，用来启动新的 worker
  • 实现 manage，在一个 worker 的生命周期中多次被调用（例如，发送中断信号）

addprocs(manager::FooManager) 需要 FooManager 实现：

  1. function launch(manager::FooManager, params::Dict, launched::Array, c::Condition)
  2.     [...]
  3. end
  5. function manage(manager::FooManager, id::Integer, config::WorkerConfig, op::Symbol)
  6.     [...]
  7. end

作为一个例子，我们来看下 LocalManager 是怎么实现的：

  1. struct LocalManager <: ClusterManager
  2.     np::Integer
  3. end
  5. function launch(manager::LocalManager, params::Dict, launched::Array, c::Condition)
  6.     [...]
  7. end
  9. function manage(manager::LocalManager, id::Integer, config::WorkerConfig, op::Symbol)
  10.     [...]
  11. end

launch 方法接收以下参数：

  • manager::ClusterManager: 调用 addprocs 时所用到的集群管理器
  • params::Dict: 所有的关键字参数都会传递到 addprocs 中
  • launched::Array: 用来存储一个或多个 WorkerConfig
  • c::Condition: 在 workers 启动后被通知的条件变量

launch 会在一个异步的task中调用，该 task 结束之后，意味着所有请求的 worker 都已经启动好了。因此，launch 函数必须在所有 worker 启动之后，尽快退出。

新启动的 worker 之间采用的是多对多的连接方式。在命令行中指定参数 --worker[=<cookie>] 会让所有启动的进程把自己当作 worker，然后通过 TCP/IP 构建连接。

集群中所有的 worker 默认使用同一个 master 的 cookie。如果 cookie 没有指定，（比如没有通过 --worker 指定），那么 worker 会尝试从它的标准输入中读取。LocalManager 和 SSHManager 都是通过标准输入来将 cookie 传递给新启动的 worker。

默认情况下，一个 worker 会监听从 getipaddr() 函数返回的地址上的一个开放端口。若要指定监听的地址，可以通过额外的参数 --bind-to bind_addr[:port] 指定，这对于多 host 的情况来说很方便。

对于非 TCP/IP 传输，可以选择 MPI 作为一种实现，此时一定不要指定 --worker 参数，另外，新启动的 worker 必须调用 init_worker(cookie) 之后再使用并行的结构体。

对于每个已经启动的 worker，launch 方法必须往 launched 中添加一个 WorkerConfig 对象（相应的值已经初始化）。

  1. mutable struct WorkerConfig
  2.     # Common fields relevant to all cluster managers
  3.     io::Union{IO, Nothing}
  4.     host::Union{AbstractString, Nothing}
  5.     port::Union{Integer, Nothing}
  7.     # Used when launching additional workers at a host
  8.     count::Union{Int, Symbol, Nothing}
  9.     exename::Union{AbstractString, Cmd, Nothing}
  10.     exeflags::Union{Cmd, Nothing}
  12.     # External cluster managers can use this to store information at a per-worker level
  13.     # Can be a dict if multiple fields need to be stored.
  14.     userdata::Any
  16.     # SSHManager / SSH tunnel connections to workers
  17.     tunnel::Union{Bool, Nothing}
  18.     bind_addr::Union{AbstractString, Nothing}
  19.     sshflags::Union{Cmd, Nothing}
  20.     max_parallel::Union{Integer, Nothing}
  22.     # Used by Local/SSH managers
  23.     connect_at::Any
  25.     [...]
  26. end

WorkerConfig 中的大多数字段都是内置的集群管理器会用到，对于自定义的管理器，通常只需要指定 io 或 host/port:

  • 如果指定了 io，那么就会用来读取 host/port 信息。每个 worker 会在启动时打印地址和端口，这样 worker 就可以自由监听可用的端口，而不必手动配置 worker 的端口。

  • 如果 io 没有指定，那么 host 和 port 就会用来连接。

  • count，exename 和 exeflags 用于从一个 worker 上启动额外的 worker。例如，一个集群管理器可能对每个节点都只启动一个 worker，然后再用它来启动额外的 worker。

    • count 可以是一个整数 n，用来指定启动 n 个 worker
    • count 还可以是 :auto，用来启动跟那台机器上 CPU 个数（逻辑上的核的个数）相同的 worker
    • exename 是 julia 可执行文件的全路径
    • exeflags 应该设置成传递给将要启动的 worker 命令行参数

  • tunnel, bind_addr, sshflags 和 max_parallel 会在从 worker 与 master 进程建立 ssh 隧道时用到

  • userdata 用来提供给自定义集群管理器存储自己的 worker 相关的信息

manage(manager::FooManager, id::Integer, config::WorkerConfig, op::Symbol) 会在一个 worker 生命周期中的不同时刻被调用，其中 op 的值可能是：

  • :register/:deregister，从 Julia 的 worker 池子中添加/删除一个 worker
  • :interrupt，当 interrupt(workers) 被调用是，此时，ClusterManager 应该给相应的 worker 发送终端信号
  • :finalize，用于清理操作。

自定义集群管理器的传输方式

将默认的 TCP/IP 多对多 socket 连接替换成一个自定义的传输层需要做很多工作。每个 Julia 进程都有与其连接的 worker 数量相同的通信 task。例如，在一个有 32 个进程的多对多集群中：

  • 每个进程都有31个通信task
  • 每个 task 在一个消息处理循环中从一个远端 worker 读取所有的输入信息
  • 每个消息处理循环等待一个 IO 对象（比如，在默认实现中是一个 TCPSocket），然后读取整个信息，处理，等待下一个
  • 发送消息则可以直接在任意 Julia task 中完成，而不只是通信 task，同样，也是通过相应的 IO 对象

要替换默认的传输方式，需要新的实现能够在远程 worker 之间建立连接，同时提供一个可以用来被消息处理循环等待的 IO 对象。集群管理器的回调函数需要实现如下函数：

  1. connect(manager::FooManager, pid::Integer, config::WorkerConfig)
  2. kill(manager::FooManager, pid::Int, config::WorkerConfig)

默认的实现（使用的是 TCP/IP socket）是 connect(manager::ClusterManager, pid::Integer, config::WorkerConfig)。

connect 需要返回一对 IO 对象，一个用于从 pid worker 读取数据，另一个用于往 pid 写数据。自定义的集群管理器可以用内存中的 BUfferStream 作为一个管道将自定义的（很可能是非 IO 的）传输与 Julia 内置的并行基础设施衔接起来。

BufferStream 是一个内存中的 IOBuffer，其表现很像 IO，就是一个流（stream），可以异步地处理。

在 Examples repository 的 clustermanager/0mq 目录中，包含一个使用 ZeroMQ 连接 Julia worker 的例子，用的是星型拓补结构。需要注意的是：Julia 的进程仍然是逻辑上相互连接的，任意 worker 都可以与其它 worker 直接相连而无需感知到 0MQ 作为传输层的存在。

在使用自定义传输的时候：

  • Julia 的 workers 必须不能通过 --worker 启动。如果启动的时候使用了 --worker，那么新启动的 worker 会默认使用基于 TCP/IP socket 的实现
  • 对于每个 worker 逻辑上的输入连接，必须调用 Base.process_messages(rd::IO, wr::IO)()，这会创建一个新的 task 来处理 worker 消息的读写
  • init_worker(cookie, manager::FooManager) 必须作为 worker 进程初始化的一部分呢被调用
  • WorkerConfig中的 connect_at::Any 字段可以被集群管理器在调用 launch 的时候设置，该字段的值会发送到所有的 connect 回调中。通常，其中包含的是如何连接到一个 worker 的信息。例如，在 TCP/IP socket 传输中，用这个字段存储 (host, port) 来声明如何连接到一个 worker。

kill(manager, pid, config) 用来从一个集群中删除一个 worker，在 master 进程中，对应的 IO 对象必须通过对应的实现来关闭，从而保证正确地释放资源。默认的实现简单地对指定的远端 worker 执行 exit() 即可。

在例子目录中，clustermanager/simple 展示了一个简单地实现，使用的是 UNIX 下的 socket。

LocalManager 和 SSHManager 的网络要求

Julia 集群设计的时候，默认是在一个安全的环境中执行，比如本地的笔记本，部门的集群，甚至是云端。这部分将介绍 LocalManager 和 SSHManager 的网络安全要点：

  • master 进程不监听任何端口，它只负责向外连接 worker
  • 每个 worker 都只绑定一个本地的接口，同时监听一个操作系统分配的临时端口。
  • addprocs(N) 使用的 LocalManager，默认只会绑定到回环接口（loopback interface），这就意味着，之后在远程主机上（恶意）启动的 worker 无法连接到集群中，在执行 addprocs(4) 之后，又跟一个 addprocs(["remote_host"]) 会失败。有些用户可能希望创建一个集群同时管理本地系统和几个远端系统，这可以通过在绑定 LocalManager 到外部网络接口的时候，指定一个 restrict 参数：addprocs(4; restrict=false)
  • addprocs(list_of_remote_hosts) 使用的 SSHManager 会通过 SSH 启动远程机上的 worker。

默认 SSH 只会用来启动 Julia 的 worker。随后的 master-worker 和 worker-worker 连接使用的是普通的、未加密的 TCP/IP 通信。 远程机必须开启免密登陆。 额外的 SSH 标记或认证信息会通过关键字参数 sshflags 指定。

  • addprocs(list_of_remote_hosts; tunnel=true, sshflags=<ssh keys and other flags>) 在我们希望给 master-worker 也使用 SSH 连接的时候很有用。 一个典型的场景是本地的笔记本 运行 Julia ERPL （做为 master）和云上的其他机器，比如 Amazon EC2，构成集群。 这时候远程机器只要开启 22 端口就可以，然后要有 SSH 客户端 通过公约基础设施（PKI）认证过。授权信息可以通过 sshflags 生效，比如 sshflags=`-i <keyfile>` 。

    在一个所有节点联通的拓扑网中（默认情况下是这样的），所有的 worker 节点都通过普通 TCP socket 通信互相连接。 这样集群的安全策略就必须允许 worker 节点间 通过操作系统分配的临时端口范围自由连接。

    所有 worker-worker 间（都是 SSH）的安全和加密或者信息的加密 都可以通过自定义 ClusterManager 完成。

集群 Cookie

集群上所有的进程都共享同一个 cookie，默认是 master 进程随机生成的字符串。

  • cluster_cookie() 返回 cookie，而 cluster_cookie(cookie)() 设置并返回新的 cookie。
  • 所有的连接都进行双向认证，从而保证只有 master 启动的 worker 才能相互连接。
  • cookie 可以在 worker 启动的时候，通过参数 --worker=<cookie> 指定，如果参数 --worker 没有指定 cookie，那么 worker 会从它的标准输入中 (stdin) 读取， stdin 会在 cookie 获取之后立即关闭。
  • ClusterManager 可以通过 cluster_cookie() 从 master 中过去 cookie，不适用默认 TCP/IP 传输的集群管理器（即没有指定 --worker）必须用于 master 相同的 cookie 调用 init_worker(cookie, manager)。

注意，在对安全性要求很高的环境中，可以通过自定义 ClusterManager 实现。例如，cookie 可以提前共享，然后不必再启动参数中指定。

指定网络拓补结构（实验性功能）

可以通过传递到 addprocs 中的参数 topology 来指定 worker 之间如何连接。

  • :all_to_all，默认的，所有 worker 之间相互都连接
  • :master_worker，只有主进程，即 pid 为 1 的进程能够与 worker 建立连接
  • :custom: 集群管理器的 launch 方法通过 WorkerConfig 中的 ident 和 connect_idents 指定连接的拓补结构。一个 worker 通过集群管理器提供的 ident 来连接到所有 connect_idents 指定的 worker。

关键字参数 lazy=true|false 只会影响 topology 选项中的 :all_to_all。如果是 true，那么集群启动的时候 master 会连接所有的 worker，然后 worker 之间的特定连接会在初次唤醒的是建立连接，这有利于降低集群初始化的时候对资源的分配。lazy 的默认值是 true。

目前，在没有建立连接的两个 worker 之间传递消息会出错，目前该行为是实验性的，未来的版本中可能会改变。

一些值得关注的外部库

除了 Julia 自带的并行机制之外，还有许多外部的库值得一提。例如 MPI.jl 提供了一个 MPI 协议的 Julia 的封装，或者是在 共享数组 提到的 DistributedArrays.jl，此外尤其值得一提的是 Julia 的 GPU 编程生态，其包括：

  1. 底层（C内核）的 OpenCL.jl 和 CUDAdrv.jl，分别提供了 OpenCL 和 CUDA 的封装。

  2. 底层（Julia 内核）的接口，如 CUDAnative.jl，提供了 Julia 原生的 CUDA 实现。

  3. 高层的特定抽象，如 CuArrays.jl 和 CLArrays.jl。

  4. 高层的库，如 ArrayFire.jl 和 GPUArrays.jl。

下面的例子将介绍如何用 DistributedArrays.jl 和 CuArrays.jl 通过 distribute() 和 CuArray() 将数组分配到多个进程。

记住在载入 DistributedArrays.jl 时，需要用 @everywhere 将其载入到多个进程中。

  1. $ ./julia -p 4
  3. julia> addprocs()
  5. julia> @everywhere using DistributedArrays
  7. julia> using CuArrays
  9. julia> B = ones(10_000) ./ 2;
  11. julia> A = ones(10_000) .* π;
  13. julia> C = 2 .* A ./ B;
  15. julia> all(C .≈ 4*π)
  16. true
  18. julia> typeof(C)
  19. Array{Float64,1}
  21. julia> dB = distribute(B);
  23. julia> dA = distribute(A);
  25. julia> dC = 2 .* dA ./ dB;
  27. julia> all(dC .≈ 4*π)
  28. true
  30. julia> typeof(dC)
  31. DistributedArrays.DArray{Float64,1,Array{Float64,1}}
  33. julia> cuB = CuArray(B);
  35. julia> cuA = CuArray(A);
  37. julia> cuC = 2 .* cuA ./ cuB;
  39. julia> all(cuC .≈ 4*π);
  40. true
  42. julia> typeof(cuC)
  43. CuArray{Float64,1}

要牢记，当前一些 Julia 的特性并没有被 CUDAnative.jl [2] 支持，尤其是一些像 sin 之类的函数需要换成 CUDAnative.sin(cc: @maleadt)。

下面的例子中，通过 DistributedArrays.jl 和 CuArrays.jl 将一个数组分配到多个进程，然后调用一个函数。

  1. function power_method(M, v)
  2.     for i in 1:100
  3.         v = M*v
  4.         v /= norm(v)
  5.     end
  7.     return v, norm(M*v) / norm(v)  # or  (M*v) ./ v
  8. end

power_method 重复创建一个新的向量然后对其归一化，这里并没有在函数中指定类型信息，来看看是否对前面提到的类型适用：

  1. julia> M = [2. 1; 1 1];
  3. julia> v = rand(2)
  4. 2-element Array{Float64,1}:
  5. 0.40395
  6. 0.445877
  8. julia> power_method(M,v)
  9. ([0.850651, 0.525731], 2.618033988749895)
  11. julia> cuM = CuArray(M);
  13. julia> cuv = CuArray(v);
  15. julia> curesult = power_method(cuM, cuv);
  17. julia> typeof(curesult)
  18. CuArray{Float64,1}
  20. julia> dM = distribute(M);
  22. julia> dv = distribute(v);
  24. julia> dC = power_method(dM, dv);
  26. julia> typeof(dC)
  27. Tuple{DistributedArrays.DArray{Float64,1,Array{Float64,1}},Float64}

最后，我们来看看 MPI.jl，这个库时 Julia 对 MPI 协议的封装。一一介绍其中的每个函数太累赘了，这里领会其实现协议的方法就够了。

考虑下面这个简单的脚本，它做的只是调用每个子进程，然后初始化其 rank，然后在 master 访问时，对 rank 求和。

  1. import MPI
  3. MPI.Init()
  5. comm = MPI.COMM_WORLD
  6. MPI.Barrier(comm)
  8. root = 0
  9. r = MPI.Comm_rank(comm)
  11. sr = MPI.Reduce(r, MPI.SUM, root, comm)
  13. if(MPI.Comm_rank(comm) == root)
  14.    @printf("sum of ranks: %s\n", sr)
  15. end
  17. MPI.Finalize()
  1. mpirun -np 4 ./julia example.jl
  • 1In this context, MPI refers to the MPI-1 standard. Beginning with MPI-2, the MPI standards committee introduced a new set of communication mechanisms, collectively referred to as Remote Memory Access (RMA). The motivation for adding rma to the MPI standard was to facilitate one-sided communication patterns. For additional information on the latest MPI standard, see https://mpi-forum.org/docs.
  • 2Julia GPU 手册