多处理和分布式计算

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

julia> rand2(2,2)
2×2 Array{Float64,2}:
 0.153756  0.368514
 1.15119   0.918912

julia> fetch(@spawnat :any rand2(2,2))
ERROR: RemoteException(2, CapturedException(UndefVarError(Symbol("#rand2"))))
Stacktrace:
[...]

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

朱莉娅>MyType(7)
类型(7)

julia>fetch（@spawnat2MyType（7））
错误：在worker2上:
UndefVarError:`Mytype`未在'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

朱莉娅>addprocs(2)
2元素数组{Int64,1}:
 2
 3

julia> A = rand(1000,1000);

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

[...]

julia> fetch(Bref);

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

[...]

朱莉娅>fetch(Bref);

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

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>@fetchfrom2InteractiveUtils.varinfo()
名称大小摘要
––––––––– ––––––––– ––––––––––––––––––––––
一个800字节10×10数组{Float64,2}
基本模块
核心模块
主模块

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    # ожидание доступных данных
           DictChannel{T}() where {T} = new(Dict(), Threads.Condition())
           DictChannel() = DictChannel{Any}()
       end

julia> begin
       function Base.put!(D::DictChannel, k, v)
           @lock D.cond_take begin
               D.d[k] = v
               notify(D.cond_take)
           end
           return D
       end
       function Base.take!(D::DictChannel, k)
           @lock D.cond_take begin
               v = fetch(D, k)
               delete!(D.d, k)
               return v
           end
       end
       Base.isready(D::DictChannel) = @lock D.cond_take !isempty(D.d)
       Base.isready(D::DictChannel, k) = @lock D.cond_take haskey(D.d, k)
       function Base.fetch(D::DictChannel, k)
           @lock D.cond_take begin
               wait(D, k)
               return D.d[k]
           end
       end
       function Base.wait(D::DictChannel, k)
           @lock D.cond_take begin
               while !isready(D, k)
                   wait(D.cond_take)
               end
           end
       end
       end;

julia> d = DictChannel();

julia> isready(d)
false

julia> put!(d, :k, :v);

julia> isready(d, :k)
true

julia> fetch(d, :k)
:v

julia> wait(d, :k)

julia> take!(d, :k)
:v

julia> isready(d, :k)
false

julia> addprocs(4); # Добавляем рабочие процессы

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

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

julia> @everywhere function do_work(jobs, results) # Определяем рабочую функцию через everywhere
           while true
               job_id = take!(jobs)
               exec_time = rand()
               sleep(exec_time) # Имитирует время, затрачиваемое на реальную работу
               put!(results, (job_id, exec_time, myid()))
           end
       end

julia> function make_jobs(n)
           for i in 1:n
               put!(jobs, i)
           end
       end;

julia> n = 12;

julia> errormonitor(@async make_jobs(n)); # Подаем в канал jobs n заданий

julia> for p in workers() # Запускаем в рабочих процессах задачи для параллельной обработки запросов
           remote_do(do_work, p, jobs, results)
       end

julia> @elapsed while n > 0 # Выводим результаты
           job_id, exec_time, where = take!(results)
           println("$job_id finished in $(round(exec_time; digits=2)) seconds on worker $where")
           global n = n - 1
       end
1 finished in 0.18 seconds on worker 4
2 finished in 0.26 seconds on worker 5
6 finished in 0.12 seconds on worker 4
7 finished in 0.18 seconds on worker 4
5 finished in 0.35 seconds on worker 5
4 finished in 0.68 seconds on worker 2
3 finished in 0.73 seconds on worker 3
11 finished in 0.01 seconds on worker 3
12 finished in 0.02 seconds on worker 3
9 finished in 0.26 seconds on worker 5
8 finished in 0.57 seconds on worker 4
10 finished in 0.58 seconds on worker 2
0.055971741

julia> using Distributed;

julia> rc = RemoteChannel(()->Channel(3));   # RemoteChannel создается в локальном узле

julia> v = [0];

julia> for i in 1:3
           v[1] = i                          # Используем `v` повторно
           put!(rc, v)
       end;

julia>结果=[拿！(rc)对于_在1:3];

julia>println（结果）;
阵列{Int64,1}[[3], [3], [3]]

julia>println("Num Unique objects:",length(unique(map(objectid,result))));
Num唯一对象：1

朱莉娅>addprocs(1);

julia>rc=RemoteChannel(()->Channel(3),workers()[1]);#RemoteChannel在远程节点中创建

朱莉娅>v=[0];

朱莉娅>对于我在1:3
           v[1]=i
           放！(rc,v)
       结束;

julia>结果=[拿！(rc)对于_在1:3];

julia>println（结果）;
阵列{Int64,1}[[1], [2], [3]]

julia>println("Num Unique objects:",length(unique(map(objectid,result))));
Num唯一对象：3

julia> using Distributed; addprocs(1);

julia> v = [0];

julia> v2 = remotecall_fetch(x->(x[1] = 1; x), myid(), v);     # Выполняется в локальном узле

julia>println（"v=$v，v2=$v2，"，v===v2）;
v=[1]，v2=[1]，真

朱莉娅>v=[0];

julia>v2=remotecall_fetch(x->(x[1]=1;x),workers()[1],v);#在远程节点上运行

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 Array{Int64,1}:
 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 SharedArray{Int64,2}:
 2  2  3  4
 2  3  3  4
 2  3  4  4

julia> S[3,2] = 7
7

julia> S
3×4 SharedArray{Int64,2}:
 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 SharedArray{Int64,2}:
 2  2  2  2
 3  3  3  3
 4  4  4  4

@sync begin
    for p in procs(S)
        @async 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 # Этому рабочему процессу часть не назначена.
               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)  # Выведем на экран, чтобы знать, что происходит.
           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)
                   @async 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
    # Поля, общие для всех диспетчеров кластеров
    io::Union{IO, Nothing}
    host::Union{AbstractString, Nothing}
    port::Union{Integer, Nothing}

    # Используется при запуске дополнительных рабочих процессов на хосте
    count::Union{Int, Symbol, Nothing}
    exename::Union{AbstractString, Cmd, Nothing}
    exeflags::Union{Cmd, Nothing}

    # Внешние диспетчеры кластеров могут использовать этот объект для хранения информации на уровне отдельного рабочего процесса
    # Для хранения нескольких полей можно использовать словарь.
    userdata::Any

    # Подключения к рабочим процессам через SSHManager или туннель SSH
    tunnel::Union{Bool, Nothing}
    bind_addr::Union{AbstractString, Nothing}
    sshflags::Union{Cmd, Nothing}
    max_parallel::Union{Integer, Nothing}

    # Используется локальными диспетчерами или диспетчерами SSH
    connect_at::Any

    [...]
end

$ ./julia -p 4

julia> addprocs()

julia> @everywhere using DistributedArrays

julia> using CUDA

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

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

julia> C = 2 .* A ./ B;

julia> all(C .≈ 4*π)
true

julia> typeof(C)
Array{Float64,1}

julia> dB = distribute(B);

julia> dA = distribute(A);

julia> dC = 2 .* dA ./ dB;

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

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

朱莉娅>幼崽=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)  # или (M*v) ./ v
end

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

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

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

julia> cuM = CuArray(M);

julia> cuv = CuArray(v);

julia>cuesult=power_method（暨，cuv）;

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

julia>dM=分布(M);

julia>dv=分布（v）;

julia>dC=power_method（dM，dv）;

朱莉娅>打字(dC)
元组{DistributedArrays.DArray{Float64,1,Array{Float64,1}}，浮64}

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