Polish ParallelExectuor constructor into small functions

[Aurelius84]

PR types

Others

PR changes

Others

Describe

Polish ParallelExectuor constructor into small functions

1. PR 背景：

ParallelExecutor执行器构造函数逻辑过于复杂，非常不利于后续其他模块复用此代码，比如动转静执行器升级PE工作。

2. PR 内容：

对ParallelExectuor构造函数进行了逻辑拆分，增强代码的可读性和可复用性。本PR为动转静升级PE的前序工作，不涉及新功能增加。

此PR后，PE的构造函数为：

ParallelExecutor::ParallelExecutor(...){
  InitP2P(places);
  ir::InitReaderQueueDeviceCount(graph, *(member_->global_scope_),
                                 member_->places_.size());
  // Initialize necessary info of member_ with strategy.
  InitExecutorPrivateMemberInfo(exec_strategy, build_strategy, places.size(),
                                *graph);

  // Step 2. Create local scopes and Clone graph into multi device
  CreateLocalScopes(scope, local_scopes, /*create_new*/ true);
  std::vector<ir::Graph *> graphs = CloneGraphToMultiDevices(graph);
  PrepareNCCLCommunicator(scope);

  auto need_broadcast = [&]() -> bool {
    if (member_->build_strategy_.num_trainers_ > 1) {
      // 1. num_tariners would be grater than 1 for nccl distributed training.
      return true;
    } else if (member_->local_scopes_.size() != 1 && local_scopes.empty()) {
      // 2. Only one trainer process, but ParallelExecutor hold multiple
      // devices.
      return true;
    }
    return false;
  };
  if (need_broadcast()) {
    BCastParamsToDevices(bcast_vars, member_->build_strategy_.trainer_id_);
  }

  // Step 2. Convert main_program to SSA form and dependency graph. Also, insert
  // ncclOp
  std::vector<ir::Graph *> async_graphs =
      CompileGraphWithBuildStrategy(graph, &graphs, loss_var_name);
  graph = member_->ApplyMemoryOptimizePass(graph);
  async_graphs[0] = graph;

  // Step 3. Create vars in each scope. Passes may also create new vars.
  //         skip control vars and empty vars
  std::vector<details::VariableInfo> var_infos;
  CreateVariableInfos(&var_infos, graph);
  std::unordered_map<Scope *, Scope *> scope_map =
      CreateLocalExecScopes(member_->local_scopes_, /*create_new*/ true);

  // Step 4. Create SSAGraph executor
  std::vector<ir::Graph *> final_graphs =
      CreateSSAGraphExecutor(exec_strategy, &async_graphs, graph);

  VLOG(3) << "use ScopeBufferedSSAGraphExecutor";
  if (!member_->build_strategy_.async_mode_) {
    member_->executor_.reset(new details::ScopeBufferedSSAGraphExecutor(
        exec_strategy, member_->local_scopes_, member_->local_exec_scopes_,
        std::move(var_infos), member_->places_, std::move(member_->executor_)));
  }

  ResetOpHandleScopeMapOfGraphs(final_graphs, scope_map);
  SetReaderOpDeviceInfoOfGraphs(final_graphs);
}

[paddle-bot-old]

Thanks for your contribution!
Please wait for the result of CI firstly. See Paddle CI Manual for details.

[zhiqiu]

LGTM

[chenwhql]

LGTM

[zhhsplendid]

LGTM