@@ -33,80 +33,69 @@ def _grad_output() -> torch.Tensor:
3333 return torch .randn (SEQ , BATCH , generator = torch .Generator ().manual_seed (11 ))
3434
3535
36- def _check_against_full_softmax (rank : int , world_size : int , port : int ) -> None :
37- init_gloo (rank , world_size , port = port )
38- try :
39- group = dist .new_group (list (range (world_size )))
40- shard = VOCAB // world_size
41- for dtype , tol in ((torch .float32 , 1e-6 ), (torch .bfloat16 , 1e-2 )):
42- full = _full_logits (dtype )
43- expected_loss , expected_grad = _reference (full , _target ())
44- local = full [..., rank * shard : (rank + 1 ) * shard ].clone ().requires_grad_ ()
45- local_before = local .detach ().clone ()
46-
47- loss = vocab_parallel_cross_entropy (local , _target (), group )
48- loss .backward (_grad_output ())
49-
50- torch .testing .assert_close (loss , expected_loss , atol = tol , rtol = tol )
51- assert local .grad .dtype == dtype
52- torch .testing .assert_close (
53- local .grad , expected_grad [..., rank * shard : (rank + 1 ) * shard ], atol = tol , rtol = tol
54- )
55- torch .testing .assert_close (local .detach (), local_before , atol = 0 , rtol = 0 )
56- finally :
57- dist .destroy_process_group ()
58-
59-
60- @pytest .mark .parametrize ("world_size" , [1 , 2 ])
61- def test_matches_the_full_vocabulary_log_softmax_on_every_shard (world_size : int ) -> None :
36+ def _check_against_full_softmax (rank : int , world_size : int , group : dist .ProcessGroup ) -> None :
6237 """Loss and gradient equal a plain log-softmax over the whole vocabulary, however it is split."""
63- run_multiprocess (_check_against_full_softmax , world_size = world_size )
64-
38+ shard = VOCAB // world_size
39+ for dtype , tol in ((torch .float32 , 1e-6 ), (torch .bfloat16 , 1e-2 )):
40+ full = _full_logits (dtype )
41+ expected_loss , expected_grad = _reference (full , _target ())
42+ local = full [..., rank * shard : (rank + 1 ) * shard ].clone ().requires_grad_ ()
43+ local_before = local .detach ().clone ()
6544
66- def _check_against_megatron ( rank : int , world_size : int , port : int ) -> None :
67- from megatron . core . fusions . fused_cross_entropy import fused_vocab_parallel_cross_entropy
45+ loss = vocab_parallel_cross_entropy ( local , _target (), group )
46+ loss . backward ( _grad_output ())
6847
69- init_gloo (rank , world_size , port = port )
70- try :
71- group = dist .new_group (list (range (world_size )))
72- shard = VOCAB // world_size
73- full = _full_logits (torch .bfloat16 )
74- results = []
75- for fn in (vocab_parallel_cross_entropy , fused_vocab_parallel_cross_entropy ):
76- local = full [..., rank * shard : (rank + 1 ) * shard ].clone ().requires_grad_ ()
77- loss = fn (local , _target (), group )
78- loss .backward (_grad_output ())
79- results .append ((loss .detach (), local .grad ))
80- (loss , grad ), (megatron_loss , megatron_grad ) = results
81- torch .testing .assert_close (loss , megatron_loss , atol = 1e-6 , rtol = 1e-6 )
82- torch .testing .assert_close (grad , megatron_grad , atol = 0 , rtol = 0 )
83- finally :
84- dist .destroy_process_group ()
48+ torch .testing .assert_close (loss , expected_loss , atol = tol , rtol = tol )
49+ assert local .grad .dtype == dtype
50+ torch .testing .assert_close (
51+ local .grad , expected_grad [..., rank * shard : (rank + 1 ) * shard ], atol = tol , rtol = tol
52+ )
53+ torch .testing .assert_close (local .detach (), local_before , atol = 0 , rtol = 0 )
8554
8655
87- @pytest .mark .parametrize ("world_size" , [1 , 2 ])
88- def test_matches_megatron_fused_cross_entropy (world_size : int ) -> None :
56+ def _check_against_megatron (rank : int , world_size : int , group : dist .ProcessGroup ) -> None :
8957 """Where Megatron is installed, the stand-in computes what its fused kernel computes."""
90- pytest .importorskip ("megatron.core.fusions.fused_cross_entropy" )
91- run_multiprocess (_check_against_megatron , world_size = world_size )
58+ from megatron .core .fusions .fused_cross_entropy import fused_vocab_parallel_cross_entropy
9259
60+ shard = VOCAB // world_size
61+ full = _full_logits (torch .bfloat16 )
62+ results = []
63+ for fn in (vocab_parallel_cross_entropy , fused_vocab_parallel_cross_entropy ):
64+ local = full [..., rank * shard : (rank + 1 ) * shard ].clone ().requires_grad_ ()
65+ loss = fn (local , _target (), group )
66+ loss .backward (_grad_output ())
67+ results .append ((loss .detach (), local .grad ))
68+ (loss , grad ), (megatron_loss , megatron_grad ) = results
69+ torch .testing .assert_close (loss , megatron_loss , atol = 1e-6 , rtol = 1e-6 )
70+ torch .testing .assert_close (grad , megatron_grad , atol = 0 , rtol = 0 )
9371
94- def _check_compute_log_probs_without_megatron (rank : int , world_size : int , port : int ) -> None :
72+
73+ def _check_compute_log_probs_without_megatron (group : dist .ProcessGroup ) -> None :
74+ """The FSDP and torchtitan log-prob path imports nothing from Megatron when it is not installed."""
9575 sys .modules ["megatron" ] = None # importing any of it now raises, as on an install without it
9676 math_utils ._MEGATRON_INSTALLED = False
77+ logits = _full_logits (torch .float32 )[:, 0 ]
78+ tokens = _target ()[:, 0 ]
79+ # [seq, 1], the shape compute_log_probs has always returned
80+ expected = torch .log_softmax (logits , dim = - 1 ).gather (- 1 , tokens .unsqueeze (- 1 ))
81+
82+ torch .testing .assert_close (compute_log_probs (logits , tokens , group ), expected , atol = 1e-6 , rtol = 1e-6 )
83+
84+
85+ def _run_checks (rank : int , world_size : int , port : int ) -> None :
9786 init_gloo (rank , world_size , port = port )
9887 try :
99- group = dist .new_group ([0 ])
100- logits = _full_logits (torch .float32 )[:, 0 ]
101- tokens = _target ()[:, 0 ]
102- # [seq, 1], the shape compute_log_probs has always returned
103- expected = torch .log_softmax (logits , dim = - 1 ).gather (- 1 , tokens .unsqueeze (- 1 ))
104-
105- torch .testing .assert_close (compute_log_probs (logits , tokens , group ), expected , atol = 1e-6 , rtol = 1e-6 )
88+ group = dist .new_group (list (range (world_size )))
89+ _check_against_full_softmax (rank , world_size , group )
90+ _check_against_megatron (rank , world_size , group )
91+ if world_size == 1 :
92+ # last: it hides Megatron from the rest of this process
93+ _check_compute_log_probs_without_megatron (group )
10694 finally :
10795 dist .destroy_process_group ()
10896
10997
110- def test_compute_log_probs_runs_without_megatron () -> None :
111- """The FSDP and torchtitan log-prob path imports nothing from Megatron when it is not installed."""
112- run_multiprocess (_check_compute_log_probs_without_megatron , world_size = 1 )
98+ @pytest .mark .parametrize ("world_size" , [1 , 2 ])
99+ def test_vocab_parallel_cross_entropy (world_size : int ) -> None :
100+ """One process group per tensor-parallel size: spawning is what costs, so every check shares it."""
101+ run_multiprocess (_run_checks , world_size = world_size )
0 commit comments