Add CachingPool method for remotecall

Author: IanButterworth

src/workerpool.jl

@@ -400,3 +400,28 @@ function remotecall_pool(rc_f, f, pool::CachingPool, args...; kwargs...)
         put!(pool, worker)
     end
 end
+
+# Specialization for remotecall. We have to wait for the Future it returns
+# before putting the worker back in the pool.
+function remotecall_pool(rc_f::typeof(remotecall), f, pool::CachingPool, args...; kwargs...)
+    worker = take!(pool)
+    f_ref = get(pool.map_obj2ref, (worker, f), (f, RemoteChannel(worker)))
+    isa(f_ref, Tuple) && (pool.map_obj2ref[(worker, f)] = f_ref[2])   # Add to tracker
+
+    local x
+    try
+        x = rc_f(exec_from_cache, worker, f_ref, args...; kwargs...)
+    catch
+        put!(pool, worker)
+        rethrow()
+    end
+
+    t = Threads.@spawn Threads.threadpool() try
+        wait(x)
+    finally
+        put!(pool, worker)
+    end
+    errormonitor(t)
+
+    return x
+end

test/distributed_exec.jl

@@ -718,6 +718,8 @@ wp = WorkerPool(workers())
 @test nworkers() == length(unique(remotecall_fetch(wp->pmap(_->myid(), wp, 1:100), id_other, wp)))
 wp = WorkerPool(2:3)
 @test sort(unique(pmap(_->myid(), wp, 1:100))) == [2,3]
+@test fetch(remotecall(myid, wp)) in wp.workers
+@test_throws RemoteException fetch(remotecall(error, wp))
 
 # wait on worker pool
 wp = WorkerPool(2:2)
@@ -743,6 +745,8 @@ status = timedwait(() -> isready(f), 10)
 # CachingPool tests
 wp = CachingPool(workers())
 @test [1:100...] == pmap(x->x, wp, 1:100)
+@test fetch(remotecall(myid, wp)) in wp.workers
+@test_throws RemoteException fetch(remotecall(error, wp))
 
 clear!(wp)
 @test length(wp.map_obj2ref) == 0
@@ -1092,16 +1096,18 @@ end
 # Test the behaviour of remotecall(f, ::AbstractWorkerPool), this should
 # keep the worker out of the pool until the underlying remotecall has
 # finished.
-let
-    remotechan = RemoteChannel(wrkr1)
-    pool = WorkerPool([wrkr1])
-    put_future = remotecall(() -> wait(remotechan), pool)
-    @test !isready(pool)
-    put!(remotechan, 1)
-    wait(put_future)
-    # The task that waits on the future to put it back into the pool runs
-    # asynchronously so we use timedwait() to check when the worker is back in.
-    @test timedwait(() -> isready(pool), 10) === :ok
+for PoolType in (WorkerPool, CachingPool)
+    let
+        remotechan = RemoteChannel(wrkr1)
+        pool = PoolType([wrkr1])
+        put_future = remotecall(() -> wait(remotechan), pool)
+        @test !isready(pool)
+        put!(remotechan, 1)
+        wait(put_future)
+        # The task that waits on the future to put it back into the pool runs
+        # asynchronously so we use timedwait() to check when the worker is back in.
+        @test timedwait(() -> isready(pool), 10) === :ok
+    end
 end
 
 # Test calling @everywhere from a module not defined on the workers