can run

paddle/fluid/framework/details/async_ssa_graph_executor.cc

@@ -80,6 +80,7 @@ void ProcessGraph(std::vector<ir::Graph *> graphs, Scope *scope) {
         }
       }
     }
+    /*
     VLOG(3) << "delete all recv ops";
     for (auto *node : nodes_to_delete) {
       // delete input edge
@@ -105,6 +106,7 @@ void ProcessGraph(std::vector<ir::Graph *> graphs, Scope *scope) {
       VLOG(3) << "delete node " << node->Name();
       graphs[i]->RemoveNode(node);
     }
+    */
   }
   // init communicator here
   if (send_varname_to_ctx.size() > 0) {

paddle/fluid/framework/details/multi_devices_graph_pass.h

@@ -127,6 +127,10 @@ class AsyncSSAGraphBuilder : public MultiDevSSAGraphBuilderBase {
   bool NeedCollectiveOps() const override { return false; }
 
   bool DealWithSpecialOp(ir::Graph *result, ir::Node *node) const override {
+    if (node->Op()->Type() == "recv") {
+      node->Op()->SetAttr("do_not_run", true);
+      node->Op()->Flush();
+    }
     return false;
   }
 

paddle/fluid/operators/distributed_ops/recv_op.cc

@@ -36,6 +36,11 @@ class RecvOp : public framework::OperatorBase {
 
   void RunImpl(const framework::Scope &scope,
                const platform::Place &place) const override {
+    bool do_not_run = Attr<bool>("do_not_run");
+    if (do_not_run) {
+      VLOG(3) << "recv do not run!";
+      return;
+    }
     std::vector<std::string> epmap = Attr<std::vector<std::string>>("epmap");
     std::vector<std::string> varnames =
         Attr<std::vector<std::string>>("varnames");
@@ -126,6 +131,7 @@ This operator can get variables from server side.
         "(vector<string>) "
         "the splited parameter varnames to be recved from pserver")
         .SetDefault(std::vector<std::string>{});
+    AddAttr<bool>("do_not_run", "").SetDefault(false);
   }
 };