Merge pull request #432 from JuliaParallel/jps/fix-init-eager

submission: Call init_eager on worker 1

Author: jpsamaroo

src/sch/eager.jl

@@ -14,7 +14,7 @@ end
 
 function init_eager()
     if myid() != 1
-        return
+        throw(ConcurrencyViolationError("init_eager can only be called on worker 1"))
     end
     if Threads.atomic_xchg!(EAGER_INIT, true)
         wait(EAGER_READY)

src/submission.jl

@@ -1,5 +1,7 @@
 # Remote
 function eager_submit_internal!(@nospecialize(payload))
+    Sch.init_eager()
+
     ctx = Dagger.Sch.eager_context()
     state = Dagger.Sch.EAGER_STATE[]
     task = current_task()
@@ -10,6 +12,8 @@ function eager_submit_internal!(ctx, state, task, tid, payload; uid_to_tid=Dict{
     @nospecialize payload
     ntasks, uid, future, ref, f, args, options, reschedule = payload
 
+    Sch.init_eager()
+
     if uid isa Vector
         thunk_ids = Sch.ThunkID[]
         for i in 1:ntasks
@@ -123,6 +127,7 @@ function eager_submit!(ntasks, uid, future, finalizer_ref, f, args, options)
     elseif myid() != 1
         return remotecall_fetch(eager_submit_internal!, 1, (ntasks, uid, future, finalizer_ref, f, args, options, true))
     else
+        Sch.init_eager()
         state = Dagger.Sch.EAGER_STATE[]
         return lock(state.lock) do
             eager_submit_internal!((ntasks, uid, future, finalizer_ref,
@@ -170,8 +175,6 @@ function eager_process_options_submission_to_local(id_map, options::NamedTuple)
     end
 end
 function eager_spawn(spec::EagerTaskSpec)
-    Dagger.Sch.init_eager()
-
     # Generate new EagerThunk
     uid = eager_next_id()
     future = ThunkFuture()

test/thunk.jl

@@ -49,6 +49,9 @@ end
 end
 
 @testset "@spawn" begin
+    @test_throws_unwrap ConcurrencyViolationError remotecall_fetch(last(workers())) do
+        Dagger.Sch.init_eager()
+    end
     @test Dagger.Sch.EAGER_CONTEXT[] === nothing
     @testset "per-call" begin
         x = 2