Ch.4 Tuning Distributed Network Communication - rFronteddu/general_wiki GitHub Wiki

By tuning bucket sizes and scheduling these transfers appropriately, one can achieve a higher degree of overlap and prevent communication delays from stalling the com‐ pute pipeline. Tools such as the PyTorch profiler and NVIDIA Nsight Systems offer insight into whether your computation and communication are overlapping, allowing engineers to adjust these parameters for maximal efficiency

By combining larger batch sizes, gradient accumulation, asynchronous transfers, compression, and bucketing into one cohesive strategy, large, distributed AI models can overcome network limitations and reduce idle time. This design minimizes syn‐ chronization events while achieving high throughput and optimal GPU utilization

AI frameworks hide most of this complexity. PyTorch’s DistributedData Parallel automatically installs hooks on the backward pass so that each gradient bucket triggers an asynchronous NCCL all-reduce on a dedicated communication CUDA stream, while the default CUDA stream continues computing gradients for subsequent layers

To maintain proper overlapping, avoid unnecessary synchronization points with torch.cuda.synchronize() or inadvertently triggering a full device sync by moving tensors to the CPU with torch.Tensor.item().

Bucketing, as implemented in PyTorch’s Distributed Data Parallel (DDP) communication mechanism, also reduces per-call overhead by grouping many small tensors into larger messages. However, bucket sizing is a trade-off. Very large buckets maxi‐ mize bandwidth utilization but delay the start of communication since you wait for more gradients to accumulate before kicking off the all-reduce. Very small buckets start transfers earlier but incur more overhead due to many small NCCL calls. As of this writing, the default bucket size in PyTorch DDP is 25 MB. This is a balance that overlaps well in most cases. However, if you have a model with very large layers, you might increase this to reduce overhead. If you have a model with many small layers, you might actually benefit from smaller buckets to start communication sooner. Ultimately, achieving maximal overlap may require profiling different bucket sizes to see which yields the best iteration time.

The takeaway is that a well-tuned DDP should overlap most of the gradient commnication with computation.

PyTorch’s DDP’s default overlap strategy is often described as wait-free back-propagation (WFBP), which bucketizes gradients and launches reductions as soon as each bucket is ready

Avoid operations that inadvertently move tensors from the GPU to the CPU (e.g., calling .item() on a tensor) until you’re sure that all asynchronous GPU work is finished. Otherwise, you will force a synchronization, stall the computation, and slow down your training or inference workload. This typically happens when adding print() or log() statements for debugging. These can be disastrous for performance.

Also, manual calls to torch.cuda.synchronize() should be minimized and used only for accurate benchmarking—or when required for correctness. Otherwise, they will serialize GPU work and negatively impact performance

DDP’s design and PyTorch’s operations are already asynchronous and handle dependencies correctly. Explicit synchronization is rarely needed in user code.

Prefer RDMA paths where available and verified. And always confirm (and continuously reconfirm) with logs and micro-benchmarks that the RDMA data path is active.

n container environments like Docker and Kubernetes, ensure the container has direct access to the host’s InfiniBand devices (e.g., /dev/infiniband). Otherwise, NCCL may silently fall back to TCP sockets instead of GPUDirect RDMA—and without any obvious errors to highlight the degradation. This results in throughput dropping from tens of GB/s to only a few Gb/s, with no obvious error messages.

A related container pitfall arises when the container’s GID assignments don’t match the host, as in some “rdma-shared” Docker images. This prevents GPUDirect registration and uses CPU-driven RDMA copies instead of using true GPU-based RDMA.

Always verify that it is true GPUDirect RDMA. Confirm that the kernel module is loaded with lsmod | grep nvidia_peermem, and check dmesg for initialization. For an end-to-end check, run NCCL with NCCL_DEBUG=INFO to confirm NET/IB paths and use RDMA perftests with --use_cuda to validate GPU-to-GPU transfers. Verifying will help prevent stealthy performance degradations

• Understand the topology: Use nvidia-smi topo -m to get a basic GPU interconnect view. For NVSwitch- and NVLink-based systems also use nvidiasmi nvlink or Nsight Systems to understand multihop switch fabric connectivity.
• Leverage NVLink Switch: Make sure jobs are placed within the same NVLink domain to fully utilize this ultrafast interconnect.
• Make sure to use RDMA
• Aggregate bandwidth with multiple NICs: NCCL can stripe traffic across multiple NICs (called multirail) to increase bandwidth.
• Utilize optimized “direct NIC”: Favor high-bandwidth, multirail NIC configurations that give each GPU or small groups of GPUs sufficient dedicated network bandwidth. With modern GPU systems, NCCL supports GPU-initiated networking with InfiniBand GPUDirect Async (IBGDA) and the direct NIC path without CPU intervention.
• Check for misconfiguration: A common pitfall is a mismatch in network configuration that causes a fallback to a slower path. Tools like NCCL’s debugging output and network interface counters (ibstat, ifstat) can help verify which interface is being used more heavily. For modern systems with large 200–400 Gbps paths, dropping to 10 Gbps would cause a severe bottleneck.

Gloo pythorch backend uses CPUs and TCP sockets. NCCL is the preferred backend for NVIDIA GPUs. If we were trying to allreduce 400MB, we would observe it taking 200ms (2GB/s)-- much lower than infiniband limit of 100GB/s, and also we would see near 100% CPU utilization. You can verify the backend in PyTorch by calling torch.distributed.get_backend(). In a production cluster environment with multiple NICs, you should explicitly set NCCL_SOCKET_IFNAME=ib0 so that NCCL’s initial TCP handshake runs over the InfiniBand host channel adapter (HCA). This ensures it bootstraps correctly and then hands off to GPUDirect RDMA on the fastest path. Ensure that all nodes can reach one another over the selected interconnect.

If you run PyTorch’s bundled NCCL (e.g., torch.cuda.nccl.version() == ()) against a different version of the system-installed libnccl, you will hang the system or fall back to a slower implementation. Make sure you have alignment by matching nvidia-nccl-cu* packages or rebuilding PyTorch against the system NCCL.

NCCL uses ephemeral TCP ports for its out‐of‐band setup, and if your OS’s net.ipv4.ip_local_port_range is too narrow, you can exhaust available ports, caus‐ ing failed or stalled handshakes. It’s recommended that you widen your port range in /proc/sys/net/ipv4/ip_local_port_range (e.g., 50000 51000) to avoid hidden bootstrap failures.

When profiling your workload under these unfortunate conditions, you will observe that scaling to multiple nodes significantly slows down training. In other words, the “per-GPU throughput” will drop. In this case, check the network links. Monitor the network throughput using nvidia-smi dmon, for instance, to collect NVLink/PCIe/Network statistics. You can also use built-in tools like ethtool -S or ip -s link show for byte/packet counters, or launch interactive monitors such as iftop or nload to watch live NIC throughput.

You can also try to utilize multiple interfaces, if available. If you’re saturating an 800 Gbps (100 GB/s) InfiniBand link, for instance, and your job needs more network throughput, consider enabling NCCL’s multi-NIC support—assuming that you have multiple NICs. Make sure that NCCL_NSOCKS_PERTHREAD and NCCL_SOCKET_NTHREADS are tuned, as these control how many parallel connections and threads NCCL uses for network transfers. In cases with multiple NICs, increasing these environment variable values from their platform-dependent defaults can help utilize both NICs

Remember that the product of threads and sockets should not exceed 64 per NVIDIA guidance since more threads mean more CPU usage. Increase these thread-related settings stepwise (e.g., 2 → 4 → 8), and continuously measure the throughput. Too many threads will contend for resources and potentially diminish returns.

In multinode training, the slowest node, or GPU, will determine the overall pace because synchronization needs to wait for every node and GPU to respond. Using monitoring tools like NVIDIA’s DCGM or InfiniBand counters on each node can help spot if one node has degraded performance due to NIC link flapping or GPU thermal throttling. It’s also useful to use collective profiling tools such as PyTorch’s torch.distributed.monitored_barrier to identify if a particular rank is consistently lagging

# barrier_straggler
import torch
import torch.distributed as dist
import os
import datetime
def run(rank, world_size):
  dist.init_process_group(backend="nccl", init_method="env://")
  local_rank = int(os.environ["LOCAL_RANK"])
  torch.cuda.set_device(local_rank)
  # ... your forward/backward work here ...
  # Before syncing at end of iteration, use a monitored barrier:
  try:
    # Wait up to 30 seconds for all ranks
    # if one lags, you’ll get a timeout on that rank
    dist.monitored_barrier(timeout=datetime.timedelta(seconds=30))
  except RuntimeError as e:
    print(f"Rank {rank} timed out at barrier: {e}")
  # Now proceed knowing all ranks are roughly in sync
  dist.destroy_process_group()

Here, dist.monitored_barrier(timeout=datetime.timedelta(seconds=30)) will raise an error on any GPU that doesn’t arrive within 30s. This will help you pinpoint stragglers. Combine this with NCCL_DEBUG=INFO and NCCL_ASYNC_ERROR_HANDLING=1 to get both PyTorch and NCCL logs around which rank or link is slow.

PyTorch’s caching allocator holds onto GPU memory across iterations. In distributed settings using UCX/RDMA, these long-lived allocations can exhaust registration pools or fragment memory, causing sporadic allocation failures or performance cliffs. Monitoring torch.cuda.memory_reserved() versus memory_allocated() helps surface these edge cases:

NVIDIA NCCL is a many-to-many communication library for operations, called collectives, used by groups of GPUs to share data. NCCL underpins most multi-GPU training workloads in NVIDIA’s ecosystem.

NCCL provides optimized implementations of collective communication operations like all-reduce, all-gather, broadcast, and reduce-scatter that scale from a few GPUs to many thousands and, someday, millions.

While NCCL can use a simple pattern communication like ring all-reduce to commu‐ nicate with each link equally, it will automatically use a topology-aware hierarchical communication pattern to maximize communication performance. For systems with multiple NUMA node domains, for instance, NCCL might first do an intranode reduce, then a cross-node reduce, then an intranode broadcast, which is effectively a hierarchical all-reduce.

It is possible to override NCCL’s algorithm selection with the environment vari‐ able NCCL_ALGO (e.g., NCCL_ALGO=NVLS,NVLSTree,Tree,Ring,PAT, etc.), but generally NCCL does a good job of automatically choosing the best path based on the topology. Manual override is usually only for specific situations like troubleshooting, research experiments, and more.

Tools like NVIDIA Nsight Systems—or NCCL’s own traces with NCCL_DEBUG=INFO and NCCL_TOPO_DUMP_FILE=—will show if NVLink paths are being utilized fully.

Internally, NCCL can employ different communication algorithms depending on the size of data, number of GPUs, and topology. The primary algorithms NCCL uses for collectives are Ring, Tree, CollTree, CollNet, and Parallel Aggregated Tree (PAT).

• Ring: bandwidth-dominated workload
• Tree and NVLSTree: latency-dominated workload
• CollTree (hierarchical tree collectives): preferred when cross node latency dominates
• CollNet (hierarchical collectives across nodes): very large, multinode GPU clusters
• Parallel aggregated tree (PAT): The result is near–ring-level throughput for large data transfers plus tree‐level latency advantages for smaller segments.

As when choosing any communication algorithm, the choice of NCCL algorithm typ‐ ically comes down to message size and topology. Small messages (on the order of 10s of megabytes) favor tree algorithms since there are fewer steps. Large messages favor ring algorithms because they provide better bandwidth utilization.

if profiling your workload reveals suboptimal communication, such as unexpectedly high cross-node latency, you can override the communication algorithm on a case-by-case basis by setting the NCCL_ALGO environment variable. This will force NCCL to use a particular algorithm on that communicator. If setting this variable in code, make sure to do it before calling ncclCommInitRank()

The key is to overlap communication with computation at every level. This includes using NCCL for all-reduce and NIXL for one-to-one transfers. Using these mecha‐ nisms, you can scale to thousands and millions of GPUs with high efficiency

Other techniques like gradient accumulation and activation check‐ pointing are also critical at ultrascale to manage the memory foot‐ print without sacrificing throughput.

When scaling to multiple GPUs on a single node, PyTorch offers both data-parallel (split the data) and model-parallel (split the model) approaches at the framework level.

let’s compare two of the most basic data-parallel strategies from a systems performance standpoint: nn.DataParallel (DP) and torch.distributed.DistributedDataParallel (DDP). It’s important to understand their differences as choosing the wrong one can severely impact performance:

• Data parallelism (DP): DP is an easy-to-use API that involves a single process, or single Python thread, controlling multiple GPUs.
• Fully sharded data parallelism (FSDP): FSDP avoids full model replicas by sharding activations, gradients, and parameters across GPUs, greatly reducing memory overhead.
• Distributed Data Parallel (DDP): DDP uses one process per GPU device and relies on NCCL to communicate gradients. Like most simple data parallel strategies (FSDP being the exception), each process has its own copy of the model.

1. Creating NCCL communicators too often: init_process_group call is designed to be called once at startup, you should avoid any design that reinitializes it on every iteration.
2. Do not create and destroy NCCL communicators on every iteration: create the subcommunicators once at the beginning using PyTorch’s torch.distributed.new_group() and reuse these communicators. If you need to create multiple communicators because, for instance, you have a dynamic runtime membership scenario or a staged initialization, NCCL provides a C++ API to initialize multiple communicators together using ncclGroupStart(), ncclCommInitRank(...), and ncclGroupEnd(). PyTorch does not support fully dynamic membership changes at runtime without a full communicator teardown. All ranks must invoke creation and destruction calls in lockstep to prevent hangs.
3. Avoid overtuning or disabling NCCL features with environment variables
4. Verify CPU-GPU NUMA-node affinity for NCCL threads: The recommended approach is to bind each GPU process to the CPU cores for its NUMA domain and then set NCCL_IGNORE_CPU_AFFINITY=1 so that NCCL can fine-tune thread placement within those cores. PyTorch’s launch utilities handle much of this automatically, but it’s good to verify.
5. Resist the temptation to ignore NCCL warnings and errors
6. NCCL communicator hangs, errors, or shuts down completely: NCCL supports asynchronous error handling and failover for cases like network errors

NIXL was designed specifically to accelerate large-scale LLM distributed and disaggregated inference. NIXL is a core component of NVIDIA’s open source Dynamo inference engine. NCCL remains the standard for many-to-many collective operations common in large-scale training such as all-reduce. NIXL, however, targets one-to-one or one-to-few data transfers that are common in large-scale inference such as moving KV cache data.

The inference path of a transformer-based model is actually split into two different stages: prefill and decode.

• Prefill, is often compute bound as it uses many matrix multiplications to build the KV cache from the incoming request data (aka prompt).
• Decode, is often memory-throughput bound, as it needs to gather the model weights from GPU HBM memory to calculate the next set of tokens (aka completion or response).

This prefill/decode split is implemented in common inference engines vLLM, SGLang, and NVIDIA’s Dynamo and TensorRT-LLM. The prefill (prompt ingestion) creates the KV cache, and the decode (generation) uses this cache. NIXL specifically accelerates the transfer of the KV cache between nodes in this workflow.

The traditional setup has each GPU node handle both the prefill (compute-bound) and decode (memory-bound, I/O-bound) phases. The disaggregated serving configuration places the prefill workers in the GPU cluster and the decode workers in another GPU cluster. A GPU in the prefill cluster generates the KV cache for the input sequence and uses NIXL to transfer it to a GPU in the decode cluster. This specialization produces higher overall throughput and advanced scaling configurations.

In such cases, the KV cache, which can run into tens of gigabytes in a long prompt, must move seamlessly from one processing unit to another in near-real time. This way, the text generation happens at speeds that are unnoticeable to end users.

NIXL provides a direct channel for transferring data from one GPU to another or a small group of GPUs across compute nodes and even across racks. The system looks at the available pathways and always selects the one that gets the data there the quickest.

NIXL offers a straightforward API. You post a transfer request with a pointer to the data and a destination—either GPUs, CPUs, or storage targets like Amazon S3. NIXL will transfer that data as fast as possible. You register memory with regis terMem, obtain transfer descriptors with trim, prepare a nonblocking request with prepXfer, and submit it with postXfer. NIXL chooses whether to perform a direct PCIe or NVLink copy, an RDMA transfer, or a storage path such as GPUDirect Storage. The NIXL library is nonblocking and returns a request handle that you poll with checkXfer to detect completion.

A nixlAgent is NIXL’s core transfer object. It encapsulates the endpoint configuration, memory registrations, and backend selection. It also manages metadata, connection information, and asynchronous transfer requests to and from other agents. You need two agents for a transfer because each nixlAgent instance represents one endpoint in the transfer. The source agent (agentSrc) encapsulates the context, memory registrations, and backends for the origin of the data. The destination agent (agentDst) does the same for the receiver side.

NIXL is not a replacement for NCCL but a complement. NCCL still handles synchronized collectives for GPUs working on a single task/stage in parallel, such as an all-reduce split across multiple GPUs. NIXL, on the other hand, performs asynchronous data transfers between tasks/stages—or between distinct components (e.g., GPUs, CPUs, storage) in a distributed system.

• NCCL (collective communication) Primary use case: Many-to-many collectives (e.g., all-reduce, all-gather) for tightly coupled GPU groups in training

• NIXL (point-to-point communication) Primary use case: One-to-one or one-to-few transfers (e.g., sending large tensors or caches) for distributed inference or pipelining.