parallel_executor.cc 22.5 KB
Newer Older
Y
Yang Yang 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15
/* Copyright (c) 2016 PaddlePaddle Authors. All Rights Reserved.

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License. */

#include "paddle/fluid/framework/parallel_executor.h"
D
dzhwinter 已提交
16
#include <algorithm>
Q
qingqing01 已提交
17
#include <memory>
C
chengduoZH 已提交
18
#include <string>
19
#include <tuple>
Q
Qiao Longfei 已提交
20
#include <utility>
Q
qiaolongfei 已提交
21
#include <vector>
Q
Qiao Longfei 已提交
22
#include "paddle/fluid/framework/details/async_ssa_graph_executor.h"
Y
yuyang18 已提交
23
#include "paddle/fluid/framework/details/fast_threaded_ssa_graph_executor.h"
24
#include "paddle/fluid/framework/details/multi_devices_helper.h"
Y
Yancey1989 已提交
25
#include "paddle/fluid/framework/details/parallel_ssa_graph_executor.h"
Y
yuyang18 已提交
26
#include "paddle/fluid/framework/details/scope_buffered_ssa_graph_executor.h"
Y
Yu Yang 已提交
27
#include "paddle/fluid/framework/details/threaded_ssa_graph_executor.h"
28 29
#include "paddle/fluid/framework/ir/graph.h"
#include "paddle/fluid/framework/ir/graph_helper.h"
30
#include "paddle/fluid/framework/ir/memory_optimize_pass/reference_count_pass_helper.h"
31
#include "paddle/fluid/platform/profiler.h"
Y
Yu Yang 已提交
32

Y
Yu Yang 已提交
33
#ifdef WITH_GPERFTOOLS
Y
Yu Yang 已提交
34
#include "gperftools/profiler.h"
Y
Yu Yang 已提交
35
#endif
Y
Yu Yang 已提交
36
DEFINE_string(pe_profile_fname, "",
Y
Yu Yang 已提交
37 38
              "Profiler filename for PE, which generated by gperftools."
              "Only valid when compiled `WITH_PRIFILER=ON`. Empty if disable.");
39
DEFINE_bool(enable_parallel_graph, false,
Y
Yancey1989 已提交
40
            "Force disable parallel graph execution mode if set false.");
Y
Yu Yang 已提交
41

Y
Yang Yang 已提交
42
namespace paddle {
Y
Yu Yang 已提交
43 44
namespace framework {

Y
Yu Yang 已提交
45
static std::once_flag gProfileOnce;
Y
Yu Yang 已提交
46
#ifdef WITH_GPERFTOOLS
Y
Yu Yang 已提交
47
static bool gProfileStarted = false;
Y
Yu Yang 已提交
48
#endif
49

Y
Yu Yang 已提交
50 51 52
class ParallelExecutorPrivate {
 public:
  explicit ParallelExecutorPrivate(const std::vector<platform::Place> &places)
Y
Yu Yang 已提交
53
      : places_(places) {
Y
Yu Yang 已提交
54
    if (!FLAGS_pe_profile_fname.empty()) {
Y
Yu Yang 已提交
55 56
      std::call_once(gProfileOnce, [] {
#ifdef WITH_GPERFTOOLS
Y
Yu Yang 已提交
57
        ProfilerStart(FLAGS_pe_profile_fname.c_str());
Y
Yu Yang 已提交
58 59 60
        gProfileStarted = true;
#else
        LOG(WARNING) << "Paddle is not compiled with gperftools. "
61
          "FLAGS_pe_profile_fname will be ignored";
Y
Yu Yang 已提交
62 63 64 65
#endif
      });
    }
  }
Y
Yu Yang 已提交
66

67 68 69 70 71 72 73 74 75 76 77
  ~ParallelExecutorPrivate() {
    if (own_local_scope_) {
      for (size_t i = 1; i < local_scopes_.size(); ++i) {
        // Skip the first scope, since it is the global scope.
        Scope *local_scope = local_scopes_[i];
        if (global_scope_->HasKid(local_scope)) {
          global_scope_->DeleteScope(local_scope);
        }
      }
    }
  }
S
sneaxiy 已提交
78

79
  ir::Graph *PrepareGCAndRefCnts(ir::Graph *graph, size_t max_memory_size);
S
sneaxiy 已提交
80 81 82 83 84 85 86 87 88 89 90 91

  inline bool HasGarbageCollectors() const { return !gcs_.empty(); }

  void ResetRuntimeReferenceCount(const std::vector<std::string> &fetch_tensors,
                                  const std::string &fetched_var_name) {
    for (size_t i = 0; i < runtime_ref_cnts_.size(); ++i) {
      for (auto &pair : global_ref_cnts_[i]) {
        runtime_ref_cnts_[i][pair.first] = pair.second;
      }

      for (auto &fetch_name : fetch_tensors) {
        runtime_ref_cnts_[i].erase(fetch_name);
S
sneaxiy 已提交
92
      }
S
sneaxiy 已提交
93
      runtime_ref_cnts_[i].erase(fetched_var_name);
S
sneaxiy 已提交
94 95 96
    }
  }

D
dzhwinter 已提交
97
  BuildStrategy build_strategy_;
Y
Yu Yang 已提交
98 99
  std::vector<platform::Place> places_;
  std::vector<Scope *> local_scopes_;
100
  Scope *global_scope_;  // not owned
Y
Yu Yang 已提交
101
  std::unique_ptr<details::SSAGraphExecutor> executor_;
Y
Yu Yang 已提交
102

P
peizhilin 已提交
103
#if defined(PADDLE_WITH_CUDA) && !defined(_WIN32)
Y
Yu Yang 已提交
104
  std::unique_ptr<platform::NCCLContextMap> nccl_ctxs_;
Y
Yu Yang 已提交
105
#endif
C
chengduoZH 已提交
106 107
  bool own_local_scope_;
  bool use_cuda_;
108
  bool use_all_reduce_;
109
  size_t nranks_;
S
sneaxiy 已提交
110

S
sneaxiy 已提交
111 112 113
  // global_ref_cnts_ is only initialized when ParallelExecutor constructs, and
  // then keeps unchanged
  // Before each iteration, runtime_ref_cnts_ is reset to global_ref_cnts_
114 115 116
  std::vector<ir::ReferenceCountMap> global_ref_cnts_;
  std::vector<ir::AtomicReferenceCountMap> runtime_ref_cnts_;
  ir::GarbageCollectorMap gcs_;
Y
Yu Yang 已提交
117 118
};

119 120
ir::Graph *ParallelExecutorPrivate::PrepareGCAndRefCnts(
    ir::Graph *graph, size_t max_memory_size) {
S
sneaxiy 已提交
121 122 123 124 125
  for (size_t i = 0; i < places_.size(); ++i) {
    auto &place = places_[i];
    if (gcs_.count(place) > 0) {
      continue;
    }
S
sneaxiy 已提交
126
    std::unique_ptr<GarbageCollector> gc;
S
sneaxiy 已提交
127
#ifdef PADDLE_WITH_CUDA
S
sneaxiy 已提交
128 129
    if (platform::is_gpu_place(place)) {
      if (IsFastEagerDeletionModeEnabled()) {
S
sneaxiy 已提交
130 131
        gc.reset(new UnsafeFastGPUGarbageCollector(
            boost::get<platform::CUDAPlace>(place), max_memory_size));
S
sneaxiy 已提交
132
      } else {
S
sneaxiy 已提交
133 134
        gc.reset(new StreamGarbageCollector(
            boost::get<platform::CUDAPlace>(place), max_memory_size));
S
sneaxiy 已提交
135 136
      }
      VLOG(10) << "Created " << i << "-th GarbageCollector at " << place;
S
sneaxiy 已提交
137
    } else {
S
sneaxiy 已提交
138
#endif
S
sneaxiy 已提交
139 140 141 142 143 144 145
      if (platform::is_cpu_place(place)) {
        gc.reset(new CPUGarbageCollector(boost::get<platform::CPUPlace>(place),
                                         max_memory_size));
        VLOG(10) << "Created GarbageCollector at " << place;
      } else {
        PADDLE_THROW("Unsupported place for garbage collection");
      }
S
sneaxiy 已提交
146 147 148 149
#ifdef PADDLE_WITH_CUDA
    }
#endif

S
sneaxiy 已提交
150
    gcs_.emplace(place, std::move(gc));
S
sneaxiy 已提交
151 152
  }

S
sneaxiy 已提交
153
  if (!gcs_.empty()) {
154
    std::vector<ir::LastLiveOpsOfVars> last_live_ops_of_vars;
S
sneaxiy 已提交
155 156 157

    auto ref_cnt_pass =
        ir::PassRegistry::Instance().Get("reference_count_pass");
158 159
    ref_cnt_pass->SetNotOwned(ir::kGlobalReferenceCount, &global_ref_cnts_);
    ref_cnt_pass->SetNotOwned(ir::kLastLiveOpsOfVars, &last_live_ops_of_vars);
160
    graph = ref_cnt_pass->Apply(graph);
S
sneaxiy 已提交
161 162 163 164
    VLOG(10) << "ReferenceCountPass Applied";

    auto eager_deletion_pass =
        ir::PassRegistry::Instance().Get("eager_deletion_pass");
165
    eager_deletion_pass->SetNotOwned(ir::kRuntimeReferenceCount,
S
sneaxiy 已提交
166
                                     &runtime_ref_cnts_);
167 168
    eager_deletion_pass->SetNotOwned(ir::kGarbageCollector, &gcs_);
    eager_deletion_pass->SetNotOwned(ir::kLastLiveOpsOfVars,
S
sneaxiy 已提交
169
                                     &last_live_ops_of_vars);
170
    eager_deletion_pass->SetNotOwned(ir::kAllPlaces, &places_);
171
    graph = eager_deletion_pass->Apply(graph);
S
sneaxiy 已提交
172 173 174 175 176
    VLOG(10) << "EagerDeletionPass Applied";
  }
  return graph;
}

177 178 179 180
std::vector<Scope *> &ParallelExecutor::GetLocalScopes() {
  return member_->local_scopes_;
}

181 182 183 184 185 186 187 188 189 190 191 192 193 194
void ParallelExecutor::DropLocalExeScopes() {
  auto executor = dynamic_cast<details::ScopeBufferedSSAGraphExecutor *>(
      member_->executor_.get());
  if (executor) {
    executor->DropLocalExeScopes();
  }
}

bool ParallelExecutor::NeedCreateLocalExeScope() {
  auto executor = dynamic_cast<details::ScopeBufferedSSAGraphExecutor *>(
      member_->executor_.get());
  return executor && executor->NeedCreateLocalExeScope();
}

Y
Yan Xu 已提交
195 196 197 198 199 200 201 202
ParallelExecutor::ParallelExecutor(const std::vector<platform::Place> &places,
                                   const std::vector<std::string> &bcast_vars,
                                   const std::string &loss_var_name,
                                   Scope *scope,
                                   const std::vector<Scope *> &local_scopes,
                                   const ExecutionStrategy &exec_strategy,
                                   const BuildStrategy &build_strategy,
                                   ir::Graph *graph)
Y
Yu Yang 已提交
203
    : member_(new ParallelExecutorPrivate(places)) {
Y
Yu Yang 已提交
204
  member_->global_scope_ = scope;
205
  member_->use_cuda_ = exec_strategy.use_cuda_;
D
dzhwinter 已提交
206
  member_->build_strategy_ = build_strategy;
207 208
  member_->use_all_reduce_ =
      build_strategy.reduce_ == BuildStrategy::ReduceStrategy::kAllReduce;
X
Xin Pan 已提交
209
  member_->nranks_ = build_strategy.num_trainers_ * places.size();
210 211 212 213
  if (!member_->use_all_reduce_) {
    PADDLE_ENFORCE(places.size() > 1,
                   "If you set build_strategy.reduce with 'Reduce',"
                   "the number of places must be greater than 1.");
Y
Yancey1989 已提交
214 215
  }

216
  // Step 1. Bcast the bcast_vars to devs.
Y
Yu Yang 已提交
217
  // Create local scopes
218
  if (local_scopes.empty()) {
C
chengduoZH 已提交
219
    member_->own_local_scope_ = true;
Y
Yu Yang 已提交
220 221
    member_->local_scopes_.emplace_back(member_->global_scope_);
    for (size_t i = 1; i < member_->places_.size(); ++i) {
Y
Debug  
Yu Yang 已提交
222
      member_->local_scopes_.emplace_back(&scope->NewScope());
223 224
    }
  } else {
C
chengduoZH 已提交
225
    member_->own_local_scope_ = false;
226 227
    PADDLE_ENFORCE_EQ(member_->places_.size(), local_scopes.size());
    for (size_t i = 0; i < member_->places_.size(); ++i) {
228
      member_->local_scopes_.emplace_back(&local_scopes[i]->NewScope());
229
    }
Y
Yu Yang 已提交
230 231
  }

Q
Qiao Longfei 已提交
232
  std::vector<ir::Graph *> graphs;
Q
Qiao Longfei 已提交
233 234 235
  if (build_strategy.async_mode_) {
    PADDLE_ENFORCE(!member_->use_cuda_,
                   "gpu mode does not support async_mode_ now!");
Q
Qiao Longfei 已提交
236
    graphs.push_back(graph);
D
dongdaxiang 已提交
237
    for (size_t i = 1; i < places.size(); ++i) {
Q
Qiao Longfei 已提交
238 239 240 241
      auto *tmp_graph = new ir::Graph(graph->OriginProgram());
      async_graphs_.emplace_back(tmp_graph);
      graphs.push_back(tmp_graph);
    }
Q
Qiao Longfei 已提交
242
  }
Q
Qiao Longfei 已提交
243

Y
Yancey1989 已提交
244 245 246
  // FIXME(Yancey1989): parallel graph mode get better performance
  // in GPU allreduce distributed training. Need an elegant way to
  // choice the execution strategy.
247 248
  build_strategy.enable_parallel_graph_ =
      EnableParallelGraphExecution(*graph, exec_strategy, build_strategy);
Y
Yancey1989 已提交
249 250 251 252
  if (build_strategy.enable_parallel_graph_)
    VLOG(0) << "The Executor would execute the graph by ParallelGraph "
               "Execution which can get better performance,"
            << "you can force it off by env FLAGS_enable_parallel_graph=0";
Y
Yancey1989 已提交
253

C
chengduoZH 已提交
254
  if (member_->use_cuda_) {
Y
Yu Yang 已提交
255
// Bcast Parameters to all GPUs
P
peizhilin 已提交
256
#if defined(PADDLE_WITH_CUDA) && !defined(_WIN32)
Y
Yancey1989 已提交
257 258 259
    ncclUniqueId *nccl_id = nullptr;
    // gen_nccl_id operator can broadcast the ncclUniqueId for nccl2 collective
    // distributed training
C
chengduoZH 已提交
260
    auto *nccl_id_var = scope->FindVar(NCCL_ID_VARNAME);
Y
Yancey1989 已提交
261
    if (nccl_id_var != nullptr) {
Y
Yancey1989 已提交
262
      nccl_id = nccl_id_var->GetMutable<ncclUniqueId>();
Y
Yancey1989 已提交
263
    }
264
    if (build_strategy.enable_parallel_graph_ && member_->nranks_ > 1UL) {
Y
Yancey1989 已提交
265 266 267 268
      if (nccl_id == nullptr) {
        local_nccl_id_.reset(new ncclUniqueId());
        platform::dynload::ncclGetUniqueId(local_nccl_id_.get());
        nccl_id = local_nccl_id_.get();
Y
Yancey1989 已提交
269
      }
C
chengduoZH 已提交
270
    }
Y
Yancey1989 已提交
271

C
chengduoZH 已提交
272
    member_->nccl_ctxs_.reset(new platform::NCCLContextMap(
273 274
        member_->places_, nccl_id, build_strategy.num_trainers_,
        build_strategy.trainer_id_));
Q
qingqing01 已提交
275

W
Wu Yi 已提交
276 277 278
    // Initialize device context's nccl comm, will be used by normal
    // Operators like sync_batch_norm, and collective ops.
    // NOTE: more than one ParallelExecutor with same place, the nccl comm will
Q
qingqing01 已提交
279
    // be rewrite and there will be some problem.
W
Wu Yi 已提交
280 281 282 283 284 285 286
    // NOTE: NCCL group-calls and non-group-calls can not use the same
    // NCCL communicator, so for ParallelGraph and Multi-Process mode, re-use
    // same communicators.
    std::unique_ptr<platform::NCCLContextMap> dev_nccl_ctxs;
    if (nccl_id == nullptr) {
      dev_nccl_ctxs.reset(new platform::NCCLContextMap(member_->places_));
    }
Q
qingqing01 已提交
287 288 289 290 291
    for (size_t dev_id = 0; dev_id < member_->places_.size(); ++dev_id) {
      platform::DeviceContextPool &pool =
          platform::DeviceContextPool::Instance();
      auto *dev_ctx = static_cast<platform::CUDADeviceContext *>(
          pool.Get(member_->places_[dev_id]));
W
Wu Yi 已提交
292 293 294 295 296 297 298
      if (nccl_id != nullptr) {
        auto &nccl_ctx = member_->nccl_ctxs_->at(member_->places_[dev_id]);
        dev_ctx->set_nccl_comm(nccl_ctx.comm());
      } else {
        auto &nccl_ctx = dev_nccl_ctxs->at(member_->places_[dev_id]);
        dev_ctx->set_nccl_comm(nccl_ctx.comm());
      }
Q
qingqing01 已提交
299
    }
C
chengduoZH 已提交
300 301
#else
    PADDLE_THROW("Not compiled with CUDA");
Y
Yu Yang 已提交
302
#endif
C
chengduoZH 已提交
303
  }
Y
Yan Xu 已提交
304 305 306 307 308 309 310 311 312 313 314 315 316 317 318
  // broadcast parameters from the 0th device to others:
  auto need_broadcast = [&]() -> bool {
    if (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, build_strategy.trainer_id_);
Y
Yu Yang 已提交
319
  }
Q
Qiao Longfei 已提交
320
  // Startup Program has been run. All local scopes has correct parameters.
Y
yuyang18 已提交
321

Q
Qiao Longfei 已提交
322 323 324
  // Step 2. Convert main_program to SSA form and dependency graph. Also, insert
  // ncclOp
  std::vector<ir::Graph *> async_graphs(places.size());
P
peizhilin 已提交
325
#if defined(PADDLE_WITH_CUDA) && !defined(_WIN32)
Q
Qiao Longfei 已提交
326
  if (build_strategy.async_mode_) {
Q
Qiao Longfei 已提交
327
    VLOG(3) << "use local async mode";
328 329 330
    graph = build_strategy.Apply(graph, {member_->places_[0]}, loss_var_name,
                                 {member_->local_scopes_[0]}, 1,
                                 member_->use_cuda_, member_->nccl_ctxs_.get());
D
dongdaxiang 已提交
331
    for (size_t i = 1; i < member_->places_.size(); ++i) {
332 333 334
      graphs[i] =
          build_strategy.Apply(graphs[i], {member_->places_[i]}, loss_var_name,
                               {member_->local_scopes_[i]}, 1,
Q
Qiao Longfei 已提交
335
                               member_->use_cuda_, member_->nccl_ctxs_.get());
336
      async_graphs[i] = graphs[i];
Q
Qiao Longfei 已提交
337
    }
Q
Qiao Longfei 已提交
338
  } else {
339 340 341
    graph = build_strategy.Apply(graph, member_->places_, loss_var_name,
                                 member_->local_scopes_, member_->nranks_,
                                 member_->use_cuda_, member_->nccl_ctxs_.get());
Q
Qiao Longfei 已提交
342
  }
C
chengduoZH 已提交
343
#else
Q
Qiao Longfei 已提交
344
  if (build_strategy.async_mode_) {
Q
Qiao Longfei 已提交
345
    VLOG(3) << "use local async mode";
346 347 348
    graph = build_strategy.Apply(graph, {member_->places_[0]}, loss_var_name,
                                 {member_->local_scopes_[0]}, 1,
                                 member_->use_cuda_);
349
    for (size_t i = 1; i < member_->places_.size(); ++i) {
350 351
      graphs[i] = build_strategy.Apply(
          graphs[i], {member_->places_[i]}, loss_var_name,
Q
Qiao Longfei 已提交
352
          {member_->local_scopes_[i]}, 1, member_->use_cuda_);
353
      async_graphs[i] = graphs[i];
Q
Qiao Longfei 已提交
354
    }
Q
can run  
Qiao Longfei 已提交
355
  } else {
356 357 358
    graph = build_strategy.Apply(graph, member_->places_, loss_var_name,
                                 member_->local_scopes_, member_->nranks_,
                                 member_->use_cuda_);
Q
can run  
Qiao Longfei 已提交
359
  }
Y
Yu Yang 已提交
360
#endif
361

Y
Yancey1989 已提交
362
  auto max_memory_size = GetEagerDeletionThreshold();
D
dzhwinter 已提交
363 364
  VLOG(10) << "Eager Deletion Threshold "
           << static_cast<float>(max_memory_size) / (1 << 30);
Y
Yancey1989 已提交
365
  if (max_memory_size >= 0) {
366 367
    graph = member_->PrepareGCAndRefCnts(graph,
                                         static_cast<size_t>(max_memory_size));
Y
Yancey1989 已提交
368 369
  }

Q
Qiao Longfei 已提交
370 371
  async_graphs[0] = graph;

372 373
  // Step 3. Create vars in each scope. Passes may also create new vars.
  //         skip control vars and empty vars
Y
Yancey1989 已提交
374
  std::vector<details::VariableInfo> var_infos;
Q
Qiao Longfei 已提交
375 376 377 378 379 380
  for (auto &node : graph->Nodes()) {
    if (node->IsVar() && !node->IsCtrlVar() && node->Var()) {
      var_infos.emplace_back();
      var_infos.back().name_ = node->Var()->Name();
      var_infos.back().type_ = node->Var()->GetType();
      var_infos.back().persistable_ = node->Var()->Persistable();
Y
Yancey1989 已提交
381 382
    }
  }
Y
Yancey1989 已提交
383

W
Wu Yi 已提交
384 385
  // If the loss_var_name is given, the number of graph should be only one.
  if (loss_var_name.size()) {
Q
Qiao Longfei 已提交
386
    size_t graph_num = ir::GraphNum(*graph);
C
chengduo 已提交
387 388 389 390
    if (graph_num > 1) {
      LOG(WARNING)
          << "The number of graph should be only one, "
             "but the current graph has "
Q
Qiao Longfei 已提交
391
          << ir::GraphNum(*graph)
C
chengduo 已提交
392 393 394 395 396
          << " sub_graphs. If you want to see the nodes of the "
             "sub_graphs, you should use 'FLAGS_print_sub_graph_dir' "
             "to specify the output dir. NOTES: if you not do training, "
             "please don't pass loss_var_name.";
    }
W
Wu Yi 已提交
397 398
  }

Q
Qiao Longfei 已提交
399
  if (build_strategy.async_mode_) {
Q
can run  
Qiao Longfei 已提交
400 401
    VLOG(3) << "use AsyncSSAGraphExecutor";
    member_->executor_.reset(new details::AsyncSSAGraphExecutor(
Q
Qiao Longfei 已提交
402
        exec_strategy, member_->local_scopes_, member_->places_, async_graphs));
Q
can run  
Qiao Longfei 已提交
403 404
  } else if (build_strategy.enable_parallel_graph_) {
    VLOG(3) << "use ParallelSSAGraphExecutor";
Y
Yancey1989 已提交
405
#ifdef PADDLE_WITH_CUDA
Y
Yancey1989 已提交
406 407
    // TODO(Yancey1989): Remove passing in the main_program when
    // allreduce_seq_pass doesn't need it as the attr.
Y
Yancey1989 已提交
408
    member_->executor_.reset(new details::ParallelSSAGraphExecutor(
X
Xin Pan 已提交
409
        exec_strategy, member_->local_scopes_, member_->places_, graph));
Y
Yancey1989 已提交
410 411 412 413
#else
    PADDLE_THROW(
        "Paddle should be compiled with CUDA for ParallelGraph Execution.");
#endif
Y
yuyang18 已提交
414
  } else {
Y
Yancey1989 已提交
415
    if (exec_strategy.type_ == ExecutionStrategy::kDefault) {
Q
can run  
Qiao Longfei 已提交
416
      VLOG(3) << "use ThreadedSSAGraphExecutor";
Y
Yancey1989 已提交
417
      member_->executor_.reset(new details::ThreadedSSAGraphExecutor(
X
Xin Pan 已提交
418
          exec_strategy, member_->local_scopes_, member_->places_, graph));
Y
Yancey1989 已提交
419
    } else {
Q
can run  
Qiao Longfei 已提交
420
      VLOG(3) << "use FastThreadedSSAGraphExecutor";
Y
Yancey1989 已提交
421
      member_->executor_.reset(new details::FastThreadedSSAGraphExecutor(
X
Xin Pan 已提交
422
          exec_strategy, member_->local_scopes_, member_->places_, graph));
Y
Yancey1989 已提交
423
    }
C
chengduoZH 已提交
424
  }
Y
yuyang18 已提交
425

Q
can run  
Qiao Longfei 已提交
426
  VLOG(3) << "use ScopeBufferedSSAGraphExecutor";
Q
Qiao Longfei 已提交
427 428 429 430 431
  if (!build_strategy.async_mode_) {
    member_->executor_.reset(new details::ScopeBufferedSSAGraphExecutor(
        exec_strategy, member_->local_scopes_, std::move(var_infos),
        member_->places_, std::move(member_->executor_)));
  }
Y
Yu Yang 已提交
432 433
}

Y
Yancey1989 已提交
434
void ParallelExecutor::BCastParamsToDevices(
Y
Yan Xu 已提交
435
    const std::vector<std::string> &vars, int trainer_id) const {
Q
Qiao Longfei 已提交
436
  VLOG(3) << "BCastParamsToDevices";
X
Xin Pan 已提交
437
  // the initializing bcast, all vars would be bcast from device(0).
438
  for (auto &var : vars) {
X
Xin Pan 已提交
439
    framework::Variable *main_var = member_->local_scopes_[0]->FindVar(var);
J
JiayiFeng 已提交
440
    if (main_var == nullptr || !main_var->IsType<LoDTensor>()) {
441 442 443 444
      continue;
    }

    auto &main_tensor = main_var->Get<LoDTensor>();
445
    if (!main_tensor.IsInitialized()) {
M
minqiyang 已提交
446
      VLOG(3) << "one in var not inited, return!";
447 448
      continue;
    }
449 450
    auto &dims = main_tensor.dims();
    if (paddle::platform::is_gpu_place(main_tensor.place())) {
P
peizhilin 已提交
451
#if defined(PADDLE_WITH_CUDA) && !defined(_WIN32)
452
      std::vector<void *> buffers;
C
chengduo 已提交
453
      buffers.reserve(member_->places_.size());
454 455 456 457 458
      size_t numel = main_tensor.numel();
      ncclDataType_t data_type = platform::ToNCCLDataType(main_tensor.type());
      for (size_t i = 0; i < member_->places_.size(); ++i) {
        auto place = member_->places_[i];
        void *buffer;
459

Y
Yan Xu 已提交
460
        if (i == 0 && trainer_id == 0) {
461 462
          buffer = const_cast<void *>(main_tensor.data<void>());
        } else {
Y
Yu Yang 已提交
463
          auto local_scope = member_->local_scopes_[i];
464
          auto *t = local_scope->Var(var)->GetMutable<LoDTensor>();
Y
Update  
Yu Yang 已提交
465
          t->Resize(dims);
466
          buffer = t->mutable_data(place, main_tensor.type());
Y
Update  
Yu Yang 已提交
467
        }
468
        buffers.push_back(buffer);
469
      }
470

471 472 473 474 475 476
      PADDLE_ENFORCE_EQ(member_->places_.size(), buffers.size(),
                        "variables' buffer size to bcast NOT equal to places");
      {
        platform::NCCLGroupGuard guard;
        for (size_t i = 0; i < member_->places_.size(); ++i) {
          auto &nccl_ctx = member_->nccl_ctxs_->at(member_->places_[i]);
X
Xin Pan 已提交
477 478
          platform::dynload::ncclBcast(buffers[i], numel, data_type, 0,
                                       nccl_ctx.comm_, nccl_ctx.stream());
479
        }
480
        member_->nccl_ctxs_->WaitAll();
481
      }
C
chengduoZH 已提交
482 483 484
#else
      PADDLE_THROW("Not compiled with CUDA");
#endif
485 486
    } else {
      platform::CPUPlace cpu;
C
chengduo 已提交
487
      for (size_t i = 1; i < member_->places_.size(); ++i) {
488 489
        auto local_scope = member_->local_scopes_[i];
        auto *t = local_scope->Var(var)->GetMutable<LoDTensor>();
C
chengduo 已提交
490

Q
Qiao Longfei 已提交
491
        auto copy_memory = [&] {
492 493 494
          t->Resize(dims);
          t->mutable_data(cpu, main_tensor.type());
          paddle::framework::TensorCopy(main_tensor, cpu, t);
Q
can run  
Qiao Longfei 已提交
495 496
        };

Q
Qiao Longfei 已提交
497
        auto share_memory = [&] { t->ShareDataWith(main_tensor); };
Q
can run  
Qiao Longfei 已提交
498 499 500 501 502 503 504

        // FIXME(zcd): LR_DECAY_COUNTER should not be shared. This is a hot fix.
        if (member_->build_strategy_.async_mode_) {
          share_memory();
        } else if (member_->use_all_reduce_ || member_->use_cuda_ ||
                   var == "@LR_DECAY_COUNTER@") {
          copy_memory();
505
        } else {
Q
can run  
Qiao Longfei 已提交
506
          share_memory();
507
        }
Y
Yu Yang 已提交
508
      }
Y
Stash  
Yu Yang 已提交
509 510
    }
  }
Y
Yu Yang 已提交
511
}
Y
Yu Yang 已提交
512

Y
Yu Yang 已提交
513 514
void ParallelExecutor::Run(const std::vector<std::string> &fetch_tensors,
                           const std::string &fetched_var_name) {
Y
Yu Yang 已提交
515 516 517
#ifdef WITH_GPERFTOOLS
  if (gProfileStarted) {
    ProfilerFlush();
S
sneaxiy 已提交
518 519
  }
#endif
Y
Yu Yang 已提交
520

X
Xin Pan 已提交
521
  platform::RecordBlock b(0);
S
sneaxiy 已提交
522 523
  if (member_->HasGarbageCollectors()) {
    member_->ResetRuntimeReferenceCount(fetch_tensors, fetched_var_name);
S
sneaxiy 已提交
524
  }
S
sneaxiy 已提交
525 526 527
  auto fetch_data = member_->executor_->Run(fetch_tensors);
  *member_->global_scope_->Var(fetched_var_name)->GetMutable<FeedFetchList>() =
      fetch_data;
Y
Yu Yang 已提交
528
}
Y
Yu Yang 已提交
529

Y
Yu Yang 已提交
530 531 532 533 534 535 536 537 538 539 540 541 542 543 544 545 546 547 548
void ParallelExecutor::FeedTensorsIntoLocalScopes(
    const std::vector<std::unordered_map<std::string, LoDTensor>> &tensors) {
  PADDLE_ENFORCE_EQ(member_->local_scopes_.size(), tensors.size());

  for (size_t i = 0; i < tensors.size(); ++i) {
    auto &map = tensors[i];
    auto *scope = member_->local_scopes_[i];
    for (auto &pair : map) {
      auto *trg = scope->Var(pair.first)->GetMutable<LoDTensor>();
      trg->ShareDataWith(pair.second);
      trg->set_lod(pair.second.lod());
    }
  }
}

void ParallelExecutor::FeedAndSplitTensorIntoLocalScopes(
    const std::unordered_map<std::string, LoDTensor> &tensors) {
  for (auto pair : tensors) {
    auto lod_tensors = pair.second.SplitLoDTensor(member_->places_);
549 550 551 552 553
    PADDLE_ENFORCE_EQ(
        member_->places_.size(), lod_tensors.size(),
        "The number of samples of current batch is less than the count of "
        "devices, currently, it is not allowed. (%d vs %d)",
        member_->places_.size(), lod_tensors.size());
X
Xin Pan 已提交
554 555
    for (size_t j = 0; j < member_->places_.size(); ++j) {
      // TODO(panxy0718): Do I need to delete this var?
556
      auto t =
Y
Yu Yang 已提交
557
          member_->local_scopes_[j]->Var(pair.first)->GetMutable<LoDTensor>();
558 559
      t->ShareDataWith(lod_tensors[j]);
      t->set_lod(lod_tensors[j].lod());
X
Xin Pan 已提交
560 561 562 563
    }
  }
}

X
Xin Pan 已提交
564 565 566 567 568 569 570
ParallelExecutor::~ParallelExecutor() {
  for (auto &p : member_->places_) {
    platform::DeviceContextPool::Instance().Get(p)->Wait();
  }
  delete member_;
}

571
bool ParallelExecutor::EnableParallelGraphExecution(
X
Xin Pan 已提交
572
    const ir::Graph &graph, const ExecutionStrategy &exec_strategy,
573
    const BuildStrategy &build_strategy) const {
Y
Yancey1989 已提交
574
  if (!FLAGS_enable_parallel_graph) return false;
575

Y
Yancey1989 已提交
576
  bool enable_parallel_graph = true;
577

X
Xin Pan 已提交
578 579 580 581 582 583 584 585 586 587 588 589 590
  for (ir::Node *node : graph.Nodes()) {
    if (node->IsVar() && node->Var()) {
      // TODO(Yancey1989): support sparse update in ParallelGraph mode.
      if (node->Var()->GetType() == proto::VarType::SELECTED_ROWS) {
        enable_parallel_graph = false;
        break;
      }
    } else if (node->IsOp() && node->Op()) {
      // TODO(Yancey1989): support pserver mode
      if (node->Op()->Type() == "send" || node->Op()->Type() == "recv") {
        enable_parallel_graph = false;
        break;
      }
591 592 593 594 595
    }
  }

  if (!member_->use_all_reduce_ || !member_->use_cuda_)

Y
Yancey1989 已提交
596 597 598
    if (build_strategy.enable_sequential_execution_ ||
        exec_strategy.type_ == ExecutionStrategy::ExecutorType::kExperimental)
      enable_parallel_graph = false;
Y
Yancey1989 已提交
599
  return enable_parallel_graph;
600 601
}

Y
Yu Yang 已提交
602
}  // namespace framework
Y
Yang Yang 已提交
603
}  // namespace paddle
S
sneaxiy 已提交
604

S
sneaxiy 已提交
605
USE_PASS(reference_count_pass);
S
sneaxiy 已提交
606
USE_PASS(eager_deletion_pass);