Source-linked AI summary
Horovod: fast and easy distributed deep learning in TensorFlow
Alexander Sergeev, Mike Del Balso
TL;DR
As models and datasets grew, TensorFlow users faced long training times, communication overhead, and difficult distributed-training code changes. Horovod uses ring-allreduce with a simplified API to make distributed TensorFlow training easier and faster. Benchmarks report strong scaling on TCP and RDMA networks, while MPI installation and large-model support remain practical scope boundaries.
Problem
Growing models and datasets created training times of a week or longer, while standard distributed TensorFlow introduced difficult code changes and communication overhead.
Method
Horovod uses ring-allreduce for gradient communication and provides a simplified TensorFlow API for distributed training.
Results
Horovod scales well on plain TCP and RDMA-capable networks, with RDMA enabling over 90 percent scaling efficiency for Inception V3 and ResNet-101.
Takeaways & Limitations
Horovod provides an easier way to start and accelerate distributed deep learning projects with TensorFlow.
Takeaways & Limitations
Cluster deployment still requires effort to install MPI and configure workload managers and network hardware.
Abstract
from arXiv · showhide
Training modern deep learning models requires large amounts of computation, often provided by GPUs. Scaling computation from one GPU to many can enable much faster training and research progress but entails two complications. First, the training library must support inter-GPU communication. Depending on the particular methods employed, this communication may entail anywhere from negligible to significant overhead. Second, the user must modify his or her training code to take advantage of inter-GPU communication. Depending on the training library's API, the modification required may be either significant or minimal. Existing methods for enabling multi-GPU training under the TensorFlow library entail non-negligible communication overhead and require users to heavily modify their model-building code, leading many researchers to avoid the whole mess and stick with slower single-GPU training. In this paper we introduce Horovod, an open source library that improves on both obstructions to scaling: it employs efficient inter-GPU communication via ring reduction and requires only a few lines of modification to user code, enabling faster, easier distributed training in TensorFlow. Horovod is available under the Apache 2.0 license at https://github.com/uber/horovod
1 Introduction
Deep learning advances increased the need for scalable computation, while TensorFlow’s broad adoption, performance, flexibility, and end-to-end support made it Uber’s preferred framework. Horovod is introduced as an open-source Michelangelo component to simplify and accelerate distributed TensorFlow training.
- Deep learning advances drove progress across image processing, speech recognition, forecasting, and Uber applications.
- TensorFlow combined wide adoption, high performance, low-level flexibility, and broad end-to-end support for Uber’s deep learning use cases.
- Horovod is an open-source Michelangelo component designed to make distributed TensorFlow projects easier to start and faster to run.
2 Going distributed
Growing models and datasets made training times unacceptably long, motivating distributed training. Standard distributed TensorFlow posed usability and scalability problems, including substantial communication overhead on large GPU configurations.
- Training times sometimes reached a week or longer as model sizes and data consumption grew, motivating distributed training.
- Standard distributed TensorFlow required users to understand workers, parameter servers, cluster specifications, and synchronization components.
- The API introduced subtle, hard-to-diagnose bugs and a steep learning curve unrelated to users’ primary goal of accelerating existing models.
- About half of the resources were lost to communication overhead when training on 128 GPUs with standard distributed TensorFlow.
- Figure 1 compares images processed per second for Inception V3 and ResNet-101 on 1–128 NVIDIA Pascal GPUs against theoretically ideal scaling.
3 Leveraging a different type of algorithm
Data-parallel training splits data across model copies, computes gradients independently, and averages them for consistent updates. Ring-allreduce replaces parameter-server communication with a bandwidth-optimal, simpler allreduce approach that requires only an allreduce operation in user code.
- Data parallelism: Data-parallel training runs multiple copies that read data chunks, compute gradients, average them, and repeat.
- Parameter servers: Parameter-server training assigns worker and parameter-server roles, with workers sending gradients to parameter servers for averaging.
- Parameter servers: Choosing parameter-server ratios can create bottlenecks or saturate network interconnects through all-to-all communication.
- Parameter servers: Distributed TensorFlow increased program complexity through explicit process startup, service discovery, server construction, device placement, and multi-GPU code restructuring.
- Ring-allreduce: Ring-allreduce communicates with two peers 2*(N −1) times and is described as bandwidth-optimal when the data buffer is sufficiently large.
- Ring-allreduce: MPI launches program copies and transparently establishes communication, leaving users to add an allreduce() operation for gradient averaging.
4 Introducing Horovod
Horovod packages ring-allreduce as a standalone Python library, uses NCCL for optimized collective communication, and extends support to multi-GPU models within one server. Its API reduces user changes to four operations and adds consistent initialization support.
- Horovod’s introduction was motivated by the usability and performance advantages of ring-allreduce.
- Horovod was converted into a standalone Python package so teams could use ring-allreduce across different TensorFlow releases.
- NCCL replaced the draft implementation with an optimized ring-allreduce capable of operating across multiple machines.
- Horovod added support for models fitting within one server while spanning multiple GPUs.
- The API added broadcast-based consistent initialization and reduced required modifications to a single-GPU program to four operations.
5 Distributing your training job with Horovod
Horovod distributes single-GPU TensorFlow programs with only a few code changes, using MPI to launch processes and ring-allreduce to average gradients. Its integration includes GPU assignment, distributed optimization, and synchronized variable initialization.
- Horovod requires only a few changes to make single-GPU TensorFlow programs distributed.The example adds initialization, GPU assignment, a distributed optimizer, and variable broadcasting.
- hvd.init() initializes Horovod, while each TensorFlow process is assigned a GPU using its local rank.The visible device list is set to str(hvd.local_rank()).
- hvd.DistributedOptimizer(opt) averages gradients using ring-allreduce while wrapping a regular TensorFlow optimizer.This replaces substantial parameter-server boilerplate with a distributed optimizer wrapper.
- BroadcastGlobalVariablesHook(0) broadcasts variables from the first process to the others to ensure consistent initialization.Without MonitoredTrainingSession, users can run the broadcast operation directly.
- MPI launches multiple copies of the program across servers, distributing the training script across the requested nodes and GPUs.The example uses mpirun to run four GPUs on each of four servers.
- Horovod is presented as effective for distributing single-GPU or single-server programs because of its ease of use, debugging efficiency, and speed.The same distribution steps also apply to Keras programs.
6 Horovod Timeline
Horovod Timeline provides a high-level view of operation timelines across worker nodes during distributed training. It is designed to help users identify bugs and debug performance issues without cross-referencing separate server profiles.
- Native TensorFlow timelines and CUDA Profiler workflows require collecting and cross-referencing profiles from multiple servers.
- Horovod Timeline shows what each node was doing at each time step throughout a training job.It is compatible with Chrome’s about:tracing trace event profiling viewer.
- Users enable Horovod Timeline with a single environment variable and view profiling results in the browser through chrome://tracing.
7 Tensor Fusion
Tensor Fusion combines small tensors before ring-allreduce because ring-allreduce is less efficient for very small tensors. The approach improved performance by up to 65 percent on large-layer-count models over an unoptimized TCP network.
- 7 Tensor Fusion: Models with many tensors, such as ResNet-101, can generate many tiny allreduce operations that reduce ring-allreduce efficiency.Ring-allreduce uses the network optimally when tensors are large enough.
- 7 Tensor Fusion: Tensor Fusion fuses tensors together before Horovod calls ring-allreduce.
- 7 Tensor Fusion: 65 percent improvement in performance was observed on models with a large number of layers running on an unoptimized TCP network.This was the upper observed improvement reported for the Tensor Fusion experiments.
- 7 Tensor Fusion: The fusion procedure selects ready tensors of the same type, copies them into a buffer, allreduces the buffer, and copies results back.The default fusion buffer size is 64 MB, and the process repeats until no tensors remain in the cycle.
- 7 Tensor Fusion: Horovod and Tensor Fusion are presented as improving efficiency, speed, and ease of use across machine-learning systems.
8 Horovod Benchmarks
Horovod improved distributed TensorFlow scaling over TCP and RDMA-capable networks, with gains depending on model communication demands. RDMA mattered most for VGG-16, while Inception V3 and ResNet-101 saw smaller improvements.
- TCP benchmarks: 88 percent efficiency made Inception V3 and ResNet-101 training about twice as fast as standard distributed TensorFlow.The comparison used distributed training over 25GbE TCP and different numbers of NVIDIA Pascal GPUs.
- RDMA results: Three to four percent increases over TCP let RDMA exceed 90 percent scaling efficiency for Inception V3 and ResNet-101.RDMA did not significantly improve performance for these two models, despite improving scaling efficiency.
- RDMA results: 30 percent speedup with RDMA occurred for VGG-16, whose parameter-heavy architecture shifted the bottleneck from computation to communication.Its fully connected layers and relatively small number of layers created a networking bottleneck.
- Network comparison: Horovod scaled well on both plain TCP and RDMA-capable networks.The benchmarks evaluated images processed per second across different GPU counts and model architectures.
- Future work: The authors identify further performance optimization as ongoing work.They describe Horovod’s reported benchmarks as only an initial exploration of possible optimizations.
9 Next steps
The next steps focus on making Horovod easier to deploy, expanding guidance for distributed-training parameter adjustments, and adding examples for larger models. The authors also invite community contributions.
- Installation: MPI installation on clusters remains an active improvement area because workload managers and network hardware require different configuration efforts.The authors are developing reference cluster designs with the MPI community and network hardware vendors.
- Training guidance: The authors plan to collect and share guidance on adjusting model hyperparameters for distributed training.They describe this research area as still being in its early stages.
- Model coverage: Horovod currently supports models fitting within one server while spanning multiple GPUs, and the authors seek examples for larger models.They encourage others to test Horovod on models spanning multiple GPUs.
- Community: The project welcomes reports of issues, shared speedups, and pull requests.Community feedback and contributions are explicitly invited.