Here seems to be a safer implementation... but I'm still not quite
confident it's optimally written. Opinions welcome!
function subdivide(v0,v1,filter::Function,middle::Function,doer::Function)
c = Channel(1000000)
results = Any[doer(v0),doer(v1)]
filter(v0,v1) || return results # nothing to subdivide
put!(c,(results[1],results[2]))
np = nprocs() # determine the number of processes available
idle = Array(Bool,np)
idle[myid()] = true
@sync begin
for p=1:np
if p != myid() || np == 1
@async begin
while true
if isready(c)
idle[p] = false
(d0,d1) = take!(c)
v = middle(d0,d1)
info("Doing task $v on process $p, queue
$c...\n")
d = remotecall_fetch(p,doer,v)
push!(results,d)
filter(d0,d) && put!(c,(d0,d))
filter(d,d1) && put!(c,(d,d1))
else
idle[p] = true
if all(idle) break end
yield()
end
end
end
end
end
end
results
end
On Friday, 4 September 2015 09:30:50 UTC+2, Laurent Bartholdi wrote:
>
> Hello world,
> I'm completely new to Julia, and advice/comments are welcome. I'd like to
> run calculations (~ 1 second per job) distributed along the different cores
> of my CPU. Each calculation may trigger new calculations.
> More specifically, each calculation works on a number in an interval, and
> may trigger a subdivision of the interval.
>
> I thought about the following implementation:
>
> function
> subdivide(v0::Number,v1::Number,filter::Function,middle::Function,doer::Function)
> c = Channel(1000)
> results = Any[doer(v0),doer(v1)]
> put!(c,(results[1],results[2]))
> np = nprocs() # determine the number of processes available
> @sync begin
> for p=1:np
> if p != myid() || np == 1
> @async begin
> for (d0,d1)=c
> v = middle(d0,d1)
> #println("Doing $v on process $p")
> d = remotecall_fetch(p,doer,v)
> push!(results,d)
> filter(d0,d) && put!(c,(d0,d))
> filter(d,d1) && put!(c,(d,d1))
> isready(c) || break
> end
> end
> end
> end
> end
> results
> end
>
> typical use:
>
> julia>
> subdivide(0,10,(x,y)->x[1]<y[1]-0.7,(x,y)->(x[1]+y[1])/2,v->[v,string(v)])
> 17-element Array{Any,1}:
> Any[0,"0"]
> Any[10,"10"]
> ⋮
> Any[8.125,"8.125"]
> Any[9.375,"9.375"]
>
> As you see, the function "subdivide" receives a "doer" which computes a
> bit of data, and a "filter" which decides if a middle point is required
> between data points. "middle" computes the argument of a middle point.
>
> My worry is: certainly there will be race conditions when a calculation
> has been removed from the channel, but the new calculations will not yet
> have been put in; then one of the cores will stop prematurely. The last
> "isready(c)" could also, potentially, return true but the upcoming "for
> (d0,d1)=c" will block because meanwhile a calculation has been removed by
> another (green) thread.
>
> How should I program this safely? It seems like a useful primitive, so
> maybe it already exists...
>
> Thanks in advance, Laurent
>
