tests and bugfixes

Author: jishnub

README.md

@@ -2,6 +2,7 @@
 
 [![Build Status](https://travis-ci.com/jishnub/ParallelUtilities.jl.svg?branch=master)](https://travis-ci.com/jishnub/ParallelUtilities.jl)
 [![Coverage Status](https://coveralls.io/repos/github/jishnub/ParallelUtilities.jl/badge.svg?branch=master)](https://coveralls.io/github/jishnub/ParallelUtilities.jl?branch=master)
+[![codecov](https://codecov.io/gh/jishnub/ParallelUtilities.jl/branch/master/graph/badge.svg)](https://codecov.io/gh/jishnub/ParallelUtilities.jl)
 
 Parallel mapreduce and other helpful functions for HPC, meant primarily for embarassingly parallel operations that often require one to split up a list of tasks into subsections that can be processed on individual cores.
 

src/ParallelUtilities.jl

@@ -675,13 +675,13 @@ function pmapreduce(fmap::Function,freduce::Function,iterable::Tuple,args...;kwa
 
 	nprocs_node_dict = nprocs_node(procs_used)
 	node_channels = Dict(
-		node=>RemoteChannel(()->Channel{Any}(nprocs_node_dict[node]),procid_node)
+		node=>RemoteChannel(()->Channel{pval}(nprocs_node_dict[node]),procid_node)
 			for (node,procid_node) in zip(nodes,procid_rank1_on_node))
 
 	# Worker at which the final reduction takes place
 	p_final = first(procid_rank1_on_node)
 
-	finalnode_reducechannel = RemoteChannel(()->Channel{Any}(length(procid_rank1_on_node)),p_final)
+	finalnode_reducechannel = RemoteChannel(()->Channel{pval}(length(procid_rank1_on_node)),p_final)
 
 	result_channel = RemoteChannel(()->Channel{Any}(1))
 
@@ -761,10 +761,12 @@ end
 # pmap in batches without reduction
 ############################################################################################
 
-function pmapbatch(f::Function,iterable::Tuple,args...;kwargs...)
+function pmapbatch(f::Function,iterable::Tuple,args...;
+	num_workers = nworkersactive(iterable),kwargs...)
+
 	procs_used = workersactive(iterable)
-	num_workers = get(kwargs,:num_workers,length(procs_used))
-	if num_workers<length(procs_used)
+
+	if num_workers < length(procs_used)
 		procs_used = procs_used[1:num_workers]
 	end
 	num_workers = length(procs_used)

test/runtests.jl

@@ -215,6 +215,7 @@ end
         ps = ProductSplit(iters,np,proc_id)
         @test whichproc(iters,first(ps),np) === nothing
         @test whichproc(iters,nothing,np) === nothing
+        @test procrange_recast(ps,2) == (0:-1)
 
         iters = (1:1,2:2)
         ps = ProductSplit(iters,1,1)
@@ -265,6 +266,20 @@ end
 	        end
 	    end
     end
+
+    @testset "getindex" 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:ntasks(iters),p=1:np
+            	ps = ProductSplit(iters,np,p)
+            	ps_col = collect(ps)
+            	for i in 1:length(ps)
+            		@test ps[i] == ps_col[i]
+            	end
+            	@test_throws ParallelUtilities.BoundsErrorPS ps[0]
+            	@test_throws ParallelUtilities.BoundsErrorPS ps[length(ps)+1]
+            end
+        end
+    end
 end
 
 @testset "ReverseLexicographicTuple" begin
@@ -327,6 +342,10 @@ end
         itp = Iterators.product(iters...)
         @test nworkersactive(itp) == nworkersactive(iters)
         @test workersactive(itp) == workersactive(iters)
+
+        ps = ProductSplit(iters,2,1)
+        @test nworkersactive(ps) == nworkersactive(iters)
+        @test workersactive(ps) == workersactive(iters)
     end
 
     @testset "hostnames" begin
@@ -354,6 +373,8 @@ end
 			    @test res == workers()
 			    res = pmapbatch(x->myid(),(iterable,1:1))
 			    @test res == workers()
+			    res = pmapbatch(x->myid(),iterable,num_workers=1)
+			    @test res == workers()[1:1]
 
 			    iterable = 1:nworkers()-1
 			    res = pmapbatch(x->myid(),iterable)
@@ -373,6 +394,11 @@ end
 			    res = pmapbatch(identity,iterable)
 			    resexp = [ProductSplit((iterable,),nworkersactive(iterable),p) for p=1:nworkersactive(iterable)]
 				@test res == resexp
+			    
+			    iterable = 1:2nworkers()
+			    res = pmapbatch(identity,Iterators.product(iterable))
+			    resexp = [ProductSplit((iterable,),nworkersactive(iterable),p) for p=1:nworkersactive(iterable)]
+				@test res == resexp
 			end
 
 			@testset "errors" begin