fix remove recv op

paddle/fluid/framework/details/async_ssa_graph_executor.cc

@@ -82,6 +82,27 @@ void ProcessGraph(std::vector<ir::Graph *> graphs, Scope *scope) {
     }
     VLOG(3) << "delete all recv ops";
     for (auto *node : nodes_to_delete) {
+      // delete input edge
+      for (auto *in : node->inputs) {
+        auto &in_outs = in->outputs;
+        for (auto iter = in_outs.begin(); iter != in_outs.end();) {
+          if (*iter == node) {
+            VLOG(3) << "delete input edge from " << in->Name() << " for "
+                    << node->Name();
+            iter = in_outs.erase(iter);
+          } else {
+            ++iter;
+          }
+        }
+      }
+      // delete output edge
+      for (auto *out : node->outputs) {
+        PADDLE_ENFORCE_EQ(out->outputs.size(), 0, "%s should have no outputs",
+                          out->Name());
+        VLOG(3) << "delete output edge to " << out->Name();
+        graphs[i]->RemoveNode(out);
+      }
+      VLOG(3) << "delete node " << node->Name();
       graphs[i]->RemoveNode(node);
     }
   }

paddle/fluid/operators/distributed/communicator.cc

@@ -31,7 +31,7 @@ namespace distributed {
 static inline void MergeVars(const std::string &var_name,
                              const std::vector<std::shared_ptr<Variable>> &vars,
                              Scope *scope) {
-  VLOG(3) << "merge " << vars.size() << " vars " << var_name << " to one";
+  VLOG(3) << "merge " << vars.size() << " vars " << var_name << " to 1";
   PADDLE_ENFORCE(!vars.empty(), "should have value to merge!");
   auto cpu_place = platform::CPUPlace();
   auto &var0 = vars[0];