Introduce AbstractWorkerPool, CachingPool

[amitmurthy]

This PR does the following:

• Introduce AbstractWorkerPool. WorkerPool is an implementation of AbstractWorkerPool
• CachingPool <: AbstractWorkerPool provides two functionalities

  1. caches functions executed via remote, remotecall_fetch(f, cp::CachingPool.....) and others.
     This helps in efficiently executing closures with large closed data multiple times on remote workers.

  2. implements a define(wp::CachingPool; kwargs...) method which broadcasts all the arguments specified onto workers in the pool. These are bound under Main with the same name as the keyword arg.

• empty!(wp::CachingPool) releases the closure cache as well as set all previously bound names to nothing. We cannot "unbind" a name in Julia yet.

Addresses #10438, #16345 and other issues on easily broadcasting variables.

Example of large closure:

const wp = Base.CachingPool(workers())

function foo(c, n)
    a = ones(n)
    f = x->sum(a) * x
    t = @elapsed results = pmap(wp, f, 1:c)
    @assert results == map(f, 1:c)
    t
end

for coll_sz in [10, 10^2, 10^4]
    for data_sz in [10, 10^4, 10^6]
        println("coll_sz:", coll_sz, ", data_sz:", data_sz)
        tt = foo(coll_sz, data_sz)
        println(tt)
    end
end

Example of broadcast (from the test file):

# CachingPool tests
wp = CachingPool(workers())

# test variable broadcasting
define(wp; wp_foo=1, wp_bar="foobar", wp_baz=ones(10^6))
for p in workers()
    @test remotecall_fetch(()->wp_foo, p) == 1
    @test remotecall_fetch(()->wp_bar, p) == "foobar"
    @test remotecall_fetch(()->wp_baz, p) == ones(10^6)
end

empty!(wp)

for p in workers()
    @test remotecall_fetch(()->wp_foo, p) == nothing
    @test remotecall_fetch(()->wp_bar, p) == nothing
    @test remotecall_fetch(()->wp_baz, p) == nothing
end

Doc updates are pending.

Edit: Means of broadcasting globals across workers have been removed

[tkelman]

This is a pretty generic name to start exporting for this. Is it more like a remoteset! or remotedefine maybe?

[oxinabox]

It is generic, but that is the beauty of multiple dispatch, no?
(That said it still might be too generic, I'm not sure either way)

[tkelman]

if we can't say "this is what this verb should mean" then maybe it's not a name we should be exporting as a generic function yet

[JeffBezanson]

Agree with @tkelman ; if we have this function it should be called something like remoteset!. Modifying global variables, especially in other processes, is not the kind of thing that gets a nice name.

[JeffBezanson]

I don't think global variables can be the way to do broadcasting. It has significant problems for composability and performance.

[amitmurthy]

I don't think global variables can be the way to do broadcasting. It has significant problems for composability and performance.

I suspect people end up doing this anyways since we don't have an easy way to achieve this. Storing variable data in remote references and using fetch all over your code isn't very pretty to look at.

Declaring them as const would mean memory cannot be freed.

Wrapping them in getter functions is slightly better - the functions could be named get_<symbol>(), for example get_foo() would return the value of foo. Still ugly but a little better than fetch(ref_foo)

[JeffBezanson]

It's well accepted that one shouldn't use global variables for everything. We need to provide a good alternative so that people don't have to do it. Not being able to combine two functions because they both have a variable called x isn't supposed to happen in the 21st century.

[amitmurthy]

I understand. Maybe we can leave it unexported for now? While an inefficient practice there are enough requests on the forums for just this, especially for smaller and simpler programs. And people end up defining via @everywhere or eval on the workers.

[amitmurthy]

The PR has been updated with an alternate implementation for remoteset! :

remoteset!(pool::CachingPool, isconst::Bool=true, mod=Symbol(:Mod_, randstring()); kwargs...)

• By default all globals are declared as const. The assumption here is variables defined thus are required for the duration of the program run. Can override via arg isconst
• By default variables are defined as globals in a new module. This avoids namespace clashes. remoteset! returns the module. Users can specify an existing (or a specific name for the new module) via arg mod.

[amitmurthy]

Added docs.

I propose we either add both CachingPool and method remoteset! or neither. The reason being while the type implements caching of closures and is useful when large amount of data is captured, closures only keep a reference to global variables and not the data itself. remoteset! is useful for the latter case.

[tkelman]

what's the difference between push! and put! here?

[amitmurthy]

My penchant for reusing verbs. take! and put! remove and add back workers from the list of workers available for remote execution - this is maintained in pool.channel.

push! adds a new worker to the pool itself. Maintained in pool.workers

[amitmurthy]

Have addressed feedback comments.

Have also changed empty!(pool) to clear(pool) and exported the same.

clear here would mean "clear the cache" and in #16774 to mean "clear the state of the serializer".

[tkelman]

clear certainly has side effects here

[amitmurthy]

Changed to clear!.

Will leave it open for a couple more days. Would like some feedback on the interface and how it can be improved. I believe Jeff's reservations have been partially addressed with defining variables as consts and in their own namespaces by default.

In case of any strong reservations I am open to have only AbstractWorkerPool defined in Base, with CachingPool added as an example or as a separate package.

[amitmurthy]

Have removed the hack to broadcast globals. Large globals can be captured in closures by wrapping them with a let block.

For global foo of 10^8 floats with a caching pool:

julia> let foo=foo
           @time pmap(wp, i->sum(foo)+i, 1:100);
       end;
  4.932819 seconds (403.47 k allocations: 19.095 MB, 0.37% gc time)

Without

julia> let foo=foo
           @time pmap(i->sum(foo)+i, 1:100);
       end;
 46.053112 seconds (1.35 M allocations: 57.517 MB, 0.05% gc time)

For very small closures, the regular workerpool is faster by around 10-20%. The crossover is around 8MB (10^6 floats) when they are equal.

Closes #10438

[tkelman]

this worker-pool signature of remotecall_wait doesn't look like it's present in the rst docs

[amitmurthy]

Merging in a short while.