Sending functions within structs to workers leads to world age problem

Specific Domains Julia at Scale
, ,
1

I have a simulation where I pass functions as parameters by putting them in a parameter object. However, when I try to run the simulation using Distributed, the functions do not run in the workers due to their world age being too new.

For example, this MWE:

using Distributed 

num = 2
addprocs(num)

@everywhere begin
    struct testParams
        f::Function
    end        

    function worker(input, output)
        while true
            f = take!(input).f
            put!(output, (myid(), f(1.0)))
        end
    end
end

# channels to send and receive data from workers
input = RemoteChannel(()->Channel{testParams}(num))
output = RemoteChannel(()->Channel{Tuple}(num))

for i in 1:num
    put!(input, testParams((x)->2*x))
end

# run workers
for w in workers()
    remote_do(worker, w, input, output)
end

# collect results
for i in 1:num
    w, ans = take!(output)
    println("worker $w: $ans")
end

produces the following errors:

From worker 3:    (::Distributed.var"#113#115"{Distributed.RemoteDoMsg})() at ./task.jl:356MethodError: no method matching (::Serialization.__deserialized_types__.var"#27#28")(::Float64)
      From worker 3:    The applicable method may be too new: running in world age 27823, while current world is 27824.
      From worker 3:    Closest candidates are:
      From worker 3:      #27(::Any) at /Users/vancleve/science/projects/food_sharing_odwyer/eees/julia/popsim.jl:63 (method too new to be called from this world context.)
      From worker 3:    worker(::RemoteChannel{Channel{testParams}}, ::RemoteChannel{Channel{Tuple}}) at /Users/vancleve/science/projects/food_sharing_odwyer/eees/julia/popsim.jl:53
      From worker 3:    (::Distributed.var"#114#116"{Distributed.RemoteDoMsg})() at /Users/julia/buildbot/worker/package_macos64/build/usr/share/julia/stdlib/v1.5/Distributed/src/process_messages.jl:315
      From worker 3:    run_work_thunk(::Distributed.var"#114#116"{Distributed.RemoteDoMsg}, ::Bool) at /Users/julia/buildbot/worker/package_macos64/build/usr/share/julia/stdlib/v1.5/Distributed/src/process_messages.jl:79
      From worker 5:    MethodError: no method matching (::Serialization.__deserialized_types__.var"#27#28")(::Float64)
      From worker 5:    The applicable method may be too new: running in world age 27817, while current world is 27818.
      From worker 5:    Closest candidates are:
      From worker 5:      #27(::Any) at /Users/vancleve/science/projects/food_sharing_odwyer/eees/julia/popsim.jl:63 (method too new to be called from this world context.)
      From worker 5:    worker(::RemoteChannel{Channel{testParams}}, ::RemoteChannel{Channel{Tuple}}) at /Users/vancleve/science/projects/food_sharing_odwyer/eees/julia/popsim.jl:53
      From worker 5:    (::Distributed.var"#114#116"{Distributed.RemoteDoMsg})() at /Users/julia/buildbot/worker/package_macos64/build/usr/share/julia/stdlib/v1.5/Distributed/src/process_messages.jl:315
      From worker 5:    run_work_thunk(::Distributed.var"#114#116"{Distributed.RemoteDoMsg}, ::Bool) at /Users/julia/buildbot/worker/package_macos64/build/usr/share/julia/stdlib/v1.5/Distributed/src/process_messages.jl:79

Is there any solution to this other than invoking the functions in the workers with Base.invokelatest?

2

The world age problem comes from this line since your workers do not know the newly created anonymous function x->2*x. But they know *. If you want to avoid invokelatest, you could else

  • define your new function with @everywhere or
  • send known functions along with parameters like that:
using Distributed

num = 2
addprocs(num)

@everywhere begin
    struct testParams{F}
        f::F
        args::Tuple

        testParams(f, args...) = new{typeof(f)}(f, args)
    end        

    function worker(input, output)
        while true
            p = take!(input)
            put!(output, (myid(), p.f(1.0, p.args...)))
        end
    end
end

# channels to send and receive data from workers
input = RemoteChannel(()->Channel{testParams}(num))
output = RemoteChannel(()->Channel{Tuple}(num))

for i in 1:num
    put!(input, testParams(*, 2))
end

# run workers
for w in workers()
    remote_do(worker, w, input, output)
end

# collect results
for i in 1:num
    w, ans = take!(output)
    println("worker $w: $ans")
end

worker 2: 2.0
worker 3: 2.0

This is synonymous with what you do but doesn’t cause the problem.

3

Here is a way using RuntimeGeneratedFunctions:

using Distributed 

num = 2
addprocs(num)

@everywhere using RuntimeGeneratedFunctions
@everywhere RuntimeGeneratedFunctions.init(@__MODULE__)

@everywhere begin
    struct testParams
        f::Function
    end        

    function worker(input, output)
        while true
            f = take!(input).f
            put!(output, (myid(), f(1.0)))
        end
    end
end

# channels to send and receive data from workers
input = RemoteChannel(()->Channel{testParams}(num))
output = RemoteChannel(()->Channel{Tuple}(num))

@everywhere f = @RuntimeGeneratedFunction( :( x -> 2*x ) )
for i in 1:num
    put!(input, testParams(f) )
end

# run workers
for w in workers()
    remote_do(worker, w, input, output)
end

# collect results
for i in 1:num
    w, ans = take!(output)
    println("worker $w: $ans")
end

julia> for i in 1:num
           w, ans = take!(output)
           println("worker $w: $ans")
       end
worker 14: 2.0
worker 27: 2.0
5

Great, thanks for the tip!!!

6

very helpful solution, thanks!