[fluid_ops] clean c_allreduce_sum if statement (#72265)

co63oc · web-flow · commit e897a9aa311b · 2025-04-17T10:11:35.000+08:00
* Fix

* Fix

* Fix
diff --git a/python/paddle/distributed/auto_parallel/static/mapper.py b/python/paddle/distributed/auto_parallel/static/mapper.py
@@ -27,7 +27,6 @@
 
 def is_collective_comm_op(op):
     comm_list = [
-        "c_allreduce_sum",
         "all_gather",
         "all_reduce",
         "broadcast",
diff --git a/python/paddle/distributed/auto_parallel/static/operators/common.py b/python/paddle/distributed/auto_parallel/static/operators/common.py
@@ -692,13 +692,8 @@ def is_amp_flag_sync_op(op):
 
 def is_global_norm_sync_op(op):
     return (
-        (
-            op.type == "c_allreduce_sum"
-            or (
-                op.type == "all_reduce"
-                and op.desc.attr("reduce_type") == dist.ReduceOp.SUM
-            )
-        )
+        op.type == "all_reduce"
+        and op.desc.attr("reduce_type") == dist.ReduceOp.SUM
         and op.desc.has_attr("op_namescope")
         and SyncMode.GlobalNormSync in op.desc.attr("op_namescope")
     )
diff --git a/python/paddle/distributed/fleet/meta_optimizers/sharding/prune.py b/python/paddle/distributed/fleet/meta_optimizers/sharding/prune.py
@@ -53,7 +53,6 @@ def _build_deps(
 
         for idx, op in enumerate(self._block.ops):
             if op.type in [
-                "c_allreduce_sum",
                 "c_sync_comm_stream",
                 "c_calc_comm_stream",
                 'all_reduce',
diff --git a/python/paddle/distributed/fleet/meta_optimizers/sharding/utils.py b/python/paddle/distributed/fleet/meta_optimizers/sharding/utils.py
@@ -119,15 +119,11 @@ def check_allreduce_sum(block, shard, sharding_ring_id, dp_ring_id=-1):
     for idx, op in enumerate(block.ops):
         # sharding use both allreduce and reduce to sync grad
         if (
-            op.type == "c_allreduce_sum"
-            or (
-                op.type == "reduce"
-                and op.desc.attr("reduce_type") == dist.ReduceOp.SUM
-            )
-            or (
-                op.type == "all_reduce"
-                and op.desc.attr("reduce_type") == dist.ReduceOp.SUM
-            )
+            op.type == "reduce"
+            and op.desc.attr("reduce_type") == dist.ReduceOp.SUM
+        ) or (
+            op.type == "all_reduce"
+            and op.desc.attr("reduce_type") == dist.ReduceOp.SUM
         ):
             if not op.all_attrs()["use_calc_stream"]:
                 ring_id = op.desc.attr("ring_id")
@@ -168,15 +164,11 @@ def check_allreduce_sum(block, shard, sharding_ring_id, dp_ring_id=-1):
                     dp_grads_status[var_name] = 1
         # check sharding allreduce and  reduce but skip megatron allreduce
         elif (
-            op.type == "c_allreduce_sum"
-            or (
-                op.type == "all_reduce"
-                and op.desc.attr("reduce_type") == dist.ReduceOp.SUM
-            )
-            or (
-                op.type == "reduce"
-                and op.desc.attr("reduce_type") == dist.ReduceOp.SUM
-            )
+            op.type == "all_reduce"
+            and op.desc.attr("reduce_type") == dist.ReduceOp.SUM
+        ) or (
+            op.type == "reduce"
+            and op.desc.attr("reduce_type") == dist.ReduceOp.SUM
         ):
             if not op.all_attrs()["use_calc_stream"]:
                 var_name = op.desc.input_arg_names()[0]
@@ -943,7 +935,7 @@ def comm_analyse(main_program):
             broadcast_vars[var_name] = (
                 get_var_size(block.var(var_name)) * 1024.0
             )
-        elif op.type == "c_allreduce_sum" or (
+        elif (
             op.type == "all_reduce"
             and op.desc.attr("reduce_type") == dist.ReduceOp.SUM
         ):
diff --git a/python/paddle/distributed/fleet/meta_optimizers/sharding_optimizer.py b/python/paddle/distributed/fleet/meta_optimizers/sharding_optimizer.py
@@ -997,9 +997,6 @@ def _prune_main_program(self, block, shard, rings):
             output_names = op.desc.output_arg_names()
             # FIXME(wangxi): need use grads, pipeline grad is @GRAD@MERGE
             if (
-                op.type == "c_allreduce_sum"
-                and op.attr('use_model_parallel') is False
-            ) or (
                 op.type == "all_reduce"
                 and op.attr('reduce_type') == paddle.distributed.ReduceOp.SUM
             ):
@@ -1021,7 +1018,6 @@ def _prune_main_program(self, block, shard, rings):
         # Prune
         for idx, op in reversed(list(enumerate(block.ops))):
             if op.type in [
-                "c_allreduce_sum",
                 "all_reduce",
                 "c_sync_comm_stream",
                 "c_calc_comm_stream",
diff --git a/python/paddle/distributed/passes/allreduce_matmul_grad_overlapping.py b/python/paddle/distributed/passes/allreduce_matmul_grad_overlapping.py
@@ -26,10 +26,10 @@
 
 # For allreduce pattern in the backward phase of column parallel linear:
 #   dX, dY = matmul_grad(X, Y, dOut)
-#   dX = c_allreduce_sum(dX)
+#   dX = all_reduce_sum(dX)
 # Split matmul_grad to 2 matmul:
 #   dX = matmul(dOut, Y^T)
-#   dX = c_allreduce_sum(dX)
+#   dX = all_reduce_sum(dX)
 #   dY = matmul(X^T, dOut)
 #
 # Then the all_reduce sum can overlap with the compute of dY.
diff --git a/python/paddle/distributed/passes/auto_parallel_data_parallel_optimization.py b/python/paddle/distributed/passes/auto_parallel_data_parallel_optimization.py
@@ -435,7 +435,6 @@ def _update_program(self, grad_groups):
 
         remove_op_types = [
             'scale',
-            'c_allreduce_sum',
             'all_reduce',
             'c_wait_compute',
         ]
@@ -490,13 +489,11 @@ def _update_program(self, grad_groups):
                 )
 
             allreduce_op = block.ops[group.allreduce_op_idx]
-            assert allreduce_op.type in [
-                'c_allreduce_sum',
-            ] or (
+            assert (
                 allreduce_op.type == 'all_reduce'
                 and allreduce_op.attr('reduce_type')
                 == paddle.distributed.ReduceOp.SUM
-            ), f"should found c_allreduce_sum op but found {allreduce_op}"
+            ), f"should found all_reduce sum op but found {allreduce_op}"
             allreduce_op_dist_attr = (
                 self.dist_context.get_op_dist_attr_for_program(allreduce_op)
             )
@@ -748,7 +745,7 @@ def add(self, grad_var, ring_id, i):
         if len(self.gradients) == 1:
             # TODO Remove this is a temporary hack for Tensor Parallel. the logic
             # for find grad_op should be more general.
-            if self.ops[grad_op_idx].type == "c_allreduce_sum" or (
+            if (
                 self.ops[grad_op_idx].type == "all_reduce"
                 and self.ops[grad_op_idx].attr("reduce_type")
                 == paddle.distributed.ReduceOp.SUM
diff --git a/python/paddle/distributed/passes/auto_parallel_fused_linear_promotion.py b/python/paddle/distributed/passes/auto_parallel_fused_linear_promotion.py
@@ -542,7 +542,7 @@ def _transform_forward_segment(
                     origin_comm_op.input_arg_names[0],
                     origin_matmul_output_new_name,
                 )
-            if origin_comm_op.type == "c_allreduce_sum" or (
+            if (
                 origin_comm_op.type == "all_reduce"
                 and origin_comm_op.attr("reduce_type")
                 == paddle.distributed.ReduceOp.SUM
diff --git a/python/paddle/distributed/passes/auto_parallel_gradient_merge.py b/python/paddle/distributed/passes/auto_parallel_gradient_merge.py
@@ -440,7 +440,7 @@ def _move_reduce_to_optimizer_ops_block(
             reduce_op_desc._set_attr(OP_ROLE_KEY, OpRole.Optimize)
             removed_op_idx.append(idx)
 
-            if op.type == "c_allreduce_sum" or (
+            if (
                 op.type == "reduce"
                 and op.attr("reduce_type") == dist.ReduceOp.SUM
             ):
diff --git a/python/paddle/distributed/passes/auto_parallel_sequence_parallel_optimization.py b/python/paddle/distributed/passes/auto_parallel_sequence_parallel_optimization.py
@@ -84,12 +84,9 @@ def is_valid_split_op(idx, block):
                 return False
             pre_op = block.ops[idx - 1]
             if not (
-                pre_op.type == "c_allreduce_sum"
-                or (
-                    pre_op.type == "all_reduce"
-                    and pre_op.attr("reduce_type")
-                    == paddle.distributed.ReduceOp.SUM
-                )
+                pre_op.type == "all_reduce"
+                and pre_op.attr("reduce_type")
+                == paddle.distributed.ReduceOp.SUM
             ):
                 return False
             pre_output_name = pre_op.output_arg_names[0]
diff --git a/python/paddle/distributed/passes/auto_parallel_sharding.py b/python/paddle/distributed/passes/auto_parallel_sharding.py
@@ -536,15 +536,11 @@ def _shard_gradient_synchronization(self, main_block):
         for idx, op in reversed(list(enumerate(main_block.ops))):
             if _is_param_grad_allreduce_op(op, main_block):
                 if (
-                    op.type == "c_allreduce_sum"
-                    or (
-                        op.type == "all_reduce"
-                        and op.attr("reduce_type") == dist.ReduceOp.SUM
-                    )
-                    or (
-                        op.type == "reduce"
-                        and op.attr("reduce_type") == dist.ReduceOp.SUM
-                    )
+                    op.type == "all_reduce"
+                    and op.attr("reduce_type") == dist.ReduceOp.SUM
+                ) or (
+                    op.type == "reduce"
+                    and op.attr("reduce_type") == dist.ReduceOp.SUM
                 ):
                     reduce_op_type = "reduce"
                     reduce_type = dist.ReduceOp.SUM
@@ -1041,15 +1037,11 @@ def op_depend_on_group(op, group):
                     param_name
                 ):
                     cur_group.is_in_local_shard = True
-                    assert ops[i + 1].type in [
-                        "c_allreduce_sum",
-                    ] or (
-                        ops[i + 1].type == 'all_reduce'
-                        and ops[i + 1].attr('reduce_type')
-                        in [
-                            paddle.distributed.ReduceOp.SUM,
-                        ]
-                    ), "Sharding should reduce grad first and than allreduce if Hybrid Sharding with Data-Parallel"
+                    assert ops[i + 1].type == 'all_reduce' and ops[i + 1].attr(
+                        'reduce_type'
+                    ) in [
+                        paddle.distributed.ReduceOp.SUM,
+                    ], "Sharding should reduce grad first and than allreduce if Hybrid Sharding with Data-Parallel"
                     assert (
                         ops[i + 1].output_arg_names[0] == grad_name
                     ), "Hybrid Sharding with Data-Parallel should sync same gradient var"
@@ -1249,13 +1241,9 @@ def _overlap_grad_comm(
         grad_comm_op_to_stream_idx = {}
         for idx, op in enumerate(ops):
             if is_data_parallel_reduce_op(op):
-                if op.type in ["c_allreduce_sum"] or (
-                    op.type == 'all_reduce'
-                    and op.attr('reduce_type')
-                    in [
-                        paddle.distributed.ReduceOp.SUM,
-                    ]
-                ):
+                if op.type == 'all_reduce' and op.attr('reduce_type') in [
+                    paddle.distributed.ReduceOp.SUM,
+                ]:
                     continue
                 stream_idx = reduce_op_count % self.grad_comm_stream_num
                 grad_comm_op_to_stream_idx[op] = stream_idx
@@ -1308,15 +1296,11 @@ def _overlap_grad_comm(
                 op._set_attr("ring_id", comm_group.id)
                 if self.sharding_hybrid_dp and grad_group.is_in_local_shard:
                     next_op = ops[idx + 1]
-                    assert next_op.type in [
-                        "c_allreduce_sum",
-                    ] or (
-                        next_op.type == 'all_reduce'
-                        and next_op.attr('reduce_type')
-                        in [
-                            paddle.distributed.ReduceOp.SUM,
-                        ]
-                    )
+                    assert next_op.type == 'all_reduce' and next_op.attr(
+                        'reduce_type'
+                    ) in [
+                        paddle.distributed.ReduceOp.SUM,
+                    ]
                     assert next_op.output("Out")[0] == reduce_varname
                     # FIXME hybrid sharding-dp support multi comm & stream in feature
                     # next_op._set_attr("ring_id", comm_group.id)
diff --git a/python/paddle/distributed/passes/fuse_all_reduce.py b/python/paddle/distributed/passes/fuse_all_reduce.py
@@ -148,7 +148,6 @@ def has_same_attrs(op1, op2, attr_names):
 def filter_all_collective_op_indices(block):
     # NOTE: should add more collective ops
     all_collective_ops = {
-        "c_allreduce_sum",
         "c_broadcast",
         "broadcast",
         "all_gather",
diff --git a/python/paddle/distributed/passes/pipeline_scheduler_pass/pipeline_1f1b.py b/python/paddle/distributed/passes/pipeline_scheduler_pass/pipeline_1f1b.py
@@ -233,14 +233,8 @@ def _split_program_for_overlapping(self, job_type, program, split_points):
 
     def is_comm_op_valid_to_overlap(self, op):
         return (
-            (
-                op.type == "c_allreduce_sum"
-                or (
-                    op.type == "all_reduce"
-                    and op.attr("reduce_type")
-                    == paddle.distributed.ReduceOp.SUM
-                )
-            )
+            op.type == "all_reduce"
+            and op.attr("reduce_type") == paddle.distributed.ReduceOp.SUM
             and op.dist_attr.execution_stream
             == AutoParallelStreamType.CALC_STREAM.value
         )
diff --git a/python/paddle/distributed/transpiler/collective.py b/python/paddle/distributed/transpiler/collective.py
@@ -352,7 +352,7 @@ def _insert_allreduce_ops(self):
                         )
                         offset += 1
 
-                    # As we search ops reversely, we should insert c_allreduce_sum
+                    # As we search ops reversely, we should insert all_reduce sum
                     # op in the same way to keep the ring_id alternate
                     ring_id = (ring_id + 1) % self.nrings
                     block._insert_op(
diff --git a/python/paddle/static/amp/fp16_lists.py b/python/paddle/static/amp/fp16_lists.py
@@ -262,7 +262,6 @@ def _update_list(self):
     'fused_bn_add_activation',
     'c_identity',
     'c_concat',
-    'c_allreduce_sum',
     'all_reduce',
     'concat',
     'split',

Original file line number	Diff line number	Diff line change
`@@ -542,7 +542,7 @@ def _transform_forward_segment(`
`542`	`542`	`origin_comm_op.input_arg_names[0],`
`543`	`543`	`origin_matmul_output_new_name,`
`544`	`544`	`)`
`545`		`- if origin_comm_op.type == "c_allreduce_sum" or (`
	`545`	`+ if (`
`546`	`546`	`origin_comm_op.type == "all_reduce"`
`547`	`547`	`and origin_comm_op.attr("reduce_type")`
`548`	`548`	`== paddle.distributed.ReduceOp.SUM`