Merge pull request #94 from JuliaParallel/vc/map_localparts

Implement the general case for map_localparts over DArray

Author: andreasnoack

src/DistributedArrays.jl

@@ -2,9 +2,6 @@ __precompile__(true)
 
 module DistributedArrays
 
-using Compat
-import Compat.view
-
 using Primes
 using Primes: factor
 

src/linalg.jl

@@ -43,7 +43,7 @@ function dot(x::DVector, y::DVector)
             @async push!(results, remotecall_fetch((x, y, i) -> dot(localpart(x), fetch(y, i)), x.pids[i], x, y, i))
         end
     end
-    return reduce(@functorize(+), results)
+    return reduce(+, results)
 end
 
 function norm(x::DVector, p::Real = 2)

src/mapreduce.jl

@@ -26,7 +26,7 @@ function _mapreduce(f, opt, d::DArray)
     end
     reduce(opt, results)
 end
-Base.mapreduce(f, opt::Union{typeof(@functorize(|)), typeof(@functorize(&))}, d::DArray) = _mapreduce(f, opt, d)
+Base.mapreduce(f, opt::Union{typeof(|), typeof(&)}, d::DArray) = _mapreduce(f, opt, d)
 Base.mapreduce(f, opt::Function, d::DArray) = _mapreduce(f, opt, d)
 Base.mapreduce(f, opt, d::DArray) = _mapreduce(f, opt, d)
 
@@ -124,26 +124,25 @@ for (fn, fr) in ((:sum, :+),
                  (:minimum, :min),
                  (:any, :|),
                  (:all, :&))
-    @eval (Base.$fn)(d::DArray) = reduce(@functorize($fr), d)
+    @eval (Base.$fn)(d::DArray) = reduce($fr, d)
 end
 
 # mapreduce like
 for (fn, fr1, fr2) in ((:maxabs, :abs, :max),
                        (:minabs, :abs, :min),
                        (:sumabs, :abs, :+),
                        (:sumabs2, :abs2, :+))
-    @eval (Base.$fn)(d::DArray) = mapreduce(@functorize($fr1), @functorize($fr2), d)
+    @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(@functorize(identity)), d::DArray) = mapreduce(f, @functorize($fr), d)
-        (Base.$fn)(f::Base.Predicate, d::DArray) = mapreduce(f, @functorize($fr), d)
-        # (Base.$fn)(f::Base.Func{1}, d::DArray) = mapreduce(f, @functorize $fr, d)
-        (Base.$fn)(f::Callable, d::DArray) = mapreduce(f, @functorize($fr), d)
+        (Base.$fn)(f::typeof(identity), d::DArray) = mapreduce(f, $fr, d)
+        (Base.$fn)(f::Base.Predicate{1}, d::DArray) = mapreduce(f, $fr, d)
+        (Base.$fn)(f::Callable, d::DArray) = mapreduce(f, $fr, d)
     end
 end
 
@@ -208,17 +207,17 @@ for f in (:+, :-, :div, :mod, :rem, :&, :|, :$)
             B = samedist(A, B)
             map_localparts($f, A, B)
         end
-        ($f){T}(A::DArray{T}, B::Array{T}) = ($f)(A, distribute(B, A))
-        ($f){T}(A::Array{T}, B::DArray{T}) = ($f)(distribute(A, B), B)
+        ($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
 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}) = ($f)(A, distribute(B, A))
-        ($f){T}(A::Array{T}, B::DArray{T}) = ($f)(distribute(A, B), B)
+        ($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/runtests.jl

@@ -1,9 +1,4 @@
-if VERSION >= v"0.5.0-dev+7720"
-    using Base.Test
-else
-    using BaseTestNext
-    const Test = BaseTestNext
-end
+using Base.Test
 
 # add at least 3 worker processes
 if nworkers() < 3
@@ -13,8 +8,6 @@ end
 @assert nprocs() > 3
 @assert nworkers() >= 3
 
-using Compat
-import Compat.view
 using DistributedArrays
 using StatsBase # for fit(Histogram, ...)
 @everywhere using StatsBase # because exported functions are not exported on workers with using