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

@@ -722,6 +722,8 @@ end
     @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)
@@ -747,6 +749,8 @@ end
     # 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
@@ -1017,15 +1021,19 @@ f16091b = () -> 1
     # Test the behaviour of remotecall(f, ::AbstractWorkerPool), this should
     # keep the worker out of the pool until the underlying remotecall has
     # finished.
-    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
     LocalBar.bar()