Entraîner sur plusieurs machines commence par une opération banale : sommer le même tenseur chez tous les participants. En parallélisme de données, chaque rang calcule le gradient sur son lot ; ce qui fait avancer le modèle n’est ni le gradient de l’un ni celui de l’autre, mais leur moyenne. C’est un all_reduce — et c’est presque toujours lui qui décide de la vitesse d’un entraînement distribué, parce qu’il s’exécute à chaque pas et qu’il fait circuler les gradients entiers.
Ce carnet écrit cet all_reduceà la main, en anneau, puis le compare à celui de la bibliothèque.
Ce que le carnet mesure, et sur quoi il se prononce
Question
Instrument
Ce qui est comparé
Le résultat est-il le bon ?
comparaison à dist.all_reduce sur des entiers
exactitude, pas plausibilité
Combien d’octets circulent ?
compteur posé dans la boucle d’anneau
anneau contre schéma naïf, en éléments
Combien de messages ?
compteur d’envois et de réceptions
la loi \(4(p-1)\)
Le résultat est-il le même ?
somme flottante contre la même somme réordonnée
l’associativité, mesurée
Est-ce plus rapide ?
chronométrage, dispersion comprise
et ce que la mesure ne permet pas de conclure
Ce que le carnet refuse de faire
Il ne présente pas l’anneau comme « plus rapide » parce que c’est la théorie. La dernière section montre une mesure où les deux implémentations sont inséparables, et une autre où la bibliothèque est plus lente à deux rangs qu’à quatre — une inversion qu’aucun modèle de volume ne prédit. Le carnet donne la mesure et nomme ce qu’il ne l’explique pas.
Le protocole : deux processus, deux rangs, un store fichier
torch.distributed a besoin d’un rendez-vous : les rangs doivent se trouver avant d’échanger. Trois façons de le donner, et sur cette machine une seule fonctionne sans friction :
Méthode
Ce qui se passe ici
torchrun --standalone
échoue — le store TCP ne s’ouvre pas sur Windows (RendezvousConnectionError)
variables d’environnement maîtrisées
possible, mais suppose un ordonnanceur
store fichier (init_method="file:///…")
retenu — aucun socket en écoute, les rangs se retrouvent par un fichier
Le prix du store fichier est qu’il faut fournir rank et world_sizeexplicitement : init_method ne les déduit pas de l’environnement. Le carnet le fait dans le lanceur.
Un point de vocabulaire, parce qu’il piège : un rang n’est pas un processus par convention, c’est un processus. Les deux rangs de ce carnet sont deux interpréteurs Python séparés, qui se parlent par gloo. C’est ce qui rend la mesure honnête — et c’est aussi ce qui la limite, comme la dernière section le montrera.
Le noyau fait partie du protocole
Avant même le rendez-vous, gloo doit choisir une interface réseau. Toutes les versions de torch n’y parviennent pas, et le noyau de ce carnet a été choisi pour cette raison — la mesure est reproducible :
L’erreur n’est pas un problème d’adresse : le nom d’hôte se résout correctement (myia-po-2023 → 192.168.0.46), et forcer l’interface par GLOO_SOCKET_IFNAME déplace l’erreur (makeDeviceForInterface(): unsupported gloo device) sans la lever. C’est une propriété de la version de torch, pas de la configuration réseau.
La leçon dépasse ce carnet : pour du code distribué, le noyau est une dépendance du protocole, pas un réglage cosmétique. Un carnet qui se contente de déclarer « Python 3 » peut fonctionner partout pour du calcul local et échouer sur sa première ligne de communication. C’est pourquoi celui-ci déclare coursia-ml-training, l’environnement d’entraînement du dépôt, explicitement CPU — ce qui suffit amplement ici, gloo ne touchant pas aux cartes.
import jsonimport osimport pathlibimport subprocessimport sysimport tempfileimport torchprint("torch", torch.__version__)print("distribue disponible :", torch.distributed.is_available())print("backend gloo :", torch.distributed.is_gloo_available())print("carte GPU presente :", torch.cuda.is_available())print("threads torch :", torch.get_num_threads(), "(intra-operateur, pas les coeurs)")# On imprime le NOM de l'environnement, pas le chemin de l'interpreteur : un# chemin absolu publie le nom d'utilisateur de la machine dans une sortie# committee. C'est le nom qui informe (quel noyau tourne), pas le chemin.print("environnement :", pathlib.Path(sys.executable).parent.name)
torch 2.14.0+cpu
distribue disponible : True
backend gloo : True
carte GPU presente : False
threads torch : 14 (intra-operateur, pas les coeurs)
environnement : coursia-ml-training
Le schéma naïf, et pourquoi l’anneau gagne
Le résultat à produire est la somme, élément par élément, des \(p\) tenseurs — chacun détenu par un rang. Une façon évidente de faire : chaque rang envoie tout à tous les autres et additionne ce qu’il reçoit.
Schéma
Éléments envoyés par rang
À \(p=4\)
À \(p=8\)
naïf (chacun à chacun)
\(n(p-1)\)
\(3n\)
\(7n\)
anneau
\(2(p-1)\dfrac{n}{p}\)
\(1{,}5\,n\)
\(1{,}75\,n\)
L’anneau découpe le tenseur en \(p\) morceaux et fait faire un tour à chacun : chaque rang envoie \(p-1\) morceaux en phase de réduction, puis \(p-1\) morceaux en phase de diffusion — d’où \(2(p-1)\) morceaux de \(n/p\) éléments.
Le rapport entre les deux vaut \(p/2\) : à deux rangs l’anneau n’apporte rien (le vérifier est un des exercices), à huit il divise le trafic par quatre. C’est cette propriété — un volume par rang qui ne dépend presque plus de \(p\) — qui rend l’entraînement distribué praticable.
La section suivante ne se contente pas d’écrire ces formules : elle les compte sur une exécution réelle.
L’anneau, écrit à la main
La cellule suivante écrit le fichier travailleur sur le disque. Deux points de mécanique méritent l’attention, parce que les deux ont effectivement cassé la première version :
Où va la somme. À l’étape \(i\), le rang \(r\) envoie le morceau \((r-i)\) et reçoit le morceau \((r-i-1)\) d’un voisin qui l’a déjà partiellement réduit. La somme doit donc s’ajouter au morceau reçu, pas à celui qu’on possède. La première version accumulait dans une variable unique : le résultat était faux, et faux silencieusement — la forme du tenseur restait bonne.
L’ordre des envois. Avec send/recv bloquants, deux rangs qui commencent tous les deux par un envoi s’attendent mutuellement : le programme se bloque sans erreur. batch_isend_irecv prend le rendez-vous par paire, ce qui lève le blocage.
Ces deux défauts ne sont pas des anecdotes : ce sont les deux façons dont un code de communication échoue — un résultat faux sans message d’erreur, et un blocage sans message d’erreur. Les nommer vaut mieux que les découvrir.
TRAVAILLEUR =r'''# Un rang de l'anneau : initie le groupe, compare anneau et appel natif, mesure.import jsonimport osimport timeimport torchimport torch.distributed as distdef _echange(ops):# Poste des operations point-a-point puis attend qu'elles aboutissent.# batch_isend_irecv prend le rendez-vous par paire : deux rangs qui# commencent tous deux par un envoi bloquant s'interbloquent avec send/recv# nus, pas ici. for req in dist.batch_isend_irecv(ops): req.wait()def anneau_allreduce(t, rang, monde):# All-reduce en anneau : reduce-scatter puis all-gather.# Rend (tenseur reduit, messages, elements_envoyes) pour le rang appelant. n = t.numel() assert n % monde == 0, "la taille doit se partager en `monde` morceaux" k = n // monde plat = t.reshape(-1).clone() messages = elements = 0 suivant, precedent = (rang + 1) % monde, (rang - 1) % monde# Phase 1 : reduce-scatter. A l'etape i le rang r envoie le morceau (r-i)# et accumule celui qu'il recoit, (r-i-1) : la somme va dans le morceau# recu, pas dans celui qu'on possede. Apres p-1 etapes, le rang r detient# le morceau (r+1) entierement reduit. for etape in range(monde - 1): i_envoi = (rang - etape) % monde i_recoi = (rang - etape - 1) % monde morceau = plat[i_envoi * k:(i_envoi + 1) * k].contiguous() tampon = torch.empty(k, dtype=t.dtype) _echange([dist.P2POp(dist.isend, morceau, suivant), dist.P2POp(dist.irecv, tampon, precedent)]) plat[i_recoi * k:(i_recoi + 1) * k]+= tampon messages += 2 elements += k# Phase 2 : all-gather. Les morceaux reduits font le tour de l'anneau. for etape in range(monde - 1): i_envoi = (rang + 1 - etape) % monde i_recoi = (rang - etape) % monde morceau = plat[i_envoi * k:(i_envoi + 1) * k].contiguous() tampon = torch.empty(k, dtype=t.dtype) _echange([dist.P2POp(dist.isend, morceau, suivant), dist.P2POp(dist.irecv, tampon, precedent)]) plat[i_recoi * k:(i_recoi + 1) * k] = tampon messages += 2 elements += k return plat.reshape(t.shape), messages, elementsdef chrono(fn, repetitions=5):# Rend la liste des durees individuelles, en secondes. Chaque mesure est# separee : sur une seule machine, des rangs qui se partagent les memes# coeurs donnent des constantes bruitees, et une moyenne seule cacherait# cette dispersion. fn() durees = [] for _ in range(repetitions): dist.barrier() debut = time.perf_counter() fn() durees.append(time.perf_counter() - debut) return dureesdef main(): init = os.environ["RING_INIT"] dist.init_process_group("gloo", init_method=init, rank=int(os.environ["RANK"]), world_size=int(os.environ["WORLD_SIZE"])) rang, monde = dist.get_rank(), dist.get_world_size() torch.manual_seed(0) sortie = {"rang": rang, "monde": monde, "backend": dist.get_backend()}# --- 1) exactitude sur une petite entree ------------------------------- n = 12 ent = torch.arange(n, dtype=torch.int64)+ 1 a, messages, elements = anneau_allreduce(ent.clone(), rang, monde) b = ent.clone() dist.all_reduce(b, op=dist.ReduceOp.SUM) sortie["entiers_identiques"] = bool(torch.equal(a, b)) sortie["attendu"] = (ent * monde).tolist() sortie["messages"] = messages sortie["elements_envoyes"] = elements sortie["elements_naif"] = n *(monde - 1) sortie["theorie_messages"] = 2 * 2 *(monde - 1) sortie["theorie_elements"] = 2 *(monde - 1)*(n // monde)# --- 2) associativite : la somme flottante n'est pas associative -------# Chaque rang detient une valeur differente (c'est le cas reel : les# gradients different d'un rang a l'autre). Les valeurs sont choisies pour# que l'ordre des additions change le resultat. table = torch.tensor([1e16, 1.0, -1e16, 3.0], dtype=torch.float64) base = torch.full((n,), float(table[rang % 4].item()), dtype=torch.float64) mien, _, _ = anneau_allreduce(base.clone(), rang, monde) natif = base.clone() dist.all_reduce(natif, op=dist.ReduceOp.SUM) sortie["assoc_anneau"] = mien[0].item() sortie["assoc_natif"] = natif[0].item() sortie["assoc_ecart"] = float((mien - natif).abs().max().item()) sortie["assoc_identique"] = bool(torch.equal(mien, natif))# Le meme fait, calcule hors de tout reseau : deux ordres de sommation sur# les memes quatre nombres, deux resultats. sortie["ordre_gauche"] = float(((table[0]+ table[1])+ table[2])+ table[3]) sortie["ordre_equilibre"] = float((table[0]+ table[2])+(table[1]+ table[3]))# --- 3) chronometrage et volume sur le fil ----------------------------- mesures = [] for puissance in (18, 22): m = 1 << puissance gros = torch.randn(m, dtype=torch.float64) t_anneau = chrono(lambda: anneau_allreduce(gros.clone(), rang, monde)) t_natif = chrono(lambda: dist.all_reduce(gros.clone(), op=dist.ReduceOp.SUM)) mesures.append({ "elements": m, "Mo": round(m * 8 / 2**20, 2), "anneau_s": [round(x, 5) for x in t_anneau], "natif_s": [round(x, 5) for x in t_natif], "rapport_median": round( sorted(t_anneau)[len(t_anneau) // 2] / sorted(t_natif)[len(t_natif) // 2], 3), "elements_envoyes": 2 *(monde - 1)*(m // monde), "elements_naif": m *(monde - 1), }) sortie["chrono"] = mesures if rang == 0: print("RESULTAT " + json.dumps(sortie)) dist.destroy_process_group()if __name__ == "__main__": main()'''dossier = pathlib.Path(tempfile.mkdtemp(prefix="collectives_"))chemin = dossier /"travailleur.py"chemin.write_text(TRAVAILLEUR, encoding="utf-8")lignes = TRAVAILLEUR.count("\n")# Meme raison que ci-dessus : le basename, jamais le chemin absolu du temporaire.print("fichier ecrit :", chemin.name)print("lignes :", lignes)print("octets :", len(TRAVAILLEUR.encode("utf-8")))print("compile :", end=" ")compile(TRAVAILLEUR, str(chemin), "exec")print("ok")
Le lanceur ci-dessous fait trois choses : créer le répertoire du store (le rendez-vous par fichier), lancer monde processus en leur passant chacun son RANK, et récupérer la ligne RESULTAT du rang 0.
Deux précautions, parce qu’un lancement de processus qui tourne mal ne rend pas de trace utile :
un délai maximal : sans lui, un blocage de rendez-vous fait attendre indéfiniment (c’est le mode d’échec du premier essai) ;
le code de retour et la fin de stderr de chaque rang : un rang qui meurt doit le dire.
def lance(monde, travailleur, delai=240):# Lance `monde` rangs relies par un store fichier et rend le dictionnaire# du rang 0. Leve une erreur explicite si un rang echoue ou bloque. store = pathlib.Path(tempfile.mkdtemp(prefix=f"store_p{monde}_")) init ="file:///"+str(store /"rendez_vous").replace("\\", "/") env =dict(os.environ, RING_INIT=init, WORLD_SIZE=str(monde)) procs = []for rang inrange(monde): procs.append(subprocess.Popen( [sys.executable, str(travailleur)], env=dict(env, RANK=str(rang)), stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, encoding="utf-8")) resultats, pannes = [], []for rang, proc inenumerate(procs):try: out, err = proc.communicate(timeout=delai)except subprocess.TimeoutExpired: proc.kill() out, err = proc.communicate() pannes.append(f"rang {rang} : blocage, tue apres {delai} s")continueif proc.returncode !=0: pannes.append(f"rang {rang} : code {proc.returncode}\n{err.strip()[-600:]}")for ligne in out.splitlines():if ligne.startswith("RESULTAT "): resultats.append(json.loads(ligne[len("RESULTAT "):]))if pannes:raiseRuntimeError("lancement en echec :\n"+"\n".join(pannes))assertlen(resultats) ==1, f"{len(resultats)} rangs ont repondu, 1 attendu"return resultats[0]mesure2 = lance(2, chemin)print("monde :", mesure2["monde"])print("backend :", mesure2["backend"])print("somme :", mesure2["attendu"])print("entiers identiques a l'appel natif :", mesure2["entiers_identiques"])print("messages :", mesure2["messages"], "(theorie", mesure2["theorie_messages"], ")")print("elements envoyes par rang :", mesure2["elements_envoyes"],"| schema naif :", mesure2["elements_naif"])
monde : 2
backend : gloo
somme : [2, 4, 6, 8, 10, 12, 14, 16, 18, 20, 22, 24]
entiers identiques a l'appel natif : True
messages : 4 (theorie 4 )
elements envoyes par rang : 12 | schema naif : 12
Premier dépouillement : à deux rangs, l’anneau n’apporte rien
Le résultat est exact — la comparaison porte sur des entiers, où l’égalité est sans ambiguïté : l’anneau et l’appel natif rendent le même tenseur, au bit près.
Le volume, lui, est instructif par ce qu’il n’apporte pas. À \(p=2\), la formule de l’anneau \(2(p-1)n/p\) vaut simplement \(n\) — exactement le coût du schéma naïf \(n(p-1)\). Il n’y a pas de tour à faire circuler quand il n’y a que deux participants : l’anneau n’a d’intérêt qu’à partir de trois.
C’est le premier endroit où la formule sert de garde-fou contre une intuition : « anneau » sonne comme « meilleur », mais le modèle dit égalité, et la mesure le confirme.
mesure4 = lance(4, chemin)champs = ["monde", "messages", "theorie_messages", "elements_envoyes","theorie_elements", "elements_naif", "entiers_identiques"]for nom in champs:print(f"{nom:20s}{mesure4[nom]}")
Deuxième dépouillement : la loi des messages et celle des octets
La sortie ci-dessus porte trois vérifications, et elles tombent toutes justes :
Grandeur
Formule
À \(p=4\)
Mesuré
messages par rang
\(4(p-1)\)
\(12\)
\(12\)
éléments envoyés par rang
\(2(p-1)n/p\)
\(1{,}5\,n\)
\(18 = 1{,}5 \times 12\)
éléments du schéma naïf
\(n(p-1)\)
\(3n\)
\(36 = 3 \times 12\)
L’anneau envoie donc exactement la moitié du schéma naïf à quatre rangs — le rapport \(p/2\) de la formule, vérifié sur un compteur posé dans la boucle plutôt que sur une estimation.
Le compteur de messages mérite un mot : \(4(p-1)\) et non \(2(p-1)\). Chaque étape comporte un envoi et une réception — deux messages — et il y a \(p-1\) étapes dans chacune des deux phases. Un implémentation qui compterait \(2(p-1)\) aurait oublié une phase.
# La somme flottante n'est pas associative : deux ordres, deux resultats.# Ce calcul est purement local, hors de tout reseau.valeurs = torch.tensor([1e16, 1.0, -1e16, 3.0], dtype=torch.float64)gauche = ((valeurs[0] + valeurs[1]) + valeurs[2]) + valeurs[3]equilibre = (valeurs[0] + valeurs[2]) + (valeurs[1] + valeurs[3])print("valeurs :", valeurs.tolist())print("ordre de gauche :", gauche.item())print("ordre equilibre :", equilibre.item())print("les deux ordres :", "different"if gauche != equilibre else"coincident")print()print("explication : 1e16 + 1.0 vaut 1e16 (le 1 est perdu), mais")print(" (1e16 - 1e16) + 1.0 vaut 1.0. L'ordre decide du resultat.")
valeurs : [1e+16, 1.0, -1e+16, 3.0]
ordre de gauche : 3.0
ordre equilibre : 4.0
les deux ordres : different
explication : 1e16 + 1.0 vaut 1e16 (le 1 est perdu), mais
(1e16 - 1e16) + 1.0 vaut 1.0. L'ordre decide du resultat.
Troisième dépouillement : le même problème, deux réponses
Le calcul précédent n’a rien de distribué — c’est une propriété de l’addition flottante. Il sert de contrôle : si deux ordres de sommation sur quatre nombres connus donnent \(3{,}0\) et \(4{,}0\), alors deux implémentations d’all_reduce qui accumulent dans des ordres différents doivent pouvoir diverger.
C’est exactement ce que la mesure distribuée montre à quatre rangs.
print("anneau (a la main) :", mesure4["assoc_anneau"])print("appel natif :", mesure4["assoc_natif"])print("ecart :", mesure4["assoc_ecart"])print("identiques :", mesure4["assoc_identique"])print()print("a deux rangs : anneau", mesure2["assoc_anneau"],"| natif", mesure2["assoc_natif"],"| identiques", mesure2["assoc_identique"])
anneau (a la main) : 3.0
appel natif : 4.0
ecart : 1.0
identiques : False
a deux rangs : anneau 1e+16 | natif 1e+16 | identiques True
Ce que cette divergence signifie — et ce qu’elle ne signifie pas
À quatre rangs, l’anneau rend \(3{,}0\) et l’appel natif \(4{,}0\) : un écart de \(1\) sur la même somme mathématique. À deux rangs, les deux coïncident — il n’y a qu’un ordre possible quand il n’y a qu’une addition à faire.
Trois conséquences, dans l’ordre d’importance :
Ce n’est pas un bug de l’anneau. Les deux résultats sont des arrondis légitimes de la même somme exacte. Le carnet n’a pas écrit un all-reduce faux : il a écrit un all-reduce différent, et la différence est de l’ordre de l’erreur d’arrondi.
Cela ne gêne pas la descente de gradient. Un gradient est une direction, pas un scalaire à préserver : une erreur relative de \(10^{-16}\) sur ses composantes ne change pas le pas suivant de manière observable. C’est pourquoi les bibliothèques se permettent de choisir l’ordre qui les arrange.
Cela interdit la reproductibilité au bit près. Deux tailles de monde, deux implémentations, deux versions de la bibliothèque : le même programme ne donnera pas exactement le même modèle. Reproduire un entraînement distribué au bit près demande de fixer aussi l’algorithme de réduction, pas seulement la graine.
L’écart de \(1{,}0\) sur cet exemple est un écart choisi par les valeurs du test — les magnitudes ont été construites pour rendre l’ordre visible. Sur des gradients réels, la divergence est d’une tout autre échelle : c’est précisément pour cela qu’on la mesure sur un cas fabriqué, où elle est lisible.
def tableau(chronos): entete =f"{'vol (Mo)':>9}{'anneau (s), 5 mesures':<34}{'natif (s), 5 mesures':<34}{'rapport':>8}"print(entete)print("-"*len(entete))for c in chronos: a =" ".join(f"{x:.5f}"for x in c["anneau_s"]) n =" ".join(f"{x:.5f}"for x in c["natif_s"])print(f"{c['Mo']:>6.1f} Mo {a:<34}{n:<34}{c['rapport_median']:>8.3f}")print()print("(rapport = mediane de l'anneau / mediane du natif ; < 1 = anneau plus rapide)")print("=== mesure a 2 rangs ===")tableau(mesure2["chrono"])print()print("=== mesure a 4 rangs ===")tableau(mesure4["chrono"])
=== mesure a 2 rangs ===
vol (Mo) anneau (s), 5 mesures natif (s), 5 mesures rapport
----------------------------------------------------------------------------------------
2.0 Mo 0.01358 0.01537 0.00259 0.00271 0.02096 0.02824 0.03065 0.00320 0.01372 0.02990 0.481
32.0 Mo 0.04779 0.05597 0.03634 0.06700 0.03760 0.16393 0.28421 0.10232 0.17302 0.24679 0.276
(rapport = mediane de l'anneau / mediane du natif ; < 1 = anneau plus rapide)
=== mesure a 4 rangs ===
vol (Mo) anneau (s), 5 mesures natif (s), 5 mesures rapport
----------------------------------------------------------------------------------------
2.0 Mo 0.00692 0.00621 0.00601 0.00724 0.00646 0.01960 0.00492 0.00425 0.00416 0.00413 1.521
32.0 Mo 0.07999 0.09281 0.08325 0.09944 0.09100 0.09754 0.08303 0.11940 0.06833 0.09833 0.933
(rapport = mediane de l'anneau / mediane du natif ; < 1 = anneau plus rapide)
Quatrième dépouillement : ce que le chronomètre permet de dire, et ce qu’il ne permet pas
Trois lectures, dont deux négatives. Les présenter ensemble est le point de cette section : un chiffre de performance isolé ne veut rien dire tant qu’on n’a pas regardé sa dispersion.
1. À \(4\) rangs et \(2\) Mo, l’anneau à la main est plus lent — et c’est net. Côté anneau, les cinq mesures sont serrées (6,0 à 7,2 ms) ; côté natif, quatre le sont (4,1 à 4,9 ms) mais la cinquième (19,6 ms) part seule — un artefact d’ordonnancement du genre de ceux que la section suivante décrit. C’est précisément pour ça que la table compare les médianes, robustes à ce point isolé : sur elles, l’écart est réel, il n’est pas un artefact de dispersion. Le facteur, lui, varie d’une exécution à l’autre — entre \(1{,}3\) et \(1{,}6\) sur les exécutions répétées pendant l’écriture de ce carnet, sans que la mesure locale permette d’attribuer cette variation à autre chose qu’à l’ordonnancement des processus sur une machine partagée. La table ci-dessus donne celui de l’exécution enregistrée ici ; c’est la certitude de l’écart, pas sa valeur exacte, qui se transporte d’une exécution à l’autre. La cause est dans le code, pas dans le réseau : six étapes écrites en Python, chacune avec son appel batch_isend_irecv et ses allocations, là où l’appel natif descend dans du C++ qui enchaîne les mêmes étapes sans repasser par l’interpréteur. Le schéma est le même ; c’est le coût par étape qui diffère.
2. À \(32\) Mo, les deux deviennent inséparables. Les cinq mesures de l’anneau recouvrent celles du natif : chaque plage mord sur celle de l’autre, et la plus lente des mesures de l’un est plus lente que la plus rapide de l’autre. La conclusion honnête est « on ne peut pas trancher sur cette machine », pas « l’anneau rattrape son retard » — un recouvrement ne prouve pas une égalité, il constate qu’on manque de résolution pour les séparer. C’est aussi ce qu’on attend d’un régime où le temps de transfert domine le coût de l’ordonnancement.
3. Une inversion que rien n’explique dans ce carnet. À \(32\) Mo, l’appel natif à deux rangs prend plus de temps que l’appel natif à quatre rangs — sur le même volume, avec moins de participants. Une implémentation qui choisit son algorithme selon la taille du monde (ce que fait gloo) peut parfaitement produire ce genre d’inversion, mais ce carnet ne descend pas dans ses entrailles : il constate, il ne l’explique pas. La formuler est plus utile que de l’habiller.
La limite de protocole, à énoncer
Les deux rangs tournent sur une seule machine, et se partagent les mêmes cœurs. Le réseau est absent, l’ordonnanceur du système voit quatre interpréteurs concurrents, et les durées dépendent de l’ordre d’ordonnancement des threads. Ce que ces mesures établissent de façon robuste, ce sont les comptes — éléments sur le fil, nombre de messages, exactitude — qui ne dépendent ni de la machine ni de la charge. Les durées, elles, sont des ordres de grandeur locaux : elles disent qu’un anneau écrit en Python coûte ce qu’il coûte, pas qu’un anneau est lent en général.
Ce que la bibliothèque fait de tout cela
DistributedDataParallel enveloppe un modèle et, à chaque rétropropagation, exécute un all_reduce sur les gradients — puis les divise par le nombre de rangs. Trois choix d’ingénierie rendent l’opération supportable, et les trois découlent de ce carnet :
Le regroupement en seaux. Réduire chaque tenseur de gradient dès qu’il est prêt multiplierait les petits messages. DDP attend d’avoir accumulé assez de gradients pour remplir un seau (25 Mo par défaut), puis réduit le seau entier. C’est le même arbitrage que la section précédente : à petit volume, c’est le coût par message qui domine ; à gros volume, le débit.
Le recouvrement. La réduction d’un seau se recouvre avec la suite de la rétropropagation — ce qui reste à calculer ne dépend pas de ce qui est déjà réduit.
L’anneau, précisément parce que son volume par rang ne dépend presque plus de \(p\). La loi \(2(p-1)n/p\) tend vers \(2n\) quand \(p\) grandit : doubler le nombre de participants double le débit agrégé disponible sans doubler le travail de chacun.
C’est le fil qui mène au carnet suivant : ce que coûte la mémoire d’un entraînement distribué, et pourquoi le découpage des états de l’optimiseur (ZeRO) ou du modèle lui-même (FSDP) change la donne autrement que le découpage du lot.
Ce que ce carnet a établi
Affirmation
Statut
l’anneau écrit à la main est correct
vérifié — égalité au bit contre l’appel natif sur des entiers
il envoie \(2(p-1)n/p\) éléments par rang
vérifié — compteur posé dans la boucle, \(18\) contre \(36\) à \(p=4\)
il échange \(4(p-1)\) messages par rang
vérifié — \(12\) à \(p=4\)
à deux rangs il n’apporte rien
vérifié — \(n\) éléments des deux côtés
il n’est pas plus rapide que le natif à \(4\) rangs
mesuré — plus lent à petit volume, inséparable à gros volume
la somme flottante n’est pas associative
vérifié — \(3{,}0\) contre \(4{,}0\), et divergence mesurée entre les deux implémentations
Exercices
Les trois exercices portent sur le même fichier travailleur. Ils se traitent en modifiant la chaîne TRAVAILLEUR, en réécrivant le fichier, puis en relançant lance.
Un rappel de méthode, parce que c’est là que se perd le temps : après avoir modifié le travailleur, il faut réécrire le fichier (chemin.write_text(TRAVAILLEUR, encoding="utf-8")) avant de relancer. lance exécute le fichier sur le disque, pas la chaîne en mémoire.
# Exercice 1 -- Decouper l'anneau en deux fonctions.# TODO etudiant : ecrire reduce_scatter(t, rang, monde) qui ne fait QUE la# phase 1, et all_gather(t, rang, monde) qui ne fait QUE la phase 2.# Indice : la phase 1 laisse le rang r proprietaire du morceau (r+1) reduit ;# la phase 2 fait circuler ces morceaux jusqu'a ce que chacun ait tout.# Critere de reussite : all_gather(reduce_scatter(t)) doit rendre exactement# ce que rend anneau_allreduce(t), sur le meme tenseur entier.def reduce_scatter(t, rang, monde):returnNonedef all_gather(t, rang, monde):returnNoneprint("Exercice a completer")
Exercice a completer
# Exercice 2 -- Verifier la loi a trois rangs.# TODO etudiant : lancer le travailleur a monde=3 et reporter les deux comptes.# Critere de reussite : messages = 4*(monde-1) = 8, et# elements_envoyes = 2*(monde-1)*(n//monde) = 2*2*4 = 16,# contre elements_naif = n*(monde-1) = 24.# Attention : n=12 ne se divise pas par 5, mais se divise par 3 -- le choix de# n dans le travailleur n'est pas arbitraire.mesure3 =None# TODO etudiant : remplacer par lance(3, chemin)print("Exercice a completer")
Exercice a completer
# Exercice 3 -- Le rapport entre anneau et schema naif.# TODO etudiant : completer la fonction ci-dessous, qui rend le rapport# (cout_naif / cout_anneau) pour un monde donne, puis l'appeler pour# monde = 2, 4, 8, 64.# Indice : les deux couts sont 2*(p-1)/p et (p-1) par element, le rapport ne# depend donc PAS de n.# Critere de reussite : le rapport doit valoir p/2.def rapport_couts(monde):returnNoneprint("Exercice a completer")
Exercice a completer
Suite
Ce carnet a isolé la brique de communication. Le suivant (3.13) monte d’un étage : comment la mémoire se répartit quand on découpe non plus les données mais le modèle et ses états — DDP contre ZeRO contre FSDP, puis le parallélisme de pipeline, où l’ordre des micro-lots (GPipe contre 1F1B) décide de l’occupation réelle des cartes.