Merge pull request JuliaParallel#94 from JuliaParallel/vc/map_localparts

Implement the general case for map_localparts over DArray

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