[fluid_ops] Change c_allgather to all_gather (#70414)

co63oc · web-flow · commit 976fdf7eb444 · 2024-12-24T10:33:49.000+08:00
diff --git a/python/paddle/distributed/fleet/meta_parallel/pp_utils/utils.py b/python/paddle/distributed/fleet/meta_parallel/pp_utils/utils.py
@@ -101,10 +101,8 @@ def _all_gather(tensor, group=None, use_calc_stream=True):
         if group is None
         else group.nranks
     )
-    return _legacy_C_ops.c_allgather(
+    return _legacy_C_ops.all_gather(
         tensor,
-        'use_calc_stream',
-        use_calc_stream,
         'ring_id',
         ring_id,
         'nranks',
diff --git a/python/paddle/distributed/ps/utils/collective_transpiler.py b/python/paddle/distributed/ps/utils/collective_transpiler.py
@@ -631,14 +631,14 @@ def _insert_allgather_ops(self):
                         )
                         offset += 1
 
-                    # As we search ops reversely, we should insert c_allgather
+                    # As we search ops reversely, we should insert all_gather
                     # op in the same way to keep the ring_id alternate
                     ring_id = (ring_id + 1) % self.nrings
                     block._insert_op(
                         offset,
-                        type='c_allgather',
-                        inputs={'X': grad},
-                        outputs={'Out': new_grad_var},
+                        type='all_gather',
+                        inputs={'x': grad},
+                        outputs={'out': new_grad_var},
                         attrs={
                             'nranks': self.allgather_ranks,
                             'ring_id': ring_id,
diff --git a/python/paddle/distributed/transpiler/collective.py b/python/paddle/distributed/transpiler/collective.py
@@ -834,14 +834,14 @@ def _insert_allgather_ops(self):
                         )
                         offset += 1
 
-                    # As we search ops reversely, we should insert c_allgather
+                    # As we search ops reversely, we should insert all_gather
                     # op in the same way to keep the ring_id alternate
                     ring_id = (ring_id + 1) % self.nrings
                     block._insert_op(
                         offset,
-                        type='c_allgather',
-                        inputs={'X': grad},
-                        outputs={'Out': new_grad_var},
+                        type='all_gather',
+                        inputs={'x': grad},
+                        outputs={'out': new_grad_var},
                         attrs={
                             'nranks': self.allgather_ranks,
                             'ring_id': ring_id,
diff --git a/python/paddle/incubate/distributed/models/moe/moe_layer.py b/python/paddle/incubate/distributed/models/moe/moe_layer.py
@@ -85,10 +85,8 @@ def _all_gather(tensor, group=None, use_calc_stream=True):
             if group is None
             else group.nranks
         )
-        return paddle._legacy_C_ops.c_allgather(
+        return paddle._legacy_C_ops.all_gather(
             tensor,
-            'use_calc_stream',
-            use_calc_stream,
             'ring_id',
             ring_id,
             'nranks',
diff --git a/test/auto_parallel/static_completion_convergence.py b/test/auto_parallel/static_completion_convergence.py
@@ -129,16 +129,16 @@ def test_reshard_mesh(self):
         assert check_ops == [
             'matmul_v2',
             'elementwise_add',
-            'c_allgather',
+            'all_gather',
             'split',
             'concat',
             'split',
             'assign',
-            'c_allgather',
+            'all_gather',
             'matmul_v2',
             'elementwise_add',
             'elementwise_sub',
-            'c_allgather',
+            'all_gather',
             'split',
             'concat',
             'assign',
@@ -156,7 +156,7 @@ def test_reshard_mesh(self):
             'c_allreduce_sum',
             'scale',
             'assign',
-            'c_allgather',
+            'all_gather',
             'split',
             'concat',
             'split',
diff --git a/test/legacy_test/collective_allgather_op.py b/test/legacy_test/collective_allgather_op.py
@@ -41,10 +41,10 @@ def get_model(self, main_prog, startup_program):
                 stop_gradient=False,
             )
             main_prog.global_block().append_op(
-                type="c_allgather",
-                inputs={'X': tindata},
+                type="all_gather",
+                inputs={'x': tindata},
                 attrs={'ring_id': ring_id, 'nranks': nranks},
-                outputs={'Out': toutdata},
+                outputs={'out': toutdata},
             )
             main_prog.global_block().append_op(
                 type="c_sync_comm_stream",
@@ -75,10 +75,10 @@ def get_model(self, main_prog, startup_program):
                 stop_gradient=False,
             )
             main_prog.global_block().append_op(
-                type="c_allgather",
-                inputs={'X': tindata},
+                type="all_gather",
+                inputs={'x': tindata},
                 attrs={'ring_id': ring_id, 'nranks': nranks},
-                outputs={'Out': toutdata},
+                outputs={'out': toutdata},
             )
             main_prog.global_block().append_op(
                 type="c_sync_comm_stream",
diff --git a/test/xpu/test_collective_allgather_xpu.py b/test/xpu/test_collective_allgather_xpu.py
@@ -32,8 +32,8 @@ def _setup_config(self):
         "run test when having at least 2 XPUs.",
     )
     def test_allgather(self):
-        support_types = get_xpu_op_support_types('c_allgather')
-        # TODO(lijin23): The XPU c_allgather does support float64 and bool,
+        support_types = get_xpu_op_support_types('all_gather')
+        # TODO(lijin23): The XPU all_gather does support float64 and bool,
         # but the split kernel corresponding with it does not.
         # To be removed when the XDNN api supports float64.
         for dtype in support_types:
@@ -48,7 +48,7 @@ def test_allgather(self):
         "run test when having at least 2 XPUs.",
     )
     def test_allgather_dygraph(self):
-        support_types = get_xpu_op_support_types('c_allgather')
+        support_types = get_xpu_op_support_types('all_gather')
         for dtype in support_types:
             if dtype in ["float64", "bool"]:
                 continue
