remotechannel type

Author: jishnub

src/ParallelUtilities.jl

@@ -242,7 +242,7 @@ end
 
 get_nprocs_node(procs_used::Vector{<:Integer}=workers()) = get_nprocs_node(get_hostnames(procs_used))
 
-function pmapsum(f::Function,iterable,args...;kwargs...)
+function pmapsum(::Type{T},f::Function,iterable,args...;kwargs...) where {T}
 
 	procs_used = workers_active(iterable)
 
@@ -252,13 +252,13 @@ function pmapsum(f::Function,iterable,args...;kwargs...)
 	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{Any}(nprocs_node[node]),pid_node)
+	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{Any}(length(pid_rank0_on_node)),p_final)
+	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
@@ -288,6 +288,11 @@ function pmapsum(f::Function,iterable,args...;kwargs...)
 	return result
 end
 
+function pmapsum(f::Function,iterable,args...;kwargs...)
+	T = promote_type(Base.return_types(f)...)
+	pmapsum(T,f,iterable,args...;kwargs...)
+end
+
 function pmap_onebatch_per_worker(f::Function,iterable,args...;kwargs...)
 
 	procs_used = workers_active(iterable)