Error with remotecall_fetch called on locally defined closures

Ref: discourse link

In cases when parallel data movement is a bottle neck, it would useful to be able to use closures with pre-computed internal state parameters that are computed on each worker only once and not serialized from the master node.

Here are two examples. The first one is a simple case where gen_foo returns a closure generated on all workers. However remotecall_fetch gives an error. @amitmurthy gave a workaround in the link above but it would be nice to have a more direct approach.

The second example tries to use planned FFT's as internal states in the closures.

First example

julia> addprocs(2)
2-element Array{Int64,1}:
 2
 3

julia> @everywhere function gen_foo(local_state)
           foo(x) = x * sum(local_state)
           return foo::Function
       end

julia> @everywhere foo = gen_foo(rand(100,100))

julia> remotecall_fetch(foo, 1, 10) #<-- works
50008.9259374688

julia> remotecall_fetch(foo, 2, 10) #<-- UndefVarError: #foo#9 not defined
ERROR: On worker 2:
UndefVarError: #foo#9 not defined
deserialize_datatype at ./serialize.jl:968
handle_deserialize at ./serialize.jl:674
deserialize at ./serialize.jl:634
handle_deserialize at ./serialize.jl:681
deserialize_msg at ./distributed/messages.jl:98
message_handler_loop at ./distributed/process_messages.jl:161
process_tcp_streams at ./distributed/process_messages.jl:118
JuliaLang/julia#99 at ./event.jl:73
Stacktrace:
 [1] #remotecall_fetch#141(::Array{Any,1}, ::Function, ::Function, ::Base.Distributed.Worker, ::Int64, ::Vararg{Int64,N} where N) at ./distributed/remotecall.jl:354
 [2] remotecall_fetch(::Function, ::Base.Distributed.Worker, ::Int64, ::Vararg{Int64,N} where N) at ./distributed/remotecall.jl:346
 [3] #remotecall_fetch#144(::Array{Any,1}, ::Function, ::Function, ::Int64, ::Int64, ::Vararg{Int64,N} where N) at ./distributed/remotecall.jl:367
 [4] remotecall_fetch(::Function, ::Int64, ::Int64, ::Vararg{Int64,N} where N) at ./distributed/remotecall.jl:367

Second example

julia> addprocs(2)
2-element Array{Int64,1}:
 2
 3

julia> @everywhere function gen_bar(local_state, Δx, n)
           FFT = Δx / (2π) * plan_rfft(rand(n); flags=FFTW.PATIENT, timelimit=4)
           function bar(x)
               y = FFT * (x .* local_state)
               return y[1]
           end
           return bar::Function
       end

julia> @everywhere Δx, n = 0.1, 1000

julia> @everywhere bar = gen_bar(rand(n), Δx, n)

julia> x = rand(n);

julia> @everywhere x=$x

julia> @everywhere println(bar(x))    # <-- works
3.9819631247716307 + 0.0im
	From worker 2:	3.8079619702395155 + 0.0im
       From worker 3:	4.008786805675184 + 0.0im

julia> remotecall_fetch(bar, 1, x)   # <-- works
3.9819631247716307 + 0.0im

However I get UndefVarError: #bar#9 not defined for these

remotecall_fetch(bar, 2, x) # <-- UndefVarError: #bar#9 not defined

out = @parallel (vcat) for i = 1:nprocs() # <-- UndefVarError: #bar#9 not defined
    bar(x)
end

out = @parallel (vcat) for i = 1:nprocs() # <-- UndefVarError: #bar#9 not defined
        remotecall_fetch(bar, i, x)
end

Reproduce the closure examples with addprocs(2), gen_foo, gen_bar, and remotecall_fetch. Start by reading distributed/remotecall.jl and the serialization paths shown in serialize.jl and distributed/messages.jl. Done means locally defined closures with worker-local state can be passed to worker 2 and used through remotecall_fetch and @parallel without the UndefVarError.