并行计算

对于多线程和并行计算的新手来说，首先了解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)

  1. and a function foo which reads items from c1, processes the item read

  2. and writes a result to c2,

  3. function foo()
  4. while true
  5. data = take!(c1)
  6. [...] # process data
  7. put!(c2, result) # write out result
  8. end

  9. end

  10. we can schedule n instances of foo to be active concurrently.

  11. for _ in 1:n
  12. @async foo()
  13. end ```
  14. Channel 可以通过 Channel{T}(sz) 构造，得到的 channel 只能存储类型 T 的数据。如果 T 没有指定，那么 channel 可以存任意类型。sz 表示该 channel 能够存储的最大元素个数。比如 Channel(32) 得到的 channel 最多可以存储32个元素。而 Channel{MyType}(64) 则可以最多存储64个 MyType 类型的数据。
  15. 如果一个 Channel 是空的，读取的 task(即执行 take! 的 task)会被阻塞直到有新的数据准备好了。
  16. 如果一个 Channel 是满的，那么写入的 task(即执行 put! 的 task)则会被阻塞，直到 Channel 有空余。
  17. isready 可以用来检查一个 channel 中是否有已经准备好的元素，而等待一个元素准备好 则用 wait
  18. 一个 Channel 一开始处于开启状态，也就是说可以被 take! 读取和 put! 写入。close 会关闭一个 Channel，对于一个已经关闭的 Channel，put! 会失败，例如：

``` 1. julia> c = Channel(2);

  1. julia> put!(c, 1) # put! on an open channel succeeds

  2. 1

  3. julia> close(c);

  4. julia> put!(c, 2) # put! on a closed channel throws an exception.

  5. ERROR: InvalidStateException("Channel is closed.",:closed)
  6. Stacktrace:
  7. [...] ```
  8. take! 和 fetch (只读取，不会将元素从 channel 中删掉)仍然可以从一个已经关闭的 channel 中读数据，直到 channel 被取空了为止。继续上面的例子：

`` 1. julia> fetch(c) # Any number offetch` calls succeed. 2. 1

  1. julia> fetch(c)

  2. 1

  3. julia> take!(c) # The first take! removes the value.

  4. 1

  5. julia> take!(c) # No more data available on a closed channel.

  6. ERROR: InvalidStateException("Channel is closed.",:closed)
  7. Stacktrace:
  8. [...] ```

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. 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

（本文超出站点单文档体积上限，已截断）