separate functions

Author: jishnub

src/ParallelUtilities.jl

@@ -242,114 +242,68 @@ 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_remotechannel(::Type{T},f::Function,iterable,args...;kwargs...) where {T}
 
-# 	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];
-
-# 	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)
-
-# 	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
-			
-# 			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
-
-# 			@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
-
-# 	return result :: T
-# end
-
-# function pmapsum(::Type{T},f::Function,iterable,args...;kwargs...) where {T<:AbstractArray}
-
-# 	# Use ArrayChannels
-
-# 	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
 
-# 			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))
-
-# 			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
+			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
 
-# 			if p==p_final
-# 				result = @fetchfrom p_final sum(take!(sum_channel) 
-# 					for i=1:length(pid_rank0_on_node))			
-# 			end
-# 		end
-# 	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
 
-# 	return result :: T
-# end
+	return result :: T
+end
 
-# function pmapsum(f::Function,iterable,args...;kwargs...)
-# 	pmapsum(Any,f,iterable,args...;kwargs...)
-# end
+function pmapsum_remotechannel(f::Function,iterable,args...;kwargs...)
+	pmapsum(Any,f,iterable,args...;kwargs...)
+end
 
-function pmapsum(f::Function,iterable,args...;kwargs...)
+function pmapsum_distributedfor(f::Function,iterable,args...;kwargs...)
 	@distributed (+) for i in 1:nworkers()
 		np = nworkers_active(iterable)
 		iter_proc = split_across_processors(iterable,np,i)
 		f(iter_proc,args...;kwargs...)
 	end
 end
 
+pmapsum(f,args...;kwargs...) = pmapsum_remotechannel(f,args...;kwargs...)
+
 function pmap_onebatch_per_worker(f::Function,iterable,args...;kwargs...)
 
 	procs_used = workers_active(iterable)