|
72 | 72 | "aclnn_input_scale_reciprocal", |
73 | 73 | "aclnn_input_offset", |
74 | 74 | ) |
75 | | -O_PROJ_INPUT_SHARDED_QUANT_PARAMS = ("weight_scale_second", "weight_scale") |
76 | 75 |
|
77 | 76 |
|
78 | 77 | def _get_indexer_types(configs: tuple[Any, ...]) -> Any | None: |
@@ -581,8 +580,12 @@ def __init__( |
581 | 580 | # Enable layer sharding via DSA-CP on the P node in the PD-disaggregated setup. |
582 | 581 | self.enable_dsa_cp_with_layer_shard = enable_dsa_cp_with_layer_shard() |
583 | 582 |
|
584 | | - # use original TP o_proj weight in PD mix stage, and full gather |
585 | | - # for o_proj weight for prefill stage. |
| 583 | + # SFA DSA-CP mixed deployments keep o_proj in the existing TP layout. |
| 584 | + # Decode can use the TP-sharded o_proj directly after an activation |
| 585 | + # all-to-all, while prefill/mixed batches temporarily gather the TP |
| 586 | + # shards into a full-weight buffer because their SFA output is not |
| 587 | + # TP-sharded. This is part of the DSA-CP mixed-mode data path rather |
| 588 | + # than an independent user-facing feature switch. |
586 | 589 | self.enable_dsa_cp_with_o_proj_tp = enable_dsa_cp_with_o_proj_tp() |
587 | 590 |
|
588 | 591 | if self.enable_dsa_cp: |
@@ -657,7 +660,7 @@ def process_weights_after_loading(self, act_dtype: torch.dtype): |
657 | 660 | for layer in self.layer_sharding_kwargs or []: |
658 | 661 | if is_hidden_layer(layer): |
659 | 662 | post_process_after_loading_for_shard_weight_series(layer) |
660 | | - else: |
| 663 | + elif self.enable_dsa_cp_with_o_proj_tp: |
661 | 664 | self._init_o_proj_tp_full_params() |
662 | 665 |
|
663 | 666 | if self.enable_mlapo: |
@@ -861,12 +864,20 @@ def rope_single( |
861 | 864 |
|
862 | 865 | def _init_o_proj_tp_full_params(self): |
863 | 866 | """ |
864 | | - Initialize TP-mode and Full-mode parameters for o_proj weight, |
865 | | - preparing for weight switching in PD mix stage. |
866 | | -
|
867 | | - For PD mix stage: |
868 | | - - Use original TP o_proj weight for decode phase |
869 | | - - Need full-gather o_proj weight from all TP ranks for prefill phase |
| 867 | + Initialize TP-mode aliases and Full-mode buffers for DSA-CP o_proj. |
| 868 | +
|
| 869 | + In SFA DSA-CP mixed execution, the same model instance can run both |
| 870 | + decode-only and prefill/mixed batches: |
| 871 | + - Decode-only batches all-to-all the SFA output in the TP group, then |
| 872 | + run the original TP-sharded o_proj. |
| 873 | + - Prefill/mixed batches produce SFA output that is not directly |
| 874 | + compatible with TP-sharded o_proj, so each rank all-gathers the TP |
| 875 | + o_proj shards and input-sharded quant params before running o_proj. |
| 876 | +
|
| 877 | + The original TP parameter storage remains the persistent source of |
| 878 | + truth. The o_proj_tp_* tensors below alias that storage, while the |
| 879 | + o_proj_full_* tensors are temporary gather destinations reused across |
| 880 | + forwards. They are not a second persistent copy of the TP weight. |
870 | 881 | """ |
871 | 882 | sample = self.o_proj.weight |
872 | 883 | self.o_proj_full_weight_gather_dim = 1 if self._is_o_proj_unquantized() else 0 |
@@ -895,36 +906,41 @@ def _init_o_proj_tp_full_params(self): |
895 | 906 | else: |
896 | 907 | self.o_proj_full_pool = self.o_proj_full_gather_pool.transpose(0, 1) |
897 | 908 |
|
898 | | - # Save TP-mode parameters (original sharded weights) |
899 | | - self.o_proj_tp_weight = self.o_proj.weight.clone().detach() |
| 909 | + # TP tensors alias the original parameter storage. The TP shard remains |
| 910 | + # the single source of truth; full-weight tensors below are temporary |
| 911 | + # gather destinations only. |
| 912 | + self.o_proj_tp_weight = self.o_proj.weight.detach() |
900 | 913 | if self.o_proj_full_weight_gather_dim == 0: |
901 | 914 | self.o_proj_tp_weight_gather_input = self.o_proj_tp_weight |
902 | 915 | else: |
| 916 | + # Communication scratch only: all_gather_into_tensor concatenates on |
| 917 | + # dim0, while unquantized row-parallel o_proj is sharded on dim1. |
903 | 918 | self.o_proj_tp_weight_gather_input = self.o_proj_tp_weight.transpose(0, 1).contiguous() |
904 | 919 | self.o_proj_tp_aclnn_input_params = {} |
905 | 920 | self.o_proj_full_aclnn_input_params = {} |
906 | 921 | for param_name in O_PROJ_ACLNN_INPUT_PARAMS: |
907 | 922 | param = getattr(self.o_proj, param_name, None) |
908 | 923 | if param is None: |
909 | 924 | continue |
910 | | - self.o_proj_tp_aclnn_input_params[param_name] = param.clone().detach() |
| 925 | + self.o_proj_tp_aclnn_input_params[param_name] = param.detach() |
911 | 926 | self.o_proj_full_aclnn_input_params[param_name] = param.repeat(self.tp_size) |
912 | 927 |
|
913 | 928 | self.o_proj_tp_input_sharded_quant_params = {} |
914 | 929 | self.o_proj_full_input_sharded_quant_params = {} |
915 | | - for param_name in O_PROJ_INPUT_SHARDED_QUANT_PARAMS: |
916 | | - param = getattr(self.o_proj, param_name, None) |
917 | | - if param is None or getattr(param, "input_dim", None) != 1: |
918 | | - continue |
919 | | - self.o_proj_tp_input_sharded_quant_params[param_name] = param.clone().detach() |
| 930 | + for param_name, param in self._iter_o_proj_input_sharded_quant_params(): |
| 931 | + self.o_proj_tp_input_sharded_quant_params[param_name] = param.detach() |
920 | 932 | self.o_proj_full_input_sharded_quant_params[param_name] = torch.empty( |
921 | 933 | (param.shape[0] * self.tp_size, *param.shape[1:]), dtype=param.dtype, device=param.device |
922 | 934 | ) |
923 | 935 |
|
924 | | - # Initially switch to TP mode for graph capture |
925 | | - self.o_proj.weight.set_(self.o_proj_tp_weight) |
926 | | - self._switch_o_proj_params(self.o_proj_tp_aclnn_input_params) |
927 | | - self._switch_o_proj_params(self.o_proj_tp_input_sharded_quant_params) |
| 936 | + def _iter_o_proj_input_sharded_quant_params(self): |
| 937 | + if not isinstance(self.o_proj, nn.Module): |
| 938 | + return |
| 939 | + for param_name, param in self.o_proj.named_parameters(recurse=False): |
| 940 | + if param_name == "weight" or param_name in O_PROJ_ACLNN_INPUT_PARAMS: |
| 941 | + continue |
| 942 | + if getattr(param, "input_dim", None) == 1: |
| 943 | + yield param_name, param |
928 | 944 |
|
929 | 945 | def _switch_o_proj_params(self, params: dict[str, torch.Tensor]): |
930 | 946 | for param_name, param in params.items(): |
@@ -960,15 +976,13 @@ def _handle_o_proj_weight_switch_and_forward( |
960 | 976 | if handle is not None: |
961 | 977 | handle.wait() |
962 | 978 |
|
963 | | - # Switch o_proj to Full-mode (gathered weight from all TP ranks) |
| 979 | + # Temporarily switch o_proj to the gathered full-weight view for |
| 980 | + # prefill/mixed DSA-CP, whose attention output is not TP-sharded. |
964 | 981 | self.o_proj.weight.set_(self.o_proj_full_pool) |
965 | 982 | self._switch_o_proj_params(self.o_proj_full_aclnn_input_params) |
966 | 983 | self._switch_o_proj_params(self.o_proj_full_input_sharded_quant_params) |
967 | | - |
968 | | - # Apply quantization method and execute forward computation |
969 | 984 | output[...] = self._apply_o_proj_full_weight(attn_output) |
970 | | - |
971 | | - # Switch o_proj back to TP-mode for subsequent decode operations |
| 985 | + # Restore TP aliases so later decode batches keep using TP storage. |
972 | 986 | self.o_proj.weight.set_(self.o_proj_tp_weight) |
973 | 987 | self._switch_o_proj_params(self.o_proj_tp_aclnn_input_params) |
974 | 988 | self._switch_o_proj_params(self.o_proj_tp_input_sharded_quant_params) |
@@ -1312,8 +1326,8 @@ def forward( |
1312 | 1326 | # all-gather o_proj weight for prefill stage of PD mix node |
1313 | 1327 | o_proj_full_handle = None |
1314 | 1328 | o_proj_full_param_handles = None |
1315 | | - # if is PD mix stage, using original TP o_proj weight, and also need to full gather for o_proj |
1316 | | - # weight for prefill stage. |
| 1329 | + # Prefill/mixed DSA-CP computes o_proj with a temporary full weight. |
| 1330 | + # Decode keeps the original TP path and only exchanges activations. |
1317 | 1331 | full_gather_o_proj_enabled = self.enable_dsa_cp_with_o_proj_tp and attn_metadata.attn_state not in { |
1318 | 1332 | AscendAttentionState.DecodeOnly, |
1319 | 1333 | AscendAttentionState.SpecDecoding, |
@@ -1621,9 +1635,9 @@ def forward( |
1621 | 1635 | ) |
1622 | 1636 |
|
1623 | 1637 | if self.enable_dsa_cp_with_o_proj_tp: |
1624 | | - # When using SFA-CP with pd mixed, o_proj has two cases: |
1625 | | - # 1. prefill: o_proj is a TP weight, we need to all-gather o_proj weight to switch TP=1. |
1626 | | - # 2. decode: all-to-all the hidden_state before the o_proj forward. |
| 1638 | + # SFA DSA-CP mixed mode keeps o_proj weight sharded in the TP domain: |
| 1639 | + # 1. prefill/mixed: gather TP shards into a temporary full weight. |
| 1640 | + # 2. decode-only: all-to-all hidden states, then run TP o_proj. |
1627 | 1641 | result, require_o_proj_forward = self._handle_o_proj_weight_switch_and_forward( |
1628 | 1642 | attn_output=attn_output, |
1629 | 1643 | output=output, |
|
0 commit comments