Julia并行计算笔记（二）

（持续修订中，最近更新于2020年8月10日。）

四、远程调用之一

上一节讲了Julia的协程(Task)级并行，本节讲的是进程级并行。协程只能在单台计算机上并行，而进程可以在多台计算机上并行。一切开始前，首先仍要using Distributed，并且要addprocs(n)或julia -p n来开启多个Worker。强调一遍，Worker特指远程进程。

远程调用，指通过主进程在Worker（远程进程）中启动某一函数或表达式。对于函数，用remotecall()实现远程调用：

remotecall(函数, Worker的PID, 函数参数)

例如：在主进程上调用rand()函数创建一个2x3的随机数组，表达式为rand(2,3)。如果要在PID=4的Worker上做这件事，那么应写成：

remotecall(rand,4,2,3)

远程调用后不会立即返回结果到本地，需要使用fetch()提取结果，例如：

julia> r = remotecall(rand,4,2,3)
Future(4, 1, 5, nothing)julia> fetch(r)
2×3 Array{Float64,2}:0.466937  0.761268  0.9755530.754082  0.025674  0.824383

注意这里的fetch()跟之前Task并行不一样，会移除Worker上的数据。如果提取不到结果，它会阻塞直到有结果为止。fetch()提取的结果会缓存在主进程上，具体来说是存在r的内部的一个类型为Future的对象中。

我们要专门讲一下Future对象。Future(远程PID, 本地PID, Future ID, 远程结果)是一个存储远程调用信息的对象，会在远程调用时立即返回，但这时最后一项只包含nothing。当fetch()从远程提取结果后，会把结果填入nothing的位置。实际上，我们可以自己创建一个Future对象，例如：

julia> Future(100)
Future(100, 1, 34, nothing)

这个Future的远程PID是100，本地PID是1，自己的编号是34（说明它是你创建的第34个Future）。不过这个Future不属于任何一个远程调用，所以不会有数据填充nothing位置。

像Future这样的存储远程调用信息的对象，称之为“远程引用”（所以远程引用是一个对象而不是一个操作）。另一个远程引用是RemoteChannel，是上一节Channel的跨进程版本，用于进程间交换数据。或者这样说，Channel是协程间管道，RemoteChannel是进程间管道。

现在回头来看fetch()。提取结果后，可以在主进程上重复使用fetch(r)来多次使用结果，或者干脆把结果赋值给一个新的对象：

julia> result = fetch(r);julia> result
2×3 Array{Float64,2}:0.466937  0.761268  0.9755530.754082  0.025674  0.824383

自然地，又有一步到位的技巧，即remotecall_fetch()，可以把上面的示例缩写为：

result = remotecall_fetch(rand,4,2,3)

同样地，它会一直阻塞直到成功提取结果。由于Future对象会消耗时间，如果不需要提取结果，那么可使用不含Future对象的remote_do()代替remotecall()。不过这里有个细节是：当多次远程调用时，remotecall()会依次执行，而remote_do()则是无序的。

对于表达式（强调一下，赋参的函数是一个表达式），我们可用宏命令@spawnat来实现远程调用。例如：

julia> s = @spawnat 4 rand(2,3)
Future(4, 1, 7, nothing)julia> fetch(s)
2×3 Array{Float64,2}:0.375975  0.844135  0.2576470.057513  0.169291  0.0544206

当然也可以这样写：

julia> fetch(@spawnat 4 rand(2,3))
2×3 Array{Float64,2}:0.57577   0.500889  0.2289970.268749  0.295895  0.0822172

再做得更复杂一点：

julia> fetch(@spawnat 3 (1).+fetch(s))
2×3 Array{Float64,2}:1.37598  1.84413  1.257651.05751  1.16929  1.05442

这里的表达式(1).+fetch(s)的意思是把fetch(s)逐个加1。注意要写成(1).+fetch(s)而不是书中的1 .+fetch(s)，否则会报错。貌似是1.1版本的变化之一。

另一个宏命令@spawn不需要指定PID，用起来更方便得多。所以一般用它代替@spawnat或remotecall()。例如：

julia> fetch(@spawn (1).+fetch(@spawn rand(2,3)))
2×3 Array{Float64,2}:1.14194  1.57693  1.900711.88392  1.31092  1.51812

小贴士：用myid()可以查询当前进程的PID。如果直接调用，会返回1；如果在远程调用，则会返回远程进程的PID。

但它不能代替remote_do()，因为后者不返回结果。此外，如果Worker是已知空闲的，指定PID会稍微快一点（可惜多数情况下我们并不清楚哪些Worker是空闲的）。

宏命令也有“一步到位”的技巧！我们可以把fetch(@spawn 表达式)简写为@fetch 表达式，把fetch(@spawnat PID 表达式)简写为@fetchfrom PID 表达式。例如：

julia> @fetch rand(2,3)
2×3 Array{Float64,2}:0.0622937  0.93881   0.4717340.576323   0.621816  0.713404

与之前的“一步到位”类似，由于把调用和提取合并为一个命令，所以主进程会阻塞，等待Worker返回结果之后才继续。这种调用方式称为“同步调用”。如果把调用和提取拆分开来，主进程就可以在调用之后去做别的事情，直到Worker通知它，再来提取结果。这种方式称为“异步调用”。

小贴士：异步调用就是你喊你朋友吃饭，你朋友说知道了，待会忙完去找你，你就去做别的了。同步调用就是你喊你朋友吃饭，你朋友在忙，你就一直在那等，等你朋友忙完了，你们一起去。

现在我们来看一组例子加深对同步性质的理解：

# 例1
julia> @time @spawn sleep(3)0.000288 seconds (112 allocations: 5.891 KiB)
Future(4, 1, 24, nothing)# 例2
julia> @time @sync @spawn sleep(3)3.014166 seconds (2.96 k allocations: 173.319 KiB)
Future(2, 1, 25, nothing)# 例3
julia> @time @sync @fetch rand(2,3)0.015145 seconds (152 allocations: 7.703 KiB)
2×3 Array{Float64,2}:0.665148  0.607531  0.5630960.506471  0.748635  0.588137

@time是计时的宏命令。可以看到，在例1中，调用后立即返回了Future对象，但此时Worker仍未执行完毕。例2添加了一个@sync宏命令，它会强制令Future对象在Worker执行完毕后才返回。例3表明@sync亦可作用于@fetch。实际上，@sync可作用于@spawn、@spawnat、@fetch、@async、@distributed（后文介绍），但不适用于异步操作如remotecall()和remote_do()。而@fecthfrom如前文所述必定为同步调用。

“同步”是一个非常重要的概念。我们再看一组例子，理解如何使用@sync实现多进程同步：

# 开辟4个进程。
julia> addprocs(3); procs()
4-element Array{Int64,1}:1234# 示例1
julia> @time for pid in procs()@spawnat pid (sleep(pid); println(myid()))end0.060870 seconds (35.07 k allocations: 1.694 MiB)# 示例2
julia> @time fetch(@sync (for pid in procs()@spawnat pid (sleep(pid); println(myid()))end))
1From worker 2:    2From worker 3:    3From worker 4:    44.066450 seconds (35.37 k allocations: 1.705 MiB)# 示例3
julia> @time for pid in procs()fetch(@spawnat pid (sleep(pid); println(myid())))end
1From worker 2:    2From worker 3:    3From worker 4:    410.113555 seconds (35.43 k allocations: 1.710 MiB)

示例1的含义是：在每个进程上发起一组动作sleep(pid); println(myid())，然后对整个循环计时。观察输出，可见整个循环结束时各进程的动作还未完成。示例2加上了@sync，保证所有进程动作结束后再进行计时。各进程的动作是并行的，总耗时约等于最慢进程的耗时。示例3对每个进程做fetch(@spawnat pid (sleep(pid); println(myid())))，等价于@fetchfrom pid (sleep(pid); println(myid()))，耗时达到10秒，证明各进程是事实上的串行而非并行。这是因为@fetchfrom本身包含了一次同步。因此，当我们把并行的任务发送到各进程时，不要着急立即提取结果，而应该先@sync，然后再逐个进程提取到本地。看上去有点麻烦，所以出现了像DistributedArrays这种东西。下面稍微介绍一下这个包，以后有空写一篇关于DistributedArrays包的详细解释。

DistributedArrays包提供了DArray类型，是一种分布式数组，即数组由存储在多个进程中的子块拼成。创建DArray的目的是方便跨进程索引。简单地讲，当你用@spawnat在远程修改DArray之后，可省略提取到本地的步骤，直接从本地访问DArray中的被修改的元素。举个例子，我们创建一个DArray类型的数组d：

julia> d = dzeros(4,3)
4×3 DArray{Float64,2,Array{Float64,2}}:0.0  0.0  0.00.0  0.0  0.00.0  0.0  0.00.0  0.0  0.0

然后用localindices获取进程2上的子数组的索引：

julia> @fetchfrom 2 localindices(d)
(1:2, 1:3)

可知元素d[1,1]是存储在进程2上。当我们要修改它时，必须在进程2操作：

julia> @spawnat 2 localpart(d)[1,1] = 666
Future(2, 1, 397, nothing)

注意这里要用localpart而不能直接写d[1,1]。进程2中的localpart(d)[1,1]对应于d[1,1]。localpart(d)[1,1]中的[1,1]是子数组的索引而不是整个数组的索引，只是凑巧重合了。

从其他进程（包括主进程）都只能读取而不能修改d[1,1]，否则会报错：

# 读取
julia> d
4×3 DArray{Float64,2,Array{Float64,2}}:666.0  0.0  0.00.0  0.0  0.00.0  0.0  0.00.0  0.0  0.0# 修改
julia> d[1,1] = 666
ERROR: setindex! not defined for DArray{Float64,2,Array{Float64,2}}
Stacktrace:[1] error(::String, ::Type) at ./error.jl:42[2] error_if_canonical_setindex(::IndexCartesian, ::DArray{Float64,2,Array{Float64,2}}, ::Int64, ::Int64) at ./abstractarray.jl:1084[3] setindex!(::DArray{Float64,2,Array{Float64,2}}, ::Int64, ::Int64, ::Int64) at ./abstractarray.jl:1073[4] top-level scope at REPL[40]:1

DistributedArrays参考资料：链接。不过它的Julia版本太旧，有些命令过时了。

最后我们讨论一下数据跨进程传输的问题。在远程调用一个表达式时，表达式中的参数会自动从主进程传输到Worker上，例如：

julia> A = rand(1000,1000);julia> @time @spawn A^20.251465 seconds (496.78 k allocations: 24.047 MiB, 6.12% gc time)
Future(3, 1, 31, nothing)

这里传输到Worker上的参数是A。换一种方式写：

julia> @time @spawn rand(1000,1000)^20.000250 seconds (123 allocations: 6.588 KiB)
Future(4, 1, 32, nothing)

这里传输的参数是1000,1000，显然比A的数据量小，于是Future对象返回得更快了。两种写法各有好处：如果你觉得跨进程传输A的代价相比于A的运算小很多，那么选第一种，否则选第二种。写法的差异会对Julia运行效率产生显著影响。

Julia并行计算笔记（二）相关推荐

Julia并行计算笔记（一）
本文是<Julia语言程序设计>(魏坤)第14章的读书笔记,加入了很多自己测试和官方文档的内容.内容基本上完整覆盖,不过对照原著风味更佳.全文很长,分为五篇,这是第一篇. 一.进程.线程 ...
qml学习笔记(二)：可视化元素基类Item详解（上半场anchors等等）
原博主博客地址:http://blog.csdn.net/qq21497936 本文章博客地址:http://blog.csdn.net/qq21497936/article/details/7851 ...
oracle直查和call哪个更快,让oracle跑的更快1读书笔记二
当前位置:我的异常网» 数据库 » <>读书笔记二 <>读书笔记二 www.myexceptions.net 网友分享于:2013-08-23 浏览:9次 <> ...
【Visual C++】游戏开发笔记二十七 Direct3D 11入门级知识介绍
游戏开发笔记二十七 Direct3D 11入门级知识介绍作者:毛星云邮箱: happylifemxy@163.com 期待着与志同道合的朋友们相互交流上一节里我们介绍了在迈入Dire ...
[转载]dorado学习笔记(二)
原文地址:dorado学习笔记(二)作者:傻掛 ·isFirst, isLast在什么情况下使用?在遍历dataset的时候会用到 ·dorado执行的顺序,首先由jsp发送请求,调用相关的ViewM ...
PyTorch学习笔记(二)——回归
PyTorch学习笔记(二)--回归本文主要是用PyTorch来实现一个简单的回归任务. 编辑器:spyder 1.引入相应的包及生成伪数据 import torch import torch.nn ...
tensorflow学习笔记二——建立一个简单的神经网络拟合二次函数
tensorflow学习笔记二--建立一个简单的神经网络 2016-09-23 16:04 2973人阅读评论(2) 收藏举报分类: tensorflow(4) 目录(?)[+] 本笔记目的 ...
趣谈网络协议笔记-二（第十九讲）
趣谈网络协议笔记-二(第十九讲) HttpDNS:网络世界的地址簿也会指错路自勉勿谓言之不预也 -- 向为祖国牺牲的先烈致敬! 引用 dns缓存刷新时间是多久?dns本地缓存时间介绍 - 东大网管 ...
趣谈网络协议笔记-二（第十八讲）
趣谈网络协议笔记-二(第十八讲) DNS协议:网络世界的地址簿自勉勿谓言之不预也 -- 向为祖国牺牲的先烈致敬! 正文 DNS用于域名解析,但也不仅仅是用于域名解析,不仅仅是将域名转换成IP. 在 ...

Julia并行计算笔记（二）

四、远程调用之一

Julia并行计算笔记（二）相关推荐

最新文章

热门文章