# Using RemoteChannels for an parallel iterable map

RemoteChannels hit with julia v0.5 There are no examples online, but I managed to get them to work.

They are still very new, and all this whole area of multithreading code continues to have running changes -- there was one just a few days ago.

"A kind of async parallel iterator map, that does not quiet  presever order."
immutable WorkFarmerIterator{T}
func :: Function #Must take an tuple of const_data, and an element of source_iter and do work on them
source_iter
const_data::T
channel_sizes::Int64
end

type WorkFarmerState
pending::RemoteChannel
complete::RemoteChannel
job_submitter::Future
end

function Base.start(iter::WorkFarmerIterator)

o_pending = RemoteChannel(()->Channel(iter.channel_sizes),myid())
o_complete = RemoteChannel(()->Channel(iter.channel_sizes),myid())

o_job_submitter = remotecall(myid(), o_pending, iter.source_iter) do pending, source_iter
#Can prob do this without a remotecall
for wk in source_iter
put!(pending, wk)
end
true
end

# Fire off workers that Read Pending (until closed exception)
# And write to complete
@sync for pid in workers()
@async remotecall(pid, iter.func, iter.const_data, o_pending,o_complete) do func, const_data, pending, complete
try
while(true)
wk = take!(pending) #Block til work arrives
res = func(const_data, wk)
put!(complete, res) #Block til I can hand work over
end
catch ee
#Break out of loop when a stream is closed and we try to read pending
#Or in the case of a unfortunte timing error we try to write to complete
if !(ee|>typeof ==InvalidStateException && ee.state == :closed)
rethrow()
end
#otherwise eat the InvalidStateException
end
end
end

#Register Finalizer
state=WorkFarmerState(o_pending,o_complete, o_job_submitter)
function finalize_state(st::WorkFarmerState)
close(st.pending)
close(st.complete)
end
finalizer(state, finalize_state)
state
end

function Base.done(iter::WorkFarmerIterator, state::WorkFarmerState)
#Check that job_submitter is done
&&  !isready(state.complete) #and complete work closed or empty
)
#We can't actually check if the RemoteChannel is closed.
if isdone
finalize(state) #Done not hurt to finalize it twice
end
isdone
end

function Base.next(iter::WorkFarmerIterator, state::WorkFarmerState)
#Read complete -- this is a blocking operation so if nothing is ready we'll sit here til it is.
#All of the sending (by job_submitter) basically has to happen here since this is the only garenteed time the local process will yield
res = take!(state.complete)
(res,state)
end

Base.iteratorsize(iter::WorkFarmerIterator) = Base.iteratorsize(iter.source_iter)

Base.length(iter::WorkFarmerIterator) = Base.length(iter.source_iter)


Example of use:

@everywhere function wkfun(data,ele)
sleep(0.5*rand())
(data*ele, myid())
end

witer  = WorkFarmerIterator(wkfun,1:1000, 1, 1024)

open("temp.txt","w") do
t0=time()
for ii in witer
println(fp,ii,"\t", time()-t0)
end
end


### Potential Issues:

Right now I see a few issues:

• I am not sure if it is a good idea to infinite loop in a WorkerProcess while reading a Channel vs firing off new tasks continually (either with remote_call, or via pgenerator etc.)
• I don't see anyway to profile it at all. Not even a good way of logging how full the Channels are. (Maybe a log call in Base.next?)
• The exact requirements for iter.func to be serialize/available are poorly known by me, and the code (nor the comments) do thus not make it clear to the user what the requirements on iter.func are.
• To my moderate surprise, this "just-works", in the case of a not having any workers. Because if you have a single process, workers()=1. vs if you have multiple (eg 5) workers()=[2,3,4,5,6]. And multiple multiple Tasks running on one cpu just works cos they are blocking at different times. – Lyndon White May 11 '16 at 10:18
• I believe this task is already covered mostly by Base.pgenerate` – Lyndon White May 13 '16 at 7:10