Point-to-Point и Collective Operations в PyTorch: как GPUs обмениваются данными

Когда одна GPU не может молчать: зачем вообще обмениваться данными

Представьте: у вас есть 8 RTX 5090 Ti (да, в 2026 они уже существуют), каждая с 48GB VRAM. Вы запускаете обучение Llama-3 405B параметров. И тут понимаете - даже на одной карте модель не помещается. Знакомо? Это момент, когда distributed training перестает быть опцией и становится необходимостью.

Но вот что бесит: большинство гайдов показывают только DataParallel или DistributedDataParallel. Как будто других способов нет. А между тем, под капотом этих абстракций скрываются два фундаментальных механизма: point-to-point (точка-точка) и collective (коллективные) операции.

Point-to-Point: когда вам нужен прямой разговор

Point-to-point операции - это как приватный чат между двумя GPU. Один отправляет, другой получает. Никто больше не видит этот разговор. Звучит просто? На практике здесь больше подводных камней, чем в Бермудском треугольнике.

1 Когда использовать send/recv вместо коллективных операций

Коллективные операции синхронизируют все процессы в группе. Все ждут самого медленного. Point-to-point позволяет двум GPU общаться, пока остальные делают что-то полезное.

Типичный сценарий: pipeline parallelism. GPU 0 обрабатывает первые слои, передает активации GPU 1, которая обрабатывает следующие слои, и так далее. Пока GPU 1 ждет данные от GPU 0, GPU 2 может уже вычислять свои градиенты.

# ПРАВИЛЬНО: асинхронная передача в pipeline
import torch.distributed as dist

# Инициализация (обязательно для всех операций)
dist.init_process_group(backend='nccl')
rank = dist.get_rank()
world_size = dist.get_world_size()

# Неправильный подход (так делают 90% новичков):
# if rank == 0:
#     tensor_to_send = torch.randn(1024, 1024, device='cuda:0')
#     dist.send(tensor_to_send, dst=1)  # БЛОКИРУЕТСЯ!
# elif rank == 1:
#     tensor_recv = torch.empty(1024, 1024, device='cuda:1')
#     dist.recv(tensor_recv, src=0)  # Тоже блокируется

# Правильный подход с PyTorch 2.4+:
if rank == 0:
    tensor_to_send = torch.randn(1024, 1024, device='cuda:0')
    req = dist.isend(tensor_to_send, dst=1)  # НЕБЛОКИРУЮЩИЙ!
    # Можем делать другие вычисления
    req.wait()  # Ждем только когда нужно
elif rank == 1:
    tensor_recv = torch.empty(1024, 1024, device='cuda:1')
    req = dist.irecv(tensor_recv, src=0)  # Тоже неблокирующий
    # Пока ждем данные, можем подготовить что-то
    req.wait()

2 Самые частые ошибки в point-to-point операциях

Я видел эти ошибки в десятках проектов. Они как грабли - наступают все, но никто не учится.

Collective Operations: когда нужно собрать всех за одним столом

Коллективные операции - это семейный ужин. Все должны быть за столом, все получают свою порцию. Никто не начинает есть, пока все не сядут.

В distributed training коллективные операции используются для синхронизации градиентов. Каждая GPU вычисляет градиенты на своем куске данных, потом все градиенты усредняются.

all_reduce: король коллективных операций

all_reduce делает две вещи: reduce (суммирование всех тензоров) и broadcast (рассылка результата всем). В DataParallel это происходит автоматически. В ручном режиме - нужно вызывать явно.

# Типичный паттерн синхронизации градиентов
def synchronize_gradients(model):
    """Синхронизирует градиенты между всеми GPU"""
    for param in model.parameters():
        if param.grad is not None:
            # all_reduce с суммированием
            dist.all_reduce(param.grad, op=dist.ReduceOp.SUM)
            # Делим на количество процессов для усреднения
            param.grad /= dist.get_world_size()

# Новое в PyTorch 2.4: fused all_reduce
# Старый способ (медленный):
# for param in model.parameters():
#     dist.all_reduce(param.grad)

# Новый способ (быстрее в 2-3 раза):
grad_buckets = []
for param in model.parameters():
    if param.grad is not None:
        grad_buckets.append(param.grad.flatten())

# Объединяем все градиенты в один тензор
all_grads = torch.cat(grad_buckets)
# Один all_reduce вместо сотен
dist.all_reduce(all_grads, op=dist.ReduceOp.SUM)
all_grads /= dist.get_world_size()

# Распаковываем обратно (осторожно с памятью!)
offset = 0
for param in model.parameters():
    if param.grad is not None:
        numel = param.grad.numel()
        param.grad.copy_(all_grads[offset:offset+numel].view_as(param.grad))
        offset += numel

Почему fused all_reduce быстрее? Меньше вызовов NCCL, лучше использование bandwidth. Но есть нюанс: если у вас mixed precision (float16/float32), нужно быть осторожным с объединением тензоров разного типа.

all_gather, broadcast, reduce_scatter: остальная семья

all_reduce - не единственная коллективная операция. У каждой свои use cases:

• broadcast: один процесс отправляет данные всем остальным. Идеально для рассылки весов модели в начале обучения.
• all_gather: каждый процесс отправляет свои данные, все получают все данные. Используется в GRPO + LoRA на нескольких GPU для сбора статистик.
• reduce_scatter: противоположность all_gather. Суммируем данные, потом раздаем куски обратно.
• barrier: просто ждем, пока все процессы дойдут до этой точки. Самый простой, но часто самый бесполезный способ синхронизации.

NCCL под капотом: что на самом деле происходит

Когда вы вызываете dist.all_reduce(), PyTorch делегирует работу NCCL (NVIDIA Collective Communications Library). NCCL - это черная магия, которая превращает медленные коммуникации в быстрые.

Но вот что важно понимать: NCCL автоматически выбирает алгоритм коммуникаций в зависимости от:

1. Количества GPU
2. Топологии соединений (NVLink, PCIe)
3. Размера данных
4. Версии NCCL (в 2026 актуальна NCCL 3.0+)

Для 2 GPU с NVLink NCCL использует простой peer-to-peer обмен. Для 8 GPU - сложные деревья или кольцевые алгоритмы. Вы можете заставить NCCL использовать конкретный алгоритм, но обычно он сам выбирает оптимальный.

# Как посмотреть, что выбрал NCCL
export NCCL_DEBUG=INFO
# Запускаете ваш скрипт
# В логах увидите что-то вроде:
# [0] NCCL INFO Channel 00/02 :    0   1   2   3   4   5   6   7
# [0] NCCL INFO Trees [0] : 1/-1/-1->0->4->6/-1/-1->2->5->7/-1/-1->3

Когда что использовать: практическое правило

За 5 лет работы с multi-GPU системами я выработал простое правило:

Производительность: как не утонуть в коммуникациях

Коммуникации между GPU - главный bottleneck в distributed training. Вот цифры, которые заставят вас задуматься:

• NVLink 4.0 (в RTX 5090): 900 GB/s между соседними GPU
• PCIe 5.0 x16: 64 GB/s (в 14 раз медленнее!)
• Сеть 400GbE: 50 GB/s (еще медленнее)

Если ваши GPU соединены через PCIe (как в 7 видеокарт на AM5), коммуникации будут в 10-20 раз медленнее, чем вычисления.

Как это исправить?

3 Техника: overlapping communications and computations

Суть: пока GPU вычисляет градиенты для слоя N+1, она уже отправляет градиенты слоя N. В PyTorch это делается через hook'и:

# Упрощенный пример overlapping
class OverlapOptimizer(torch.optim.Optimizer):
    def step(self):
        # 1. Вычисляем градиенты
        loss = self.compute_loss()
        loss.backward()
        
        # 2. Для каждого параметра запускаем all_reduce асинхронно
        reqs = []
        for param in self.model.parameters():
            if param.grad is not None:
                req = dist.all_reduce(param.grad, async_op=True)  # Ключ: async_op=True
                reqs.append((param, req))
        
        # 3. Пока градиенты синхронизируются, делаем что-то полезное
        self.prepare_next_batch()  # Загрузка данных
        
        # 4. Ждем завершения только когда нужно
        for param, req in reqs:
            req.wait()
            # Теперь param.grad синхронизирован
            param.data -= self.lr * param.grad

В реальности overlapping сложнее. Нужно учитывать dependencies между операциями. Но выигрыш может быть 30-40% на больших моделях.

Отладка: когда все ломается

Distributed код ломается всегда. Не иногда, а всегда. Вот чеклист, который спас мне сотни часов:

1. Проверяем инициализацию: dist.init_process_group() вызван на всех процессах с одинаковыми параметрами
2. Проверяем ранги: rank и world_size правильные на каждом процессе
3. Проверяем устройство: тензоры находятся на правильных GPU (cuda:0, cuda:1 и т.д.)
4. Включаем NCCL debug: export NCCL_DEBUG=INFO (но готовьтесь к огромным логам)
5. Используем torch.distributed.barrier() для синхронизации в отладочных целях
6. Проверяем deadlocks с помощью NVIDIA Nsight Systems

Что будет дальше: тенденции 2026-2027

За последний год в PyTorch distributed появилось несколько важных изменений:

• Compressed communications: all_reduce со сжатием градиентов (1-bit, 2-bit). Экономит 4-8x bandwidth.
• Topology-aware collectives: NCCL теперь лучше учитывает физическую топологию (кто с кем соединен NVLink).
• Async error handling: раньше если один процесс падал, все висли. Теперь есть механизмы graceful degradation.
• Better MIG support: для разделения GPU на части улучшилась поддержка в distributed.

Мой прогноз: к 2027 году point-to-point операции почти исчезнут из пользовательского кода. Их заменят высокоуровневые абстракции, которые автоматически выбирают оптимальный паттерн коммуникаций. Но понимать, что происходит под капотом, все равно будет необходимо - иначе вы не сможете оптимизировать действительно сложные случаи.

Последний совет: если ваша модель помещается на одну GPU - не используйте distributed. Сложность растет экспоненциально, а выигрыш может быть линейным. Но если не помещается... теперь вы знаете, как заставить 8 GPU работать как одна слаженная команда.