# Parallel reductions on CPUs with different speeds
**URL:** https://discourse.julialang.org/t/parallel-reductions-on-cpus-with-different-speeds/40722
**Category:** Performance
**Tags:** parallel, cluster
**Created:** [June 4, 2020, 12:15pm UTC](https://discourse.julialang.org/t/parallel-reductions-on-cpus-with-different-speeds/40722 "2020-06-04T12:15:37Z")
**Posts on this page:** 7
**Page:** 1
### Author:  [@hcarlsso](https://discourse.julialang.org/u/hcarlsso)
#### Post date: [June 4, 2020, 12:15pm UTC](https://discourse.julialang.org/t/parallel-reductions-on-cpus-with-different-speeds/40722/1 "2020-06-04T12:15:37Z")
I am looking for how to “load balance” parallel reduction on multiple CPUs that have different speeds. The following snippet illustrates the problem:
```julia
using Distributed
addprocs(3)
res = @time @distributed (+) for (t, n) in collect(zip(1:9, vcat([10], ones(8))))
println("Task $(t) sleeping: $(n) s")
sleep(n)
n
end
```
which yields
```julia
From worker 2: Task 1 sleeping: 10.0 s
From worker 3: Task 4 sleeping: 1.0 s
From worker 4: Task 7 sleeping: 1.0 s
From worker 3: Task 5 sleeping: 1.0 s
From worker 4: Task 8 sleeping: 1.0 s
From worker 3: Task 6 sleeping: 1.0 s
From worker 4: Task 9 sleeping: 1.0 s
From worker 2: Task 2 sleeping: 1.0 s
From worker 2: Task 3 sleeping: 1.0 s
12.243273 seconds (510.34 k allocations: 25.608 MiB)
18.0
```
That is, worker 2 still performs task 2 and 3, while workers 3 and 4 are doing nothing. What can be done in this situation so that worker 3 and 4 performs task 2 and 3?
---
### Author:  [@eaubanel](https://discourse.julialang.org/u/eaubanel)
#### Post date: [June 4, 2020, 1:45pm UTC](https://discourse.julialang.org/t/parallel-reductions-on-cpus-with-different-speeds/40722/2 "2020-06-04T13:45:06Z")
To quote from a comment in the source associated with `@distributed`: “Statically split range [1,N] into equal sized chunks for np processors”, so here worker 2 gets tasks 1-3, 3 gets 4,5,6, and 4 gets 7,8,9. You could deal with this yourself using `@spawn`, but it would be more work for you than using `@distributed`.
---
### Author:  [@jishnub](https://discourse.julialang.org/u/jishnub)
#### Post date: [June 4, 2020, 3:02pm UTC](https://discourse.julialang.org/t/parallel-reductions-on-cpus-with-different-speeds/40722/3 "2020-06-04T15:02:39Z")
`pmap` does this load balancing for you, but it works a bit differently – jobs are submitted one by one to free workers instead of in batches as in the distributed for loop.
---
### Author:  [@hcarlsso](https://discourse.julialang.org/u/hcarlsso)
#### Post date: [June 4, 2020, 3:48pm UTC](https://discourse.julialang.org/t/parallel-reductions-on-cpus-with-different-speeds/40722/4 "2020-06-04T15:48:52Z")
but can one do the reduction on each worker with `pmap`? with `pmap`, isnt the result of each task sent back to master?
---
### Author:  [@jishnub](https://discourse.julialang.org/u/jishnub)
#### Post date: [June 4, 2020, 4:56pm UTC](https://discourse.julialang.org/t/parallel-reductions-on-cpus-with-different-speeds/40722/5 "2020-06-04T16:56:27Z")
This is true. One solution is to not return the value in the function applied, but to push it to a `RemoteChannel` on each worker, and asynchronously run a reduction operation. I’m not sure if there is a standard function for this.
---
### Author:  [@lungben](https://discourse.julialang.org/u/lungben)
#### Post date: [June 4, 2020, 5:51pm UTC](https://discourse.julialang.org/t/parallel-reductions-on-cpus-with-different-speeds/40722/6 "2020-06-04T17:51:41Z")
Maybe this is helpful for you (disclaimer: I have not used it so far):
> **[GitHub - tro3/ThreadPools.jl: Improved thread management for background and...](https://github.com/tro3/ThreadPools.jl)**
>
> Improved thread management for background and nonuniform tasks in Julia. Docs at https://tro3.github.io/ThreadPools.jl - GitHub - tro3/ThreadPools.jl: Improved thread management for background and...
Edit: just realized that your question is about multiprocessing, not multithreading. Therefore this is probably not helpful for you.
---
### Author:  [@hcarlsso](https://discourse.julialang.org/u/hcarlsso)
#### Post date: [June 5, 2020, 1:34pm UTC](https://discourse.julialang.org/t/parallel-reductions-on-cpus-with-different-speeds/40722/7 "2020-06-05T13:34:02Z")
Ok, so this is an implementation `mapreduce` using `RemoteChannel`.
```julia
@everywhere function do_work(f, op, jobs, res)
args = try
take!(jobs)
catch InvalidStateException
# No work to do
return nothing
end
v = f(args...)
while true
args = try
take!(jobs)
catch InvalidStateException
# We are done
break
end
v = op(v, f(args...))
end
put!(res, v)
end
function pmapreduce(f, op, itr; nbuf::Int = 100)
jobs = RemoteChannel(()-> Channel(length(itr)))
all_w = workers()[1:min(nworkers(), length(itr))]
res = RemoteChannel(()-> Channel(length(all_w)))
for pid in all_w
remotecall(do_work, pid, f, op, jobs, res)
end
# Make jobs
for job_arg in itr
put!(jobs, job_arg)
end
close(jobs)
# Collect results
v = take!(res)
if length(all_w) > 1
for n in 2:length(all_w)
v = op(v, take!(res))
end
end
close(res)
return v
end
@everywhere function f(n)
println("Task sleeping: $(n) s")
sleep(n)
n
end
input = vcat([10], ones(8))
res = @time pmapreduce(f, +, input)
println(res)
res = @time @distributed (+) for (t, n) in collect(zip(1:9, input))
println("Task $(t) sleeping: $(n) s")
sleep(n)
n
end
```
yielding
```julia
julia> include("pmapreduce.jl")
From worker 3: Task sleeping: 1.0 s
From worker 4: Task sleeping: 10.0 s
From worker 2: Task sleeping: 1.0 s
From worker 3: Task sleeping: 1.0 s
From worker 2: Task sleeping: 1.0 s
From worker 3: Task sleeping: 1.0 s
From worker 2: Task sleeping: 1.0 s
From worker 3: Task sleeping: 1.0 s
From worker 2: Task sleeping: 1.0 s
10.297645 seconds (232.16 k allocations: 11.238 MiB, 0.15% gc time)
18.0
From worker 3: Task 4 sleeping: 1.0 s
From worker 4: Task 7 sleeping: 1.0 s
From worker 2: Task 1 sleeping: 10.0 s
From worker 3: Task 5 sleeping: 1.0 s
From worker 4: Task 8 sleeping: 1.0 s
From worker 3: Task 6 sleeping: 1.0 s
From worker 4: Task 9 sleeping: 1.0 s
From worker 2: Task 2 sleeping: 1.0 s
From worker 2: Task 3 sleeping: 1.0 s
12.090770 seconds (65.97 k allocations: 3.238 MiB)
18.0
```