Skip to content

Commit e8e39ab

Browse files
authored
Refactor post dataloading process (pytorch#4116)
# Update Aug 18 `BaseModel` defines `preprocess_inputs` and `_build_forward_inputs`. `preprocess_inputs` does the following: 1. prepares additional inputs using `_build_forward_inputs` which also returns `input_sharding`, an information on how to shard the tensor inputs. 2. CP shard the inputs based on `input_sharding` 3. annotate spmd_types or full_dtensor-ize the inputs based on `input_sharding` on the tensors 4. Also return `local_ntokens` counted after CP sharding and before dtensorizing `labels`. Trainer's `post_dataloading_process` is replaced with `_prepare_batch` for now to count `ntokens_seen`. This can be removed after dtensor support is removed, so the `local_ntokens` can be counted outside of `preprocess_inputs`. When a model has additional inputs to CP shard/annotate, one can override `_build_forward_inputs` as in [Kimi](https://github.com/pytorch/torchtitan/pull/4116/changes#diff-de024fcd87b762920e1b5c44bdd9759cfe65bcd766dc0b1d40205e235fb9c538R88-R103). The above recipe can handle most of the type annotation except [Qwen3.5](https://github.com/pytorch/torchtitan/pull/4116/changes#diff-ce202ea253afff036554347e92b8e31ec401bab12bcb34c4a58f31cb5e842c12R772-R797), which annotates `cu_seq_q` inside `VarlenMetadata`. In such cases one can override `preprocess_inputs`. ## Remark This could allow CP in multimodal models (assuming the decoder part allows CP like Muse Glimmer) by building a index tensor in `_build_forward_inputs` that indicates for each token position, which vision encoder output it should get. Then we CP shard this tensor in the same way as tokens, and scatter it inside `forward`. <del> Remove decoder specific logic in `Trainer.post_dataloading_process` to be configurable. 1. Each model defines `preprocess_inputs` as a function in `BaseModel.preprocess_inputs`. This reads the input batch given from the dataloader and produces additional input that the model needs, such as attention masks. 5. Each model defines `input_sharding: dict[str, SpmdLayout]` in `BaseModel.Config` that is used to give correct DTensor or spmd_types for each argument in the `model.forward`. 6. `prepare_context_parallel_input` is also generalized to infer from `input_sharding` on which inputs and which dimension to CP shard. This would allow multimodal models to annotate spmd_types for additional inputs such as `pixel_values` with a simple configuration. <\del> ## Test Plan `python -m tests.integration_tests.run_tests <output_dir> --test_suit models --ngpu 8` runs without error.
1 parent cdf0c99 commit e8e39ab

30 files changed

Lines changed: 848 additions & 443 deletions

‎scripts/checkpoint_conversion/numerical_tests_qwen3_5.py‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -308,7 +308,7 @@ def forward(self, q, k, v, **kwargs):
308308
tokens.to(device),
309309
pixel_values=pixel_values.half().to(device),
310310
grid_thw=grid_thw.to(device),
311-
mrope_positions=mrope_positions.to(device),
311+
positions=mrope_positions.to(device),
312312
special_tokens=special_tokens,
313313
)
314314
outputs.append(logits[:, -1:, :].cpu())

‎scripts/checkpoint_conversion/numerical_tests_qwen3_5_shard.py‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -115,8 +115,8 @@ def run_worker(args):
115115
dist.broadcast(tokens, src=0)
116116

117117
# Text-only inputs: plain sequential positions. The flex backend requires a
118-
# BlockMask, which the trainer normally builds in post_dataloading_process;
119-
# build it here directly since we call the model outside the trainer.
118+
# BlockMask, which the model normally builds in its preprocess_inputs; build
119+
# it here directly since we call the model outside the trainer.
120120
positions = torch.arange(seq_len, device="cuda").unsqueeze(0)
121121
attention_masks = cast(Qwen35Model, model).get_attention_masks(positions=positions)
122122

‎tests/unit_tests/cpu/test_trainer.py‎

Lines changed: 50 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -11,7 +11,6 @@
1111

1212
import pytest
1313
import torch
14-
1514
from torchtitan.distributed.cudagraph import wrap_with_cuda_graph
1615
from torchtitan.trainer import Trainer
1716

@@ -24,11 +23,18 @@ def test_pp_forward_backward_step_returns_sentinel_without_last_stage():
2423
pp_has_last_stage=False,
2524
pp_schedule=SimpleNamespace(step=lambda **kwargs: None),
2625
train_context=nullcontext,
27-
post_dataloading_process=lambda input_dict, labels: (
28-
input_dict["input"],
29-
labels,
30-
{},
31-
),
26+
model_parts=[
27+
SimpleNamespace(
28+
preprocess_inputs=lambda input_dict, **kw: (
29+
input_dict["input"],
30+
input_dict["labels"],
31+
{},
32+
)
33+
)
34+
],
35+
parallel_dims=SimpleNamespace(pp_enabled=True),
36+
config=SimpleNamespace(parallelism="PARA"),
37+
ntokens_seen=0,
3238
device=torch.device("cpu"),
3339
),
3440
)
@@ -43,6 +49,44 @@ def test_pp_forward_backward_step_returns_sentinel_without_last_stage():
4349
torch.testing.assert_close(loss, torch.tensor([-1.0]))
4450

4551

52+
def test_forward_backward_step_accumulates_tokens_and_forwards_triple():
53+
captured = {}
54+
55+
class _FakeModel:
56+
def preprocess_inputs(self, input_dict, **kw):
57+
captured["preprocess_kwargs"] = kw
58+
return ("INPUTS", torch.ones(7), {"positions": 1})
59+
60+
def fwd_bwd_fn(inputs, labels, global_valid_tokens, extra_kwargs):
61+
captured["fwd_bwd_args"] = (inputs, labels, extra_kwargs)
62+
return torch.tensor(0.0)
63+
64+
fake = SimpleNamespace(
65+
model_parts=[_FakeModel()],
66+
parallel_dims=SimpleNamespace(pp_enabled=False),
67+
config=SimpleNamespace(parallelism="PARA"),
68+
ntokens_seen=100,
69+
fwd_bwd_fn=fwd_bwd_fn,
70+
)
71+
72+
Trainer.forward_backward_step(
73+
fake,
74+
input_dict={"input": 0},
75+
labels=torch.zeros(1),
76+
global_valid_tokens=torch.tensor(1),
77+
)
78+
79+
inputs, labels, extra = captured["fwd_bwd_args"]
80+
assert inputs == "INPUTS"
81+
assert extra == {"positions": 1}
82+
assert labels.numel() == 7
83+
assert fake.ntokens_seen == 107 # labels.numel() (7) folded in
84+
assert captured["preprocess_kwargs"] == {
85+
"parallel_dims": fake.parallel_dims,
86+
"parallelism": "PARA",
87+
}
88+
89+
4690
def test_cuda_graph_wrapper_returns_graph_owned_output():
4791
class PassthroughCUDAGraphWrapper:
4892
def __init__(self, fn, example_inputs):

‎tests/unit_tests/cpu/test_validate.py‎

Lines changed: 13 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -30,7 +30,17 @@ def close(self):
3030
self.closed = True
3131

3232

33-
class _FailingModel(nn.Module):
33+
class _EchoModel(nn.Module):
34+
def preprocess_inputs(self, input_dict, *, parallel_dims, parallelism):
35+
del parallel_dims, parallelism
36+
return input_dict["input"], input_dict["labels"], {}
37+
38+
def forward(self, inputs, **kwargs):
39+
del kwargs
40+
return inputs
41+
42+
43+
class _FailingModel(_EchoModel):
3444
def forward(self, *args, **kwargs):
3545
del args, kwargs
3646
raise RuntimeError("validation failed")
@@ -61,11 +71,7 @@ def _generic_validator(loader):
6171
)
6272
validator.validation_context = nullcontext
6373
validator.loss_fn = lambda predictions, labels: (predictions.sum(), None)
64-
validator.post_dataloading_process = lambda input_dict, labels, model_parts: (
65-
input_dict["input"],
66-
labels,
67-
{},
68-
)
74+
validator.parallelism = SimpleNamespace()
6975
return validator
7076

7177

@@ -74,7 +80,7 @@ def test_generic_validator_closes_temporary_loader(monkeypatch, raises):
7480
row = ({"input": torch.ones(1, 1)}, torch.ones(1, 1, dtype=torch.long))
7581
loader = _ClosableLoader([row, row])
7682
validator = _generic_validator(loader)
77-
model = _FailingModel() if raises else nn.Identity()
83+
model = _FailingModel() if raises else _EchoModel()
7884
monkeypatch.setattr(validate_module.utils, "device_type", "cpu")
7985

8086
if raises:
Lines changed: 134 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,134 @@
1+
# Copyright (c) Meta Platforms, Inc. and affiliates.
2+
# All rights reserved.
3+
#
4+
# This source code is licensed under the BSD-style license found in the
5+
# LICENSE file in the root directory of this source tree.
6+
7+
"""Characterization tests for Qwen3.5 position routing.
8+
9+
Locks the observable contract of the ``preprocess_inputs`` -> ``forward``
10+
pipeline: for a folded (batch-flattened) dataloader batch, the position
11+
tensor that reaches every transformer layer is the ``(num_tokens, 3)``
12+
``mrope_positions`` when present, else the plain 1D ``(num_tokens,)``
13+
``positions`` -- while attention masks are always built from the 1D
14+
``positions``.
15+
16+
These go through the public ``preprocess_inputs`` seam and call
17+
``model(inputs, **batch)``, so they stay valid whether the mrope/positions
18+
resolution lives in ``forward`` or in ``preprocess_inputs``.
19+
"""
20+
21+
import unittest
22+
23+
import torch
24+
from torch import nn
25+
26+
27+
def _build_config_modules():
28+
# torchtitan.models.qwen3_5 imports the FLA (flash-linear-attention)
29+
# kernels at module scope. FLA is a triton/CUDA-only optional dependency,
30+
# so skip instead of erroring on environments without it.
31+
try:
32+
from torchtitan.config import ParallelismConfig
33+
from torchtitan.distributed.parallel_dims import ParallelDims
34+
from torchtitan.models.qwen3_5 import model_registry
35+
except ModuleNotFoundError as exc:
36+
raise unittest.SkipTest(
37+
f"Qwen3.5 optional dependency unavailable: {exc.name}"
38+
) from exc
39+
return model_registry, ParallelDims, ParallelismConfig
40+
41+
42+
class _RecordingLayer(nn.Module):
43+
"""Layer stub that records the positions it is handed and passes x through.
44+
45+
The mrope/positions resolution is independent of the layer internals, so
46+
stubbing the layers keeps these tests on CPU without the FLA kernels while
47+
still exercising the real ``preprocess_inputs`` and ``forward`` glue.
48+
"""
49+
50+
def __init__(self, sink: dict):
51+
super().__init__()
52+
self._sink = sink
53+
54+
def forward(self, x, attention_masks=None, positions=None):
55+
self._sink["positions"] = positions
56+
return x
57+
58+
59+
class TestQwen35MRoPEPositions(unittest.TestCase):
60+
def _build_stub_model(self):
61+
model_registry, ParallelDims, ParallelismConfig = _build_config_modules()
62+
# varlen backend keeps mask construction to pure tensor ops (no flex
63+
# compile) so the pipeline runs on CPU.
64+
model = model_registry("debugmodel", attn_backend="varlen").model.build()
65+
sink: dict = {}
66+
for key in list(model.layers.keys()):
67+
model.layers[key] = _RecordingLayer(sink)
68+
parallel_dims = ParallelDims(
69+
dp_replicate=1, dp_shard=1, cp=1, tp=1, pp=1, ep=1, world_size=1
70+
)
71+
# partial_dtensor avoids the spmd_types annotation path, which is
72+
# orthogonal to position routing.
73+
parallelism = ParallelismConfig(spmd_backend="partial_dtensor")
74+
return model, sink, parallel_dims, parallelism
75+
76+
def _run(self, model, parallel_dims, parallelism, input_dict):
77+
inputs, _labels, batch = model.preprocess_inputs(
78+
input_dict,
79+
parallel_dims=parallel_dims,
80+
parallelism=parallelism,
81+
)
82+
model(inputs, **batch)
83+
return batch
84+
85+
def test_text_batch_routes_1d_positions_to_layers(self):
86+
model, sink, parallel_dims, parallelism = self._build_stub_model()
87+
# Folded 1D token stream packing docs of length 3, 2, and 5.
88+
positions = torch.tensor([0, 1, 2, 0, 1, 0, 1, 2, 3, 4], dtype=torch.int32)
89+
input_dict = {
90+
"input": torch.randint(0, 100, (10,)),
91+
"positions": positions.clone(),
92+
"labels": torch.zeros(10),
93+
}
94+
95+
batch = self._run(model, parallel_dims, parallelism, input_dict)
96+
97+
# No mrope: layers see the plain 1D positions.
98+
self.assertTrue(torch.equal(sink["positions"], positions))
99+
# Masks come from the 1D positions.
100+
self.assertEqual(
101+
batch["attention_masks"]["deltanet"].cu_seq_q_host, (0, 3, 5, 10)
102+
)
103+
104+
def test_multimodal_batch_routes_mrope_to_layers(self):
105+
model, sink, parallel_dims, parallelism = self._build_stub_model()
106+
positions = torch.tensor([0, 1, 2, 0, 1, 0, 1, 2, 3, 4], dtype=torch.int32)
107+
# Folded (num_tokens, 3) T/H/W positions whose H/W channels differ from
108+
# the 1D positions, so routing the wrong tensor to the layers is
109+
# detectable.
110+
mrope_positions = torch.stack(
111+
[positions, positions + 7, positions + 13], dim=-1
112+
)
113+
input_dict = {
114+
"input": torch.randint(0, 100, (10,)),
115+
"positions": positions.clone(),
116+
"mrope_positions": mrope_positions.clone(),
117+
"labels": torch.zeros(10),
118+
}
119+
120+
batch = self._run(model, parallel_dims, parallelism, input_dict)
121+
122+
# mrope present: layers see the (num_tokens, 3) mrope positions, not the
123+
# 1D positions.
124+
self.assertEqual(sink["positions"].ndim, 2)
125+
self.assertEqual(sink["positions"].shape[-1], 3)
126+
self.assertTrue(torch.equal(sink["positions"], mrope_positions))
127+
# Masks are still built from the 1D positions, not the mrope positions.
128+
self.assertEqual(
129+
batch["attention_masks"]["deltanet"].cu_seq_q_host, (0, 3, 5, 10)
130+
)
131+
132+
133+
if __name__ == "__main__":
134+
unittest.main()

‎torchtitan/components/validate.py‎

Lines changed: 13 additions & 77 deletions
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,6 @@
1212
import torch
1313
import torch.nn as nn
1414
from torch.distributed.pipelining.schedules import _PipelineSchedule
15-
1615
from torchtitan.components.data import ConcatThenSplitPackingConfig, GrainDataLoader
1716
from torchtitan.components.data.collators import TrainerBatch
1817
from torchtitan.components.data.loader import BaseDataLoader
@@ -21,11 +20,9 @@
2120
from torchtitan.components.tokenizer import BaseTokenizer
2221
from torchtitan.config import Configurable, ParallelismConfig
2322
from torchtitan.distributed import ParallelDims, utils as dist_utils
24-
from torchtitan.distributed.context_parallel import prepare_context_parallel_input
2523
from torchtitan.hf_datasets.text_datasets import DATASETS
26-
from torchtitan.models.common.attention import FlexAttention, VarlenAttention
27-
from torchtitan.models.common.decoder import Decoder
2824
from torchtitan.observability import structured_logger as sl
25+
from torchtitan.protocols.model import BaseModel
2926
from torchtitan.tools import utils
3027

3128
ValidationContext: TypeAlias = Callable[[], AbstractContextManager[None]]
@@ -140,74 +137,6 @@ def __init__(
140137
self.pp_has_first_stage = pp_has_first_stage
141138
self.pp_has_last_stage = pp_has_last_stage
142139

143-
def post_dataloading_process(
144-
self,
145-
input_dict: dict[str, torch.Tensor],
146-
labels: torch.Tensor,
147-
model_parts: list[nn.Module],
148-
) -> tuple[torch.Tensor, torch.Tensor, dict[str, Any]]:
149-
"""
150-
Post-processing hook after data loading and before model forward pass.
151-
152-
This method processes the raw data from the dataloader and prepares it for
153-
the model's forward pass. It separates the main input tensor from auxiliary
154-
inputs and constructs additional keyword arguments (e.g., attention masks).
155-
156-
Args:
157-
input_dict: Dictionary containing tensors from the dataloader. Must
158-
contain an "input" key with the main input tensor. May contain
159-
additional keys for auxiliary inputs (e.g., position ids).
160-
labels: Target labels for the batch.
161-
model_parts: List of model parts for accessing model methods.
162-
163-
Returns:
164-
A tuple of (inputs, labels, extra_kwargs) where:
165-
- inputs: Main input tensor extracted from input_dict["input"].
166-
- labels: Target labels (potentially modified by CP sharding).
167-
- extra_kwargs: Additional keyword arguments for the model forward
168-
(e.g. positions, attention_masks), forwarded to every
169-
pipeline-parallel stage.
170-
"""
171-
inputs = input_dict["input"]
172-
extra_kwargs: dict[str, Any] = {
173-
k: v for k, v in input_dict.items() if k != "input"
174-
}
175-
176-
# TODO: deduplicate with Trainer.post_dataloading_process which has
177-
# the same logic; extract a shared function to prevent further drift.
178-
# The dataloader always provides per-document positions, which drive
179-
# both RoPE and block_causal attention masking.
180-
model_config = getattr(model_parts[0], "config", None)
181-
182-
positions = extra_kwargs.get("positions", None)
183-
# positions and attention_masks are optional (Decoder.forward defaults
184-
# both to None). Build masks only for the masked backends (Flex/Varlen),
185-
# which is where get_attention_masks is defined. A maskless backend (the
186-
# SDPA config used by the graph_trainer tests) still receives positions
187-
# for RoPE but no masks — it relies on is_causal instead.
188-
if isinstance(model_config, Decoder.Config) and positions is not None:
189-
attention_backend = model_config.first_full_attention_backend
190-
if isinstance(
191-
attention_backend, (FlexAttention.Config, VarlenAttention.Config)
192-
):
193-
model = cast(Decoder, model_parts[0])
194-
extra_kwargs["attention_masks"] = model.get_attention_masks(
195-
positions=positions,
196-
)
197-
198-
if self.parallel_dims.cp_enabled:
199-
inputs, labels, extra_kwargs = prepare_context_parallel_input(
200-
inputs,
201-
labels,
202-
extra_kwargs,
203-
self.parallel_dims.get_mesh("cp"),
204-
inputs.device,
205-
self.parallelism.context_parallel_load_balancer,
206-
self.parallelism.context_parallel_ptrr_mask_key,
207-
)
208-
209-
return inputs, labels, extra_kwargs
210-
211140
@sl.log_trace_span("eval")
212141
@torch.no_grad()
213142
def validate(
@@ -283,8 +212,12 @@ def validate(
283212
)
284213

285214
for input_dict, labels in microbatches:
286-
inputs, labels, extra_kwargs = self.post_dataloading_process(
287-
input_dict, labels, model_parts
215+
inputs, labels, extra_kwargs = cast(
216+
BaseModel, model_parts[0]
217+
).preprocess_inputs(
218+
{**input_dict, "labels": labels},
219+
parallel_dims=self.parallel_dims,
220+
parallelism=self.parallelism,
288221
)
289222
if self.pp_has_first_stage:
290223
arg_mbs.append((inputs,))
@@ -312,9 +245,12 @@ def validate(
312245
else:
313246
assert len(microbatches) == 1
314247
input_dict, labels = microbatches[0]
315-
# Process data (extract inputs, handle attention masks, CP sharding)
316-
inputs, labels, extra_kwargs = self.post_dataloading_process(
317-
input_dict, labels, model_parts
248+
inputs, labels, extra_kwargs = cast(
249+
BaseModel, model_parts[0]
250+
).preprocess_inputs(
251+
{**input_dict, "labels": labels},
252+
parallel_dims=self.parallel_dims,
253+
parallelism=self.parallelism,
318254
)
319255
with self.validation_context():
320256
assert len(model_parts) == 1

0 commit comments

Comments
 (0)