多处理和分布式计算

julia> function rand2(dims...)
           return 2*rand(dims...)
       end

朱莉娅>rand2（2,2）
2×2矩阵{Float64}:
 0.153756  0.368514
 1.15119   0.918912

julia>fetch（@spawnat：any rand2（2,2））
错误：RemoteException（2，CapturedException（UndefVarError（Symbol（"＃rand2"）））））
[医]堆垛机:
[...]

module DummyModule

export MyType, f

mutable struct MyType
    a::Int
end

f(x) = x^2+1

println("loaded")

end

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

julia> using .DummyModule

julia> MyType(7)
MyType(7)

julia> fetch(@spawnat 2 MyType(7))
ERROR: On worker 2:
UndefVarError: `MyType` not defined in `Main`
⋮

julia>fetch（@spawnat2DummyModule。MyType(7))
类型(7)

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

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

julia> using Distributed

julia> addprocs(2)
2-element Vector{Int64}:
 2
 3

julia> A = rand(1000,1000);

julia> Bref = @spawnat :any A^2;

[...]

julia> fetch(Bref);

julia> Bref = @spawnat :any rand(1000,1000)^2;

[...]

julia> fetch(Bref);

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

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

julia> A = rand(10,10);

julia> remotecall_fetch(()->A, 2);

julia> B = rand(10,10);

julia> let B = B
           remotecall_fetch(()->B, 2)
       end;

julia> @fetchfrom 2 InteractiveUtils.varinfo()
name           size summary
––––––––– ––––––––– ––––––––––––––––––––––
A         800 bytes 10×10 Array{Float64,2}
Base                Module
Core                Module
Main                Module

function count_heads(n)
    c::Int = 0
    for i = 1:n
        c += rand(Bool)
    end
    c
end

julia> @everywhere include_string(Main, $(read("count_heads.jl", String)), "count_heads.jl")

julia> a = @spawnat :any count_heads(100000000)
Future(2, 1, 6, nothing)

julia> b = @spawnat :any count_heads(100000000)
Future(3, 1, 7, nothing)

julia> fetch(a)+fetch(b)
100001564

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

a = zeros(100000)
@distributed for i = 1:100000
    a[i] = i
end

using SharedArrays

a = SharedArray{Float64}(10)
@distributed for i = 1:10
    a[i] = i
end

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

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

julia> pmap(svdvals, M);

julia> struct DictChannel{T} <: AbstractChannel{T}
           d::Dict
           cond_take::Threads.Condition    # waiting for data to become available
           DictChannel{T}() where {T} = new(Dict(), Threads.Condition())
           DictChannel() = DictChannel{Any}()
       end

朱莉娅>开始
       功能基。放！(D::DictChannel,k,v)
           @锁D.cond_take开始
               D.d[k]=v
               通知(d.cond_take)
           结束
           返回D
       结束
       功能基。拿！(D::DictChannel,k)
           @锁D.cond_take开始
               v=取(D,k)
               删除！（D.d，k）
               返回v
           结束
       结束
       基地。isready（D::DictChannel）=@lock D.cond_take！isempty(d.d)
       基地。isready(D::DictChannel,k)=@lock d.cond_take haskey(D.d,k)
       功能基。获取(D::DictChannel,k)
           @锁D.cond_take开始
               等待(D,k)
               返回D.d[k]
           结束
       结束
       功能基。等待(D::DictChannel,k)
           @锁D.cond_take开始
               而！已准备好(D,k)
                   等待（d.cond_take）
               结束
           结束
       结束
       结束;

朱莉娅>d=DictChannel（）;

朱莉娅>已准备好(d)
错误

朱莉娅>放！(d,:k,:v);

朱莉娅>isready(d,:k)
真的

朱莉娅>fetch(d,:k)
:v

julia>等待(d,:k)

茱莉亚>拿！(d,:k)
:v

朱莉娅>isready(d,:k)
错误

julia> addprocs(4); # add worker processes

julia> const jobs = RemoteChannel(()->Channel{Int}(32));

julia> const results = RemoteChannel(()->Channel{Tuple}(32));

julia>@everywhere function do_work(jobs,results)#定义工作函数everywhere
           虽然真实
               job_id=拿！(职位)
               exec_time=兰德（）
               sleep(exec_time)#模拟实际工作经过的时间
               放！（结果，（job_id，exec_time，myid（）））
           结束
       结束

julia>函数make_jobs(n)
           对于我在1：n
               放！(工作,我)
           结束
       结束;

朱莉娅>n=12;

julia>errormonitor（线程。@spawn make_jobs(n));#用"n"个作业馈送作业通道

julia>for p in workers()#在workers上启动任务以并行处理请求
           remote_do（do_work，p，jobs，results）
       结束

julia>@elased while n>0#打印出结果
           job_id，exec_time，where=take！(结果)
           println("worker job_id在worker(round(exec_time;digits=2))seconds on worker where where")上完成
           全局n=n-1
       结束
1在工人4上以0.18秒完成
2在工人5上以0.26秒完成
6在工人4上以0.12秒完成
7在工人4上以0.18秒完成
工人5在0.35秒内完成
4在工人2上以0.68秒完成
3在工人3上以0.73秒完成
11在工人3上以0.01秒完成
12在工人3上以0.02秒完成
9在工人5上以0.26秒完成
8在工人4上以0.57秒完成
10在工人2上以0.58秒完成
0.055971741

julia> using Distributed

julia> rc = RemoteChannel(()->Channel(3));   # RemoteChannel created on local node

julia> v = [0];

julia> for i in 1:3
           v[1] = i                          # Reusing `v`
           put!(rc, v)
       end;

julia> result = [take!(rc) for _ in 1:3];

julia> println(result);
[[3], [3], [3]]

julia> println("Num Unique objects : ", length(unique(map(objectid, result))));
Num Unique objects : 1

julia> addprocs(1);

julia> rc = RemoteChannel(()->Channel(3), workers()[1]);   # RemoteChannel created on remote node

julia> v = [0];

julia> for i in 1:3
           v[1] = i
           put!(rc, v)
       end;

julia> result = [take!(rc) for _ in 1:3];

julia> println(result);
[[1], [2], [3]]

julia> println("Num Unique objects : ", length(unique(map(objectid, result))));
Num Unique objects : 3

julia> using Distributed; addprocs(1);

julia> v = [0];

julia> v2 = remotecall_fetch(x->(x[1] = 1; x), myid(), v);     # Executed on local node

julia> println("v=$v, v2=$v2, ", v === v2);
v=[1], v2=[1], true

julia> v = [0];

julia> v2 = remotecall_fetch(x->(x[1] = 1; x), workers()[1], v); # Executed on remote node

julia> println("v=$v, v2=$v2, ", v === v2);
v=[0], v2=[1], false

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

julia> using Distributed

julia> addprocs(3)
3-element Vector{Int64}:
 2
 3
 4

julia> @everywhere using SharedArrays

julia> S = SharedArray{Int,2}((3,4), init = S -> S[localindices(S)] = repeat([myid()], length(localindices(S))))
3×4 SharedMatrix{Int64}:
 2  2  3  4
 2  3  3  4
 2  3  4  4

julia> S[3,2] = 7
7

julia> S
3×4 SharedMatrix{Int64}:
 2  2  3  4
 2  3  3  4
 2  7  4  4

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

@sync begin
    for p in procs(S)
        Threads.@spawn begin
            remotecall_wait(fill!, p, S, p)
        end
    end
end

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

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

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

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

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

julia> function advection_parallel!(q, u)
           for t = 1:size(q,3)-1
               @sync @distributed for j = 1:size(q,2)
                   for i = 1:size(q,1)
                       q[i,j,t+1]= q[i,j,t] + u[i,j,t]
                   end
               end
           end
           q
       end;

julia> function advection_shared!(q, u)
           @sync begin
               for p in procs(q)
                   Threads.@spawn remotecall_wait(advection_shared_chunk!, p, q, u)
               end
           end
           q
       end;

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

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

julia> @time advection_serial!(q, u);
(irange,jrange,trange) = (1:500,1:500,1:499)
 830.220 milliseconds (216 allocations: 13820 bytes)

julia> @time advection_parallel!(q, u);
   2.495 seconds      (3999 k allocations: 289 MB, 2.09% gc time)

julia> @time advection_shared!(q,u);
        From worker 2:       (irange,jrange,trange) = (1:500,1:125,1:499)
        From worker 4:       (irange,jrange,trange) = (1:500,251:375,1:499)
        From worker 3:       (irange,jrange,trange) = (1:500,126:250,1:499)
        From worker 5:       (irange,jrange,trange) = (1:500,376:500,1:499)
 238.119 milliseconds (2264 allocations: 169 KB)

function launch(manager::FooManager, params::Dict, launched::Array, c::Condition)
    [...]
end

function manage(manager::FooManager, id::Integer, config::WorkerConfig, op::Symbol)
    [...]
end

struct LocalManager <: ClusterManager
    np::Integer
end

function launch(manager::LocalManager, params::Dict, launched::Array, c::Condition)
    [...]
end

function manage(manager::LocalManager, id::Integer, config::WorkerConfig, op::Symbol)
    [...]
end

mutable struct WorkerConfig
    # Common fields relevant to all cluster managers
    io::Union{IO, Nothing}
    host::Union{AbstractString, Nothing}
    port::Union{Integer, Nothing}

    # Used when launching additional workers at a host
    count::Union{Int, Symbol, Nothing}
    exename::Union{AbstractString, Cmd, Nothing}
    exeflags::Union{Cmd, Nothing}

    # External cluster managers can use this to store information at a per-worker level
    # Can be a dict if multiple fields need to be stored.
    userdata::Any

    # SSHManager / SSH tunnel connections to workers
    tunnel::Union{Bool, Nothing}
    bind_addr::Union{AbstractString, Nothing}
    sshflags::Union{Cmd, Nothing}
    max_parallel::Union{Integer, Nothing}

    # Used by Local/SSH managers
    connect_at::Any

    [...]
end

$ ./julia -p 4

julia> addprocs()

julia> @everywhere using DistributedArrays

julia>使用CUDA

julia>B=ones(10_000)。/ 2;

julia>A=ones(10_000)。* π;

朱莉娅>C=2。*A．/B;

朱莉娅>所有（C。≈ 4*π)
真的

朱莉娅>打字(C)
向量资料{Float64} （数组的别名{Float64, 1})

julia>dB=distribute(B);

julia>dA=分发(A);

julia>dC=2。*达。/分贝;

朱莉娅>所有（dC。≈ 4*π)
真的

朱莉娅>打字(dC)
分发的。阵列;阵列{Float64,1,Vector{Float64}}

朱莉娅>幼崽=CuArray（B）;

julia>cuA=CuArray（A）;

朱莉娅>cuC=2。*cuA。/幼崽;

朱莉娅>所有（cuC。≈ 4*π);
真的

朱莉娅>typeof(cuC)
[医]阵列{Float64,1}

function power_method(M, v)
    for i in 1:100
        v = M*v
        v /= norm(v)
    end

    return v, norm(M*v) / norm(v)  # or  (M*v) ./ v
end

julia> M = [2. 1; 1 1];

julia> v = rand(2)
2-element Vector{Float64}:
0.40395
0.445877

julia> power_method(M,v)
([0.850651, 0.525731], 2.618033988749895)

julia> cuM = CuArray(M);

julia> cuv = CuArray(v);

julia> curesult = power_method(cuM, cuv);

julia> typeof(curesult)
CuArray{Float64,1}

julia> dM = distribute(M);

julia> dv = distribute(v);

julia> dC = power_method(dM, dv);

julia> typeof(dC)
Tuple{DistributedArrays.DArray{Float64,1,Vector{Float64}},Float64}

import MPI

MPI.Init()

comm = MPI.COMM_WORLD
MPI.Barrier(comm)

root = 0
r = MPI.Comm_rank(comm)

sr = MPI.Reduce(r, MPI.SUM, root, comm)

if(MPI.Comm_rank(comm) == root)
   @printf("sum of ranks: %s\n", sr)
end

MPI.Finalize()

mpirun -np 4 ./julia example.jl