distributed for in pmapsum

Author: jishnub

src/ParallelUtilities.jl

@@ -242,104 +242,112 @@ end
 
 get_nprocs_node(procs_used::Vector{<:Integer}=workers()) = get_nprocs_node(get_hostnames(procs_used))
 
-function pmapsum(::Type{T},f::Function,iterable,args...;kwargs...) where {T}
+# function pmapsum(::Type{T},f::Function,iterable,args...;kwargs...) where {T}
 
-	procs_used = workers_active(iterable)
+# 	procs_used = workers_active(iterable)
 
-	num_workers = length(procs_used);
-	hostnames = get_hostnames(procs_used);
-	nodes = get_nodes(hostnames);
-	pid_rank0_on_node = [procs_used[findfirst(isequal(node),hostnames)] for node in nodes];
+# 	num_workers = length(procs_used);
+# 	hostnames = get_hostnames(procs_used);
+# 	nodes = get_nodes(hostnames);
+# 	pid_rank0_on_node = [procs_used[findfirst(isequal(node),hostnames)] for node in nodes];
 
-	nprocs_node = get_nprocs_node(procs_used)
-	node_channels = Dict(node=>RemoteChannel(()->Channel{T}(nprocs_node[node]),pid_node)
-		for (node,pid_node) in zip(nodes,pid_rank0_on_node))
+# 	nprocs_node = get_nprocs_node(procs_used)
+# 	node_channels = Dict(node=>RemoteChannel(()->Channel{T}(nprocs_node[node]),pid_node)
+# 		for (node,pid_node) in zip(nodes,pid_rank0_on_node))
 
-	# Worker at which final reduction takes place
-	p_final = first(pid_rank0_on_node)
+# 	# Worker at which final reduction takes place
+# 	p_final = first(pid_rank0_on_node)
 
-	sum_channel = RemoteChannel(()->Channel{T}(length(pid_rank0_on_node)),p_final)
-	result = nothing
+# 	sum_channel = RemoteChannel(()->Channel{T}(length(pid_rank0_on_node)),p_final)
+# 	result = nothing
 
-	# Run the function on each processor and compute the sum at each node
-	@sync for (rank,(p,node)) in enumerate(zip(procs_used,hostnames))
-		@async begin
+# 	# Run the function on each processor and compute the sum at each node
+# 	@sync for (rank,(p,node)) in enumerate(zip(procs_used,hostnames))
+# 		@async begin
 
-			node_remotechannel = node_channels[node]
-			np_node = nprocs_node[node]
+# 			node_remotechannel = node_channels[node]
+# 			np_node = nprocs_node[node]
 
-			iterable_on_proc = split_across_processors(iterable,num_workers,rank)
-			@spawnat p put!(node_remotechannel,
-				f(iterable_on_proc,args...;kwargs...))
-
-			@async if p in pid_rank0_on_node
-				f = @spawnat p put!(sum_channel,
-					sum(take!(node_remotechannel) for i=1:np_node))
-				wait(f)
-				@spawnat p finalize(node_remotechannel)
-			end
+# 			iterable_on_proc = split_across_processors(iterable,num_workers,rank)
+# 			@spawnat p put!(node_remotechannel,
+# 				f(iterable_on_proc,args...;kwargs...))
 
-			@async if p==p_final
-				result = @fetchfrom p_final sum(take!(sum_channel)
-					for i=1:length(pid_rank0_on_node))
-				@spawnat p finalize(sum_channel)
-			end
-		end
-	end
+# 			@async if p in pid_rank0_on_node
+# 				f = @spawnat p put!(sum_channel,
+# 					sum(take!(node_remotechannel) for i=1:np_node))
+# 				wait(f)
+# 				@spawnat p finalize(node_remotechannel)
+# 			end
 
-	return result :: T
-end
+# 			@async if p==p_final
+# 				result = @fetchfrom p_final sum(take!(sum_channel)
+# 					for i=1:length(pid_rank0_on_node))
+# 				@spawnat p finalize(sum_channel)
+# 			end
+# 		end
+# 	end
 
-function pmapsum(::Type{T},f::Function,iterable,args...;kwargs...) where {T<:AbstractArray}
+# 	return result :: T
+# end
 
-	# Use ArrayChannels
+# function pmapsum(::Type{T},f::Function,iterable,args...;kwargs...) where {T<:AbstractArray}
 
-	procs_used = workers_active(iterable)
+# 	# Use ArrayChannels
 
-	num_workers = length(procs_used);
-	hostnames = get_hostnames(procs_used);
-	nodes = get_nodes(hostnames);
-	pid_rank0_on_node = [procs_used[findfirst(isequal(node),hostnames)] for node in nodes];
+# 	procs_used = workers_active(iterable)
 
-	nprocs_node = get_nprocs_node(procs_used)
-	node_channels = Dict(node=>RemoteChannel(()->Channel{T}(nprocs_node[node]),pid_node)
-		for (node,pid_node) in zip(nodes,pid_rank0_on_node))
+# 	num_workers = length(procs_used);
+# 	hostnames = get_hostnames(procs_used);
+# 	nodes = get_nodes(hostnames);
+# 	pid_rank0_on_node = [procs_used[findfirst(isequal(node),hostnames)] for node in nodes];
 
-	# Worker at which final reduction takes place
-	p_final = first(pid_rank0_on_node)
+# 	nprocs_node = get_nprocs_node(procs_used)
+# 	node_channels = Dict(node=>RemoteChannel(()->Channel{T}(nprocs_node[node]),pid_node)
+# 		for (node,pid_node) in zip(nodes,pid_rank0_on_node))
 
-	sum_channel = RemoteChannel(()->Channel{T}(length(pid_rank0_on_node)),p_final)
-	result = nothing
+# 	# Worker at which final reduction takes place
+# 	p_final = first(pid_rank0_on_node)
 
-	# Run the function on each processor and compute the sum at each node
-	@sync for (rank,(p,node)) in enumerate(zip(procs_used,hostnames))
-		@async begin
+# 	sum_channel = RemoteChannel(()->Channel{T}(length(pid_rank0_on_node)),p_final)
+# 	result = nothing
+
+# 	# Run the function on each processor and compute the sum at each node
+# 	@sync for (rank,(p,node)) in enumerate(zip(procs_used,hostnames))
+# 		@async begin
 
-			iterable_on_proc = split_across_processors(iterable,num_workers,rank)
-			r = @spawnat p f(iterable_on_proc,args...;kwargs...)
+# 			iterable_on_proc = split_across_processors(iterable,num_workers,rank)
+# 			r = @spawnat p f(iterable_on_proc,args...;kwargs...)
 
-			node_remotechannel = node_channels[node]
-			np_node = nprocs_node[node]
+# 			node_remotechannel = node_channels[node]
+# 			np_node = nprocs_node[node]
 
-			@spawnat p put!(node_remotechannel,fetch(r))
+# 			@spawnat p put!(node_remotechannel,fetch(r))
 
-			if p in pid_rank0_on_node
-				s = @spawnat p sum(take!(node_remotechannel) for i=1:np_node)
-				@spawnat p put!(sum_channel,fetch(s))
-			end
+# 			if p in pid_rank0_on_node
+# 				s = @spawnat p sum(take!(node_remotechannel) for i=1:np_node)
+# 				@spawnat p put!(sum_channel,fetch(s))
+# 			end
 
-			if p==p_final
-				result = @fetchfrom p_final sum(take!(sum_channel) 
-					for i=1:length(pid_rank0_on_node))			
-			end
-		end
-	end
+# 			if p==p_final
+# 				result = @fetchfrom p_final sum(take!(sum_channel) 
+# 					for i=1:length(pid_rank0_on_node))			
+# 			end
+# 		end
+# 	end
 
-	return result :: T
-end
+# 	return result :: T
+# end
+
+# function pmapsum(f::Function,iterable,args...;kwargs...)
+# 	pmapsum(Any,f,iterable,args...;kwargs...)
+# end
 
 function pmapsum(f::Function,iterable,args...;kwargs...)
-	pmapsum(Any,f,iterable,args...;kwargs...)
+	@distributed (+) for i in 1:nworkers()
+		np = workers_active(iterable)
+		iter_proc = split_across_processors(iterable,np,i)
+		f(iter_proc,args...;kwargs...)
+	end
 end
 
 function pmap_onebatch_per_worker(f::Function,iterable,args...;kwargs...)