distributed_fused_lamb.py replace c_allreduce_sum (#70737)

Author: co63oc

python/paddle/incubate/optimizer/distributed_fused_lamb.py

@@ -89,10 +89,13 @@ def init_communicator(block, rank, ranks, ring_id):
         type='fill_constant', outputs={'Out': tmp_var}, attrs={'value': 1}
     )
     block.append_op(
-        type='c_allreduce_sum',
-        inputs={'X': tmp_var},
-        outputs={'Out': tmp_var},
-        attrs={'ring_id': ring_id, 'use_calc_stream': True},
+        type='all_reduce',
+        inputs={'x': tmp_var},
+        outputs={'out': tmp_var},
+        attrs={
+            'ring_id': ring_id,
+            'reduce_type': paddle.distributed.ReduceOp.SUM,
+        },
     )
     block.append_op(
         type='c_sync_calc_stream',

python/paddle/incubate/optimizer/pipeline.py

@@ -165,17 +165,17 @@ def _insert_allreduce_op(self, op_idx, block):
             offset += 1
         block._insert_op(
             op_idx + 1 + offset,
-            type=(
-                'c_allreduce_max'
-                if op.type == "reduce_any"
-                else 'c_allreduce_sum'
-            ),
-            inputs={'X': temp_var if op.type == "reduce_any" else out_var},
-            outputs={'Out': temp_var if op.type == "reduce_any" else out_var},
+            type='all_reduce',
+            inputs={'x': temp_var if op.type == "reduce_any" else out_var},
+            outputs={'out': temp_var if op.type == "reduce_any" else out_var},
             attrs={
                 'ring_id': self.global_ring_id,
                 self._op_role_key: self._op_role.Optimize,
-                'use_calc_stream': True,
+                'reduce_type': (
+                    paddle.distributed.ReduceOp.MAX
+                    if op.type == "reduce_any"
+                    else paddle.distributed.ReduceOp.SUM
+                ),
             },
         )
         offset += 1