Repository files navigation

Horovod is a distributed training framework for TensorFlow, Keras, and PyTorch. The goal of Horovod is to makedistributed Deep Learning fast and easy to use.

Table of Contentsgenerated withDocToc

• Why not traditional Distributed TensorFlow?
• Install
• Concepts
• Usage
• Running Horovod
• Keras
• Estimator API
• PyTorch
• mpi4py
• Inference
• Tensor Fusion
• Analyzing Horovod Performance
• Guides
• Troubleshooting
• Citation
• Publications
• References

The primary motivation for this project is to make it easy to take a single-GPU TensorFlow program and successfully trainit on many GPUs faster. This has two aspects:

1. How much modifications does one have to make to a program to make it distributed, and how easy is it to run it.
2. How much faster would it run in distributed mode?

Internally at Uber we found the MPI model to be much more straightforward and require far less code changes than theDistributed TensorFlow with parameter servers. See theUsage section for more details.

In addition to being easy to use, Horovod is fast. Below is a chart representing the benchmark that was done on 128servers with 4 Pascal GPUs each connected by RoCE-capable 25 Gbit/s network:

Horovod achieves 90% scaling efficiency for both Inception V3 and ResNet-101, and 68% scaling efficiency for VGG-16.See theBenchmarks page to find out how to reproduce these numbers.

While installing MPI and NCCL itself may seem like an extra hassle, it only needs to be done once by the team dealingwith infrastructure, while everyone else in the company who builds the models can enjoy the simplicity of training them atscale.

To install Horovod:

1. InstallOpen MPI or another MPI implementation.

Steps to install Open MPI are listedhere.

2. Install thehorovod pip package.

$ pip install horovod

This basic installation is good for laptops and for getting to know Horovod.If you're installing Horovod on a server with GPUs, read theHorovod on GPU page.If you want to use Docker, read theHorovod in Docker page.

Horovod core principles are based onMPI concepts such assize,rank,local rank,allreduce,allgather andbroadcast. Seehere for more details.

To use Horovod, make the following additions to your program:

1. Runhvd.init().

2. Pin a server GPU to be used by this process usingconfig.gpu_options.visible_device_list.With the typical setup of one GPU per process, this can be set tolocal rank. In that case, the first process onthe server will be allocated the first GPU, second process will be allocated the second GPU and so forth.

3. Scale the learning rate by number of workers. Effective batch size in synchronous distributed training is scaled bythe number of workers. An increase in learning rate compensates for the increased batch size.

4. Wrap optimizer inhvd.DistributedOptimizer. The distributed optimizer delegates gradient computationto the original optimizer, averages gradients usingallreduce orallgather, and then applies those averagedgradients.

5. Addhvd.BroadcastGlobalVariablesHook(0) to broadcast initial variable states from rank 0 to all other processes.This is necessary to ensure consistent initialization of all workers when training is started with random weights orrestored from a checkpoint. Alternatively, if you're not usingMonitoredTrainingSession, you can simply executethehvd.broadcast_global_variables op after global variables have been initialized.

6. Modify your code to save checkpoints only on worker 0 to prevent other workers from corrupting them.This can be accomplished by passingcheckpoint_dir=None totf.train.MonitoredTrainingSession ifhvd.rank() != 0.

Example (see theexamples directory for full training examples):

importtensorflowastfimporthorovod.tensorflowashvd# Initialize Horovodhvd.init()# Pin GPU to be used to process local rank (one GPU per process)config=tf.ConfigProto()config.gpu_options.visible_device_list=str(hvd.local_rank())# Build model...loss= ...opt=tf.train.AdagradOptimizer(0.01*hvd.size())# Add Horovod Distributed Optimizeropt=hvd.DistributedOptimizer(opt)# Add hook to broadcast variables from rank 0 to all other processes during# initialization.hooks= [hvd.BroadcastGlobalVariablesHook(0)]# Make training operationtrain_op=opt.minimize(loss)# Save checkpoints only on worker 0 to prevent other workers from corrupting them.checkpoint_dir='/tmp/train_logs'ifhvd.rank()==0elseNone# The MonitoredTrainingSession takes care of session initialization,# restoring from a checkpoint, saving to a checkpoint, and closing when done# or an error occurs.withtf.train.MonitoredTrainingSession(checkpoint_dir=checkpoint_dir,config=config,hooks=hooks)asmon_sess:whilenotmon_sess.should_stop():# Perform synchronous training.mon_sess.run(train_op)

The example commands below show how to run distributed training. See theRunning Horovodpage for more instructions, including RoCE/InfiniBand tweaks and tips for dealing with hangs.

1. To run on a machine with 4 GPUs:

$ mpirun -np 4 \    -H localhost:4 \    -bind-to none -map-by slot \    -x NCCL_DEBUG=INFO -x LD_LIBRARY_PATH -x PATH \    -mca pml ob1 -mca btl ^openib \    python train.py

2. To run on 4 machines with 4 GPUs each:

$ mpirun -np 16 \    -H server1:4,server2:4,server3:4,server4:4 \    -bind-to none -map-by slot \    -x NCCL_DEBUG=INFO -x LD_LIBRARY_PATH -x PATH \    -mca pml ob1 -mca btl ^openib \    python train.py

3. To run in Docker, see theHorovod in Docker page.

4. To run in Kubernetes, seeKubeflow,MPI Operator,Helm Chart, andFfDL.

Horovod supports Keras and regular TensorFlow in similar ways.

See full trainingsimple andadvanced examples.

Note: Keras 2.0.9 has aknown issue that makes each worker allocateall GPUs on the server, instead of the GPU assigned by thelocal rank. If you have multiple GPUs per server, upgradeto Keras 2.1.2, or downgrade to Keras 2.0.8.

Horovod supports Estimator API and regular TensorFlow in similar ways.

See a full trainingexample.

Horovod supports PyTorch and TensorFlow in similar ways.

Example (also see a full trainingexample):

importtorchimporthorovod.torchashvd# Initialize Horovodhvd.init()# Pin GPU to be used to process local rank (one GPU per process)torch.cuda.set_device(hvd.local_rank())# Define dataset...train_dataset= ...# Partition dataset among workers using DistributedSamplertrain_sampler=torch.utils.data.distributed.DistributedSampler(train_dataset,num_replicas=hvd.size(),rank=hvd.rank())train_loader=torch.utils.data.DataLoader(train_dataset,batch_size=...,sampler=train_sampler)# Build model...model= ...model.cuda()optimizer=optim.SGD(model.parameters())# Add Horovod Distributed Optimizeroptimizer=hvd.DistributedOptimizer(optimizer,named_parameters=model.named_parameters())# Broadcast parameters from rank 0 to all other processes.hvd.broadcast_parameters(model.state_dict(),root_rank=0)forepochinrange(100):forbatch_idx, (data,target)inenumerate(train_loader):data,target=Variable(data),Variable(target)optimizer.zero_grad()output=model(data)loss=F.nll_loss(output,target)loss.backward()optimizer.step()ifbatch_idx%args.log_interval==0:print('Train Epoch: {} [{}/{}]\tLoss: {}'.format(epoch,batch_idx*len(data),len(train_sampler),loss.data[0]))

Note: PyTorch support requires NCCL 2.2 or later. It also works with NCCL 2.1.15 if you are not using RoCE or InfiniBand.

Horovod supports mixing and matching Horovod collectives with other MPI libraries, such asmpi4py,provided that the MPI was built with multi-threading support.

You can check for MPI multi-threading support by querying thehvd.mpi_threads_supported() function.

importhorovod.tensorflowashvd# Initialize Horovodhvd.init()# Verify that MPI multi-threading is supported.asserthvd.mpi_threads_supported()frommpi4pyimportMPIasserthvd.size()==MPI.COMM_WORLD.Get_size()

Learn how to optimize your model for inference and remove Horovod operations from the graphhere.

One of the unique things about Horovod is its ability to interleave communication and computation coupled with the abilityto batch smallallreduce operations, which results in improved performance. We call this batching feature Tensor Fusion.

Seehere for full details and tweaking instructions.

Horovod has the ability to record the timeline of its activity, called Horovod Timeline.

Seehere for full details and usage instructions.

1. Run distributed training in Microsoft Azure usingBatch AI and Horovod.

See theTroubleshooting page and please submit theticketif you can't find an answer.

Please cite Horovod in your publications if it helps your research:

@article{sergeev2018horovod,  Author = {Alexander Sergeev and Mike Del Balso},  Journal = {arXiv preprint arXiv:1802.05799},  Title = {Horovod: fast and easy distributed deep learning in {TensorFlow}},  Year = {2018}}

1. Sergeev, A., Del Balso, M. (2017)Meet Horovod: Uber’s Open Source Distributed Deep Learning Framework for TensorFlow.Retrieved fromhttps://eng.uber.com/horovod/
2. Sergeev, A. (2017)Horovod - Distributed TensorFlow Made Easy. Retrieved fromhttps://www.slideshare.net/AlexanderSergeev4/horovod-distributed-tensorflow-made-easy
3. Sergeev, A., Del Balso, M. (2018)Horovod: fast and easy distributed deep learning in TensorFlow.arXiv:1802.05799

The Horovod source code was based off the Baidutensorflow-allreducerepository written by Andrew Gibiansky and Joel Hestness. Their original work is described in the articleBringing HPC Techniques to Deep Learning.