fixed bug in whichproc for empty iterators, updated readme

Author: jishnub

README.md

@@ -35,7 +35,7 @@ There are a total of 36 possible `(x,y,z)` combinations possible given these ran
 
 The package provides versions of `pmap` with an optional reduction. These differ from the one provided by `Distributed` in a few key aspects: firstly, the iterator product of the argument is what is passed to the function and not the arguments by elementwise, so the i-th task will be `Iterator.product(args...)[i]` and not `[x[i] for x in args]`. Specifically the second set of parameters in the example above will be `(2,2,3)` and not `(2,3,4)`.
 
-Secondly, the iterator is passed to the function in batches and not elementwise, and it is left to the function to iterate over the collection. Secondly, the tasks are passed on to processors sorted by rank, so the first task is passed to the first processor and the last to the last processor. The tasks are also approximately evenly distributed across processors, assuming that the function takes an equal amount of time to run for each set of parameters. The function `pmapbatch_elementwise` is also exported that passes the elements to the function one-by-one as unwrapped tuples.
+Secondly, the iterator is passed to the function in batches and not elementwise, and it is left to the function to iterate over the collection. Secondly, the tasks are passed on to processors sorted by rank, so the first task is passed to the first processor and the last to the last worker that has any tasks assigned to it. The tasks are also approximately evenly distributed across processors, assuming that the function takes an equal amount of time to run for each set of parameters. The function `pmapbatch_elementwise` is also exported that passes the elements to the function one-by-one as unwrapped tuples. This is the same as `pmap` where each worker is assigned batches of approximately equal sizes taken from the iterator product.
 
 ### pmapbatch and pmapbatch_elementwise
 
@@ -99,11 +99,11 @@ julia> pmapreduce(x->ones(2).*myid(),x->hcat(x...),1:nworkers())
 
 ## ProductSplit
 
-The package provides an iterator `ProductSplit` that lists that ranges of parameters that would be passed on to each core. This may be achieved using an
+In the above examples we have talked about the tasks being distributed approximately equally among the workers without going into details about the distribution, which is what we describe here. The package provides an iterator `ProductSplit` that lists that ranges of parameters that would be passed on to each core. This may equivalently be achieved using an
 
 ```Iterator.Take{Iterator.Drop{Iterator.ProductIterator}}```
 
-with appropriately chosen parameters, and in many ways a `ProductSplit` behaves similarly. However this iterator supports several extra features such as `O(1)` indexing, which eliminates the need to actually iterate over it in many scenarios.
+with appropriately chosen parameters, and in many ways a `ProductSplit` behaves similarly. However a `ProductSplit` supports several extra features such as `O(1)` indexing, which eliminates the need to actually iterate over it in many scenarios.
 
 The signature of the constructor is 
 

src/ParallelUtilities.jl

@@ -4,28 +4,28 @@ using Reexport
 @reexport using Distributed
 
 export  ProductSplit,
-		evenlyscatterproduct,
-		whichproc,
-		newprocrange,
-		whichproc,
-		procidrange,
-		localindex,
-		procid_and_index,
-		extremadims,
-		extrema_commonlastdim,
-		workersactive,
-		nworkersactive,
-		workerrank,
-		nodenames,
-		gethostnames,
-		nprocs_node,
-		pmapbatch,
-		pmapbatch_elementwise,
-		pmapsum,
-		pmapsum_elementwise,
-		pmapreduce,
-		pmapreduce_commutative,
-		pmapreduce_commutative_elementwise
+	evenlyscatterproduct,
+	ntasks,
+	whichproc,
+	whichproc,
+	procrange_recast,
+	localindex,
+	procid_and_localindex,
+	extremadims,
+	extrema_commonlastdim,
+	workersactive,
+	nworkersactive,
+	workerrank,
+	nodenames,
+	gethostnames,
+	nprocs_node,
+	pmapbatch,
+	pmapbatch_elementwise,
+	pmapsum,
+	pmapsum_elementwise,
+	pmapreduce,
+	pmapreduce_commutative,
+	pmapreduce_commutative_elementwise
 
 # The fundamental iterator that behaves like an Iterator.Take{Iterator.Drop{Iterator.ProductIterator}}
 
@@ -407,13 +407,20 @@ function whichproc(iterators::Tuple,val::Tuple,np::Int)
 	# We may carry out a binary search as the iterators are sorted
 	left,right = 1,np
 
+	val_t = ReverseLexicographicTuple(val)
+
 	while left <= right
 		mid = floor(Int,(left+right)/2)
 		ps = ProductSplit(iterators,np,mid)
 
-		if ReverseLexicographicTuple(val) < ReverseLexicographicTuple(first(ps))
+		# If np is greater than the number of ntasks then it's possible
+		# that ps is empty. In this case the value must be somewhere in
+		# the previous workers. Otherwise each worker has some tasks and 
+		# these are sorted, so carry out a binary seaarch
+
+		if isempty(ps) || val_t < ReverseLexicographicTuple(first(ps))
 			right = mid - 1
-		elseif ReverseLexicographicTuple(val) > ReverseLexicographicTuple(last(ps))
+		elseif val_t > ReverseLexicographicTuple(last(ps))
 			left = mid + 1
 		else
 			return mid
@@ -423,8 +430,10 @@ function whichproc(iterators::Tuple,val::Tuple,np::Int)
 	return nothing
 end
 
+whichproc(iterators::Tuple,::Nothing,np::Int) = nothing
+
 # This function is necessary when we're changing np
-function newprocrange(ps::ProductSplit,np_new::Int)
+function procrange_recast(ps::ProductSplit,np_new::Int)
 
 	if isempty(ps)
 		return 0:-1 # empty range
@@ -450,13 +459,15 @@ function localindex(ps::ProductSplit{T},val::T) where {T}
 	val == first(ps) && return left
 	val == last(ps) && return right
 
+	val_t = ReverseLexicographicTuple(val)
+
 	while left <= right
 		mid = floor(Int,(left+right)/2)
 		val_mid = @inbounds ps[mid]
 
-		if ReverseLexicographicTuple(val) < ReverseLexicographicTuple(val_mid)
+		if val_t < ReverseLexicographicTuple(val_mid)
 			right = mid - 1
-		elseif ReverseLexicographicTuple(val) > ReverseLexicographicTuple(val_mid)
+		elseif val_t > ReverseLexicographicTuple(val_mid)
 			left = mid + 1
 		else
 			return mid
@@ -466,12 +477,14 @@ function localindex(ps::ProductSplit{T},val::T) where {T}
 	return nothing
 end
 
+localindex(::ProductSplit,::Nothing) = nothing
+
 function localindex(iterators::Tuple,val::Tuple,np::Integer,procid::Integer)
 	ps = evenlyscatterproduct(iterators,np,procid)
 	localindex(ps,val)
 end
 
-function procid_and_index(iterators::Tuple,val::Tuple,np::Integer)
+function procid_and_localindex(iterators::Tuple,val::Tuple,np::Integer)
 	procid = whichproc(iterators,val,np)
 	index = localindex(iterators,val,np,procid)
 	return procid,index

test/runtests.jl

@@ -34,8 +34,8 @@ end
 			for np = 1:npmax, p = 1:np
 		        ps = ProductSplit(iters,np,p)
 		        @test collect(ps) == collect(split_product_across_processors_iterators(iters,np,p))
-		        @test ParallelUtilities.ntasks(ps) == ntasks_total
-		        @test ParallelUtilities.ntasks(ps.iterators) == ntasks_total
+		        @test ntasks(ps) == ntasks_total
+		        @test ntasks(ps.iterators) == ntasks_total
 		    end
 
 		    @test_throws ParallelUtilities.ProcessorNumberError ProductSplit(iters,npmax,npmax+1)
@@ -75,24 +75,36 @@ end
     	    ps = ProductSplit(iters,2,2)
     	    @test ps.firstind == div(length(iters[1]),2) + 1
     	    @test ps.lastind == length(iters[1])
+
+    	    for np in length(iters[1])+1:length(iters[1])+10,
+    	    	p in length(iters[1])+1:np
+
+	    	    ps = ProductSplit(iters,np,p)
+	    	    @test ps.firstind == length(iters[1]) + 1
+	    	    @test ps.lastind == length(iters[1])
+	    	end
     	end
     end
 
     @testset "firstlast" begin
         @testset "first" begin
-            for iters in [(1:10,),(1:10,4:6),(1:10,4:6,1:4),(1:2:10,4:1:6)],np=1:10
+            for iters in [(1:10,),(1:10,4:6),(1:10,4:6,1:4),(1:2:10,4:1:6)],
+            	np=1:10ntasks(iters)
+
 	            ps = ProductSplit(iters,np,1)
-	            @test first(ps) == map(first,iters)
+	            @test first(ps) == ( isempty(ps) ? nothing : map(first,iters) )
 	        end
 
 	        iters = (1:1,)
 	        ps = ProductSplit(iters,2length(iters[1]),length(iters[1])+1) # must be empty
 	        @test first(ps) === nothing
         end
         @testset "last" begin
-            for iters in [(1:10,),(1:10,4:6),(1:10,4:6,1:4),(1:2:10,4:1:6)],np=1:10
+            for iters in [(1:10,),(1:10,4:6),(1:10,4:6,1:4),(1:2:10,4:1:6)],
+            	np=1:10ntasks(iters)
+
 	            ps = ProductSplit(iters,np,np)
-	            @test last(ps) == map(last,iters)
+	            @test last(ps) == ( isempty(ps) ? nothing : map(last,iters) )
 	        end
 
 	        iters = (1:1,)
@@ -172,35 +184,45 @@ end
         end
     end
 
-    @testset "whichproc + newprocrange" begin
+    @testset "whichproc + procrange_recast" begin
         np,proc_id = 5,3
         iters = (1:10,4:6,1:4)
         ps = ProductSplit(iters,np,proc_id)
         @test whichproc(iters,first(ps),1) == 1
         @test whichproc(iters,(100,100,100),1) === nothing
-        @test newprocrange(ps,1) == 1:1
+        @test procrange_recast(ps,1) == 1:1
 
-        for np_new in 1:np
+        iters = (1:1,2:2)
+        ps = ProductSplit(iters,np,proc_id)
+        @test whichproc(iters,first(ps),np) === nothing
+        @test whichproc(iters,nothing,np) === nothing
+
+        for np_new in 1:10ntasks(iters)
         	for proc_id_new=1:np_new
 	        	ps_new = ProductSplit(iters,np_new,proc_id_new)
+
 	        	for val in ps_new
+	        		# Should loop only if ps_new is non-empty
 	        		@test whichproc(iters,val,np_new) == proc_id_new
 	        	end
 	        end
 	        procid_new_first = whichproc(iters,first(ps),np_new)
 	        proc_new_last = whichproc(iters,last(ps),np_new)
-        	@test newprocrange(ps,np_new) == procid_new_first:proc_new_last
+        	@test procrange_recast(ps,np_new) == (isempty(ps) ? (0:-1) : (procid_new_first:proc_new_last))
         end
     end
 
     @testset "localindex" begin
 
         for iters in [(1:10,),(1:10,4:6),(1:10,4:6,1:4),(1:2:10,4:1:6)]
-	        for np=1:10,proc_id=1:np
+	        for np=1:10ntasks(iters),proc_id=1:np
 	        	ps = ProductSplit(iters,np,proc_id)
 	        	for (ind,val) in enumerate(ps)
 	        		@test localindex(ps,val) == ind
 	        	end
+	        	if isempty(ps)
+	        		@test localindex(ps,first(ps)) === nothing
+	        	end
 	        end
 	    end
     end