parallel_executor.cc 18.9 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"
16
#include <algorithm>
C
chengduoZH 已提交
17
#include <string>
18
#include <tuple>
Q
qiaolongfei 已提交
19
#include <vector>
C
chengduo 已提交
20
#include "paddle/fluid/framework/ir/graph_helper.h"
21

X
clean  
Xin Pan 已提交
22
#include "paddle/fluid/framework/ir/graph.h"
X
Xin Pan 已提交
23

24
#include "paddle/fluid/framework/details/all_reduce_deps_pass.h"
Y
yuyang18 已提交
25
#include "paddle/fluid/framework/details/fast_threaded_ssa_graph_executor.h"
26
#include "paddle/fluid/framework/details/multi_devices_helper.h"
Y
Yancey1989 已提交
27
#include "paddle/fluid/framework/details/parallel_ssa_graph_executor.h"
S
sneaxiy 已提交
28
#include "paddle/fluid/framework/details/reference_count_pass_helper.h"
29
#include "paddle/fluid/framework/details/scope_buffered_ssa_graph_executor.h"
30
#include "paddle/fluid/framework/details/threaded_ssa_graph_executor.h"
31
#include "paddle/fluid/platform/profiler.h"
32

33
#ifdef WITH_GPERFTOOLS
Y
Yu Yang 已提交
34
#include "gperftools/profiler.h"
35
#endif
Y
Yu Yang 已提交
36
DEFINE_string(pe_profile_fname, "",
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.");
41

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

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

66 67 68 69 70 71 72 73 74 75 76
  ~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 已提交
77

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

  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
    }
  }

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 114 115 116
  // 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_
  std::vector<details::ReferenceCountMap> global_ref_cnts_;
  std::vector<details::AtomicReferenceCountMap> runtime_ref_cnts_;
  details::GarbageCollectorMap gcs_;
Y
Yu Yang 已提交
117 118
};

S
sneaxiy 已提交
119 120 121 122 123 124 125
std::unique_ptr<ir::Graph> ParallelExecutorPrivate::PrepareGCAndRefCnts(
    std::unique_ptr<ir::Graph> graph, size_t max_memory_size) {
  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;
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
  }

153
  if (!gcs_.empty()) {
S
sneaxiy 已提交
154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179
    std::vector<details::LastLiveOpsOfVars> last_live_ops_of_vars;

    auto ref_cnt_pass =
        ir::PassRegistry::Instance().Get("reference_count_pass");
    ref_cnt_pass->SetNotOwned(details::kGlobalReferenceCount,
                              &global_ref_cnts_);
    ref_cnt_pass->SetNotOwned(details::kLastLiveOpsOfVars,
                              &last_live_ops_of_vars);
    graph = ref_cnt_pass->Apply(std::move(graph));
    VLOG(10) << "ReferenceCountPass Applied";

    auto eager_deletion_pass =
        ir::PassRegistry::Instance().Get("eager_deletion_pass");
    eager_deletion_pass->SetNotOwned(details::kRuntimeReferenceCount,
                                     &runtime_ref_cnts_);
    eager_deletion_pass->SetNotOwned(details::kGarbageCollector, &gcs_);
    eager_deletion_pass->SetNotOwned(details::kLastLiveOpsOfVars,
                                     &last_live_ops_of_vars);
    eager_deletion_pass->SetNotOwned(details::kAllPlaces, &places_);
    graph = eager_deletion_pass->Apply(std::move(graph));
    VLOG(10) << "EagerDeletionPass Applied";
  }

  return graph;
}

180 181 182 183
std::vector<Scope *> &ParallelExecutor::GetLocalScopes() {
  return member_->local_scopes_;
}

184 185 186 187 188 189 190 191
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 已提交
192
    : member_(new ParallelExecutorPrivate(places)) {
Y
Yu Yang 已提交
193
  member_->global_scope_ = scope;
194
  member_->use_cuda_ = exec_strategy.use_cuda_;
195
  member_->build_strategy_ = build_strategy;
196 197
  member_->use_all_reduce_ =
      build_strategy.reduce_ == BuildStrategy::ReduceStrategy::kAllReduce;
X
Xin Pan 已提交
198
  member_->nranks_ = build_strategy.num_trainers_ * places.size();
199 200 201 202
  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 已提交
203 204
  }

205
  // Step 1. Bcast the bcast_vars to devs.
206
  // Create local scopes
207
  if (local_scopes.empty()) {
C
chengduoZH 已提交
208
    member_->own_local_scope_ = true;
Y
Yu Yang 已提交
209 210
    member_->local_scopes_.emplace_back(member_->global_scope_);
    for (size_t i = 1; i < member_->places_.size(); ++i) {
Y
Debug  
Yu Yang 已提交
211
      member_->local_scopes_.emplace_back(&scope->NewScope());
212 213
    }
  } else {
C
chengduoZH 已提交
214
    member_->own_local_scope_ = false;
215 216
    PADDLE_ENFORCE_EQ(member_->places_.size(), local_scopes.size());
    for (size_t i = 0; i < member_->places_.size(); ++i) {
217
      member_->local_scopes_.emplace_back(&local_scopes[i]->NewScope());
218
    }
219 220
  }

X
Xin Pan 已提交
221 222
  std::unique_ptr<ir::Graph> temp_owned_graph(graph);

Y
Yancey1989 已提交
223 224 225
  // FIXME(Yancey1989): parallel graph mode get better performance
  // in GPU allreduce distributed training. Need an elegant way to
  // choice the execution strategy.
X
Xin Pan 已提交
226 227
  build_strategy.enable_parallel_graph_ = EnableParallelGraphExecution(
      *temp_owned_graph, exec_strategy, build_strategy);
Y
Yancey1989 已提交
228 229 230 231
  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 已提交
232

C
chengduoZH 已提交
233
  if (member_->use_cuda_) {
Y
Yu Yang 已提交
234
// Bcast Parameters to all GPUs
P
peizhilin 已提交
235
#if defined(PADDLE_WITH_CUDA) && !defined(_WIN32)
Y
Yancey1989 已提交
236 237 238
    ncclUniqueId *nccl_id = nullptr;
    // gen_nccl_id operator can broadcast the ncclUniqueId for nccl2 collective
    // distributed training
C
chengduoZH 已提交
239
    auto *nccl_id_var = scope->FindVar(NCCL_ID_VARNAME);
Y
Yancey1989 已提交
240
    if (nccl_id_var != nullptr) {
Y
Yancey1989 已提交
241
      nccl_id = nccl_id_var->GetMutable<ncclUniqueId>();
Y
Yancey1989 已提交
242
    }
X
Xin Pan 已提交
243
    if (build_strategy.enable_parallel_graph_ && member_->nranks_ > 1UL) {
Y
Yancey1989 已提交
244 245 246 247
      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 已提交
248
      }
C
chengduoZH 已提交
249
    }
Y
Yancey1989 已提交
250

C
chengduoZH 已提交
251
    member_->nccl_ctxs_.reset(new platform::NCCLContextMap(
252 253
        member_->places_, nccl_id, build_strategy.num_trainers_,
        build_strategy.trainer_id_));
C
chengduoZH 已提交
254 255
#else
    PADDLE_THROW("Not compiled with CUDA");
Y
Yu Yang 已提交
256
#endif
C
chengduoZH 已提交
257
  }
258 259 260 261 262 263 264 265 266 267 268 269 270 271 272
  // 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_);
273
  }
274

X
Xin Pan 已提交
275
// Startup Program has been run. All local scopes has correct parameters.
276

X
Xin Pan 已提交
277 278
// Step 2. Convert main_program to SSA form and dependency graph. Also, insert
// ncclOp
P
peizhilin 已提交
279
#if defined(PADDLE_WITH_CUDA) && !defined(_WIN32)
X
Xin Pan 已提交
280

X
Xin Pan 已提交
281 282 283 284
  temp_owned_graph = build_strategy.Apply(
      std::move(temp_owned_graph), member_->places_, loss_var_name,
      member_->local_scopes_, member_->nranks_, member_->use_cuda_,
      member_->nccl_ctxs_.get());
X
Xin Pan 已提交
285
#else
X
Xin Pan 已提交
286 287
  temp_owned_graph = build_strategy.Apply(
      std::move(temp_owned_graph), member_->places_, loss_var_name,
X
Xin Pan 已提交
288
      member_->local_scopes_, member_->nranks_, member_->use_cuda_);
X
Xin Pan 已提交
289

Y
Yu Yang 已提交
290
#endif
291
  auto max_memory_size = GetEagerDeletionThreshold();
D
dzhwinter 已提交
292 293
  VLOG(10) << "Eager Deletion Threshold "
           << static_cast<float>(max_memory_size) / (1 << 30);
Y
Yancey1989 已提交
294
  if (max_memory_size >= 0) {
X
Xin Pan 已提交
295 296 297 298 299 300
    graph = member_
                ->PrepareGCAndRefCnts(std::move(temp_owned_graph),
                                      static_cast<size_t>(max_memory_size))
                .release();
  } else {
    graph = temp_owned_graph.release();
301 302
  }

303 304
  // Step 3. Create vars in each scope. Passes may also create new vars.
  //         skip control vars and empty vars
Y
Yancey1989 已提交
305
  std::vector<details::VariableInfo> var_infos;
Y
Yancey1989 已提交
306 307 308 309 310 311
  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 已提交
312 313
    }
  }
Y
Yancey1989 已提交
314

315 316
  // If the loss_var_name is given, the number of graph should be only one.
  if (loss_var_name.size()) {
Y
Yancey1989 已提交
317
    size_t graph_num = ir::GraphNum(*graph);
318 319 320 321
    if (graph_num > 1) {
      LOG(WARNING)
          << "The number of graph should be only one, "
             "but the current graph has "
Y
Yancey1989 已提交
322
          << ir::GraphNum(*graph)
323 324 325 326 327
          << " 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.";
    }
328 329
  }

Y
Yancey1989 已提交
330
  if (build_strategy.enable_parallel_graph_) {
Y
Yancey1989 已提交
331
#ifdef PADDLE_WITH_CUDA
332 333
    // TODO(Yancey1989): Remove passing in the main_program when
    // allreduce_seq_pass doesn't need it as the attr.
Y
Yancey1989 已提交
334
    member_->executor_.reset(new details::ParallelSSAGraphExecutor(
X
Xin Pan 已提交
335
        exec_strategy, member_->local_scopes_, member_->places_, graph));
Y
Yancey1989 已提交
336 337 338 339
#else
    PADDLE_THROW(
        "Paddle should be compiled with CUDA for ParallelGraph Execution.");
#endif
X
Xin Pan 已提交
340 341 342 343 344 345 346
  } else {
    if (exec_strategy.type_ == ExecutionStrategy::kDefault) {
      member_->executor_.reset(new details::ThreadedSSAGraphExecutor(
          exec_strategy, member_->local_scopes_, member_->places_, graph));
    } else {
      member_->executor_.reset(new details::FastThreadedSSAGraphExecutor(
          exec_strategy, member_->local_scopes_, member_->places_, graph));
Y
Yancey1989 已提交
347
    }
C
chengduoZH 已提交
348
  }
349 350

  member_->executor_.reset(new details::ScopeBufferedSSAGraphExecutor(
Y
Yancey1989 已提交
351
      exec_strategy, member_->local_scopes_, std::move(var_infos),
352
      member_->places_, std::move(member_->executor_)));
353 354
}

Y
Yancey1989 已提交
355
void ParallelExecutor::BCastParamsToDevices(
356
    const std::vector<std::string> &vars, int trainer_id) const {
357
  // the initializing bcast, all vars would be bcast from device(0).
358
  for (auto &var : vars) {
359
    framework::Variable *main_var = member_->local_scopes_[0]->FindVar(var);
J
JiayiFeng 已提交
360
    if (main_var == nullptr || !main_var->IsType<LoDTensor>()) {
361 362 363 364
      continue;
    }

    auto &main_tensor = main_var->Get<LoDTensor>();
365
    if (!main_tensor.IsInitialized()) {
M
minqiyang 已提交
366
      VLOG(3) << "one in var not inited, return!";
367 368
      continue;
    }
369 370
    auto &dims = main_tensor.dims();
    if (paddle::platform::is_gpu_place(main_tensor.place())) {
P
peizhilin 已提交
371
#if defined(PADDLE_WITH_CUDA) && !defined(_WIN32)
372
      std::vector<void *> buffers;
373
      buffers.reserve(member_->places_.size());
374 375 376 377 378
      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;
379

380
        if (i == 0 && trainer_id == 0) {
381 382
          buffer = const_cast<void *>(main_tensor.data<void>());
        } else {
Y
Yu Yang 已提交
383
          auto local_scope = member_->local_scopes_[i];
384
          auto *t = local_scope->Var(var)->GetMutable<LoDTensor>();
Y
Update  
Yu Yang 已提交
385
          t->Resize(dims);
386
          buffer = t->mutable_data(place, main_tensor.type());
Y
Update  
Yu Yang 已提交
387
        }
388
        buffers.push_back(buffer);
389
      }
390

391 392 393 394 395 396
      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]);
397 398
          platform::dynload::ncclBcast(buffers[i], numel, data_type, 0,
                                       nccl_ctx.comm_, nccl_ctx.stream());
399
        }
400
        member_->nccl_ctxs_->WaitAll();
401
      }
C
chengduoZH 已提交
402 403 404
#else
      PADDLE_THROW("Not compiled with CUDA");
#endif
405 406
    } else {
      platform::CPUPlace cpu;
407
      for (size_t i = 1; i < member_->places_.size(); ++i) {
408 409
        auto local_scope = member_->local_scopes_[i];
        auto *t = local_scope->Var(var)->GetMutable<LoDTensor>();
410 411 412 413

        // FIXME(zcd): LR_DECAY_COUNTER should not be shared. This is a hot fix.
        if (member_->use_all_reduce_ || member_->use_cuda_ ||
            var == "@LR_DECAY_COUNTER@") {
414 415 416 417 418 419
          t->Resize(dims);
          t->mutable_data(cpu, main_tensor.type());
          paddle::framework::TensorCopy(main_tensor, cpu, t);
        } else {
          t->ShareDataWith(main_tensor);
        }
420
      }
Y
Stash  
Yu Yang 已提交
421 422
    }
  }
Y
Yu Yang 已提交
423
}
424

425 426
void ParallelExecutor::Run(const std::vector<std::string> &fetch_tensors,
                           const std::string &fetched_var_name) {
427 428 429
#ifdef WITH_GPERFTOOLS
  if (gProfileStarted) {
    ProfilerFlush();
S
sneaxiy 已提交
430 431
  }
#endif
432

433
  platform::RecordBlock b(0);
S
sneaxiy 已提交
434 435
  if (member_->HasGarbageCollectors()) {
    member_->ResetRuntimeReferenceCount(fetch_tensors, fetched_var_name);
S
sneaxiy 已提交
436
  }
S
sneaxiy 已提交
437 438 439
  auto fetch_data = member_->executor_->Run(fetch_tensors);
  *member_->global_scope_->Var(fetched_var_name)->GetMutable<FeedFetchList>() =
      fetch_data;
440
}
441

442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460
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_);
461 462 463 464 465
    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 已提交
466 467
    for (size_t j = 0; j < member_->places_.size(); ++j) {
      // TODO(panxy0718): Do I need to delete this var?
468
      auto t =
469
          member_->local_scopes_[j]->Var(pair.first)->GetMutable<LoDTensor>();
470 471
      t->ShareDataWith(lod_tensors[j]);
      t->set_lod(lod_tensors[j].lod());
X
Xin Pan 已提交
472 473 474 475
    }
  }
}

X
Xin Pan 已提交
476 477 478 479 480 481 482
ParallelExecutor::~ParallelExecutor() {
  for (auto &p : member_->places_) {
    platform::DeviceContextPool::Instance().Get(p)->Wait();
  }
  delete member_;
}

X
Xin Pan 已提交
483 484 485
bool ParallelExecutor::EnableParallelGraphExecution(
    const ir::Graph &graph, const ExecutionStrategy &exec_strategy,
    const BuildStrategy &build_strategy) const {
Y
Yancey1989 已提交
486
  if (!FLAGS_enable_parallel_graph) return false;
487

Y
Yancey1989 已提交
488
  bool enable_parallel_graph = true;
489

X
Xin Pan 已提交
490 491 492 493 494 495 496 497 498 499 500 501 502
  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;
      }
503 504 505
    }
  }

Y
Yancey1989 已提交
506
  if (!member_->use_all_reduce_ || !member_->use_cuda_)
507

Y
Yancey1989 已提交
508 509 510
    if (build_strategy.enable_sequential_execution_ ||
        exec_strategy.type_ == ExecutionStrategy::ExecutorType::kExperimental)
      enable_parallel_graph = false;
Y
Yancey1989 已提交
511
  return enable_parallel_graph;
512 513
}

514
}  // namespace framework
Y
Yang Yang 已提交
515
}  // namespace paddle
S
sneaxiy 已提交
516

S
sneaxiy 已提交
517
USE_PASS(reference_count_pass);
S
sneaxiy 已提交
518
USE_PASS(eager_deletion_pass);
新手
引导
客服 返回
顶部