Fix

Author: co63oc

paddle/fluid/framework/new_executor/interpreter/interpreter_util.cc

@@ -169,7 +169,7 @@ bool IsCommunicationOp(const ::pir::Operation* op) {
   }
   const std::set<std::string> special_comm_op_set = {
       paddle::dialect::SendV2Op::name(),
-      paddle::dialect::RecvV2Op::name(),
+      paddle::dialect::PRecvOp::name(),
   };
   const std::string communication_op_prefix = "c_";
   if (op_name.find(communication_op_prefix) != std::string::npos ||

paddle/fluid/pir/dialect/operator/utils/utils.cc

@@ -39,7 +39,7 @@ namespace dialect {
 const std::unordered_set<std::string> LegacyOpList = {
     DistributedPushSparseOp::name(),
     SendV2Op::name(),
-    RecvV2Op::name(),
+    PRecv::name(),
     CAllreduceSumOp::name(),
     CAllreduceSum_Op::name(),
 };

paddle/phi/ops/yaml/inconsistent/static_ops.yaml

@@ -597,6 +597,18 @@
   interfaces : paddle::dialect::InferSymbolicShapeInterface
   traits : paddle::dialect::ForwardOnlyTrait
 
+- op : p_recv
+  args : (int[] out_shape = {}, DataType dtype = DataType::FLOAT32, int peer = 0, int ring_id = 0, bool use_calc_stream = false, bool dynamic_shape = false)
+  output : Tensor(out)
+  infer_meta:
+    func: RecvV2InferMeta
+    param: [ring_id, dynamic_shape, peer, out_shape, dtype]
+  kernel :
+    func : recv_v2
+    param : [ring_id, dynamic_shape, peer, out_shape, dtype, use_calc_stream]
+    data_type : dtype
+  interfaces : paddle::dialect::InferSymbolicShapeInterface
+
 - op : partial_recv
   args : (int ring_id = 0, int peer = 0, DataType dtype=DataType::FLOAT32, int[] out_shape= {}, int num = 1, int id = 0)
   output : Tensor(out)
@@ -699,18 +711,6 @@
   inplace : (scale -> out_scale, in_accum -> out_accum, in_state -> out_state)
   interfaces : paddle::dialect::InferSymbolicShapeInterface
 
-- op : recv_v2
-  args : (int[] out_shape = {}, DataType dtype = DataType::FLOAT32, int peer = 0, int ring_id = 0, bool use_calc_stream = false, bool dynamic_shape = false)
-  output : Tensor(out)
-  infer_meta:
-    func: RecvV2InferMeta
-    param: [ring_id, dynamic_shape, peer, out_shape, dtype]
-  kernel :
-    func : recv_v2
-    param : [ring_id, dynamic_shape, peer, out_shape, dtype, use_calc_stream]
-    data_type : dtype
-  interfaces : paddle::dialect::InferSymbolicShapeInterface
-
 - op : remainder
   args : (Tensor x, Tensor y)
   output : Tensor (out)

paddle/phi/ops/yaml/op_compat.yaml

@@ -4469,6 +4469,14 @@
   outputs :
     out : Out
 
+- op: p_recv
+  outputs :
+    out : Out
+
+- op: p_send
+  inputs :
+    x : X
+
 - op: partial_send
   inputs :
     x : X

python/paddle/distributed/auto_parallel/static/reshard.py

@@ -387,7 +387,7 @@ def insert_send_op(block, idx, tensor, src, dst, op_role, sync=True):
     @staticmethod
     def insert_recv_op(block, idx, tensor, src, dst, op_role, sync=True):
         """Insert recv op into block at the given index."""
-        op_type = 'recv_v2'
+        op_type = 'p_recv'
         insert_operation = (
             block._insert_op if sync else block._insert_op_without_sync
         )

test/collective/collective_sendrecv_api.py

@@ -47,7 +47,7 @@ def send_new(tensor, dst, group=None, sync_op=True):
     helper = framework.LayerHelper(op_type, **locals())
     helper.append_op(
         type=op_type,
-        inputs={'x': [tensor]},
+        inputs={'X': [tensor]},
         attrs={
             'ring_id': ring_id,
             'peer': dst,
@@ -78,7 +78,7 @@ def recv_new(tensor, src, group=None, sync_op=True, dtype='float32'):
     helper = framework.LayerHelper(op_type, **locals())
     helper.append_op(
         type=op_type,
-        outputs={'out': [tensor]},
+        outputs={'Out': [tensor]},
         attrs={
             'ring_id': ring_id,
             'peer': src,

test/collective/collective_sendrecv_op.py

@@ -38,7 +38,7 @@ def get_model(self, main_prog, startup_program):
             tindata.desc.set_need_check_feed(False)
             if self.rank == 0:
                 main_prog.global_block().append_op(
-                    type="send_v2",
+                    type="p_send",
                     inputs={'X': tindata},
                     attrs={
                         'ring_id': ring_id,
@@ -48,7 +48,7 @@ def get_model(self, main_prog, startup_program):
                 )
             else:
                 main_prog.global_block().append_op(
-                    type="recv_v2",
+                    type="p_recv",
                     outputs={'Out': tindata},
                     attrs={
                         'peer': 0,

test/deprecated/legacy_test/test_auto_parallel_reshard_deprecated.py

@@ -223,7 +223,7 @@ def check_send_recv_result(dist_main_prog, rank_id):
             if op.type == "send_v2" and "gelu_0.tmp_0" in op.input_arg_names:
                 send_result = True
             if (
-                op.type == "recv_v2"
+                op.type == "p_recv"
                 and "gelu_0.tmp_0@GRAD" in op.output_arg_names[0]
             ):
                 recv_result = True
@@ -234,10 +234,7 @@ def check_send_recv_result(dist_main_prog, rank_id):
                 and "gelu_0.tmp_0@GRAD" in op.input_arg_names
             ):
                 send_result = True
-            if (
-                op.type == "recv_v2"
-                and "gelu_0.tmp_0" in op.output_arg_names[0]
-            ):
+            if op.type == "p_recv" and "gelu_0.tmp_0" in op.output_arg_names[0]:
                 recv_result = True
 
     return send_result and recv_result

test/deprecated/legacy_test/test_auto_parallel_reshard_dpmppp_deprecated.py

@@ -163,7 +163,7 @@ def check_send_recv_result(dist_main_prog, rank_id):
             if op.type == "send_v2" and "gelu_0.tmp_0" in op.input_arg_names:
                 send_result = True
             if (
-                op.type == "recv_v2"
+                op.type == "p_recv"
                 and "gelu_0.tmp_0@GRAD" in op.output_arg_names[0]
             ):
                 recv_result = True
@@ -174,10 +174,7 @@ def check_send_recv_result(dist_main_prog, rank_id):
                 and "gelu_0.tmp_0@GRAD" in op.input_arg_names
             ):
                 send_result = True
-            if (
-                op.type == "recv_v2"
-                and "gelu_0.tmp_0" in op.output_arg_names[0]
-            ):
+            if op.type == "p_recv" and "gelu_0.tmp_0" in op.output_arg_names[0]:
                 recv_result = True
 
     return send_result and recv_result

test/deprecated/legacy_test/test_auto_parallel_reshard_mppp_deprecated.py

@@ -176,7 +176,7 @@ def check_send_recv_result(dist_main_prog, rank_id):
             if op.type == "send_v2" and "gelu_0.tmp_0" in op.input_arg_names:
                 send_result = True
             if (
-                op.type == "recv_v2"
+                op.type == "p_recv"
                 and "gelu_0.tmp_0@GRAD" in op.output_arg_names[0]
             ):
                 recv_result = True
@@ -187,10 +187,7 @@ def check_send_recv_result(dist_main_prog, rank_id):
                 and "gelu_0.tmp_0@GRAD" in op.input_arg_names[0]
             ):
                 send_result = True
-            if (
-                op.type == "recv_v2"
-                and "gelu_0.tmp_0" in op.output_arg_names[0]
-            ):
+            if op.type == "p_recv" and "gelu_0.tmp_0" in op.output_arg_names[0]:
                 recv_result = True
 
     return send_result and recv_result