Add Dagger.spawn, allow on any node

Author: jpsamaroo

src/thunk.jl

@@ -123,6 +123,18 @@ function Base.show(io::IO, t::EagerThunk)
     print(io, "EagerThunk ($(isready(t) ? "finished" : "running"))")
 end
 
+function spawn(f, args...; kwargs...)
+    if myid() == 1
+        Dagger.Sch.init_eager()
+        future = ThunkFuture()
+        uid = next_id()
+        put!(Dagger.Sch.EAGER_THUNK_CHAN, (future, uid, f, (args...,), (kwargs...,)))
+        EagerThunk(future, uid)
+    else
+        remotecall_fetch(spawn, 1, f, args...; kwargs...)
+    end
+end
+
 """
     @par [opts] f(args...) -> Thunk
 
@@ -172,11 +184,8 @@ function _par(ex::Expr; lazy=true, recur=true, opts=())
             return :(Dagger.delayed($(esc(f)); $(opts...))($(_par.(args; lazy=lazy, recur=false)...)))
         else
             return quote
-                Dagger.Sch.init_eager()
-                future = $ThunkFuture()
-                uid = $next_id()
-                put!(Dagger.Sch.EAGER_THUNK_CHAN, (future, uid, $(esc(f)), ($(_par.(args; lazy=lazy, recur=false)...),), ($(opts...),)))
-                EagerThunk(future, uid)
+                args = ($(_par.(args; lazy=lazy, recur=false)...),)
+                $spawn($(esc(f)), args...; $(opts...))
             end
         end
     else

test/thunk.jl

@@ -1,4 +1,4 @@
-import Dagger: @par, @spawn
+import Dagger: @par, @spawn, spawn
 
 @everywhere checkwid() = myid()==1
 
@@ -32,7 +32,8 @@ end
         a = @spawn x + x
         @test a isa Dagger.EagerThunk
         b = @spawn sum([x,1,2])
-        c = @spawn a * b
+        c = spawn(*, a, b)
+        @test c isa Dagger.EagerThunk
         @test fetch(a) == 4
         @test fetch(b) == 5
         @test fetch(c) == 20
@@ -131,4 +132,11 @@ end
             @test_throws_unwrap Dagger.ThunkFailedException fetch(d)
         end
     end
+    @testset "remote spawn" begin
+        a = fetch(Distributed.@spawnat 2 Dagger.spawn(+, 1, 2))
+        @test Dagger.Sch.EAGER_INIT[]
+        @test fetch(Distributed.@spawnat 2 !(Dagger.Sch.EAGER_INIT[]))
+        @test a isa Dagger.EagerThunk
+        @test fetch(a) == 3
+    end
 end