Merge pull request #120 from JuliaParallel/anj/bc

Define general broadcasting of DArrays

Author: andreasnoack

.travis.yml

@@ -3,11 +3,10 @@ os:
   - linux
   - osx
 julia:
-  - 0.5
   - nightly
 matrix:
-  allow_failures:
-    - julia: nightly
+  # allow_failures:
+    # - julia: nightly
 notifications:
   email: false
 before_install:

REQUIRE

@@ -1,2 +1,2 @@
-julia 0.5-
+julia 0.6-
 Primes

src/darray.jl

@@ -61,7 +61,9 @@ empty_localpart(T,N,A) = convert(A, Array(T, ntuple(zero, N)))
 typealias SubDArray{T,N,D<:DArray} SubArray{T,N,D}
 typealias SubOrDArray{T,N} Union{DArray{T,N}, SubDArray{T,N}}
 
-localtype{T,N,S}(A::DArray{T,N,S}) = S
+localtype{T,N,S}(::Type{DArray{T,N,S}}) = S
+localtype{T,N,D}(::Type{SubDArray{T,N,D}}) = localtype(D)
+localtype(A::SubOrDArray) = localtype(typeof(A))
 localtype(A::AbstractArray) = typeof(A)
 
 ## core constructors ##
@@ -272,7 +274,7 @@ end
 # get array of start indexes for dividing sz into nc chunks
 function defaultdist(sz::Int, nc::Int)
     if sz >= nc
-        return round(Int, linspace(1, sz+1, nc+1))
+        return round.(Int, linspace(1, sz+1, nc+1))
     else
         return [[1:(sz+1);], zeros(Int, nc-sz);]
     end

src/mapreduce.jl

@@ -2,15 +2,38 @@
 
 Base.map(f, d::DArray) = DArray(I->map(f, localpart(d)), d)
 
-Base.map!{F}(f::F, d::DArray) = begin
-    @sync for p in procs(d)
-        @async remotecall_fetch((f,d)->(map!(f, localpart(d)); nothing), p, f, d)
+Base.map!{F}(f::F, dest::DArray, src::DArray) = begin
+    @sync for p in procs(dest)
+        @async remotecall_fetch(() -> (map!(f, localpart(dest), src[localindexes(dest)...]); nothing), p)
     end
-    return d
+    return dest
 end
 
-# FixMe! We'll have to handle the general n args case but it seems tricky
-Base.broadcast(f, d::DArray) = DArray(I -> broadcast(f, localpart(d)), d)
+Base.Broadcast.containertype{D<:DArray}(::Type{D}) = 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.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...)
+    T     = Base.Broadcast._broadcast_eltype(f, As...)
+    shape = Base.Broadcast.broadcast_indices(As...)
+    iter  = Base.CartesianRange(shape)
+    D     = DArray(map(length, shape)) do I
+        Base.Broadcast.broadcast_c(f, Array,
+            map(a -> isa(a, Union{Number,Ref}) ? a :
+                localtype(a)(a[ntuple(i -> i > ndims(a) ? 1 : (size(a, i) == 1 ? (1:1) : I[i]), length(shape))...]), As)...)
+    end
+    return D
+end
 
 function Base.reduce(f, d::DArray)
     results=[]
@@ -148,15 +171,17 @@ end
 # Unary vector functions
 (-)(D::DArray) = map(-, D)
 
-# scalar ops
-(+)(A::DArray{Bool}, x::Bool) = A .+ x
-(+)(x::Bool, A::DArray{Bool}) = x .+ A
-(-)(A::DArray{Bool}, x::Bool) = A .- x
-(-)(x::Bool, A::DArray{Bool}) = x .- A
-(+)(A::DArray, x::Number) = A .+ x
-(+)(x::Number, A::DArray) = x .+ A
-(-)(A::DArray, x::Number) = A .- x
-(-)(x::Number, A::DArray) = x .- A
+@static if VERSION < v"0.6.0-dev.1731"
+    # scalar ops
+    (+)(A::DArray{Bool}, x::Bool) = A .+ x
+    (+)(x::Bool, A::DArray{Bool}) = x .+ A
+    (-)(A::DArray{Bool}, x::Bool) = A .- x
+    (-)(x::Bool, A::DArray{Bool}) = x .- A
+    (+)(A::DArray, x::Number) = A .+ x
+    (+)(x::Number, A::DArray) = x .+ A
+    (-)(A::DArray, x::Number) = A .- x
+    (-)(x::Number, A::DArray) = x .- A
+end
 
 map_localparts(f::Callable, d::DArray) = DArray(i->f(localpart(d)), d)
 map_localparts(f::Callable, d1::DArray, d2::DArray) = DArray(d1) do I
@@ -192,10 +217,12 @@ end
 # the same size and distribution
 map_localparts(f::Callable, As::DArray...) = DArray(I->f(map(localpart, As)...), As[1])
 
-for f in (:.+, :.-, :.*, :./, :.%, :.<<, :.>>, :div, :mod, :rem, :&, :|, :$)
-    @eval begin
-        ($f){T}(A::DArray{T}, B::Number) = map_localparts(r->($f)(r, B), A)
-        ($f){T}(A::Number, B::DArray{T}) = map_localparts(r->($f)(A, r), B)
+@static if VERSION < v"0.6.0-dev.1632"
+    for f in (:.+, :.-, :.*, :./, :.%, :.<<, :.>>, :div, :mod, :rem, :&, :|, :$)
+        @eval begin
+            ($f){T}(A::DArray{T}, B::Number) = map_localparts(r->($f)(r, B), A)
+            ($f){T}(A::Number, B::DArray{T}) = map_localparts(r->($f)(A, r), B)
+        end
     end
 end
 
@@ -217,13 +244,15 @@ for f in (:+, :-, :div, :mod, :rem, :&, :|, :$)
         ($f){T}(A::Array{T}, B::DArray{T}) = map_localparts($f, A, B)
     end
 end
-for f in (:.+, :.-, :.*, :./, :.%, :.<<, :.>>)
-    @eval begin
-        function ($f){T}(A::DArray{T}, B::DArray{T})
-            map_localparts($f, A, B)
+@static if VERSION < v"0.6.0-dev.1632"
+    for f in (:.+, :.-, :.*, :./, :.%, :.<<, :.>>)
+        @eval begin
+            function ($f){T}(A::DArray{T}, B::DArray{T})
+                map_localparts($f, A, B)
+            end
+            ($f){T}(A::DArray{T}, B::Array{T}) = map_localparts($f, A, B)
+            ($f){T}(A::Array{T}, B::DArray{T}) = map_localparts($f, A, B)
         end
-        ($f){T}(A::DArray{T}, B::Array{T}) = map_localparts($f, A, B)
-        ($f){T}(A::Array{T}, B::DArray{T}) = map_localparts($f, A, B)
     end
 end
 

test/darray.jl

@@ -171,7 +171,7 @@ t=@testset "test DArray reduce" begin
     end
 
     @testset "test map! / reduce" begin
-        map!(x->1, D)
+        map!(x->1, D, D)
         @test reduce(+, D) == 100
     end
     close(D)
@@ -343,16 +343,16 @@ end
 check_leaks(t)
 
 t=@testset "test max / min / sum" begin
-    a = map(x->Int(round(rand() * 100)) - 50, Array(Int, 100,1000))
+    a = map(x -> Int(round(rand() * 100)) - 50, Array(Int, 100,1000))
     d = distribute(a)
 
-    @test sum(d) == sum(a)
-    @test maximum(d) == maximum(a)
-    @test minimum(d) == minimum(a)
-    @test maxabs(d) == maxabs(a)
-    @test minabs(d) == minabs(a)
-    @test sumabs(d) == sumabs(a)
-    @test sumabs2(d) == sumabs2(a)
+    @test sum(d)          == sum(a)
+    @test maximum(d)      == maximum(a)
+    @test minimum(d)      == minimum(a)
+    @test maximum(abs, d) == maximum(abs, a)
+    @test minimum(abs, d) == minimum(abs, a)
+    @test sum(abs, d)     == sum(abs, a)
+    @test sum(abs2, d)    == sum(abs2, a)
     close(d)
 end
 
@@ -720,22 +720,22 @@ t=@testset "test scalar ops" begin
     c = drand(20,20)
     d = convert(Array, c)
 
-    @testset "$f" for f in (+, -, .+, .-, .*, ./, .%, div, mod)
+    @testset "$f" for f in (:+, :-, :.+, :.-, :.*, :./, :.%)
         x = rand()
-        @test f(a, x) == f(b, x)
-        @test f(x, a) == f(x, b)
-        @test f(a, c) == f(b, d)
+        @test @eval ($f)($a, $x) == ($f)($b, $x)
+        @test @eval ($f)($x, $a) == ($f)($x, $b)
+        @test @eval ($f)($a, $c) == ($f)($b, $d)
     end
 
     close(a)
     close(c)
 
     a = dones(Int, 20, 20)
     b = convert(Array, a)
-    @testset "$f" for f in (.<<, .>>)
-        @test f(a, 2) == f(b, 2)
-        @test f(2, a) == f(2, b)
-        @test f(a, a) == f(b, b)
+    @testset "$f" for f in (:.<<, :.>>)
+        @test @eval ($f)($a, 2)  == ($f)($b, 2)
+        @test @eval ($f)(2, $a)  == ($f)(2, $b)
+        @test @eval ($f)($a, $a) == ($f)($b, $b)
     end
 
     @testset "$f" for f in (rem,)

test/runtests.jl

@@ -20,8 +20,6 @@ const OTHERIDS = filter(id-> id != MYID, procs())[rand(1:(nprocs()-1))]
 
 # On 0.6, @testset does not display the test description automatically anymore.
 function print_test_desc(t, n=0)
-    VERSION < v"0.6-" && return
-
     println(repeat(" ", n), "Passed : ", t.description)
     for t2 in t.results
         if isa(t2, Base.Test.DefaultTestSet)