[tune] horovod trainable

[richardliaw]

Why are these changes needed?

This PR allows users to utilize Horovod with Ray Tune.

Caveats:

1. It depends on the GLOO communicator - horovod must be installed with HOROVOD_WITH_GLOO.
2. It assumes that workers will be placed symmetrically across nodes (i.e., workers_per_node).
3. It has an unsafe way of handling identity strings. But there is a large warning in the docs.
4. NICs isolation/choice is currently unsupported. Not quite familiar with what to do here, but would be happy to push a fix if given tips.
5. Function checkpointing is currently unsupported (but would not be hard to do so).

def train(config):
    horovod.init()
    horovod.allreduce()

from ray.tune.integration.horovod import DistributedTrainableCreator
trainable_cls = DistributedTrainableCreator(
    train, num_nodes=1, num_workers_per_node=2, use_gpu=True)

tune.run(trainable_cls)

TODO:

• Add tests
• Add dependencies for tests
• add documentation
• add another example

cc @tgaddair

Related issue number

Checks

• I've run scripts/format.sh to lint the changes in this PR.
• I've included any doc changes needed for https://docs.ray.io/en/latest/.
• I've made sure the tests are passing. Note that there might be a few flaky tests, see the recent failure rates at https://ray-travis-tracker.herokuapp.com/.
• Testing Strategy
  • Unit tests
  • Release tests
  • This PR is not tested (please justify below)

[tgaddair]

Nice! Awesome to see so much progress so quickly.

[tgaddair]

Some of this we can probably move into Horovod so we don't have to expose these internals (which may change) to Ray.

I think it would be good to have a simple horovod.ray.run API to use. Would that be sufficient for Ray Tune?

[richardliaw]

I think so. However, I would need a bit lower level than spark.run. Specifically I would need to be able to pass in an arbitrary class and obtain a list of ray Actors with horovod started on them:

trainable = wrap_function(self.__class__.function)
assert type(trainable) == type  # Ray Tune specific construct.

# encapsulate logic in horovod repo
actors = hvd.ray.start_actors(trainable, num_workers=100, elastic=False, use_gpu=True)

ray.get([a.method_foo.remote() for a in actors])

Note that this actually gives you a lot of flexibility. For example,

class CustomExecutor:
    def execute(self, fn, args):
        return fn(args)

actors = hvd.ray.start_actors(CustomExecutor)

def ray_hvd_run(*args, **kwargs):
    return ray.get([a.execute.remote(*args, **kwargs) for a in actors])

def train_func(args):
    hvd.init()
    ...

# something similar to spark run
ray_hvd_run(train_func, args="foobar")

[richardliaw]

This is basically what I do in _HorovodTrainable.setup(). If this sounds good, I could easily factor it into its own object/move it into Horovod.

[tgaddair]

That sounds good. The two things I would add would be:

1. I think it would be useful to wrap everything in some kind of "Job" object to manage the lifecycle of all the components.
2. We could then build a higher level API on top of this for users who don't need the lower-level control.

So something like:

# module horovod.ray

def create_job(num_hosts, num_slots, executor_cls=_default_executor_cls):
    ...

def run(train_fn, args, kwargs, num_hosts, num_slots):
    job = create_job(num_hosts, num_slots)
    try:
        job.start()
        return job.execute(lambda w: w.execute(train_fn, *args, **kwargs))
    finally:
        job.stop()

Something like that. What do you think? Would that give you enough flexibility for this use case?

[richardliaw]

Yeah, that definitely sounds good; let me push a refactor.

[krfricke]

Looks good so far, but I will have to look more into the horovod job

[krfricke]

Will this fail if we add more than 100 workers per node?

[richardliaw]

Yeah; though most likely GPUs on a node will be limited to 16.

[krfricke]

Is that fixture imported here? Seems only to be defined in test_horovod.py.

We should also probably move this file to tests

[richardliaw]

Yeah, this should be included upstream in horovod.