pmapreduce progress bar

show number of map and reduce operations completed in progress bar

Author: jishnub

Project.toml

@@ -1,14 +1,16 @@
 name = "ParallelUtilities"
 uuid = "fad6cfc8-4f83-11e9-06cc-151124046ad0"
 authors = ["Jishnu Bhattacharya <jishnuonline@gmail.com>"]
-version = "0.5.0"
+version = "0.6.0"
 
 [deps]
 Distributed = "8ba89e20-285c-5b6f-9357-94700520ee1b"
+ProgressMeter = "92933f4c-e287-5a05-a399-4b506db050ca"
 Reexport = "189a3867-3050-52da-a836-e630ba90ab69"
 
 [compat]
 Reexport = "0.2"
+ProgressMeter = "1.2"
 julia = "1.2"
 
 [extras]

src/ParallelUtilities.jl

@@ -1,4 +1,5 @@
 module ParallelUtilities
+using ProgressMeter
 
 using Reexport
 @reexport using Distributed
@@ -763,7 +764,8 @@ end
 	put!(pipe.selfchannels.out,valT)
 end
 
-function mapTreeNode(fmap::Function,iterator,pipe::BranchChannel,args...;kwargs...)
+function mapTreeNode(fmap::Function,iterator,pipe::BranchChannel,
+	progress::RemoteChannel,args...;kwargs...)
 	# Evaluate the function
 	# Store the error flag locally
 	# If there are no errors then store the result locally
@@ -775,6 +777,8 @@ function mapTreeNode(fmap::Function,iterator,pipe::BranchChannel,args...;kwargs.
 	catch
 		put!(pipe.selfchannels.err,true)
 		rethrow()
+	finally
+		put!(progress,(true,false))
 	end
 end
 
@@ -813,7 +817,8 @@ function reducedvalue(freduce::Function,pipe::BranchChannel{Tmap,Tred},::Sorted)
 	Tred(freduce(value(v) for v in vals))
 end
 
-function reduceTreeNode(freduce::Function,pipe::BranchChannel{Tmap,Tred},ifsort::Ordering) where {Tmap,Tred}
+function reduceTreeNode(freduce::Function,pipe::BranchChannel{Tmap,Tred},
+	ifsort::Ordering,progress::RemoteChannel) where {Tmap,Tred}
 	# This function that communicates with the parent and children
 
 	# Start by checking if there is any error locally in the map,
@@ -831,9 +836,12 @@ function reduceTreeNode(freduce::Function,pipe::BranchChannel{Tmap,Tred},ifsort:
 		catch e
 			put!(pipe.parentchannels.err,true)
 			rethrow()
+		finally
+			put!(progress,(false,true))
 		end
 	else
 		put!(pipe.parentchannels.err,true)
+		put!(progress,(false,true))
 	end
 
 	finalize(pipe)
@@ -853,15 +861,34 @@ function pmapreduceworkers(fmap::Function,freduce::Function,iterators::Tuple,
 
 	num_workers_active = nworkersactive(iterators)
 
-	# Run the function on each processor and compute the reduction at each node
-	@sync for (rank,mypipe) in enumerate(branches)
-		@async begin
-			p = mypipe.p
-			iterable_on_proc = evenlyscatterproduct(iterators,num_workers_active,rank)
+	nmap,nred = 0,0
+	progresschannel = RemoteChannel(()->Channel{Tuple{Bool,Bool}}(2num_workers_active))
+	progressbar = Progress(2num_workers_active,1,"Progress in pmapreduce : ")
 
-			@spawnat p mapTreeNode(fmap,iterable_on_proc,mypipe,args...;kwargs...)
-			@spawnat p reduceTreeNode(freduce,mypipe,ord)
+	# Run the function on each processor and compute the reduction at each node
+	@sync begin
+		for (rank,mypipe) in enumerate(branches)
+			@async begin
+				p = mypipe.p
+				iterable_on_proc = evenlyscatterproduct(iterators,num_workers_active,rank)
+
+				@spawnat p mapTreeNode(fmap,iterable_on_proc,mypipe,
+										progresschannel,args...;kwargs...)
+				@spawnat p reduceTreeNode(freduce,mypipe,ord,progresschannel)
+			end
+		end
+		
+		for i = 1:2num_workers_active
+			mapdone,reddone = take!(progresschannel)
+			if mapdone
+				nmap += 1
+			end
+			if reddone
+				nred += 1
+			end
+			next!(progressbar;showvalues=[(:map,nmap),(:reduce,nred)])
 		end
+		finish!(progressbar)
 	end
 
 	return_unless_error(first(branches))

test/runtests.jl

@@ -8,7 +8,7 @@ addprocs(2)
     Pkg.activate(".")
     using ParallelUtilities
     import ParallelUtilities: BinaryTreeNode, RemoteChannelContainer, BranchChannel, 
-	Sorted, Unsorted, Ordering, pval, value, reducedvalue, reduceTreeNode,
+	Sorted, Unsorted, Ordering, pval, value, reducedvalue, reduceTreeNode, mapTreeNode,
     BinaryTree, parentnoderank, nchildren, infer_returntypes, maybepvalput!
 end
 
@@ -861,7 +861,8 @@ end
 		end
 
 		function test_on_pipe(fn,iterator,pipe,result_expected)
-			@test_throws ErrorException ParallelUtilities.mapTreeNode(x->error(""),iterator,pipe)
+            progressrc = RemoteChannel(()->Channel{Tuple{Bool,Bool}}(2nworkers()))
+			@test_throws ErrorException mapTreeNode(x->error(""),iterator,pipe,progressrc)
 			@test !isready(pipe.selfchannels.out) # should not have any result as there was an error
 			@test isready(pipe.selfchannels.err)
 			@test take!(pipe.selfchannels.err) # error flag should be true
@@ -871,7 +872,7 @@ end
 			@test !isready(pipe.childrenchannels.out)
 			@test !isready(pipe.childrenchannels.err)
 
-			ParallelUtilities.mapTreeNode(fn,iterator,pipe)
+			mapTreeNode(fn,iterator,pipe,progressrc)
 			@test isready(pipe.selfchannels.err)
 			@test !take!(pipe.selfchannels.err) # error flag should be false
 			@test !isready(pipe.selfchannels.err)
@@ -1001,14 +1002,16 @@ end
 			@testset "reduceTreeNode" begin
 
 		    	function testreduction(freduce::Function,pipe::BranchChannel,
-			    		ifsorted::Ordering,res_exp)
+		    		ifsorted::Ordering,res_exp)
 
 		    		@test !isready(pipe.parentchannels.out)
 		    		@test !isready(pipe.parentchannels.err)
 
+                    progressrc = RemoteChannel(()->Channel{Tuple{Bool,Bool}}(2nworkers()))
+
 		    		try
 			    		wait(@spawnat pipe.p putselfchildren!(pipe,ifsorted))
-			    		reduceTreeNode(freduce,pipe,ifsorted)
+			    		reduceTreeNode(freduce,pipe,ifsorted,progressrc)
 			    	catch
 			    		rethrow()
 			    	end