Skip to content

Commit 8442714

Browse files
committed
[distributed] Add context parallelism: DistributedConfig(cp_size=N)
`cp_size` splits every sequence across the ranks of a `cp` mesh, which get the same batch and keep a contiguous slice each. The plan follows the other parallelisms: a `base_model_cp_plan` default in `PreTrainedConfig`, a `cp_plan` override merged and expanded like `tp_plan`, and `apply_context_parallelism` installing the plan's styles after TP/EP and before FSDP2. The model keeps taking whole sequences: the first decoder layer cuts `hidden_states`, every layer cuts the rotary embeddings, the model-built mask and positions stay whole, and the loss is the one of the whole sequences. FSDP2 shards over `fsdp` and `cp` together (`fsdp_cp`), and `dp_size` counts the cp ranks. `tp_size`, `pp_size` and `ep_size` are refused with it for now, and so is the `Trainer`, whose dataloader would split the batch across a cp group.
1 parent 318ef65 commit 8442714

17 files changed

Lines changed: 167 additions & 43 deletions

‎src/transformers/configuration_utils.py‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -320,6 +320,9 @@ class PreTrainedConfig(PushToHubMixin, RotaryEmbeddingConfigMixin, Heterogeneous
320320
- **base_model_fsdp_plan** (`dict[Any, str]`) -- A dict that maps sub-modules of a base model to an FSDP2
321321
sharding strategy (e.g. `"free_full_weight"` / `"keep_full_weight"`). Keys can be wildcard module paths
322322
(e.g. `"layers.*"`) or tuples of paths (grouped into a single `fully_shard` call).
323+
- **base_model_cp_plan** (`dict[str, str]`) -- A dict that maps sub-modules of a base model to a context parallel
324+
style, e.g. `"layers.*": "cp_split_sequence"` to keep this rank's slice of the sequence in the decoder layers
325+
and `"layers.*.self_attn": "cp_ulysses"` to give their attention the whole sequence with Ulysses.
323326
- **base_model_pp_plan** (`dict[str, tuple[list[str]]]`) -- A dict that maps child-modules of a base model to a
324327
pipeline parallel plan that enables users to place the child-module on the appropriate device.
325328
@@ -400,6 +403,10 @@ class PreTrainedConfig(PushToHubMixin, RotaryEmbeddingConfigMixin, Heterogeneous
400403
"layers.*": "free_full_weight",
401404
"norm": "keep_full_weight",
402405
}
406+
base_model_cp_plan: ClassVar[dict[str, str]] = {
407+
"layers.*": "cp_split_sequence",
408+
"layers.*.self_attn": "cp_ulysses",
409+
}
403410
base_model_pp_plan: ClassVar[dict[str, Sequence[list[str]]] | None] = None
404411
base_model_ep_plan: ClassVar[dict[str, Sequence[list[str]]] | None] = None
405412
_auto_class: ClassVar[str | None] = None
@@ -1426,6 +1433,7 @@ def _remove_keys_not_serialized(self, d: dict[str, Any]) -> None:
14261433
"base_model_tp_plan",
14271434
"base_model_pp_plan",
14281435
"base_model_fsdp_plan",
1436+
"base_model_cp_plan",
14291437
"distributed_config",
14301438
]:
14311439
d.pop(key_to_remove, None)

‎src/transformers/distributed/configuration_utils.py‎

Lines changed: 38 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -25,7 +25,7 @@
2525
@dataclass
2626
class DistributedConfig:
2727
"""
28-
Configuration for native distributed inference and training with tensor, pipeline, or FSDP2 parallelism.
28+
Configuration for native distributed inference and training with tensor, pipeline, context, or FSDP2 parallelism.
2929
3030
Args:
3131
tp_size (`int`, *optional*):
@@ -59,6 +59,13 @@ class DistributedConfig:
5959
Whether to enable CPU offloading for FSDP2.
6060
fsdp_mixed_precision (`bool`, *optional*, defaults to `False`):
6161
Whether to enable mixed precision for FSDP2.
62+
cp_size (`int`, *optional*):
63+
Number of devices that split each sequence (context parallelism). The ranks of a cp group get the same
64+
batch, and each runs on a contiguous slice of every sequence. Requires `torch>=2.10`.
65+
cp_plan (`dict[str, str]`, *optional*):
66+
Context parallel plan. Leave as `None` to use the model's predefined `base_model_cp_plan` and class-level
67+
`_cp_plan`. Pass a dictionary to override individual rules; unspecified rules are kept, so
68+
`{"model.layers.*.self_attn": "cp_ring"}` only changes the attention style. Applied only when `cp_size > 1`.
6269
pp_size (`int`, *optional*):
6370
Number of devices for pipeline parallelism. If `None` and another parallel mode is set, defaults to 1.
6471
ep_size (`int`, *optional*):
@@ -81,6 +88,8 @@ class DistributedConfig:
8188
fsdp_plan: dict[str, str] | None = None
8289
fsdp_cpu_offload: bool = False
8390
fsdp_mixed_precision: bool = False
91+
cp_size: int | None = None
92+
cp_plan: dict[str, str] | None = None
8493
pp_size: int | None = None
8594
ep_size: int | None = None
8695
ep_plan: dict[str, str] | None = None
@@ -91,7 +100,7 @@ def efsdp_size(self) -> int:
91100
The number of ranks that own the same experts and FSDP-shard them between each other.
92101
Dense and expert parameters are laid out over the same world size:
93102
94-
pp x fsdp x tp == pp x efsdp x ep => efsdp = fsdp x tp // ep
103+
pp x fsdp x cp x tp == pp x efsdp x ep => efsdp = fsdp x cp x tp // ep
95104
96105
Experts are not tensor-parallel, so EP and expert-FSDP together cover all the ranks that
97106
dense parameters split between FSDP and TP. Example with 16 ranks, ep=8:
@@ -101,41 +110,46 @@ def efsdp_size(self) -> int:
101110
efsdp groups : {0,8} {1,9} ... {7,15} each pair holds the same experts, FSDP-sharded
102111
on dim 0 (the expert dim) and all-gathered for compute
103112
104-
- ep == tp : efsdp == fsdp
105-
- ep == fsdp * tp: efsdp == 1, every expert lives whole on a single rank
113+
- ep == tp : efsdp == fsdp * cp
114+
- ep == fsdp * cp * tp: efsdp == 1, every expert lives whole on a single rank
106115
107116
Sharding over `efsdp` is applied by the EP token-dispatch path (`ep_dispatch_experts`).
108117
"""
109-
return self.fsdp_size * self.tp_size // self.ep_size
118+
return self.fsdp_size * self.cp_size * self.tp_size // self.ep_size
110119

111120
@property
112121
def dp_size(self) -> int:
113-
"""The number of ranks that each see a distinct batch: `dp_replicate_size * fsdp_size`."""
114-
return self.dp_replicate_size * self.fsdp_size
122+
"""
123+
The number of ranks FSDP2 reduces gradients over: `dp_replicate_size * fsdp_size * cp_size`. The ranks of a
124+
cp group share one batch, but each processes different tokens of it.
125+
"""
126+
return self.dp_replicate_size * self.fsdp_size * self.cp_size
115127

116128
def __post_init__(self):
117129
self._resolve_parallelism()
118130
self._validate_mesh_config()
119131

120132
def _resolve_parallelism(self):
121133
"""Resolve parallel sizes and legacy EP settings."""
122-
for value in (self.tp_size, self.fsdp_size, self.dp_replicate_size, self.pp_size, self.ep_size):
134+
for value in (self.tp_size, self.fsdp_size, self.dp_replicate_size, self.cp_size, self.pp_size, self.ep_size):
123135
if value is not None and value < 1:
124136
raise ValueError(f"Parallelism sizes must be >= 1, got {value}.")
125137

126138
if self.fsdp_size is None:
127139
self.fsdp_size = 1
128140
if self.dp_replicate_size is None:
129141
self.dp_replicate_size = 1
142+
if self.cp_size is None:
143+
self.cp_size = 1
130144
if self.pp_size is None:
131145
self.pp_size = 1
132146
if self.tp_size is None and self.tp_plan is not None:
133147
world_size = int(os.environ.get("WORLD_SIZE", 1))
134-
other_parallel_size = self.dp_replicate_size * self.fsdp_size * self.pp_size
148+
other_parallel_size = self.dp_replicate_size * self.fsdp_size * self.cp_size * self.pp_size
135149
if world_size % other_parallel_size != 0:
136150
raise ValueError(
137-
f"WORLD_SIZE ({world_size}) must be divisible by dp_replicate_size * fsdp_size * pp_size "
138-
f"({other_parallel_size}) to derive tp_size."
151+
f"WORLD_SIZE ({world_size}) must be divisible by dp_replicate_size * fsdp_size * cp_size * "
152+
f"pp_size ({other_parallel_size}) to derive tp_size."
139153
)
140154
self.tp_size = world_size // other_parallel_size
141155
elif self.tp_size is None:
@@ -165,12 +179,23 @@ def _validate_mesh_config(self):
165179
raise ValueError("`ep_plan` must be a dictionary or None.")
166180
if self.fsdp_plan is not None and not isinstance(self.fsdp_plan, dict):
167181
raise ValueError("`fsdp_plan` must be a dictionary or None.")
182+
if self.cp_plan is not None and not isinstance(self.cp_plan, dict):
183+
raise ValueError("`cp_plan` must be a dictionary or None.")
168184

169185
if self.ep_size > 1:
170186
if self.ep_size % self.tp_size:
171187
raise ValueError("`ep_size` must be a multiple of `tp_size`.")
172-
if (self.fsdp_size * self.tp_size) % self.ep_size:
173-
raise ValueError("`ep_size` must divide `fsdp_size * tp_size`.")
188+
if (self.fsdp_size * self.cp_size * self.tp_size) % self.ep_size:
189+
raise ValueError("`ep_size` must divide `fsdp_size * cp_size * tp_size`.")
190+
191+
if self.cp_size > 1:
192+
if not is_torch_greater_or_equal("2.10"):
193+
raise OSError("Context parallelism requires `torch>=2.10` (`DeviceMesh._unflatten`).")
194+
if unsupported := [name for name in ("pp_size", "tp_size", "ep_size") if getattr(self, name) > 1]:
195+
raise ValueError(
196+
f"Context parallelism (`cp_size={self.cp_size}`) is not supported yet with `{unsupported[0]}`. "
197+
"Use `DistributedConfig(fsdp_size=N, cp_size=M)`."
198+
)
174199

175200
if self.dp_size > 1 and self.pp_size > 1:
176201
raise ValueError(

‎src/transformers/distributed/fsdp.py‎

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -180,16 +180,18 @@ def apply_fully_sharded_data_parallelism(
180180
ep_plan: dict[str, str],
181181
) -> nn.Module:
182182
"""
183-
Apply FSDP2 (fully_shard) to a model: dispatched experts on `efsdp` and the rest of the modules on `fsdp`.
183+
Apply FSDP2 (fully_shard) to a model: dispatched experts on `efsdp` and the rest of the modules on `fsdp`, which
184+
spans `cp` too when context parallelism splits the sequences.
184185
185186
Torch availability, distributed initialization and the version requirement
186187
are asserted upstream by `initialize_distributed_mesh`.
187188
"""
188189
distributed_config = getattr(model.config, "distributed_config", None)
190+
fsdp_dim = "fsdp_cp" if distributed_config is not None and distributed_config.cp_size > 1 else "fsdp"
189191
if distributed_config is not None and distributed_config.dp_replicate_size > 1:
190-
fsdp_mesh = mesh_manager.get_mesh(("dp_replicate", "fsdp"))
192+
fsdp_mesh = mesh_manager.get_mesh(("dp_replicate", fsdp_dim))
191193
else:
192-
fsdp_mesh = mesh_manager.get_mesh("fsdp")
194+
fsdp_mesh = mesh_manager.get_mesh(fsdp_dim)
193195
fsdp_policy_kwargs = _get_fsdp_policy_kwargs(distributed_config)
194196

195197
if distributed_config is not None and distributed_config.ep_size > 1:

‎src/transformers/distributed/mixin.py‎

Lines changed: 58 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,12 @@
3232
consolidate_distributed_checkpoint,
3333
)
3434
from .configuration_utils import DistributedConfig
35+
from .context_parallel import (
36+
ContextParallelAttention,
37+
SplitSequenceParallel,
38+
apply_context_parallelism,
39+
context_parallel_styles,
40+
)
3541
from .expert_parallel import apply_expert_parallelism
3642
from .fsdp import ALL_FSDP_STYLES, _resolve_tied_embed_lm_head_plan, apply_fully_sharded_data_parallelism
3743
from .parallel_utils import ALL_PARALLEL_STYLES, _get_parameter_plan, replace_layer_number_by_wildcard
@@ -67,6 +73,7 @@ class DistributedMixin:
6773
_fsdp_size = None
6874
_pp_plan: dict[str, tuple[str, str]] | None = None
6975
_fsdp_plan: dict[str, str] | None = None
76+
_cp_plan: dict[str, str] | None = None
7077

7178
@property
7279
def is_distributed_loading_by_transformers(self) -> bool:
@@ -83,12 +90,14 @@ def init_parallel_plans(self) -> None:
8390
self._ep_plan = dict(getattr(model_cls, "_ep_plan", None) or {})
8491
self._pp_plan = dict(getattr(model_cls, "_pp_plan", None) or {})
8592
self._fsdp_plan = dict(getattr(model_cls, "_fsdp_plan", None) or {})
93+
self._cp_plan = dict(getattr(model_cls, "_cp_plan", None) or {})
8694

8795
if self.base_model is self:
8896
self._pp_plan.update(self.config.base_model_pp_plan or {})
8997
self._tp_plan.update(self.config.base_model_tp_plan or {})
9098
self._ep_plan.update(self.config.base_model_ep_plan or {})
9199
self._fsdp_plan.update(self.config.base_model_fsdp_plan or {})
100+
self._cp_plan.update(self.config.base_model_cp_plan or {})
92101
# Shard the input embeddings on the vocabulary unless the plan already names them.
93102
if self._tp_plan and isinstance(getattr(self, "embed_tokens", None), torch.nn.Embedding):
94103
self._tp_plan = {"embed_tokens": "embedding_rowwise", **self._tp_plan}
@@ -102,6 +111,8 @@ def init_parallel_plans(self) -> None:
102111
self._pp_plan.update({f"{name}.{k}": v for k, v in plan.copy().items()})
103112
if plan := getattr(module, "_fsdp_plan", None):
104113
self._fsdp_plan.update({f"{name}.{k}": v for k, v in plan.copy().items()})
114+
if plan := getattr(module, "_cp_plan", None):
115+
self._cp_plan.update({f"{name}.{k}": v for k, v in plan.copy().items()})
105116

106117
@property
107118
def tp_plan(self) -> dict[str, str]:
@@ -117,6 +128,11 @@ def ep_plan(self) -> dict[str, str]:
117128
def fsdp_plan(self) -> dict[str, str]:
118129
return self._fsdp_plan
119130

131+
@property
132+
def cp_plan(self) -> dict[str, str]:
133+
"""The full context parallel plan for the model's modules."""
134+
return self._cp_plan
135+
120136
@property
121137
def pp_plan(self) -> dict[str, tuple[str, str]]:
122138
return self._pp_plan
@@ -154,6 +170,17 @@ def fsdp_plan(self, plan: dict[str, str] | None):
154170
self._validate_parallel_plan(plan, ALL_FSDP_STYLES, "fsdp_plan")
155171
self._fsdp_plan = plan
156172

173+
@cp_plan.setter
174+
def cp_plan(self, plan: dict[str, str] | None):
175+
if plan is None:
176+
self._cp_plan = {}
177+
return
178+
if not isinstance(plan, dict):
179+
raise ValueError("Can only set a dictionary as `cp_plan`")
180+
181+
self._validate_parallel_plan(plan, context_parallel_styles(), "cp_plan")
182+
self._cp_plan = plan
183+
157184
@pp_plan.setter
158185
def pp_plan(self, plan: dict[str, tuple[str, str]] | None):
159186
if plan is None:
@@ -211,12 +238,13 @@ def _validate_parallel_plan(
211238
def _expand_parallel_plan(self, plan: dict[str, str]) -> dict[str, str]:
212239
"""Replace the `*` of the plan's rules with the layer numbers of the model, one entry per module or parameter."""
213240
expanded_plan = {}
241+
cp_styles = context_parallel_styles()
214242

215243
for module_name, module in self.named_modules():
216244
module_style = _get_parameter_plan(parameter_name=module_name, plan=plan, is_weight=False)
217245

218-
if module_style in ALL_FSDP_STYLES:
219-
# FSDP: list the module to wrap.
246+
if module_style in ALL_FSDP_STYLES or module_style in cp_styles:
247+
# FSDP2 and context parallelism wrap the module itself.
220248
expanded_plan[module_name] = module_style
221249
else:
222250
# TP/EP: list the module's own parameters (to shard) before the module (to hook).
@@ -232,16 +260,16 @@ def _expand_parallel_plan(self, plan: dict[str, str]) -> dict[str, str]:
232260

233261
def _resolve_parallel_plans(
234262
self, distributed_config: DistributedConfig
235-
) -> tuple[dict[str, str], dict[str, str], dict[str, str]]:
236-
"""Build the final TP, EP and FSDP plans: apply the config overrides, then replace `*` with the layer numbers.
263+
) -> tuple[dict[str, str], dict[str, str], dict[str, str], dict[str, str]]:
264+
"""Build the final TP, EP, FSDP and CP plans: apply the config overrides, then replace `*` by the layer numbers.
237265
238266
Each returned plan maps one concrete module or parameter name to a style, e.g. `model.layers.3.mlp.up_proj`
239267
instead of `model.layers.*.mlp.up_proj`. A module that has an EP rule keeps no TP rule.
240268
"""
241269
module_names = {replace_layer_number_by_wildcard(n) for n, _ in self.named_modules()}
242270
param_names = {replace_layer_number_by_wildcard(n) for n, _ in self.named_parameters()}
243271
names = module_names | param_names
244-
for plan_name in ("tp_plan", "ep_plan", "fsdp_plan"):
272+
for plan_name in ("tp_plan", "ep_plan", "fsdp_plan", "cp_plan"):
245273
override = getattr(distributed_config, plan_name)
246274
if isinstance(override, dict):
247275
plan = getattr(self, plan_name)
@@ -253,8 +281,9 @@ def _resolve_parallel_plans(
253281
ep_plan = dict(self.ep_plan) if distributed_config.ep_size > 1 else {}
254282
# HSDP shards each replica with the FSDP plan.
255283
fsdp_plan = dict(self.fsdp_plan) if distributed_config.dp_size > 1 else {}
284+
cp_plan = dict(self.cp_plan) if distributed_config.cp_size > 1 else {}
256285

257-
for kind, plan in (("tp", tp_plan), ("ep", ep_plan), ("fsdp", fsdp_plan)):
286+
for kind, plan in (("tp", tp_plan), ("ep", ep_plan), ("fsdp", fsdp_plan), ("cp", cp_plan)):
258287
size = distributed_config.dp_size if kind == "fsdp" else getattr(distributed_config, f"{kind}_size")
259288
if size > 1 and not plan:
260289
raise ValueError(
@@ -275,6 +304,7 @@ def _resolve_parallel_plans(
275304
self._validate_parallel_plan(tp_plan, ALL_PARALLEL_STYLES, "tp_plan", names)
276305
self._validate_parallel_plan(ep_plan, ALL_PARALLEL_STYLES, "ep_plan", names)
277306
self._validate_parallel_plan(fsdp_plan, ALL_FSDP_STYLES, "fsdp_plan", names)
307+
self._validate_parallel_plan(cp_plan, context_parallel_styles(), "cp_plan", names)
278308
distributed_config._validate_resolved_ep_plan(ep_plan)
279309

280310
# Tied embed/lm_head weights are wrapped once: fold the head's FSDP rule into the embedding's.
@@ -289,10 +319,24 @@ def _resolve_parallel_plans(
289319
tp_plan = self._expand_parallel_plan(tp_plan)
290320
ep_plan = self._expand_parallel_plan(ep_plan)
291321
fsdp_plan = self._expand_parallel_plan(fsdp_plan)
322+
cp_plan = self._expand_parallel_plan(cp_plan)
292323
# FSDP2 needs a submodule sharded before the module containing it, e.g. `mlp.gate` before its layer.
293324
fsdp_plan = dict(sorted(fsdp_plan.items(), key=lambda item: item[0].count("."), reverse=True))
294325

295-
return tp_plan, ep_plan, fsdp_plan
326+
if distributed_config.cp_size > 1:
327+
styles = [ALL_PARALLEL_STYLES[style] for style in cp_plan.values()]
328+
# a split style on the decoder layers, and an attention style (Ulysses, or another registered one)
329+
for name, kind in (("cp_split_sequence", SplitSequenceParallel), ("cp_ulysses", ContextParallelAttention)):
330+
if not any(isinstance(style, kind) for style in styles):
331+
raise ValueError(
332+
f"`cp_plan` has no `{name}` rule matching a module of {type(self).__name__}. Context "
333+
"parallelism needs `cp_split_sequence` on the decoder layers, to keep this rank's slice of the "
334+
"sequence, and an attention style such as `cp_ulysses` on their attention modules, to give "
335+
"attention the whole sequence. Add them to `base_model_cp_plan` of the model's config, pass "
336+
"`cp_plan`, or set `cp_size=1`."
337+
)
338+
339+
return tp_plan, ep_plan, fsdp_plan, cp_plan
296340

297341
@classmethod
298342
def maybe_distribute_model(
@@ -301,7 +345,9 @@ def maybe_distribute_model(
301345
distributed_config: DistributedConfig | None,
302346
mesh_manager: TransformersDeviceMesh | None,
303347
):
304-
"""Apply pipeline, tensor and expert parallelism, then FSDP2, after model init and before weight loading."""
348+
"""
349+
Apply pipeline, tensor, expert and context parallelism, then FSDP2, after model init and before weight loading.
350+
"""
305351
if mesh_manager is None:
306352
return model
307353

@@ -312,7 +358,7 @@ def maybe_distribute_model(
312358
model._tp_size = distributed_config.tp_size
313359
model._fsdp_size = distributed_config.fsdp_size
314360

315-
tp_plan, ep_plan, fsdp_plan = model._resolve_parallel_plans(distributed_config)
361+
tp_plan, ep_plan, fsdp_plan, cp_plan = model._resolve_parallel_plans(distributed_config)
316362

317363
if distributed_config.pp_size > 1:
318364
model = apply_pipeline_parallelism(model, mesh_manager.get_mesh("pp"))
@@ -333,6 +379,9 @@ def maybe_distribute_model(
333379
# we will have to slice the batch here in order to avoid computing tp_size times the same batch.
334380
model = apply_expert_parallelism(model, ep_mesh, tp_mesh, ep_plan)
335381

382+
if distributed_config.cp_size > 1:
383+
model = apply_context_parallelism(model, mesh_manager.get_mesh("cp"), cp_plan)
384+
336385
if distributed_config.dp_size > 1:
337386
model = apply_fully_sharded_data_parallelism(model, mesh_manager, fsdp_plan, ep_plan)
338387

0 commit comments

Comments
 (0)