Merge pull request #10659 from panyx0718/fix_rpc

Delete prefetch_ctx_ after use.

Author: panyx0718

paddle/fluid/operators/detail/grpc_server.cc

@@ -306,7 +306,7 @@ void AsyncGRPCServer::TryToRegisterNewPrefetchOne() {
  }
  RequestPrefetch* prefetch =
  new RequestPrefetch(&service_, cq_prefetch_.get(), sync_mode_, scope_,
- dev_ctx_, executor_, program_, prefetch_ctx_);
+ dev_ctx_, executor_, program_, prefetch_ctx_.get());
 
  VLOG(4) << "Create RequestPrefetch status:" << prefetch->Status();
 }

paddle/fluid/operators/detail/grpc_server.h

@@ -64,8 +64,9 @@ class AsyncGRPCServer final {
 
  void SetExecutor(framework::Executor *executor) { executor_ = executor; }
 
- void SetPrefetchPreparedCtx(framework::ExecutorPrepareContext *prepared) {
- prefetch_ctx_ = prepared;
+ void SetPrefetchPreparedCtx(
+ std::unique_ptr<framework::ExecutorPrepareContext> prepared) {
+ prefetch_ctx_.reset(prepared.release());
  }
 
  int GetSelectedPort() const { return selected_port_; }
@@ -116,7 +117,7 @@ class AsyncGRPCServer final {
  std::unique_ptr<std::thread> t_get_;
  std::unique_ptr<std::thread> t_prefetch_;
 
- framework::ExecutorPrepareContext *prefetch_ctx_;
+ std::unique_ptr<framework::ExecutorPrepareContext> prefetch_ctx_;
  framework::ProgramDesc *program_;
  framework::Executor *executor_;
  int selected_port_;

paddle/fluid/operators/detail/grpc_server_test.cc

@@ -100,7 +100,7 @@ void StartServer(const std::string& endpoint) {
  InitTensorsOnServer(&scope, &place, 10);
 
  rpc_service_->SetProgram(&program);
- rpc_service_->SetPrefetchPreparedCtx(prepared.get());
+ rpc_service_->SetPrefetchPreparedCtx(std::move(prepared));
  rpc_service_->SetDevCtx(&ctx);
  rpc_service_->SetScope(&scope);
  rpc_service_->SetExecutor(&exe);

paddle/fluid/operators/listen_and_serv_op.cc

@@ -322,8 +322,7 @@ void ListenAndServOp::RunImpl(const framework::Scope &scope,
  // prepare for prefetch
  VLOG(3) << "prefetch block id is " << prefetch_block->ID();
  auto prefetch_prepared = executor.Prepare(*program, prefetch_block->ID());
- rpc_service_->SetPrefetchPreparedCtx(prefetch_prepared.get());
- prefetch_prepared.release();
+ rpc_service_->SetPrefetchPreparedCtx(std::move(prefetch_prepared));
 
  // start the server listening after all member initialized.
  server_thread_.reset(new std::thread(RunServer, rpc_service_));