Hugging Face's logo

Accelerate documentation

Crochets de communication DDP

You are viewing main version, which requires installation from source. If you'd like regular pip install, checkout the latest stable version (v1.15.0).
Hugging Face's logo
Join the Hugging Face community

and get access to the augmented documentation experience

to get started

Crochets de communication DDP

Les hooks de communication de Distributed Data Parallel (DDP) fournissent une interface générique pour contrôler la manière dont les gradients sont communiqués entre les workers en remplaçant l'allreduce standard de DistributedDataParallel. Quelques hooks de communication intégrés sont fournis, et les utilisateurs peuvent facilement appliquer l'un de ces hooks pour optimiser la communication.

  • Hook de compression FP16 : compresse les gradients en les convertissant au format flottant demi-précision (torch.float16), réduisant ainsi la surcharge de communication.
  • Hook de compression BF16 : Similaire à FP16, mais utilise le format Brain Floating Point (torch.bfloat16), qui peut être plus efficace sur certains matériels.
  • PowerSGD Hook : un algorithme avancé de compression des gradients qui offre des taux de compression élevés et peut accélérer l’entraînement distribué limité par la bande passante.

Dans ce tutoriel, vous verrez comment configurer rapidement les hooks de communication DDP et effectuer l’entraînement à l’aide des utilitaires fournis par Accelerate, ce qui peut se résumer à l’ajout d’une seule nouvelle ligne de code ! Cela illustre comment utiliser les hooks de communication DDP pour optimiser la communication des gradients dans l’entraînement distribué avec la bibliothèque Accelerate.

Hook de compression FP16

PyTorch
Accelerate
import torch
from torch.nn.parallel import DistributedDataParallel as DDP
from torch.distributed.algorithms.ddp_comm_hooks import default_hooks
from accelerate.test_utils.testing import get_backend

device_type, _, _ = get_backend()
device_id = getattr(torch, device_type, torch.cuda).current_device()

class MyModel(torch.nn.Module):
    def __init__(self):
        super().__init__()
        self.layer = torch.nn.Linear(10, 10)

    def forward(self, x):
        return self.layer(x)

model = MyModel()
model = DDP(model, device_ids=[device_id])
model.register_comm_hook(state=None, hook=default_hooks.fp16_compress_hook)

# Training loop
for data, targets in data_loader:
    outputs = model(data)
    loss = criterion(outputs, targets)
    loss.backward()
    optimizer.step()
    optimizer.zero_grad()

Hook de compression BF16

L'API du hook de compression BF16 est expérimentale et nécessite une version de NCCL supérieure à 2.9.6.

PyTorch
Accelerate
import torch
from torch.nn.parallel import DistributedDataParallel as DDP
from torch.distributed.algorithms.ddp_comm_hooks import default_hooks
from accelerate.test_utils.testing import get_backend

device_type, _, _ = get_backend()
device_id = getattr(torch, device_type, torch.cuda).current_device()

class MyModel(torch.nn.Module):
    def __init__(self):
        super().__init__()
        self.layer = torch.nn.Linear(10, 10)

    def forward(self, x):
        return self.layer(x)

model = MyModel()
model = DDP(model, device_ids=[device_id])
model.register_comm_hook(state=None, hook=default_hooks.bf16_compress_hook)

# Training loop
for data, targets in data_loader:
    outputs = model(data)
    loss = criterion(outputs, targets)
    loss.backward()
    optimizer.step()
    optimizer.zero_grad()

PowerSGD Hook

PowerSGD nécessite généralement une mémoire supplémentaire de la même taille que les gradients du modèle pour permettre le retour d'erreur, ce qui peut compenser la communication compressée biaisée et améliorer la précision.

PyTorch
Accelerate
import torch
from torch.nn.parallel import DistributedDataParallel as DDP
from torch.distributed.algorithms.ddp_comm_hooks import powerSGD_hook
from accelerate.test_utils.testing import get_backend

device_type, _, _ = get_backend()
device_id = getattr(torch, device_type, torch.cuda).current_device()

class MyModel(torch.nn.Module):
    def __init__(self):
        super().__init__()
        self.layer = torch.nn.Linear(10, 10)

    def forward(self, x):
        return self.layer(x)

model = MyModel()
model = DDP(model, device_ids=[device_id])
state = powerSGD_hook.PowerSGDState(process_group=None)
model.register_comm_hook(state=state, hook=powerSGD_hook.powerSGD_hook)

# Training loop
for data, targets in data_loader:
    outputs = model(data)
    loss = criterion(outputs, targets)
    loss.backward()
    optimizer.step()
    optimizer.zero_grad()

Utilitaires des hooks de communication DDP

Deux utilitaires supplémentaires permettent de prendre en charge les fonctionnalités optionnelles avec les hooks de communication.

comm_wrapper

comm_wrapper est une option permettant d’encapsuler un hook de communication avec des fonctionnalités supplémentaires. Par exemple, il peut être utilisé pour combiner la compression FP16 avec d’autres stratégies de communication. Les wrappers actuellement pris en charge sont no, fp16 et bf16.

from accelerate import Accelerator, DDPCommunicationHookType, DistributedDataParallelKwargs
import torch

class MyModel(torch.nn.Module):
    def __init__(self):
        super().__init__()
        self.layer = torch.nn.Linear(10, 10)

    def forward(self, x):
        return self.layer(x)

# DDP Communication Hook setup
ddp_kwargs = DistributedDataParallelKwargs(
    comm_hook=DDPCommunicationHookType.POWER_SGD,
    comm_wrapper=DDPCommunicationHookType.FP16
)
accelerator = Accelerator(kwargs_handlers=[ddp_kwargs])

model = MyModel()
optimizer = torch.optim.Adam(model.parameters())
data_loader = DataLoader(dataset, batch_size=16)

model, optimizer, data_loader = accelerator.prepare(model, optimizer, data_loader)

# Training loop
for data, targets in data_loader:
    outputs = model(data)
    loss = criterion(outputs, targets)
    accelerator.backward(loss)
    optimizer.step()
    optimizer.zero_grad()

comm_state_option

comm_state_option permet de transmettre des informations d’état supplémentaires requises par certains hooks de communication. Cela est particulièrement utile pour les hooks à état, comme PowerSGD, qui nécessitent de maintenir des hyperparamètres et des états internes entre les étapes d’entraînement. Voici un exemple illustrant l’utilisation de comm_state_option avec le hook PowerSGD.

from accelerate import Accelerator, DDPCommunicationHookType, DistributedDataParallelKwargs
import torch

class MyModel(torch.nn.Module):
    def __init__(self):
        super().__init__()
        self.layer = torch.nn.Linear(10, 10)

    def forward(self, x):
        return self.layer(x)

# DDP Communication Hook setup
ddp_kwargs = DistributedDataParallelKwargs(
    comm_hook=DDPCommunicationHookType.POWER_SGD,
    comm_state_option={"matrix_approximation_rank": 2}
)
accelerator = Accelerator(kwargs_handlers=[ddp_kwargs])

model = MyModel()
optimizer = torch.optim.Adam(model.parameters())
data_loader = DataLoader(dataset, batch_size=16)

model, optimizer, data_loader = accelerator.prepare(model, optimizer, data_loader)

# Training loop
for data, targets in data_loader:
    outputs = model(data)
    loss = criterion(outputs, targets)
    accelerator.backward(loss)
    optimizer.step()
    optimizer.zero_grad()

Pour un usage plus avancé et des hooks supplémentaires, reportez-vous à la documentation des hooks de communication PyTorch DDP.

Update on GitHub

←Utiliser plusieurs modèles avec DeepSpeed Fully Sharded Data Parallel→