JuliaLang/julia — pull request #9046 by amitmurthy (2014-11-18 05:53:10+00:00)
Title: WIP/RFC: User definable transports using ClusterManagers

This is an attempt to make it possible to use alternate transport mechanisms for a Julia cluster.

NOTE: `multi.jl` has been split into `manager.jl` and `multi.jl`. The former has all code related to connection
setup and cluster managers, while the latter has the multi-processing part.

Concept
-------
Currently, all nodes are setup in a mesh network with each node connected to every other node via TCP sockets.

From a scaling point of view, in some cases, a star configuration or even a tree configuration may be preferred.

This patch makes it possible to, for example,

- Use 0MQ for transport. For example, use a 0MQ router in the middle for a star configuration.
- Better integrate with MPI.
- Use MPI as transport, or
- Have the same set of Julia processes participate in both the MPI cluster as well as the Julia cluster. This will allow for folks
to use Julia libraries that leverage MPI as well as the regular Julia parallel infrastructure

NOTE: I am unfamiliar with MPI, so the MPI comments are based on a sketchy knowledge of MPI.

As an example, say we write a ZMQClusterManager and we interconnect thousands of julia processes
via a 0MQ router into a star network.

So while a Julia worker connects to the 0MQ router with a single 0MQ socket,
the interface between ZMQClusterManager and multi.jl is via a pair of "StreamBuffer"s. A "StreamBuffer" is a
non-OS `AsyncStream`. It essentially wraps a `PipeBuffer` and read/write `Condition` variables to
provide waitable in-process stream objects.

While the implementaion may have similarities with a `Channel{Uint8}` (https://github.com/JuliaLang/julia/pull/8507),
it is different in the sense that `StreamBuffer`s implement `AsynsStream`, while `Channel`
implements single element `put!`, `take!`, etc.

Implementation
--------------
Custom cluster managers should implement the following methods:

`launch` - launches multiple worker processes
`connect_m2w` - sets up connections between the master and workers. Optional.
`connect_w2w` - sets up connections between the workers. Optional.
`manage` - bookkeeping / interrupting workers

A default implementation of `connect_m2w` and `connect_w2w` using `TCpSocket` is provided in Base.
NOTE: Custom implementations which return "StreamBuffer"s may not even setup any actual "connection"s.
They may just associate the Julia pid with the appropriate routing information.

Base provides:
```
launch{LocalManager}
launch{SSHManager}
connect_m2w{ClusterManager}
connect_w2w{ClusterManager}
manage{LocalManager}
manage{SSHManager}
```

The control flow starts with an `addprocs(np::Int; kwargs...)` where one of the
keyword args is `manager` which defaults to `LocalManager()`

- `addprocs` calls `launch(::Type, config::Dict, launched::Array, c::Condition)` to start julia workers.
The `launch` method, which is run in a separate task, should
- launch and add a Dict to `launched::Array` for every worker launched. This will typically be a copy of
`config` with additional fields that may be required to connect to the worker.

- `notify` the Condition variable `c`, as and when a worker is launched, so that connections to the launched workers can be setup in parallel

Base provides two `launch` methods :

`launch{T<:LocalManager}(::Type{T}, ....` and `launch{T<:SSHManager}(::Type{T}, ....`

- Next, `connect_m2w(::Type, pid::Int, config::Dict)` is called to setup
connections between the master and workers. It is called for each and every
`config` returned by the `launch` method above. `config` also has an additional
key `:pid` which is the pid of the worker.

`connect_m2w` should return a tuple, `(read_stream::AsyncStream, write_stream::AsyncStream)`.
It can also add any additonal key-value pairs to `config`.

Any bytes read from `read_stream` must have been sent from process `pid`
Any bytes written to `write_stream` should be delivered to process `pid`

Base provides a default implementation which uses regular TCP sockets to
setup the mesh network :
`connect{T<:ClusterManager}(::Type{T}, ....)`

Both `read_stream` and `write_stream` are the same `TCPSocket` in this case.

`connect{T<:ClusterManager}` expects either `:io` or `:host`/ `:port` to be defined
in config, where io is the stdout of the launched worker.

If `:io` is defined, it is used to read the host/port information that is printed by the worker.
If `:host` is defined, it overrides the host printed from reading io.

Both `launch` and `connect_m2w` could add a key `:connect_at`. The value of this
field is sent to all workers to be used in worker-to-worker connection setups.

- `connect_w2w(::Type, pid::Int, config::Dict)` is called to setup worker-to-worker connections.
Since this is called from the workers, very limited config information is available.

`config` in this case has a field `:connect_at` which can be used by the
cluster managers's `connect_w2w` implementation to connect to `pid`.

- `manage(::Type{T}, id::Integer, config::Dict, op::Symbol)` is called from the master
process only with `op` either of `:register`, `:deregister`, `:interrupt` and `"finalize`

startup and specifying the cluster manager in workers
-----------------------------------------------------

- worker processes are started with `--worker ` where `cluster_manager`
is the type of cluster manager.

- ClusterManagers implementing their own transport, should launch Julia workers using
the `-e` or program options.

- `Base.start_worker{T<:ClusterManager}(::Type{T}, out::IO)` is the current entry point
for a worker, which listens on a free port and prints out connection information.

- custom ClusterManagers are free to write their own entry points, which can be called
after loading the cluster manager using `-e` or or program options.

- custom ClusterManagers should call `process_messages(r_stream::AsyncStream, w_stream::AsyncStream; kwargs...)`,
once for every incoming connection/data from another node.

TODO
----

- [ ] Implement StreamBuffer
- [ ] Reference implementation of a ClusterManager using `StreamBuffer`s
- [ ] Documentation