simplify parameter send and recv

paddle/fluid/operators/distributed/communicator.cc

@@ -74,9 +74,9 @@ void Communicator::SendThread() {
       merged_var_num++;
     }
     MergeVars(var_name, vars, send_scope_.get());
-    auto send_functor = distributed::ParameterSend<float>();
-    //    send_functor(var_name, send_varname_to_ctx_[var_name], exe_ctx,
-    //    send_scope_, true);
+    // auto send_functor = distributed::ParameterSend<float>();
+    // send_functor(var_name, send_varname_to_ctx_[var_name], exe_ctx,
+    // send_scope_, true);
   }
 }
 
@@ -85,8 +85,8 @@ void Communicator::RecvThread() {
   for (auto &iter : recv_varname_to_ctx_) {
     auto &var_name = iter.first;
     VLOG(3) << "recv var " << iter.first;
-    auto recv_functor = distributed::ParameterRecv<float>();
-    //    recv_functor(var_name, iter.second, exe_ctx, recv_scope_);
+    // auto recv_functor = distributed::ParameterRecv<float>();
+    // recv_functor(var_name, iter.second, exe_ctx, recv_scope_);
   }
 }
 

paddle/fluid/operators/distributed_ops/recv_op.cc

@@ -54,12 +54,6 @@ class RecvOp : public framework::OperatorBase {
         Attr<std::vector<std::string>>("recv_varnames");
 
     if (recv_varnames.size() > 0) {
-      framework::RuntimeContext ctx(Inputs(), Outputs(), scope);
-      platform::DeviceContextPool &pool =
-          platform::DeviceContextPool::Instance();
-      auto *dev_ctx = pool.Get(place);
-      auto exe_ctx =
-          framework::ExecutionContext(*this, scope, *dev_ctx, ctx, nullptr);
       auto recv_functor = distributed::ParameterRecv<float>();
       auto rpc_ctx = distributed::RpcContext(outs[0], recv_varnames, epmap, {});
       recv_functor(rpc_ctx, scope);

paddle/fluid/operators/distributed_ops/send_op.cc

@@ -47,12 +47,6 @@ class SendOp : public framework::OperatorBase {
 
     if (send_varnames.size() > 0) {
       PADDLE_ENFORCE_EQ(ins.size(), 1, "");
-      framework::RuntimeContext ctx(Inputs(), Outputs(), scope);
-      platform::DeviceContextPool& pool =
-          platform::DeviceContextPool::Instance();
-      auto* dev_ctx = pool.Get(place);
-      auto exe_ctx =
-          framework::ExecutionContext(*this, scope, *dev_ctx, ctx, nullptr);
       auto send_functor = distributed::ParameterSend<float>();
       auto rpc_ctx = distributed::RpcContext(ins[0], send_varnames, epmap,
                                              height_sections);