Распределённое обучение

Зачем это нужно

Посчитаем по правилу $C \approx 6ND$ из урока 5.1.1. DeepSeek-V3: 37B активных параметров, 14.8T токенов, итого $6 \cdot 37 \cdot 10^9 \cdot 14.8 \cdot 10^{12} \approx 3.3 \cdot 10^{24}$ FLOPs. По техотчёту претрейн занял 2.664 млн GPU-часов, значит каждая H800 в среднем выдавала около 340 TFLOP/s полезной работы. Одна такая карта считала бы 300 лет. 2048 карт справились меньше чем за два месяца.

Вычисления — только половина проблемы. Модель с AdamW занимает около 16 байт на параметр (урок 5.4.1). Для 671B это больше 10 ТБ, а у H800 — 80 ГБ памяти: разница более чем в сто раз.

Значит, и вычисления, и память нужно делить между сотнями GPU. Делить можно по-разному, и у каждого способа своя цена — пересылки данных между картами. распределённое обучениераспределённое обучениеEN distributed training — это искусство выбрать такое деление, при котором GPU считают, а не ждут сеть. Для лаборатории с урезанной сетью (а это DeepSeek на H800) это вопрос выживания.

Четыре оси параллелизма

Ось	Что делим	Что пересылаем	Когда	Где уместно
По даннымпараллелизм по даннымEN data parallelism (DP)	батч	градиенты (all-reduce)	раз за шаг	всегда; масштабирует до тысяч GPU
Тензорнаятензорный параллелизмEN tensor parallelism (TP)	матрицы внутри слоя	активации (all-reduce)	дважды за слой в каждую сторону	внутри узла с NVLink
Конвейернаяконвейерный параллелизмEN pipeline parallelism (PP)	слои по стадиям	активации на границах стадий	на каждый микробатч	между узлами
По экспертампараллелизм экспертовEN expert parallelism (EP)Разные эксперты MoE-слоя живут на разных GPU; токены пересылаются к своим экспертам и обратно через all-to-all-обмен. (EP)	экспертов MoE	токены (all-to-all)	дважды за MoE-слой	MoE-модели

Большие прогоны совмещают несколько осей сразу. Разберём каждую по очереди.

Параллелизм по данным и all-reduce

Самый простой способ. На каждой из $N$ GPU лежит полная копия модели. Батч делится на $N$ частей, каждая GPU считает прямой и обратный проход на своей части. Затем градиенты усредняются операцией all-reduceall-reduce (суммирование со всех на все)EN all-reduceКоллективная операция — каждый GPU отдаёт свой тензор и получает сумму (или среднее) тензоров всех GPU. Так DDP усредняет градиенты.: каждая GPU отдаёт свой тензор и получает среднее по всем. После этого все делают одинаковый шаг оптимизатора, и копии остаются одинаковыми.

Почему среднее градиентов равно градиенту на всём батче? Loss — среднее по примерам, а производная суммы — сумма производных. Пример: батч из 4 примеров с градиентами $1, 3, 5, 7$ — средний градиент 4. Первая GPU берёт примеры 1–2 и получает среднее 2, вторая — примеры 3–4 и получает 6; среднее этих двух чисел снова 4.

Несколько новых слов, без которых код ниже не прочитать. Распределённое обучение запускает несколько процессов — копий одной программы, работающих одновременно (обычно по одному на GPU). У каждого процесса свой номер — ранг (rank, от 0), а их общее число называют world size. Процессы обмениваются тензорами через библиотеку связи — бэкенд: nccl для видеокарт NVIDIA, gloo для обычного процессора. Модуль torch.distributed даёт сами операции вроде all_reduce, а mp.spawn (или утилита torchrun) запускает нужное число процессов.

Проверим «руками» на CPU: два процесса, бэкенд gloo:

# ddp_by_hand.py — параллелизм по данным «руками» на CPU: 2 процесса, бэкенд gloo
import torch
import torch.distributed as dist
import torch.multiprocessing as mp
import torch.nn as nn


def make_model():
    torch.manual_seed(0)                              # одинаковые веса во всех процессах
    return nn.Sequential(nn.Linear(16, 32), nn.Tanh(), nn.Linear(32, 1))


def worker(rank, world):
    # подключиться к «группе»: все процессы находят друг друга по адресу 127.0.0.1:29533 (этот же компьютер)
    dist.init_process_group("gloo", init_method="tcp://127.0.0.1:29533", rank=rank, world_size=world)
    torch.manual_seed(1)
    X, y = torch.randn(64, 16), torch.randn(64, 1)   # общий батч из 64 примеров
    model = make_model()

    # 1. Каждый процесс считает градиент на СВОЕЙ половине батча
    part = slice(rank * 32, (rank + 1) * 32)
    loss = ((model(X[part]) - y[part]) ** 2).mean()
    loss.backward()

    # 2. All-reduce: складываем градиенты всех процессов и делим на их число
    for p in model.parameters():
        dist.all_reduce(p.grad, op=dist.ReduceOp.SUM)
        p.grad /= world

    # 3. Сравниваем с градиентом на всём батче в одном процессе
    ref = make_model()
    ((ref(X) - y) ** 2).mean().backward()
    diff = max((p.grad - q.grad).abs().max().item() for p, q in zip(model.parameters(), ref.parameters()))
    print(f"процесс {rank}: max |g_DDP - g_полный| = {diff:.2e}")
    dist.destroy_process_group()


if __name__ == "__main__":
    mp.spawn(worker, args=(2,), nprocs=2)              # запустить worker(0, 2) и worker(1, 2) в двух процессах

процесс 1: max |g_DDP - g_полный| = 2.98e-08
процесс 0: max |g_DDP - g_полный| = 2.98e-08

Разница $3 \cdot 10^{-8}$ — это округления float32: суммы посчитаны в другом порядке. В PyTorch то же самое делает DistributedDataParallel, но умнее: градиенты собираются в «корзины» по 25 МБ, и all-reduce первой корзины начинается, пока обратный проход ещё считает следующие слои. Это первый пример перекрытие вычислений и коммуникацийперекрытие вычислений и коммуникацийEN computation-communication overlapПока GPU считает один кусок работы, по сети пересылаются данные другого. Если пересылка короче вычислений, она становится «бесплатной». — главного приёма этого урока.

Сколько стоит all-reduce. Кольцевой алгоритм (ring all-reduce) пересылает с каждой GPU $2 \cdot \frac{N-1}{N}$ объёма тензора — почти не зависит от числа GPU. Модель на 1.5B параметров в BF16 даёт 3 ГБ градиентов. На 4 картах при пропускной способности 10 ГБ/с это $2 \cdot \frac{3}{4} \cdot 3 / 10 \approx 0.45$ с на каждый шаг. В разделе «Железо» вы измеряли эту скорость на своей машине — подставьте своё число.

ZeRO и FSDP: не храним лишнего

У DDP есть очевидная расточительность: все $N$ GPU хранят одинаковые 16 байт на параметр. Идея ZeROZeRO (оптимизатор без избыточности)EN ZeRO (Zero Redundancy Optimizer)Вместо того чтобы хранить на каждом GPU полную копию, состояние оптимизатора (ZeRO-1), градиенты (ZeRO-2) и веса (ZeRO-3) делят между GPU параллелизма по данным. (Microsoft, 2019): разрезать это состояние на $N$ частей, чтобы каждая GPU хранила и обновляла только свою.

Стадия	Что делится между GPU	Память на GPU ($\Psi$ параметров)	7.5B, $N = 64$
DDP	ничего	$16\Psi$	120 ГБ
ZeRO-1	состояние оптимизатора (12 байт: мастер-веса, $m$, $v$)	$4\Psi + 12\Psi / N$	31.4 ГБ
ZeRO-2	+ градиенты	$2\Psi + 14\Psi / N$	16.6 ГБ
ZeRO-3	+ сами веса	$16\Psi / N$	1.9 ГБ

Числа в последнем столбце — из рисунка 1 статьи ZeRO. Активации сюда не входят.

Как это работает. В ZeRO-1 и ZeRO-2 all-reduce заменяется парой операций с тем же объёмом: reduce-scatter (каждая GPU получает сумму только своей части градиентов), шаг оптимизатора над своей частью и all-gather (все собирают обновлённые веса). Коммуникаций столько же, сколько у DDP, а памяти в разы меньше. ZeRO-3 идёт дальше: веса слоя собираются со всех GPU прямо перед его вычислением и сразу освобождаются. Это стоит примерно в 1.5 раза больше пересылок.

FSDPFSDP (полностью шардированный параллелизм по данным)EN Fully Sharded Data Parallel (FSDP)Реализация идеи ZeRO-3 в PyTorch. Веса слоя собираются со всех GPU прямо перед его вычислением и сразу освобождаются. — реализация ZeRO-3 в PyTorch. Вот рабочий пример (FSDP2, PyTorch 2.6+). На вашей машине запускайте на 4 GPU, без GPU он работает на CPU в двух процессах:

# fsdp_train.py — запуск: torchrun --nproc_per_node=4 fsdp_train.py   (на CPU: --nproc_per_node=2)
import os

import torch
import torch.distributed as dist
import torch.nn as nn
from torch.distributed.fsdp import fully_shard   # FSDP2, PyTorch 2.6+

use_cuda = torch.cuda.is_available()
dist.init_process_group("nccl" if use_cuda else "gloo")
rank, world = dist.get_rank(), dist.get_world_size()
dev = torch.device(f"cuda:{int(os.environ['LOCAL_RANK'])}" if use_cuda else "cpu")
if use_cuda:
    torch.cuda.set_device(dev)


class Block(nn.Module):
    """Остаточный MLP-блок, как FFN в трансформере."""

    def __init__(self, d):
        super().__init__()
        self.mlp = nn.Sequential(nn.LayerNorm(d), nn.Linear(d, 4 * d), nn.GELU(), nn.Linear(4 * d, d))

    def forward(self, x):
        return x + self.mlp(x)


torch.manual_seed(0)                              # одинаковая инициализация на всех рангах
d = 1024 if use_cuda else 256
model = nn.Sequential(*[Block(d) for _ in range(8)]).to(dev)
target = torch.randn(d, d, device=dev) / d**0.5   # «правильное» линейное преобразование
n_params = sum(p.numel() for p in model.parameters())
for block in model:                               # каждый блок — своя единица шардирования
    fully_shard(block)
fully_shard(model)
local = sum(p.to_local().numel() for p in model.parameters())   # параметры стали DTensor: у ранга только свой кусок
print(f"ранг {rank}: всего параметров {n_params:,}, хранится локально {local:,} ({local / n_params:.0%})")

opt = torch.optim.AdamW(model.parameters(), lr=3e-4)
torch.manual_seed(1 + rank)                       # у каждого ранга свои данные
x = torch.randn(64, d, device=dev)
for step in range(31):
    loss = (model(x) - x @ target).pow(2).mean()  # игрушечная регрессия
    opt.zero_grad()
    loss.backward()                               # градиенты reduce-scatter'ятся: каждому рангу — свой кусок
    opt.step()                                    # каждый ранг обновляет только свои параметры
    if rank == 0 and step % 10 == 0:
        print(f"шаг {step:2d}: loss {loss.item():.4f}")
dist.destroy_process_group()

Вывод на CPU (torchrun --nproc_per_node=2 fsdp_train.py; могут появиться предупреждения о сокетах — это не ошибка):

ранг 0: всего параметров 4,208,640, хранится локально 2,104,320 (50%)
ранг 1: всего параметров 4,208,640, хранится локально 2,104,320 (50%)
шаг  0: loss 2.3124
шаг 10: loss 0.5224
шаг 20: loss 0.1520
шаг 30: loss 0.0490

Каждый ранг хранит ровно половину параметров. На 4 GPU будет четверть.

Если памяти всё равно мало, есть перевычисление активацийперевычисление активаций (activation checkpointing)EN activation checkpointing (recomputation)Не хранить часть активаций прямого прохода, а пересчитать их при обратном. Экономит память ценой дополнительных вычислений.: активации части слоёв не хранят, а пересчитывают при обратном проходе. Это стоит примерно ещё одного прямого прохода, зато память на активации падает в разы.

Тензорный параллелизм

DP и ZeRO делят данные и хранение, но каждый слой всё равно считается на одной GPU целиком. Если одна матрица не помещается или слой считается слишком долго, её режут на части. Классическая схема — из Megatron-LM (NVIDIA).

Возьмём MLP: $Y = \mathrm{GELU}(X W_1) W_2$. Разрежем $W_1$ по столбцам на $n$ кусков, а $W_2$ — по строкам. GPU $j$ считает $\mathrm{GELU}(X W_1^{(j)}) W_2^{(j)}$ без всякого обмена: GELU действует поэлементно, поэтому столбцы независимы. Полный ответ — сумма частичных результатов, то есть один all-reduce на весь MLP:

import torch
import torch.nn.functional as F

torch.manual_seed(0)
d, d_ff, n_gpu = 8, 32, 4
x = torch.randn(5, d)                       # 5 токенов
W1, W2 = torch.randn(d, d_ff), torch.randn(d_ff, d)

full = F.gelu(x @ W1) @ W2                  # обычный MLP на одном устройстве

# Megatron: W1 режем по столбцам, W2 — по строкам; каждая «GPU» получает свой кусок
W1_parts = W1.chunk(n_gpu, dim=1)           # 4 куска по (8, 8)
W2_parts = W2.chunk(n_gpu, dim=0)           # 4 куска по (8, 8)
partial = [F.gelu(x @ a) @ b for a, b in zip(W1_parts, W2_parts)]   # каждая GPU считает сама, без обмена
out = sum(partial)                          # all-reduce: единственная пересылка на весь MLP

print("форма частичного результата:", tuple(partial[0].shape))
print("max |TP - обычный| =", (out - full).abs().max().item())

форма частичного результата: (5, 8)
max |TP - обычный| = 3.814697265625e-06

С вниманием ещё проще: головы независимы, и их раздают по GPU целиком, а выходную проекцию режут по строкам, как $W_2$. Итого два all-reduce активаций на слой в прямом проходе и два в обратном — на каждом слое и каждом микробатче. Такой трафик выдерживает только NVLink, поэтому TP почти всегда ограничивают одним узлом (до 8 GPU). На вашей машине с PCIe TP для обучения невыгоден. Для инференса больших моделей (vLLM на 4 картах, модуль 7) его используют, потому что там другая арифметика.

Конвейерный параллелизм и пузыри

Конвейер режет модель по слоям: GPU 0 держит слои 1–8, GPU 1 — слои 9–16 и так далее. Между стадиями передаются только активации на границе — пересылок мало, и конвейер хорошо работает между узлами.

Проблема в простое. Пока первый батч не дошёл до последней стадии, она ждёт. Пока градиент не вернулся к первой стадии, ждёт она. Решение — резать батч на $m$ микробатчеймикробатчEN micro-batchЧасть батча, которая проходит по конвейеру как единое целое. Чем больше микробатчей, тем меньше доля простоя конвейера. и пускать их друг за другом. Простои всё равно остаются в начале и в конце — это пузырь конвейерапузырь конвейераEN pipeline bubbleВремя, когда стадии конвейера простаивают в ожидании данных. У GPipe и 1F1B доля пузыря — (p − 1) / (m + p − 1) при p стадиях и m микробатчах.. Вот симуляция для $p = 4$ стадий и $m = 8$ микробатчей, прямой проход занимает 1 единицу времени, обратный — 2 (цифра — прямой проход микробатча, буква — обратный, точка — простой):

GPipe: время 33, пузырь 27.3%, микробатчей в памяти по стадиям [8, 8, 8, 8]
GPU0 |01234567.........aabbccddeeffgghh|
GPU1 |.01234567......aabbccddeeffgghh..|
GPU2 |..01234567...aabbccddeeffgghh....|
GPU3 |...01234567aabbccddeeffgghh......|
1F1B: время 33, пузырь 27.3%, микробатчей в памяти по стадиям [4, 3, 2, 1]
GPU0 |0123......aa4bb5cc6dd7ee.ff.gg.hh|
GPU1 |.012....aa3bb4cc5dd6ee7ff.gg.hh..|
GPU2 |..01..aa2bb3cc4dd5ee6ff7gg.hh....|
GPU3 |...0aa1bb2cc3dd4ee5ff6gg7hh......|

GPipe (Google, 2018) сначала гонит все прямые проходы, потом все обратные. 1F1B (one forward, one backward) после короткого разогрева чередует: один прямой, один обратный. Время у них одинаковое, а доля пузыря выводится из картинки:

$$\text{пузырь} = \frac{(p - 1)(F + B)}{(m + p - 1)(F + B)} = \frac{p - 1}{m + p - 1}.$$

При $p = 4$, $m = 8$ это $3/11 \approx 27\%$. Выигрыш 1F1B — в памяти: каждая стадия держит активации максимум $p$ микробатчей, а не $m$. Отсюда рецепт: больше микробатчей — меньше пузырь, и 1F1B позволяет их увеличивать без роста памяти.

Есть и более хитрые расписания. Чередующийся конвейер (Megatron) даёт каждой GPU несколько несмежных кусков модели, и пузырь падает пропорционально числу кусков. Работа Zero Bubble (Sea AI Lab, 2023) делит обратный проход на две части: $B$ — градиент по входу, он нужен предыдущей стадии срочно, и $W$ — градиент по весам, его можно отложить и вставить в пузыри. DeepSeek взяла эту идею в DualPipe.

Параллелизм экспертов

Его вы видели в уроке 5.3.2: эксперты MoE-слоя лежат на разных GPU, а токены пересылаются к ним и обратно двумя операциями «все со всеми» (all-to-all): dispatch и combine. В отличие от TP, пересылаются не все активации, а только токены, выбранные для удалённых экспертов. Зато пересылки идут на каждом MoE-слое и часто между узлами.

У DeepSeek-V3 именно здесь было узкое место. По оценке авторов, на межузловой all-to-all уходило примерно столько же времени, сколько на вычисления (около 1 : 1). Делай одно после другого — и половину времени GPU простаивают.

DualPipe: спрятать коммуникации за вычислениями

DualPipeDualPipe (двунаправленный конвейер)EN DualPipeРасписание конвейера из DeepSeek-V3 — микробатчи подаются с обоих концов, а прямой и обратный проходы разных микробатчей перекрываются так, что коммуникация прячется за вычислениями. решает две задачи сразу: уменьшает пузырь конвейера и прячет all-to-all.

Идея 1: перекрыть пару чанков. Каждый кусок работы (chunk) делится на четыре части: внимание, dispatch, MLP и combine. Обратный проход внимания и MLP дополнительно делится на «градиент по входу» и «градиент по весам», как в Zero Bubble. Затем прямой чанк одного микробатча и обратный чанк другого переставляют так, чтобы пока одна часть считает, другая пересылает (упрощённо, по рис. 4 техотчёта):

Слот	Вычисления (около 112 SM)	Коммуникация (около 20 SM)
1	MLP: обратный проход микробатча A	dispatch прямого прохода микробатча B
2	MLP: прямой проход B	dispatch обратного прохода A
3	внимание: обратный проход A	combine прямого прохода B
4	внимание: прямой проход B	combine обратного прохода A

Четыре слота соответствуют одному слою, дальше цикл повторяется. Проверьте зависимости. Прямой проход B идёт в порядке «внимание → dispatch → MLP → combine → внимание следующего слоя»: внимание в слоте 4, его dispatch в слоте 1 следующего цикла, MLP в слоте 2, combine в слоте 3. Обратный проход A идёт в обратном порядке: combine (слот 4) → MLP (слот 1) → dispatch (слот 2) → внимание (слот 3). Каждая пересылка одного микробатча идёт одновременно с вычислениями другого и успевает закончиться к моменту, когда её результат понадобится.

Из 132 потоковых мультипроцессоров (SM) H800 на коммуникации выделено всего около 20. Для этого DeepSeek написала свои ядра all-to-all, спустившись до PTX — низкоуровневого «виртуального ассемблера» GPU NVIDIA.

Идея 2: двунаправленный конвейер. Микробатчи подаются с обоих концов конвейера одновременно: одна половина идёт от GPU 0 к GPU 7, другая — от GPU 7 к GPU 0. Каждая GPU держит две стадии: «свою» для прямого направления и зеркальную для обратного. Пока один поток разогревается, другой уже работает, и пузыри частично заполняют друг друга.

flowchart LR
subgraph D1["направление 1"]
  A0["GPU0: стадия 0"] --> A1["GPU1: стадия 1"] --> A2["…"] --> A7["GPU7: стадия 7"]
end
subgraph D2["направление 2"]
  B7["GPU7: стадия 0'"] --> B6["GPU6: стадия 1'"] --> B5["…"] --> B0["GPU0: стадия 7'"]
end

DualPipe при PP = 8: каждая GPU хранит две стадии модели, микробатчи входят с обоих концов

Сравнение из таблицы 2 техотчёта ($F$ — прямой чанк, $B$ — полный обратный, $W$ — обратный по весам, $F\&B$ — перекрытая пара прямого и обратного):

Расписание	Время пузыря	Копий параметров	Активаций (в микробатчах)
1F1B	$(PP - 1)(F + B)$	1×	$PP$
ZB1P (Zero Bubble)	$(PP - 1)(F + B - 2W)$	1×	$PP$
DualPipe	$(PP/2 - 1)(F\&B + B - 3W)$	2×	$PP + 1$

Цена DualPipe — вдвое больше памяти под веса. DeepSeek может себе это позволить: при 64-путевом параллелизме экспертов на каждой GPU лежит лишь малая доля экспертов. Важное свойство: ни пузырь, ни активации не растут с числом микробатчей.

Почему H800 сделали это необходимым

H800 — экспортная версия H100 для Китая: матричные вычисления для нейросетей у неё такие же быстрые, но межчиповая связь NVLink урезана. По открытым данным, около 400 ГБ/с против 900 ГБ/с у H100 (подробно — в уроке china/02). В кластере DeepSeek внутри узла из 8 GPU данные идут по NVLink с эффективной скоростью около 160 ГБ/с, между узлами — по InfiniBand (быстрой сети между серверами), около 50 ГБ/с. MoE с 256 экспертами требует постоянных межузловых all-to-all.

Ответ DeepSeek собран из всего, что мы прошли:

• без тензорного параллелизма — его all-reduce на каждом слое съел бы связь. Схема V3: 16-путевой конвейер, 64-путевой параллелизм экспертов на 8 узлах и ZeRO-1 по данным;
• токен уходит не более чем на 4 узла (урок 5.3.2) — межузловой трафик ограничен по построению;
• DualPipe и свои ядра all-to-all — оставшиеся пересылки прячутся за вычислениями;
• экономия памяти, чтобы обойтись без TP: RMSNorm и повышающие проекции MLA пересчитываются при обратном проходе, копия весов для EMAэкспоненциальное скользящее среднее (EMA)EN exponential moving average (EMA)m ← βm + (1 − β)x. Помнит примерно последние 1/(1 − β) значений; при β = 0.9 — около 10. (скользящего среднего весов, урок 1.4.2) лежит в памяти CPU, эмбеддинги и выходной слой физически общие у основной модели и MTP-модуля (урок 5.4.2).

Посмотрите на симуляцию, которую вы напишете ниже: если добавить пересылку длиной 0.5 между стадиями, время 1F1B растёт с 33 до 41. Коммуникация на критическом пути (в цепочке операций, которая определяет общее время шага) — это прямые потери. DualPipe убирает её оттуда.

Практика

Напишите симулятор расписаний конвейера: GPipe и 1F1B, зависимости между стадиями, доля пузыря, пиковое число микробатчей в памяти. Затем — формулы таблицы 2 DeepSeek-V3. Кнопка «Запустить» нарисует диаграммы как выше.

Симулятор конвейера: пузыри GPipe и 1F1B7 тестов

# ── Готовый помощник ─────────────────────────────────────────────────────────
def draw(result, schedule, F=1, B=2):
    """ASCII-диаграмма: цифра — прямой проход микробатча, буква — обратный (a = 0, b = 1, …), точка — простой."""
    total = result["total"]
    scale = 2 if any(isinstance(x, float) and x % 1 for x in result["start"].values()) else 1
    for s, ops in enumerate(schedule):
        row = ["."] * int(round(total * scale))
        for kind, i in ops:
            t0 = result["start"][(s, kind, i)]
            dur = F if kind == "F" else B
            ch = str(i % 10) if kind == "F" else chr(ord("a") + i % 26)
            for t in range(int(round(t0 * scale)), int(round((t0 + dur) * scale))):
                row[t] = ch
        print(f"GPU{s} |{''.join(row)}|")


# ── Ваша часть ───────────────────────────────────────────────────────────────
def schedule_gpipe(p, m):
    """Расписание GPipe для p стадий и m микробатчей.

    Возвращает список из p списков операций. Операция — кортеж ("F", i) или ("B", i).
    На каждой стадии: ("F", 0), ..., ("F", m-1), затем ("B", 0), ..., ("B", m-1).
    """
    # TODO: ваш код
    ...


def schedule_1f1b(p, m):
    """Расписание 1F1B (one forward, one backward).

    Для стадии s (0 — первая, p-1 — последняя):
      1) разогрев: w = min(p - s - 1, m) прямых проходов ("F", 0) ... ("F", w-1);
      2) устойчивый режим: для j = 0 .. m-w-1 пара ("F", w + j), ("B", j);
      3) остывание: оставшиеся ("B", m-w) ... ("B", m-1).
    """
    # TODO: ваш код
    ...


def simulate(schedule, F=1, B=2, comm=0):
    """Симулятор конвейера. Стадия выполняет свои операции строго по порядку из schedule.

    Зависимости:
      * ("F", i) на стадии s начинается не раньше конца ("F", i) на стадии s-1 плюс comm;
      * ("B", i) на стадии s — не раньше конца ("F", i) на той же стадии
        и конца ("B", i) на стадии s+1 плюс comm;
      * и, конечно, не раньше, чем стадия освободится от предыдущей операции.
    Длительности: F для прямого прохода, B для обратного.

    Возвращает словарь:
      "start"  — {(s, kind, i): время начала},
      "total"  — время окончания последней операции,
      "bubble" — доля простоя: 1 - (сумма длительностей всех операций) / (p * total),
      "peak"   — список по стадиям: максимум микробатчей «в полёте»
                 (+1 после F, -1 после B, идём по порядку операций стадии).
    Подсказка: крутите цикл по стадиям, пока все операции не получат время;
    операцию можно запланировать, когда все её зависимости уже имеют время конца.
    """
    # TODO: ваш код
    ...


def bubble_formula(p, m):
    """Доля пузыря у GPipe и 1F1B при нулевой коммуникации: (p - 1) / (m + p - 1)."""
    # TODO: ваш код
    ...


def deepseek_bubbles(pp, F, B, W, FB):
    """Таблица 2 техотчёта DeepSeek-V3: время пузыря для трёх расписаний.

    {"1F1B": (pp-1)(F+B), "ZB1P": (pp-1)(F+B-2W), "DualPipe": (pp/2-1)(FB+B-3W)}
    F — прямой проход, B — полный обратный, W — обратный «по весам», FB — перекрытая пара прямой+обратный.
    """
    # TODO: ваш код
    ...


# Проверьте себя кнопкой «Запустить»:
p, m = 4, 8
for name, make in [("GPipe", schedule_gpipe), ("1F1B", schedule_1f1b)]:
    sched = make(p, m)
    res = simulate(sched) if sched else None
    if res is None:
        print(f"{name}: пока не реализовано")
        continue
    print(f"{name}: время {res['total']}, пузырь {res['bubble']:.1%}, микробатчей в памяти по стадиям {res['peak']}")
    draw(res, sched)

ЗапуститьПроверитьПодсказкаСбросить

После решения поэкспериментируйте: вызовите simulate(schedule_1f1b(4, 8), comm=0.5) и draw — где появились новые точки? Сравните с GPipe при той же задержке. Почему 1F1B чувствительнее к задержкам связи? (Подсказка: посчитайте, сколько раз за прогон пересылка оказывается на критическом пути — у GPipe только при заполнении и опустошении конвейера, а у 1F1B?)

🖥 На вашей машине (4×GPU)

У вас та же задача, что у DeepSeek на H800, только в миниатюре: четыре быстрые карты и узкая связь между ними (PCIe 4.0, без NVLink). Четыре опыта: измерить связь, спрятать её за вычислениями в DDP, увидеть, где она перестаёт прятаться, и разрезать модель через FSDP.

1. Сколько стоит связь. На странице «Ваша лаборатория (4×GPU)» вы измеряли all-reduce одного тензора на 256 МБ. Теперь пройдитесь по размерам:

# allreduce_sweep.py — скорость all-reduce в зависимости от размера сообщения
# запуск: torchrun --nproc_per_node=4 allreduce_sweep.py
import os
import time

import torch
import torch.distributed as dist

dist.init_process_group("nccl")
rank, world = dist.get_rank(), dist.get_world_size()
torch.cuda.set_device(int(os.environ["LOCAL_RANK"]))

for mb in [1, 4, 16, 64, 256, 1024]:                  # размер тензора градиентов, МБ
    x = torch.ones(mb * 2**20 // 4, device="cuda")    # float32 — 4 байта на число
    for _ in range(3):                                # прогрев
        dist.all_reduce(x)
    torch.cuda.synchronize()
    n = 20
    t0 = time.perf_counter()
    for _ in range(n):
        dist.all_reduce(x)
    torch.cuda.synchronize()
    dt = (time.perf_counter() - t0) / n
    bus = mb * 2**20 * 2 * (world - 1) / world / dt / 1e9   # «шинная» скорость, как в nccl-tests
    if rank == 0:
        print(f"{mb:5d} МБ: {dt * 1000:8.2f} мс, bus bandwidth {bus:6.1f} ГБ/с")
dist.destroy_process_group()

torchrun --nproc_per_node=4 allreduce_sweep.py
NCCL_DEBUG=INFO torchrun --nproc_per_node=4 allreduce_sweep.py 2>&1 | grep " via "   # каким путём идут данные
NCCL_P2P_DISABLE=1 torchrun --nproc_per_node=4 allreduce_sweep.py                     # запретить прямой обмен P2P

• На мелких сообщениях (1–4 МБ) скорость низкая: время съедает задержка запуска операции, а не передача байтов. Поэтому DDP и складывает градиенты в «корзины» по 25 МБ.
• На крупных скорость выходит на плато — это реальная пропускная способность вашей связи. Сравните её с 10 ГБ/с из примера выше и с эффективными 160 ГБ/с NVLink внутри узла у DeepSeek.
• В логе NCCL SHM означает обмен через память процессора, P2P — напрямую по PCIe. Если с NCCL_P2P_DISABLE=1 скорость не изменилась, прямой обмен у вас и так выключен драйвером, что обычно для GeForce.

Сопоставьте связь с вычислениями. Пик карты около 97 TFLOPS в BF16 при связи около 10 ГБ/с — это примерно 10 тыс. FLOPs на каждый переданный байт. У DeepSeek на H800 — около 6 тыс. внутри узла (989 TFLOPS на 160 ГБ/с) и около 20 тыс. между узлами (на 50 ГБ/с). Порядок тот же, поэтому её приёмы — прятать обмен за вычислениями и не гонять активации между картами на каждом слое — имеют смысл и у вас. Подставьте вместо 10 ГБ/с своё число.

2. DDP: как спрятать связь. Переделайте train.py из урока 4.6.3 (GPU-версию из блока «На вашей машине») в train_ddp.py. Главные отличия:

# train_ddp.py — отличия от train.py.  Запуск: torchrun --nproc_per_node=4 train_ddp.py
import contextlib

import torch.distributed as dist
from torch.distributed.algorithms.ddp_comm_hooks import default_hooks
from torch.nn.parallel import DistributedDataParallel as DDP

dist.init_process_group("nccl")
rank, world = dist.get_rank(), dist.get_world_size()
local_rank = int(os.environ["LOCAL_RANK"])
device = f"cuda:{local_rank}"
torch.cuda.set_device(device)
torch.manual_seed(1337 + rank)        # свои случайные окна на каждой карте; веса DDP сам скопирует с rank 0

model = DDP(GPT(cfg).to(device), device_ids=[local_rank])
if os.environ.get("BF16_GRADS") == "1":
    model.register_comm_hook(None, default_hooks.bf16_compress_hook)   # градиенты по PCIe в BF16: вдвое меньше байт
use_no_sync = os.environ.get("NO_SYNC", "1") == "1"

# внутри шага обучения:
for micro in range(grad_accum):
    last = micro == grad_accum - 1
    ctx = model.no_sync() if (use_no_sync and not last) else contextlib.nullcontext()
    with ctx:                         # внутри no_sync градиенты копятся локально, без all-reduce
        xb, yb = get_batch("train")
        with torch.autocast(device_type="cuda", dtype=torch.bfloat16):
            _, loss = model(xb, yb)
        (loss / grad_accum).backward()

# оценку делайте через model.module (без DDP), печатайте и сохраняйте только при rank == 0;
# в конце — dist.destroy_process_group()

Обучите одну и ту же модель четырьмя способами и для каждого запишите токены в секунду, время шага и torch.cuda.max_memory_allocated():

Запуск	Что меняется
python train.py на одной карте	база для сравнения
NO_SYNC=0 torchrun --nproc_per_node=4 train_ddp.py	all-reduce после каждого микробатча
torchrun --nproc_per_node=4 train_ddp.py	all-reduce раз за шаг оптимизатора
BF16_GRADS=1 torchrun --nproc_per_node=4 train_ddp.py	по PCIe идёт вдвое меньше байт

Эффективность масштабирования — это токены в секунду на 4 картах, делённые на учетверённые токены в секунду на одной. Эффективный батч на 4 картах тоже вчетверо больше: для замера скорости это не важно, а loss сравнивайте с поправкой. У «baby GPT» (10.8M параметров, 43 МБ градиентов в FP32) строки будут различаться мало: обмен короткий, и DDP прячет его за обратным проходом. Возьмите модель покрупнее — n_layer=24, n_head=16, n_embd=1024, около 300M параметров и 1.2 ГБ градиентов. По формуле выше это $2 \cdot \frac{3}{4} \cdot 1.2 / 10 \approx 0.18$ с на каждый all-reduce при 10 ГБ/с — уже заметная доля шага (оценка). Вот теперь no_sync и сжатие градиентов дадут ощутимую прибавку.

3. FSDP: когда модель не помещается. У модели n_layer=24, n_head=16, n_embd=2048 около 1.2B параметров, и в DDP только веса, градиенты и состояние AdamW займут около 19 ГБ на каждой карте, плюс активации. Разрежьте её через FSDP2, как в примере выше, по блокам нашей GPT:

from torch.distributed.fsdp import fully_shard

model = GPT(cfg).to(device)
for block in model.blocks:
    fully_shard(block)        # каждый блок — своя единица шардирования
fully_shard(model)            # остальное: эмбеддинги, ln_f, lm_head

Сравните память на карту и время шага с DDP. Затем поставьте fully_shard(block, reshard_after_forward=False): веса блока останутся собранными между прямым и обратным проходом (это ближе к ZeRO-2). Пересылок станет меньше, памяти уйдёт больше. Каждый сбор весов идёт по тому же PCIe, поэтому у вас разница между режимами, скорее всего, будет заметнее, чем на кластере с NVLink.

4. Что поместится. По таблице ZeRO: 7B с AdamW — 112 ГБ состояния; ZeRO-3 на 4 картах — 28 ГБ на карту плюс активации. Это впритык к 32 ГБ: понадобятся перевычисление активаций, маленький микробатч, а возможно, и выгрузка состояния в 128 ГБ оперативной памяти (fully_shard(..., offload_policy=CPUOffloadPolicy())) ценой скорости. Модели 1.5–3B помещаются с запасом. Для 7B на вашей машине правильный путь — LoRA (ступень 6); там же ZeRO и FSDP включаются одной строкой конфига в LLaMA-Factory и ms-swift.

Полный скрипт с DDP и FSDP2, проверкой, что кривая loss на 4 картах совпадает с одной, распределёнными чекпоинтами и замером масштабирования — в уроке 5.7.4.

Китайский угол: инженерия против дефицита

DualPipe открыт. В феврале 2025 года на Open Source Week DeepSeek выложила DualPipe и DeepEP — библиотеку all-to-all для экспертов с ядрами под NVLink и RDMA. Можно читать §3.2 техотчёта и сразу смотреть в код. Через несколько месяцев DeepSeek опубликовала статью Insights into DeepSeek-V3 (2025) о совместном проектировании модели и железа: как ограничения кластера повлияли на архитектуру и чего не хватает будущим чипам.

Без TP — сознательный выбор. Большинство западных прогонов того времени опирались на тензорный параллелизм внутри узла. DeepSeek отказалась от него, потому что на H800 связь — самый дефицитный ресурс. Хороший урок для инженера: схема параллелизма выводится из железа, а не копируется из чужой статьи.

Где учиться по-китайски. Курс ZOMI AIInfra (раздел 04Train) подробно разбирает DDP, ZeRO, тензорный и конвейерный параллелизм с расписанием 1F1B. В «Погружении в глубокое обучение» (d2l-zh) — глава «Обучение на нескольких GPU» с параллелизмом по данным с нуля.

Термины. Параллелизм по данным, тензорный и по экспертам по-китайски называются дословно. Конвейер — буквально «линия текущей воды», то есть сборочная линия, а пузырь конвейера — «пузырь в сборочной линии»: образ тот же, что в английском bubble.

Проверь себя

Проверь себя5 вопросов·

1 Конвейер из $p = 8$ стадий, $m = 24$ микробатча, расписание 1F1B, связь мгновенная. Какая доля времени уходит на пузырь? (до двух знаков)

2 Чем 1F1B лучше GPipe при тех же $p$ и $m$?

Пузырь в несколько раз меньшеПузырь тот же, но каждая стадия держит активации не более $p$ микробатчей вместо $m$Не нужен обратный проходНе нужна пересылка активаций между стадиями

3 Модель на 7B параметров, AdamW со смешанной точностью (16 байт на параметр), ZeRO-3 на 4 GPU. Сколько ГБ состояния на одну GPU без учёта активаций?

4 Почему тензорный параллелизм обычно держат внутри одного узла?

Он требует all-reduce активаций на каждом слое — нужен быстрый NVLinkОн не работает больше чем на 8 GPU математическиОн хранит полную копию модели на каждой GPUОн несовместим с MoE

5 Какие утверждения о DualPipe верны? Выберите все.Можно выбрать несколько ответов

Микробатчи подаются с обоих концов конвейераКоммуникация all-to-all одного микробатча перекрывается вычислениями другогоКаждая GPU хранит две копии параметров (две стадии)Пузырь растёт пропорционально числу микробатчей

Проверить ответы

Итоги

• Четыре оси: DP делит батч (all-reduce градиентов), TP — матрицы слоя (all-reduce активаций, только внутри узла), PP — слои (активации на границах), EP — экспертов (all-to-all токенов).
• ZeRO/FSDP убирают дублирование: 16 байт на параметр превращаются в $16/N$ при ZeRO-3 ценой примерно в 1.5 раза большего трафика.
• Пузырь конвейера $(p - 1)/(m + p - 1)$; 1F1B его не уменьшает, но экономит память на активации.
• DualPipe подаёт микробатчи с двух концов и перекрывает all-to-all с вычислениями. Это ответ DeepSeek на урезанную связь H800, вместе с отказом от TP и маршрутизацией не более чем на 4 узла.

Дальше: Muon и стабильность обучения — как Moonshot заменила AdamW на оптимизатор с ортогонализацией и обучила Kimi K2 на 15.5T токенов без единого всплеска loss.