WIP: make it work on julia-0.7

Author: traktofon

.travis.yml

@@ -3,7 +3,7 @@ os:
   - linux
   - osx
 julia:
-  - 0.6
+  - 0.7
   - nightly
 matrix:
   # allow_failures:

src/DistributedArrays.jl

@@ -8,7 +8,7 @@ using LinearAlgebra
 
 import Base: +, -, *, div, mod, rem, &, |, xor
 import Base.Callable
-import LinearAlgebra.BLAS: axpy!
+import LinearAlgebra: axpy!, dot, norm,
 
 import Primes
 import Primes: factor

src/darray.jl

@@ -28,8 +28,7 @@ mutable struct DArray{T,N,A} <: AbstractArray{T,N}
     pids::Array{Int,N}                          # pids[i]==p ⇒ processor p has piece i
     indices::Array{NTuple{N,UnitRange{Int}},N}  # indices held by piece i
     cuts::Vector{Vector{Int}}                   # cuts[d][i] = first index of chunk i in dimension d
-    localpart::Union{A,Missing}
-
+    localpart::Union{A,Nothing}
     release::Bool
 
     function DArray{T,N,A}(id, dims, pids, indices, cuts, lp) where {T,N,A}
@@ -64,7 +63,7 @@ function d_from_weakref_or_d(id)
 end
 
 Base.eltype(::Type{DArray{T}}) where {T} = T
-empty_localpart(T,N,A) = A(Array{T}(ntuple(zero, N)))
+empty_localpart(T,N,A) = A(Array{T}(undef, ntuple(zero, N)))
 
 const SubDArray{T,N,D<:DArray} = SubArray{T,N,D}
 const SubOrDArray{T,N} = Union{DArray{T,N}, SubDArray{T,N}}
@@ -148,7 +147,7 @@ function ddata(;T::Type=Any, init::Function=I->nothing, pids=workers(), data::Ve
         d = registry[id]
         d = isa(d, WeakRef) ? d.value : d
     else
-        d = DArray{T,1,T}(id, (npids,), pids, idxs, cuts, missing)
+        d = DArray{T,1,T}(id, (npids,), pids, idxs, cuts, nothing)
     end
     d
 end
@@ -196,10 +195,10 @@ function DArray(refs)
         end
     end
 
-    nindices = Array{NTuple{length(dimdist),UnitRange{Int}}}(dimdist...)
+    nindices = Array{NTuple{length(dimdist),UnitRange{Int}}}(undef, dimdist...)
 
     for i in 1:length(nindices)
-        subidx = CartesianIndices(dimdist)[i]
+        subidx = CartesianIndices(dimdist)[i] #ind2sub(dimdist, i)
         nindices[i] = ntuple(length(subidx)) do x
             idx_in_dim = subidx[x]
             startidx = 1
@@ -243,7 +242,7 @@ DArray(init, d::DArray) = DArray(next_did(), init, size(d), procs(d), d.indices,
 
 sz_localpart_ref(ref, id) = size(fetch(ref))
 
-Base.similar(d::DArray, T::Type, dims::Dims) = DArray(I->Array{T}(map(length,I)), dims, procs(d))
+Base.similar(d::DArray, T::Type, dims::Dims) = DArray(I->Array{T}(undef, map(length,I)), dims, procs(d))
 Base.similar(d::DArray, T::Type) = similar(d, T, size(d))
 Base.similar(d::DArray{T}, dims::Dims) where {T} = similar(d, T, dims)
 Base.similar(d::DArray{T}) where {T} = similar(d, T, size(d))
@@ -288,7 +287,7 @@ function defaultdist(sz::Int, nc::Int)
     if sz >= nc
         return round.(Int, range(1, stop=sz+1, length=nc+1))
     else
-        return [[1:(sz+1);]; zeros(Int, nc-sz);]
+        return [[1:(sz+1);]; zeros(Int, nc-sz)]
     end
 end
 
@@ -488,7 +487,7 @@ function (::Type{Array{S,N}})(s::SubDArray{T,N}) where {S,T,N}
         end
     end
     a = Array{S}(undef, size(s))
-    a[[1:size(a,i) for i=1:N]...] = s
+    a[[1:size(a,i) for i=1:N]...] .= s
     return a
 end
 
@@ -509,7 +508,7 @@ function Base.reshape(A::DArray{T,1,S}, d::Dims) where {T,S<:Array}
         d1offs = first(I[1])
         nd = length(I)
 
-        B = Array{T}(sz)
+        B = Array{T}(undef, sz)
         nr = size(B,1)
         sztail = size(B)[2:end]
 
@@ -565,14 +564,15 @@ function Base.isassigned(D::DArray, i::Integer...)
 end
 
 
-function Base.copy!(dest::SubOrDArray, src::SubOrDArray)
+Base.copyto!(dest::SubOrDArray, src::SubOrDArray) = begin
     asyncmap(procs(dest)) do p
         remotecall_fetch(p) do
             localpart(dest)[:] = src[localindices(dest)...]
         end
     end
     return dest
 end
+Base.copy!(dest::SubOrDArray, src::SubOrDArray) = copyto!(dest, src)
 
 # local copies are obtained by convert(Array, ) or assigning from
 # a SubDArray to a local Array.

src/linalg.jl

@@ -221,7 +221,7 @@ function _matmatmul!(α::Number, A::DMatrix, B::AbstractMatrix, β::Number, C::D
                     if tA == 'T'
                         return transpose(localpart(A))*convert(localtype(B), Bjk)
                     elseif tA == 'C'
-                        return localpart(A)'*convert(localtype(B), Bjk)
+                        return ctranspose(localpart(A))*convert(localtype(B), Bjk)
                     else
                         return localpart(A)*convert(localtype(B), Bjk)
                     end

src/mapreduce.jl

@@ -1,5 +1,7 @@
 ## higher-order functions ##
 
+import Base: +, -, div, mod, rem, &, |, xor
+
 Base.map(f, d0::DArray, ds::AbstractArray...) = broadcast(f, d0, ds...)
 
 function Base.map!(f::F, dest::DArray, src::DArray) where {F}
@@ -12,22 +14,21 @@ function Base.map!(f::F, dest::DArray, src::DArray) where {F}
     return dest
 end
 
-# Base.Broadcast._containertype(::Type{D}) where {D<:DArray} = DArray
-Base.BroadcastStyle(::Type{<:DArray}) = Broadcast.ArrayStyle{DArray}()
+#Base.Broadcast._containertype(::Type{D}) where {D<:DArray} = DArray
 
-Base.Broadcast.promote_containertype(::Type{DArray}, ::Type{DArray}) = DArray
-Base.Broadcast.promote_containertype(::Type{DArray}, ::Type{Array})  = DArray
-Base.Broadcast.promote_containertype(::Type{DArray}, ct)             = DArray
-Base.Broadcast.promote_containertype(::Type{Array}, ::Type{DArray})  = DArray
-Base.Broadcast.promote_containertype(ct, ::Type{DArray})             = DArray
+Base.BroadcastStyle(::Type{DArray}, ::Type{DArray}) = DArray
+Base.BroadcastStyle(::Type{DArray}, ::Type{Array})  = DArray
+Base.BroadcastStyle(::Type{DArray}, ct)             = DArray
+#Base.Broadcast.promote_containertype(::Type{Array}, ::Type{DArray})  = DArray
+#Base.Broadcast.promote_containertype(ct, ::Type{DArray})             = DArray
 
 Base.Broadcast.broadcast_indices(::Type{DArray}, A)      = indices(A)
 Base.Broadcast.broadcast_indices(::Type{DArray}, A::Ref) = ()
 
 # FixMe!
 ## 1. Support for arbitrary indices including OneTo
 ## 2. This is as type unstable as it can be. Overhead might not matter too much for DArrays though.
-function Base.Broadcast.broadcast_c(f, ::Type{DArray}, As...)
+function Base.broadcast(f, ::Type{DArray}, ::Nothing, ::Nothing, As...)
     T     = Base.Broadcast._broadcast_eltype(f, As...)
     shape = Base.Broadcast.broadcast_indices(As...)
     iter  = Base.CartesianIndices(shape)
@@ -89,7 +90,8 @@ function mapreducedim_within(f, op, A::DArray, region)
     gridsize = [size(A.indices)...]
     arraysize[[region...]] = gridsize[[region...]]
     indx = similar(A.indices)
-    for i in CartesianIndices(size(indx))
+
+    for i in CartesianIndices(indx)
         indx[i] = ntuple(j -> j in region ? (i.I[j]:i.I[j]) : A.indices[i][j], ndims(A))
     end
     cuts = [i in region ? collect(1:arraysize[i] + 1) : A.cuts[i] for i in 1:ndims(A)]
@@ -157,24 +159,6 @@ function Base.extrema(d::DArray)
     return reduce((t,s) -> (min(t[1], s[1]), max(t[2], s[2])), r)
 end
 
-# mapreduce like
-# for (fn, fr1, fr2) in ((:abs, :max),
-#                        (:minabs, :abs, :min),
-#                        (:sumabs, :abs, :+),
-#                        (:sumabs2, :abs2, :+))
-#     @eval (Base.$fn)(d::DArray) = mapreduce($fr1, $fr2, d)
-# end
-
-# semi mapreduce
-# for (fn, fr) in ((:any, :|),
-#                  (:all, :&),
-#                  (:count, :+))
-#     @eval begin
-#         (Base.$fn)(f::typeof(identity), d::DArray) = mapreduce(f, $fr, d)
-#         (Base.$fn)(f::Callable, d::DArray) = mapreduce(f, $fr, d)
-#     end
-# end
-
 # Unary vector functions
 (-)(D::DArray) = map(-, D)
 
@@ -361,5 +345,5 @@ function ppeval(f, D...; dim::NTuple = map(t -> isa(t, DArray) ? ndims(t) : 0, D
     # the DArray constructor.
     sd = [size(D[1].pids)...]
     nd = remotecall_fetch((r)->ndims(fetch(r)), refs[1].where, refs[1])
-    DArray(reshape(refs, tuple([sd[1:nd - 1]; sd[end];]...)))
+    DArray(reshape(refs, tuple([sd[1:nd - 1]; sd[end]]...)))
 end

src/serialize.jl

@@ -1,8 +1,8 @@
 function Serialization.serialize(S::AbstractSerializer, d::DArray{T,N,A}) where {T,N,A}
     # Only send the ident for participating workers - we expect the DArray to exist in the
     # remote registry. DO NOT send the localpart.
-    destpid = Distributed.worker_id_from_socket(S.io)
-    Serializer.serialize_type(S, typeof(d))
+    destpid = worker_id_from_socket(S.io)
+    serialize_type(S, typeof(d))
     if (destpid in d.pids) || (destpid == d.id[1])
         serialize(S, (true, d.id))    # (id_only, id)
     else
@@ -43,28 +43,28 @@ end
 
 # Serialize only those parts of the object as required by the destination worker.
 mutable struct DestinationSerializer
-    generate::Union{Function,Missing}     # Function to generate the part to be serialized
-    pids::Union{Array,Missing}            # MUST have the same shape as the distribution
-
-    deser_obj::Union{Any,Missing}         # Deserialized part
+    generate::Union{Function,Nothing}  # Function to generate the part to be serialized
+    pids::Union{Array,Nothing}         # MUST have the same shape as the distribution
+    deser_obj::Any                     # Deserialized part
 
     DestinationSerializer(f,p,d) = new(f,p,d)
 end
 
-DestinationSerializer(f::Function, pids::Array) = DestinationSerializer(f, pids, missing)
+DestinationSerializer(f::Function, pids::Array) = DestinationSerializer(f, pids, nothing)
 
 # contructs a DestinationSerializer after verifying that the shape of pids.
 function verified_destination_serializer(f::Function, pids::Array, verify_size)
     @assert size(pids) == verify_size
     return DestinationSerializer(f, pids)
 end
 
-DestinationSerializer(deser_obj::Any) = DestinationSerializer(missing, missing, deser_obj)
+DestinationSerializer(deser_obj::Any) = DestinationSerializer(nothing, nothing, deser_obj)
 
 function Serialization.serialize(S::AbstractSerializer, s::DestinationSerializer)
-    pid = Distributed.worker_id_from_socket(S.io)
+    pid = worker_id_from_socket(S.io)
     pididx = findfirst(isequal(pid), s.pids)
-    Serialization.serialize_type(S, typeof(s))
+    @assert pididx !== nothing
+    serialize_type(S, typeof(s))
     serialize(S, s.generate(pididx))
 end
 
@@ -75,9 +75,9 @@ end
 
 
 function localpart(s::DestinationSerializer)
-    if !ismissing(s.deser_obj)
+    if s.deser_obj !== nothing
         return s.deser_obj
-    elseif  !ismissing(s.generate) && (myid() in s.pids)
+    elseif s.generate !== nothing && (myid() in s.pids)
         # Handle the special case where myid() is part of s.pids.
         # In this case serialize/deserialize is not called as the remotecall is executed locally
         return s.generate(findfirst(isequal(myid()), s.pids))

src/spmd.jl

@@ -4,15 +4,15 @@ using Distributed
 
 import DistributedArrays: gather, next_did, close
 export sendto, recvfrom, recvfrom_any, barrier, bcast, scatter, gather
-export context_local_storage, context, spmd, close
+export context_local_storage, context, spmd
 
 
 mutable struct WorkerDataChannel
     pid::Int
-    rc::Union{RemoteChannel,Missing}
+    rc::Union{RemoteChannel,Nothing}
     lock::ReentrantLock
 
-    WorkerDataChannel(pid) = new(pid, missing, ReentrantLock())
+    WorkerDataChannel(pid) = new(pid, nothing, ReentrantLock())
 end
 
 mutable struct SPMDContext
@@ -61,7 +61,7 @@ const map_ctxts = Dict{Tuple, SPMDContext}()
 function get_dc(wc::WorkerDataChannel)
     lock(wc.lock)
     try
-        if ismissing(wc.rc)
+        if wc.rc === nothing
             if wc.pid == myid()
                 myrc = RemoteChannel(()->Channel(typemax(Int)))
                 wc.rc = myrc
@@ -254,7 +254,7 @@ function delete_ctxt_id(ctxt_id)
     nothing
 end
 
-function close(ctxt::SPMDContext)
+function Base.close(ctxt::SPMDContext)
     for p in ctxt.pids
         Base.remote_do(delete_ctxt_id, p, ctxt.id)
     end

test/darray.jl

@@ -118,7 +118,8 @@ check_leaks()
     end
 
     @testset "test invalid use of @DArray" begin
-        @test_throws ArgumentError eval(:((@DArray [1,2,3,4])))
+        #@test_throws ArgumentError eval(:((@DArray [1,2,3,4])))
+        @test_throws LoadError eval(:((@DArray [1,2,3,4])))
     end
 end
 

test/runtests.jl

@@ -11,6 +11,7 @@ end
 @everywhere using DistributedArrays
 @everywhere using DistributedArrays.SPMD
 @everywhere using Random
+@everywhere using LinearAlgebra
 
 @everywhere srand(1234 + myid())