Repository navigation
🚨 TP: give embeddings their own style, shard input embeddings by default, and shard widths that don't divide by the TP size #49570
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Changes from all commits
10a27d8
eeeb479
ebf818b
3bbb894
94d8f78
71929e4
087c77f
0f4c42d
11051a6
1fa6155
0b17870
d702e02
ad65de5
3ae95ff
549aae4
73370f9
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -170,6 +170,15 @@ def context_around_forward(self, module, mesh): | |
| def transform_output_post_forward(self, module, output, mesh): | ||
| return output | ||
|
|
||
| @staticmethod | ||
| def _from_local_last_dim_shard(local, mesh, size): | ||
| """Create a `DTensor` for a tensor sharded across its last dimension. | ||
| The local shard can have a different size from the other ranks when `size` is not divisible by the TP size.""" | ||
| shape = (*local.shape[:-1], size) | ||
| return DTensor.from_local( | ||
| local, mesh, [Shard(-1)], run_check=False, shape=shape, stride=torch.empty(shape, device="meta").stride() | ||
| ) | ||
|
|
||
| def install_forward(self, module, mesh): | ||
| """Install pre / around / post transforms by replacing module.forward.""" | ||
| original_forward = module.forward | ||
|
|
@@ -185,7 +194,9 @@ def tp_forward(*args, **kwargs): | |
|
|
||
|
|
||
| class ColwiseParallel(TensorParallelLayer): | ||
| """Column-wise: weight & bias → Shard(0) (Embedding: Shard(1)); input replicated, output Shard(-1).""" | ||
| """Column-wise for linear layers: weight & bias → Shard(0); input replicated, output Shard(-1). | ||
|
|
||
| The output width (`out_features`) need not divide by the TP size: the last rank then holds a shorter slice.""" | ||
|
|
||
| def __init__(self, *, input_layouts=None, output_layouts=None, use_local_output: bool = True): | ||
| self.input_layouts = input_layouts or Replicate() | ||
|
|
@@ -197,28 +208,16 @@ def should_use_local_tensors(self, module): | |
| uses_local_inference_kernel = isinstance(module, torch.nn.Linear) and not torch.is_grad_enabled() | ||
| return use_local_quantized_path or uses_local_inference_kernel | ||
|
|
||
| def validate_param(self, module, param, mesh, parameter_name=None): | ||
| meta = module._parameters.get(param) | ||
| gathers_output = isinstance(self.output_layouts, Replicate) | ||
| if meta is None or not gathers_output: | ||
| return | ||
|
|
||
| shard_dim = 1 if isinstance(module, torch.nn.Embedding) else meta.ndim - 2 | ||
| output_size = meta.shape[shard_dim] | ||
| tp_size = mesh.size() | ||
| if output_size % tp_size != 0: | ||
| parameter_name = parameter_name or param | ||
| layer_name = parameter_name.rsplit(".", 1)[0] | ||
| raise ValueError( | ||
| f"The output size of `{layer_name}` ({output_size}) must be divisible by the tensor parallel size " | ||
| f"({tp_size}) when gathering a colwise output." | ||
| ) | ||
| def install_forward(self, module, mesh): | ||
| if not hasattr(module, "out_features"): | ||
| raise ValueError(f"{type(module).__name__} has no `out_features`: colwise shards linear layers.") | ||
| return super().install_forward(module, mesh) | ||
|
|
||
| def shard_param(self, module, param, mesh): | ||
| meta = module._parameters.get(param) | ||
| if meta is None: | ||
| return | ||
| placement = Shard(1) if isinstance(module, torch.nn.Embedding) else Shard(meta.ndim - 2) | ||
| placement = Shard(meta.ndim - 2) | ||
| module._parameters[param] = torch.nn.Parameter( | ||
| distribute_tensor(meta, mesh, [placement], src_data_rank=None), | ||
| requires_grad=meta.requires_grad, | ||
|
|
@@ -261,18 +260,19 @@ def transform_output_post_forward(self, module, output, mesh): | |
| if self.should_use_local_tensors(module) and self.use_local_output and output_is_local_shard: | ||
| return output | ||
| if not isinstance(output, DTensor): | ||
| output = DTensor.from_local(output, mesh, [Shard(-1)], run_check=False) | ||
| # `out_features`, not the weight's shape: a packed or quantized weight doesn't show the full width | ||
| output = self._from_local_last_dim_shard(output, mesh, module.out_features) | ||
| if output.placements != (self.output_layouts,): | ||
| output = output.redistribute(placements=[self.output_layouts]) | ||
| return output.to_local() if self.use_local_output else output | ||
|
|
||
|
|
||
| class RowwiseParallel(TensorParallelLayer): | ||
| """Row-wise: weight → Shard(1), bias → Replicate (Embedding: weight → Shard(0)). | ||
| """Row-wise for linear layers: weight → Shard(1), bias → Replicate; input sharded on the last dim. | ||
|
|
||
| Linear input is sharded on the last dim; Embedding input is replicated. The module | ||
| forward produces a Partial output which the boundary redistribute reduces to | ||
| output_layouts (Replicate → allreduce, Shard(1) → reduce-scatter). | ||
| The module forward produces a Partial output which the boundary redistribute reduces to | ||
| output_layouts (Replicate → allreduce, Shard(1) → reduce-scatter). The input width (`in_features`) need not | ||
| divide by the TP size. | ||
| """ | ||
|
|
||
| def __init__(self, *, input_layouts=None, output_layouts=None, use_local_output: bool = True): | ||
|
|
@@ -289,27 +289,32 @@ def shard_param(self, module, param, mesh): | |
| meta = module._parameters.get(param) | ||
| if meta is None: | ||
| return | ||
| if isinstance(module, torch.nn.Embedding): | ||
| placement = Shard(0) | ||
| else: | ||
| # bias is replicated (added after the row-reduce); weight shards on input dim (-1) | ||
| placement = Replicate() if param == "bias" else Shard(-1) | ||
| # bias is replicated (added after the row-reduce); weight shards on input dim (-1) | ||
| placement = Replicate() if param == "bias" else Shard(-1) | ||
| module._parameters[param] = torch.nn.Parameter( | ||
| distribute_tensor(meta, mesh, [placement], src_data_rank=None), | ||
| requires_grad=meta.requires_grad, | ||
| ) | ||
|
|
||
| def install_forward(self, module, mesh): | ||
| if not hasattr(module, "in_features"): | ||
| raise ValueError(f"{type(module).__name__} has no `in_features`: rowwise shards linear layers.") | ||
| return super().install_forward(module, mesh) | ||
|
|
||
| def transform_inputs_pre_forward(self, module, args, kwargs, mesh): | ||
| # Embedding runtime sharding needs a replicated input; Linear needs Shard(-1). | ||
| desired = Replicate() if isinstance(module, torch.nn.Embedding) else Shard(-1) | ||
| desired = Shard(-1) | ||
| x = args[0] | ||
| # A local kernel can use a plain input when the previous layer already split it. | ||
| # avoid a redundant Tensor -> DTensor -> Tensor round trip. | ||
| input_has_desired_layout = self.input_layouts == desired | ||
| if self.should_use_local_tensors(module) and input_has_desired_layout and not isinstance(x, DTensor): | ||
| return args, kwargs | ||
| if not isinstance(x, DTensor): | ||
| x = DTensor.from_local(x, mesh, [self.input_layouts], run_check=False) | ||
| if isinstance(self.input_layouts, Shard): | ||
| # the input is this rank's slice of the module's declared `in_features` | ||
| x = self._from_local_last_dim_shard(x, mesh, module.in_features) | ||
| else: | ||
| x = DTensor.from_local(x, mesh, [self.input_layouts], run_check=False) | ||
| if x.placements != (desired,): | ||
| x = x.redistribute(placements=[desired]) | ||
| if self.should_use_local_tensors(module): | ||
|
|
@@ -360,6 +365,50 @@ def transform_output_post_forward(self, module, output, mesh): | |
| return output | ||
|
|
||
|
|
||
| class EmbeddingParallel(TensorParallelLayer): | ||
|
Collaborator
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. ❤️ exactly what we needed for al ong time ty! |
||
| """Embedding with the token ids replicated on every rank. `shard_dim=0` is vocab-parallel: each rank holds a slice | ||
| of the vocabulary, and the ranks' partial lookups are summed. `shard_dim=1` shards the hidden dim: each rank looks | ||
| up a slice of every vector, and the slices are gathered. The sharded dim need not divide by the TP size.""" | ||
|
|
||
| def __init__(self, *, shard_dim: int = 0, use_local_output: bool = True): | ||
| self.shard_dim = shard_dim | ||
| self.use_local_output = use_local_output | ||
|
|
||
| def should_use_local_tensors(self, module): | ||
| return getattr(module, "_hf_quantized_needs_local_tp", False) | ||
|
|
||
| def shard_param(self, module, param, mesh): | ||
| meta = module._parameters.get(param) | ||
| if meta is None: | ||
| return | ||
| module._parameters[param] = torch.nn.Parameter( | ||
| distribute_tensor(meta, mesh, [Shard(self.shard_dim)], src_data_rank=None), | ||
| requires_grad=meta.requires_grad, | ||
| ) | ||
|
|
||
| def transform_inputs_pre_forward(self, module, args, kwargs, mesh): | ||
| x = args[0] | ||
| if self.should_use_local_tensors(module) and not isinstance(x, DTensor): | ||
| return args, kwargs | ||
| if not isinstance(x, DTensor): | ||
| x = DTensor.from_local(x, mesh, [Replicate()], run_check=False) | ||
| elif x.placements != (Replicate(),): | ||
| x = x.redistribute(placements=[Replicate()]) | ||
| if self.should_use_local_tensors(module): | ||
| x = x.to_local() | ||
| return (x,) + args[1:], kwargs | ||
|
|
||
| def transform_output_post_forward(self, module, output, mesh): | ||
| if not isinstance(output, DTensor): | ||
| if self.shard_dim == 0: | ||
| output = DTensor.from_local(output, mesh, [Partial()], run_check=False) | ||
| else: | ||
| output = self._from_local_last_dim_shard(output, mesh, module.embedding_dim) | ||
| if output.placements != (Replicate(),): | ||
| output = output.redistribute(placements=[Replicate()]) | ||
| return output.to_local() if self.use_local_output else output | ||
|
|
||
|
|
||
| class ReplicatedWithGradAllReduce(TensorParallelLayer): | ||
| """Replicated parameter whose gradient is partial. | ||
|
|
||
|
|
@@ -945,7 +994,8 @@ class ParallelInterface(GeneralInterface): | |
|
|
||
| _global_mapping = ( | ||
| { | ||
| "embedding_rowwise": RowwiseParallel(input_layouts=Replicate(), output_layouts=Replicate()), | ||
| "embedding_rowwise": EmbeddingParallel(shard_dim=0), | ||
| "embedding_colwise": EmbeddingParallel(shard_dim=1), | ||
| "colwise_gather_output": ColwiseParallel(input_layouts=Replicate(), output_layouts=Replicate()), | ||
| "colwise_rep": ColwiseParallel(input_layouts=Replicate(), output_layouts=Replicate()), | ||
| "colwise": ColwiseParallel(input_layouts=Replicate(), output_layouts=Shard(-1)), | ||
|
|
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -256,15 +256,15 @@ class Gemma4VisionConfig(PreTrainedConfig): | |
|
|
||
| model_type = "gemma4_vision" | ||
| base_model_tp_plan = { | ||
| "encoder.layers.*.self_attn.q_proj": "colwise", | ||
| "encoder.layers.*.self_attn.k_proj": "colwise", | ||
| "encoder.layers.*.self_attn.v_proj": "colwise", | ||
| "encoder.layers.*.self_attn.q_proj.linear": "colwise", | ||
|
Collaborator
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. ow what?
Member
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. yep 🥲 q_proj is a Gemma4ClippableLinear wrapper; the actual nn.Linear and weight live at q_proj.linear
Member
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
Collaborator
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. was not deeply involved in gemma4 but wow indeedd.. super weird 😢
Collaborator
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. yeah I am surprised it even worked before
Member
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. we will be able to catch those errors/regression soon with the distributed CI dashboard
Member
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
yeh normally we have a test that checks for plans matching params/modules, will check how it got past it. the encoder.layers.*.self_attn.q_proj is a module but now with the distinct linear/embedding colwise/rowwise plans we can check each plan is used with the class it's meant for for example. |
||
| "encoder.layers.*.self_attn.k_proj.linear": "colwise", | ||
| "encoder.layers.*.self_attn.v_proj.linear": "colwise", | ||
| "encoder.layers.*.self_attn.q_norm": "replicated_with_grad_allreduce", | ||
| "encoder.layers.*.self_attn.k_norm": "replicated_with_grad_allreduce", | ||
| "encoder.layers.*.self_attn.o_proj": "rowwise", | ||
| "encoder.layers.*.mlp.gate_proj": "colwise", | ||
| "encoder.layers.*.mlp.up_proj": "colwise", | ||
| "encoder.layers.*.mlp.down_proj": "rowwise", | ||
| "encoder.layers.*.self_attn.o_proj.linear": "rowwise", | ||
| "encoder.layers.*.mlp.gate_proj.linear": "colwise", | ||
| "encoder.layers.*.mlp.up_proj.linear": "colwise", | ||
| "encoder.layers.*.mlp.down_proj.linear": "rowwise", | ||
| } | ||
|
|
||
| default_theta = 100.0 | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
arf yeah would be last resort!
I expect us to move more default plan for defaults so finding a solution is welcome 👁️
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
the explicit solution is to name it in every plan 😆 see conv in #49570 (comment)
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I Know !