If someone else needs something like this I ended up copying and modifying
pmap to suit my needs. I'm sure it can be simplified and shortened but it
works.
Example code below and I get healthy speedups all the way up to 20 procs:
feldt:~/tmp$ julia03 test_pfind_colwise.jl
0.5952118622068966
feldt:~/tmp$ julia03 -p 2 test_pfind_colwise.jl
0.3214775730689655
feldt:~/tmp$ julia03 -p 4 test_pfind_colwise.jl
0.1769931201724138
feldt:~/tmp$ julia03 -p 8 test_pfind_colwise.jl
0.10350024568965517
feldt:~/tmp$ julia03 -p 16 test_pfind_colwise.jl
0.07088396200000001
feldt:~/tmp$ julia03 -p 32 test_pfind_colwise.jl
0.05338699896551725
Julia sure is great! :)
Cheers,
Robert
N = 100
Rows = 3
MaxValue = 10
@everywhere function fake_slow_condition(x)
sleep(0.01)
sum(x) == 30 ? x : false # Return x iff it fulfills the poperty
end
# Similar to Base.findfirst(predicate, A) but process the array "A"
columnwise in parallel
# and opt out of further processing if found. Returns both the index at
which found
# as well as the result returned.
@everywhere function pfind_colwise(predicate, A)
np = nprocs() # determine the number of processes available
n = size(A, 2)
i = 1
results = Any[false for i in 1:n]
found = false
wasfound(idx) = found = true
isfound() = found
# function to produce the next work item from the queue.
# in this case it's just an index to a column.
nextidx() = (idx=i; i+=1; idx)
@sync begin
for p=1:np
if p != myid() || np == 1
@async begin
while true
idx = nextidx()
if idx > n
break
end
results[idx] = remotecall_fetch(p, predicate, A[:,idx])
if isfound()
break # Someone else already found it
end
if results[idx] != false
wasfound(idx)
break # We found it
end
end
end
end
end
end
fidx = findfirst((r)->(r != false), results)
(fidx, (fidx != 0 ? results[fidx] : nothing))
end
Reps = 30
times = zeros(Reps)
for rep in 1:Reps
array = ifloor(MaxValue * rand(Rows, N))
# We will be looking for the first column that sums to 30. Make sure it
exists.
suma = sum(array, 1)
i = findfirst(suma, 30)
if i == 0
# Put it somewhere around the middle
i = ifloor(rand(0.45N:0.55*N))
array[:,i] = MaxValue * ones(Int64, Rows, 1)
end
tic()
idx, res = pfind_colwise(fake_slow_condition, array)
times[rep] = toq()
end
println(mean(times[2:Reps]))
