Fix broadcasting on Julia 1.0

Author: andreasnoack

src/darray.jl

@@ -568,9 +568,9 @@ function Base.reshape(A::DArray{T,1,S}, d::Dims) where {T,S<:Array}
             i2 = CartesianIndices(sztail)[i]
             globalidx = [ I[j][i2[j-1]] for j=2:nd ]
 
-            a = sub2ind(d, d1offs, globalidx...)
+            a = LinearIndices(d)[d1offs, globalidx...]
 
-            B[:,i] = A[a:(a+nr-1)]
+            B[:,i] = Array(A[a:(a+nr-1)])
         end
         B
     end
@@ -706,15 +706,15 @@ end
 Base.size(P::ProductIndices) = P.sz
 # This gets passed to map to avoid breaking propagation of inbounds
 Base.@propagate_inbounds propagate_getindex(A, I...) = A[I...]
-Base.@propagate_inbounds Base.getindex(P::ProductIndices{_,N}, I::Vararg{Int, N}) where {_,N} =
+Base.@propagate_inbounds Base.getindex(P::ProductIndices{J,N}, I::Vararg{Int, N}) where {J,N} =
     Bool((&)(map(propagate_getindex, P.indices, I)...))
 
 struct MergedIndices{I,N} <: AbstractArray{CartesianIndex{N}, N}
     indices::I
     sz::NTuple{N,Int}
 end
 Base.size(M::MergedIndices) = M.sz
-Base.@propagate_inbounds Base.getindex(M::MergedIndices{_,N}, I::Vararg{Int, N}) where {_,N} =
+Base.@propagate_inbounds Base.getindex(M::MergedIndices{J,N}, I::Vararg{Int, N}) where {J,N} =
     CartesianIndex(map(propagate_getindex, M.indices, I))
 # Additionally, we optimize bounds checking when using MergedIndices as an
 # array index since checking, e.g., A[1:500, 1:500] is *way* faster than

src/mapreduce.jl

@@ -20,37 +20,25 @@ end
 Base.BroadcastStyle(::Type{<:DArray}) = Broadcast.ArrayStyle{DArray}()
 Base.BroadcastStyle(::Type{<:DArray}, ::Any) = Broadcast.ArrayStyle{DArray}()
 
-function Base.similar(bc::Broadcast.Broadcasted{Broadcast.ArrayStyle{DArray}}, ::Type{ElType}) where {ElType}
-   DA = find_darray(bc)
-   DArray(I -> Array{ElType}(undef, map(length,I)), DA)
-end
-
-"`DA = find_darray(As)` returns the first DArray among the arguments."
-find_darray(bc::Base.Broadcast.Broadcasted) = find_darray(bc.args)
-find_darray(args::Tuple) = find_darray(find_darray(args[1]), Base.tail(args))
-find_darray(x) = x
-find_darray(a::DArray, rest) = a
-find_darray(::Any, rest) = find_darray(rest)
-
-function Base.copyto!(dest::DArray, bc::Broadcast.Broadcasted{Broadcast.ArrayStyle{DArray}})
-    @sync for p in procs(dest)
-       @async remotecall_fetch(p) do
-          copyto!(localpart(dest), rewrite_local(bc))
-       end
+function Base.copy(bc::Broadcast.Broadcasted{Broadcast.ArrayStyle{DArray}})
+    T     = Base.Broadcast.combine_eltypes(bc.f, bc.args)
+    shape = Base.Broadcast.combine_axes(bc.args...)
+    iter  = Base.CartesianIndices(shape)
+    D     = DArray(map(length, shape)) do I
+        A = map(bc.args) do a
+            if isa(a, Union{Number,Ref})
+                return a
+            else
+                return localtype(a)(
+                    a[ntuple(i -> i > ndims(a) ? 1 : (size(a, i) == 1 ? (1:1) : I[i]), length(shape))...]
+                    )
+            end
+        end
+        broadcast(bc.f, A...)
     end
-    dest
+    return D
 end
 
-"""
-Transform a Broadcasted{Broadcast.ArrayStyle{DArray}} object into an equivalent
-Broadcasted{Broadcast.DefaultArrayStyle} object for the localparts.
-"""
-rewrite_local(bc::Broadcast.Broadcasted{Broadcast.ArrayStyle{DArray}}) = Broadcast.broadcasted(bc.f, rewrite_local(bc.args)...)
-rewrite_local(args::Tuple) = map(rewrite_local, args)
-rewrite_local(a::DArray) = localpart(a)
-rewrite_local(x) = x
-
-
 function Base.reduce(f, d::DArray)
     results = asyncmap(procs(d)) do p
         remotecall_fetch(p) do

test/darray.jl

@@ -835,6 +835,10 @@ check_leaks()
     c = a .- m
     d = convert(Array, a) .- convert(Array, m)
     @test c == d
+    e = @DArray [ones(10) for i=1:4]
+    f = 2 .* e
+    @test Array(f) == 2 .* Array(e)
+    @test Array(map(x -> sum(x) .+ 2, e)) == map(x -> sum(x) .+ 2, e)
     d_closeall()
 end
 

test/runtests.jl

@@ -20,7 +20,7 @@ end
 @everywhere using Random
 @everywhere using LinearAlgebra
 
-@everywhere srand(1234 + myid())
+@everywhere Random.seed!(1234 + myid())
 
 const MYID = myid()
 const OTHERIDS = filter(id-> id != MYID, procs())[rand(1:(nprocs()-1))]