Error with remotecall_fetch called on locally defined closures

[EthanAnderes]

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

[amitmurthy]

For this particular use case, remote references can be used to store locally defined closures on each node and fetch it locally before execution. Something like:

# create
foo_refs = Dict()
for p in procs()
  foo_refs[p] = RemoteChannel(p)
end 

# initialize on all nodes
for p in procs()
  remote_do(x->put!(x, gen_foo(rand(100,100))), p, foo_refs[p])
end

# access on worker 2
remotecall_fetch(2, foo_refs[2], 10) do rr, v
  fetch(rr)(v)
end

You can use https://github.com/JuliaParallel/DistributedArrays.jl#working-with-distributed-non-array-data-requires-julia-06 in DistributedArrays to do the same more efficiently and in fewer lines of code.

An issue I see is the difference in the way we treat a global function definition vs closures bound to global variables.

With julia -p 4

julia> @everywhere foov = rand()

julia> @everywhere func_foo() = foov

julia> @everywhere glb_var_foo = ()->foov

julia> [remotecall_fetch(func_foo, p) for p in procs()]
5-element Array{Float64,1}:
 0.0138725
 0.451707 
 0.41358  
 0.36171  
 0.331541 

julia> 

julia> [remotecall_fetch(glb_var_foo, p) for p in procs()]
5-element Array{Float64,1}:
 0.0138725
 0.0138725
 0.0138725
 0.0138725
 0.0138725

In the second case it is actually the closure defined on node 1 that is serialized and executed on the remote node while one may intuitively expect it to be the locally defined glb_var_foo.

At this time I think we just need to better highlight this difference and add examples to the documentation.

[amitmurthy]

The original issue is also related to gen_foo defining a local foo which is then bound to a global variable foo. Consider the following 2 cases:

The first one errors as in the original issue:

julia> gen_foo() = (foo() = 1; foo)
gen_foo (generic function with 1 method)

julia> foo = gen_foo()
foo (generic function with 1 method)

julia> remotecall_fetch(foo, 2)
ERROR: On worker 2:
UndefVarError: #foo#1 not defined
deserialize_datatype at ./serialize.jl:975

However, if we change gen_foo to now return a locally defined bar, it works.

julia> gen_foo() = (bar() = 1; bar)
gen_foo (generic function with 1 method)

julia> foo = gen_foo()
(::bar) (generic function with 1 method)

julia> remotecall_fetch(foo, 2)
1

How is the name of the locally defined function affecting the way in which global variable foo is serialized? @vtjnash / @JeffBezanson ? Anyone?

[StefanKarpinski]

Bump...

[magerton]

Hi -- Thanks for the helpful work-around. I'd also much appreciate the ability to call remote-specific functions. I'm currently computing a likelihood in parallel and am using sets of pre-allocated temporary variables on each worker. Using @everywhere lets me get part of the way there. My work-around to call remote-specific closures extends your code, @amitmurthy.

# to zero-out all tmpvars
@everywhere zero_tmp() = zero!(remote_tmp::TmpVar)

function call_remote_fun(f::Symbol)
    for p in workers()
        remotecall_fetch(() -> eval(Main, :($(f)())), p)
    end
end

call_remote_fun(:zero_tmp)

# to sum over remote tmpvars
@everywhere remote_f() = get_from_remote_tmpvars(remote_tmp::TmpVar)

function mapreduce_remotevars(f::Symbol, R::Function)
    mapreduce((p::Int) -> remotecall_fetch(() -> eval(Main, :($(f)())), p), R, workers() )
end

mapreduce_remotevars(:remote_f, +)