KI in mehreren GPUs: Punkt-zu-Punkt- und kollektive Operationen

Bei diesen Vorgängen sendet ein Rang Daten an alle anderen Ränge in der Gruppe.

• torch.distributed.broadcast(tensor, src): Kopiert einen Tensor aus einem einzelnen Quellrang (src) zu allen anderen Rängen. Jeder Prozess führt zu einer identischen Kopie des Tensors. Der tensor Der Parameter dient zwei Zwecken: (1) wenn der Rang des Prozesses mit dem übereinstimmt srcDie tensor werden die Daten gesendet? (2) andernfalls, tensor dient der Speicherung der empfangenen Daten.

rank = torch.distributed.get_rank()
if rank == 0: # supply rank
  tensor = torch.tensor((1,2,3), dtype=torch.int64, system=system)
else: # vacation spot ranks
  tensor = torch.empty(3, dtype=torch.int64, system=system)
torch.distributed.broadcast(tensor, src=0)

• torch.distributed.scatter(tensor, scatter_list, src): Verteilt Datenblöcke von einem Quellrang über alle Ränge. Der scatter_list auf dem Quellrang enthält mehrere Tensoren, und jeder Rang (einschließlich der Quelle) erhält einen Tensor aus dieser Liste in seinen tensor Variable. Die Zielränge gehen einfach vorbei None für die scatter_list.

# The scatter_list should be None for all non-source ranks.
scatter_list = None if rank != 0 else (torch.tensor((i, i+1)).to(system) for i in vary(0,4,2))
tensor = torch.empty(2, dtype=torch.int64).to(system)
torch.distributed.scatter(tensor, scatter_list, src=0)
print(f'Rank {rank} obtained: {tensor}')

Diese Vorgänge sammeln Daten von allen Rängen und konsolidieren sie auf einem einzigen Zielrang.

• torch.distributed.cut back(tensor, dst, op): Nimmt einen Tensor von jedem Rang und wendet eine Reduktionsoperation an (wie SUM, MAX, MIN) und speichert das Endergebnis auf dem Zielrang (dst) nur.

rank = torch.distributed.get_rank()
tensor = torch.tensor((rank+1, rank+2, rank+3), system=system)
torch.distributed.cut back(tensor, dst=0, op=torch.distributed.ReduceOp.SUM)
print(tensor)

• torch.distributed.collect(tensor, gather_list, dst): Sammelt einen Tensor von jedem Rang in einer Liste von Tensoren auf dem Zielrang. Der gather_list muss eine Liste von Tensoren (richtige Größe und Typ) am Ziel und sein None überall sonst.

# The gather_list should be None for all non-destination ranks.
rank = torch.distributed.get_rank()
world_size = torch.distributed.get_world_size()
gather_list = None if rank != 0 else (torch.zeros(3, dtype=torch.int64).to(system) for _ in vary(world_size))
t = torch.tensor((0+rank, 1+rank, 2+rank), dtype=torch.int64).to(system)
torch.distributed.collect(t, gather_list, dst=0)
print(f'After op, Rank {rank} has: {gather_list}')

Die Variable world_size ist die Gesamtzahl der Ränge. Es kann mit erhalten werden torch.distributed.get_world_size(). Aber machen Sie sich zunächst keine Gedanken über die Implementierungsdetails. Das Wichtigste ist, die Konzepte zu verstehen.

Bei diesen Operationen sendet und empfängt jeder Rang Daten von allen anderen Rängen.

• torch.distributed.all_reduce(tensor, op): Das Gleiche wie cut backaber das Ergebnis wird gespeichert jederRang statt nur ein Ziel.

# Instance for torch.distributed.all_reduce
rank = torch.distributed.get_rank()
tensor = torch.tensor((rank+1, rank+2, rank+3), dtype=torch.float32, system=system)
torch.distributed.all_reduce(tensor, op=torch.distributed.ReduceOp.SUM)
print(f"Rank {rank} after all_reduce: {tensor}")

• torch.distributed.all_gather(tensor_list, tensor): Das Gleiche wie collectaber die gesammelte Liste der Tensoren ist auf verfügbar jeder Rang.

# Instance for torch.distributed.all_gather
rank = torch.distributed.get_rank()
world_size = torch.distributed.get_world_size()
input_tensor = torch.tensor((rank), dtype=torch.float32, system=system)
tensor_list = (torch.empty(1, dtype=torch.float32, system=system) for _ in vary(world_size))
torch.distributed.all_gather(tensor_list, input_tensor)
print(f"Rank {rank} gathered: {(t.merchandise() for t in tensor_list)}")

• torch.distributed.reduce_scatter(output, input_list): Äquivalent zur Durchführung einer Reduzierungsoperation für eine Liste von Tensoren und der anschließenden Streuung der Ergebnisse. Jeder Rang erhält einen anderen Teil der reduzierten Leistung.

# Instance for torch.distributed.reduce_scatter
rank = torch.distributed.get_rank()
world_size = torch.distributed.get_world_size()
input_list = (torch.tensor((rank + i), dtype=torch.float32, system=system) for i in vary(world_size))
output = torch.empty(1, dtype=torch.float32, system=system)
torch.distributed.reduce_scatter(output, input_list, op=torch.distributed.ReduceOp.SUM)
print(f"Rank {rank} obtained lowered worth: {output.merchandise()}")

Synchronisation

Die beiden am häufigsten verwendeten Operationen sind request.wait() Und torch.cuda.synchronize(). Es ist wichtig, den Unterschied zwischen diesen beiden zu verstehen:

• request.wait(): Dies wird für asynchrone Vorgänge verwendet. Es synchronisiert den aktuell aktiven CUDA-Stream für diesen Vorgang und stellt sicher, dass der Stream auf den Abschluss der Kommunikation wartet, bevor er fortfährt. Mit anderen Worten: Es blockiert den aktuell aktiven CUDA-Stream, bis die Datenübertragung abgeschlossen ist. Auf der Host-Seite bewirkt es nur, dass der Host wartet, bis der Kernel in die Warteschlange gestellt wird; der Gastgeber tut es nicht Warten Sie, bis die Datenübertragung abgeschlossen ist.
• torch.cuda.synchronize(): Dies ist ein energischerer Befehl, der den Host-CPU-Thread bis pausiert alle Zuvor auf der GPU eingereihte Aufgaben sind abgeschlossen. Es stellt sicher, dass die GPU vollständig im Leerlauf ist, bevor die CPU weitermacht, kann jedoch bei unsachgemäßer Verwendung zu Leistungsengpässen führen. Wann immer Sie Benchmark-Messungen durchführen müssen, sollten Sie diese verwenden, um sicherzustellen, dass Sie genau den Second erfassen, in dem die GPUs fertig sind.

Abschluss

Herzlichen Glückwunsch, dass Sie es bis zum Ende geschafft haben! In diesem Beitrag haben Sie Folgendes erfahren:

• Punkt-zu-Punkt-Operationen
• Synchronisierung und Asynchronisierung in NCCL
• Kollektive Operationen
• Synchronisationsmethoden

Im nächsten Blogbeitrag befassen wir uns mit PCIe, NVLink und anderen Mechanismen, die die Kommunikation in einer verteilten Umgebung ermöglichen!

Referenzen