fixups in PartitionedSerializer. Handle case where ser/deser is not (#92)

called on the PartitionedSerializer object.

Author: amitmurthy

src/DistributedArrays.jl

@@ -1523,7 +1523,7 @@ type PartitionedSerializer
 
    PartitionedSerializer(obj, local_idxs::Tuple) = new(obj, Nullable{Array}(), Nullable{Array}(), local_idxs)
    function PartitionedSerializer(obj, pids::Array, idxs::Array)
-        pas = new(obj,pids,idxs,Nullable{Array}())
+        pas = new(obj,pids,idxs,Nullable{Tuple}())
 
         if myid() in pids
             pas.local_idxs = idxs[findfirst(pids, myid())]
@@ -1540,16 +1540,22 @@ function Base.serialize(S::AbstractSerializer, pas::PartitionedSerializer)
     serialize(S, I)
 end
 
-
 function Base.deserialize{T<:PartitionedSerializer}(S::AbstractSerializer, t::Type{T})
     obj_part = deserialize(S)
     I = deserialize(S)
     return PartitionedSerializer(obj_part, I)
 end
 
 function verify_and_get(pas::PartitionedSerializer, I)
-    @assert I == get(pas.local_idxs, [])
-    return pas.indexable_obj
+    # Handle the special case where myid() is part of pas.pids.
+    # For this case serialize/deserialize is not called as the remotecall is executed locally
+    if myid() in get(pas.pids, [])
+        @assert I == get(pas.idxs)[findfirst(get(pas.pids),myid())]
+        return pas.indexable_obj[I...]
+    else
+        @assert I == get(pas.local_idxs, ())
+        return pas.indexable_obj
+    end
 end
 
 end # module

test/darray.jl

@@ -9,23 +9,26 @@ function check_leaks()
 end
 
 facts("test distribute") do
-    A = randn(100,100)
+    A = rand(1:100, (100,100))
 
     context("test default distribute") do
         DA = distribute(A)
         @fact length(procs(DA)) --> nworkers()
+        @fact sum(DA) --> sum(A)
         close(DA)
     end
 
     context("test distribute with procs arguments") do
-        DA = distribute(A, procs = [1, 2])
-        @fact length(procs(DA)) --> 2
+        DA = distribute(A, procs = procs())
+        @fact length(procs(DA)) --> nprocs()
+        @fact sum(DA) --> sum(A)
         close(DA)
     end
 
     context("test distribute with procs and dist arguments") do
         DA = distribute(A, procs = [1, 2], dist = [1,2])
         @fact size(procs(DA)) --> (1,2)
+        @fact sum(DA) --> sum(A)
         close(DA)
     end