parallel_executor.cc 14.6 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"
Q
qiaolongfei 已提交
16

C
chengduoZH 已提交
17
#include <string>
18
#include <tuple>
Q
qiaolongfei 已提交
19
#include <vector>
20

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

Y
Yu Yang 已提交
24
#ifdef PADDLE_WITH_CUDA
25
#include "paddle/fluid/platform/nccl_helper.h"
Y
Yu Yang 已提交
26
#endif
Y
Yang Yang 已提交
27

Y
yuyang18 已提交
28
#include "paddle/fluid/framework/details/fast_threaded_ssa_graph_executor.h"
X
Xin Pan 已提交
29 30
#include "paddle/fluid/framework/details/multi_devices_graph_check_pass.h"
#include "paddle/fluid/framework/details/multi_devices_graph_print_pass.h"
31
#include "paddle/fluid/framework/details/scope_buffered_ssa_graph_executor.h"
32
#include "paddle/fluid/framework/details/threaded_ssa_graph_executor.h"
33
#include "paddle/fluid/platform/profiler.h"
34

Y
Yang Yang 已提交
35
namespace paddle {
36 37
namespace framework {

X
Xin Pan 已提交
38 39 40 41
std::unique_ptr<ir::Graph> ApplyParallelExecutorPass(
    const ProgramDesc &main_program, const std::vector<platform::Place> &places,
    const std::string &loss_var_name,
    const std::unordered_set<std::string> &param_names,
42
    const std::vector<Scope *> &local_scopes, const bool use_cuda,
X
Xin Pan 已提交
43 44 45 46 47
#ifdef PADDLE_WITH_CUDA
    const BuildStrategy &strategy, platform::NCCLContextMap *nccl_ctxs) {
#else
    const BuildStrategy &strategy) {
#endif
X
Xin Pan 已提交
48
  // Convert the program to graph.
X
Xin Pan 已提交
49
  std::unique_ptr<ir::Graph> graph(new ir::Graph(main_program));
X
Xin Pan 已提交
50 51

  // Apply a graph viz pass to record a graph.
X
Xin Pan 已提交
52 53 54 55 56 57 58 59
  if (!strategy.debug_graphviz_path_.empty()) {
    auto viz_pass = ir::PassRegistry::Instance().Get("graph_viz_pass");
    const std::string graph_path = string::Sprintf(
        "%s%s", strategy.debug_graphviz_path_.c_str(), "_original_graph");
    viz_pass->Set<std::string>("graph_viz_path", new std::string(graph_path));
    graph = viz_pass->Apply(std::move(graph));
  }

60 61 62 63 64 65 66 67 68 69 70 71 72 73 74
  // Apply op fusion.
  if (strategy.fuse_elewise_add_act_ops_) {
    auto fuse_elewise_add_act_pass =
        ir::PassRegistry::Instance().Get("fuse_elewise_add_act_pass");
    graph = fuse_elewise_add_act_pass->Apply(std::move(graph));
    // Apply a graph viz pass to record a graph.
    if (!strategy.debug_graphviz_path_.empty()) {
      auto viz_pass = ir::PassRegistry::Instance().Get("graph_viz_pass");
      const std::string graph_path = string::Sprintf(
          "%s%s", strategy.debug_graphviz_path_.c_str(), "_fused_graph");
      viz_pass->Set<std::string>("graph_viz_path", new std::string(graph_path));
      graph = viz_pass->Apply(std::move(graph));
    }
  }

X
Xin Pan 已提交
75
  // Convert graph to run on multi-devices.
X
Xin Pan 已提交
76 77 78 79 80 81 82
  auto multi_devices_pass =
      ir::PassRegistry::Instance().Get("multi_devices_pass");
  multi_devices_pass->SetNotOwned<const std::vector<platform::Place>>("places",
                                                                      &places);
  multi_devices_pass->SetNotOwned<const std::string>("loss_var_name",
                                                     &loss_var_name);
  multi_devices_pass->SetNotOwned<const std::unordered_set<std::string>>(
X
Xin Pan 已提交
83
      "params", &param_names);
84 85
  multi_devices_pass->SetNotOwned<const std::vector<Scope *>>("local_scopes",
                                                              &local_scopes);
X
Xin Pan 已提交
86
  multi_devices_pass->SetNotOwned<const BuildStrategy>("strategy", &strategy);
X
Xin Pan 已提交
87 88 89

#ifdef PADDLE_WITH_CUDA
  platform::NCCLContextMap *nctx = use_cuda ? nccl_ctxs : nullptr;
X
Xin Pan 已提交
90
  multi_devices_pass->SetNotOwned<platform::NCCLContextMap>("nccl_ctxs", nctx);
X
Xin Pan 已提交
91
#endif
X
Xin Pan 已提交
92
  graph = multi_devices_pass->Apply(std::move(graph));
X
Xin Pan 已提交
93

X
Xin Pan 已提交
94
  // Apply a graph print pass to record a graph with device info.
X
Xin Pan 已提交
95
  if (!strategy.debug_graphviz_path_.empty()) {
X
Xin Pan 已提交
96 97 98
    auto multi_devices_print_pass =
        ir::PassRegistry::Instance().Get("multi_devices_print_pass");
    multi_devices_print_pass->SetNotOwned<const std::string>(
X
Xin Pan 已提交
99
        "debug_graphviz_path", &strategy.debug_graphviz_path_);
X
Xin Pan 已提交
100
    multi_devices_print_pass->Set<details::GraphvizSSAGraphPrinter>(
X
Xin Pan 已提交
101
        "graph_printer", new details::GraphvizSSAGraphPrinter);
X
Xin Pan 已提交
102
    graph = multi_devices_print_pass->Apply(std::move(graph));
X
Xin Pan 已提交
103 104
  }

X
Xin Pan 已提交
105
  // Verify that the graph is correct for multi-device executor.
X
Xin Pan 已提交
106 107 108
  auto multi_devices_check_pass =
      ir::PassRegistry::Instance().Get("multi_devices_check_pass");
  graph = multi_devices_check_pass->Apply(std::move(graph));
X
Xin Pan 已提交
109 110 111
  return graph;
}

Y
Yu Yang 已提交
112 113 114
class ParallelExecutorPrivate {
 public:
  explicit ParallelExecutorPrivate(const std::vector<platform::Place> &places)
Y
Yu Yang 已提交
115
      : places_(places) {}
Y
Yu Yang 已提交
116 117 118 119

  std::vector<platform::Place> places_;
  std::vector<Scope *> local_scopes_;
  Scope *global_scope_;
Y
Yu Yang 已提交
120
  std::unique_ptr<details::SSAGraphExecutor> executor_;
Y
Yu Yang 已提交
121

Y
Yu Yang 已提交
122
#ifdef PADDLE_WITH_CUDA
Y
Yu Yang 已提交
123
  std::unique_ptr<platform::NCCLContextMap> nccl_ctxs_;
Y
Yu Yang 已提交
124
#endif
C
chengduoZH 已提交
125 126
  bool own_local_scope_;
  bool use_cuda_;
127
  bool use_all_reduce_;
Y
Yu Yang 已提交
128 129
};

130 131 132 133
std::vector<Scope *> &ParallelExecutor::GetLocalScopes() {
  return member_->local_scopes_;
}

134
ParallelExecutor::ParallelExecutor(
135
    const std::vector<platform::Place> &places,
136
    const std::unordered_set<std::string> &params,
137 138
    const std::unordered_set<std::string> &bcast_vars,
    const ProgramDesc &main_program, const std::string &loss_var_name,
139
    Scope *scope, const std::vector<Scope *> &local_scopes,
140
    const ExecutionStrategy &exec_strategy, const BuildStrategy &build_strategy,
141
    size_t num_trainers, size_t trainer_id)
Y
Yu Yang 已提交
142
    : member_(new ParallelExecutorPrivate(places)) {
Y
Yu Yang 已提交
143
  member_->global_scope_ = scope;
144
  member_->use_cuda_ = exec_strategy.use_cuda_;
145 146 147 148 149 150 151 152
  member_->use_all_reduce_ =
      build_strategy.reduce_ == BuildStrategy::ReduceStrategy::kAllReduce;

  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.");
  }
153

154
  // Step 1. Bcast the params to devs.
155
  // Create local scopes
156
  if (local_scopes.empty()) {
C
chengduoZH 已提交
157
    member_->own_local_scope_ = true;
Y
Yu Yang 已提交
158 159
    member_->local_scopes_.emplace_back(member_->global_scope_);
    for (size_t i = 1; i < member_->places_.size(); ++i) {
Y
Debug  
Yu Yang 已提交
160
      member_->local_scopes_.emplace_back(&scope->NewScope());
161 162
    }
  } else {
C
chengduoZH 已提交
163
    member_->own_local_scope_ = false;
164 165
    PADDLE_ENFORCE_EQ(member_->places_.size(), local_scopes.size());
    for (size_t i = 0; i < member_->places_.size(); ++i) {
166
      member_->local_scopes_.emplace_back(&local_scopes[i]->NewScope());
167
    }
168 169
  }

C
chengduoZH 已提交
170
  if (member_->use_cuda_) {
Y
Yu Yang 已提交
171 172
// Bcast Parameters to all GPUs
#ifdef PADDLE_WITH_CUDA
C
chengduoZH 已提交
173 174 175 176 177 178 179 180 181
    auto *nccl_id_var = scope->FindVar(NCCL_ID_VARNAME);
    ncclUniqueId *nccl_id = nullptr;
    if (nccl_id_var != nullptr) {
      nccl_id = nccl_id_var->GetMutable<ncclUniqueId>();
    }
    member_->nccl_ctxs_.reset(new platform::NCCLContextMap(
        member_->places_, nccl_id, num_trainers, trainer_id));
#else
    PADDLE_THROW("Not compiled with CUDA");
Y
Yu Yang 已提交
182
#endif
C
chengduoZH 已提交
183 184 185
  }

  if (member_->local_scopes_.size() != 1 && local_scopes.empty()) {
Y
Yancey1989 已提交
186
    BCastParamsToDevices(bcast_vars);
187
  }
188 189 190 191 192 193 194 195 196 197
  // Startup Program has been run. All local scopes has correct parameters.

  // Step 2. Create vars in each scope;
  std::vector<details::VariableInfo> var_infos;
  for (auto *var : main_program.Block(0).AllVars()) {
    var_infos.emplace_back();
    var_infos.back().name_ = var->Name();
    var_infos.back().type_ = var->GetType();
    var_infos.back().persistable_ = var->Persistable();
  }
198

X
Xin Pan 已提交
199 200
// Step 3. Convert main_program to SSA form and dependency graph. Also, insert
// ncclOp
Y
yuyang18 已提交
201
#ifdef PADDLE_WITH_CUDA
X
Xin Pan 已提交
202 203 204 205
  std::unique_ptr<ir::Graph> graph = ApplyParallelExecutorPass(
      main_program, member_->places_, loss_var_name, params,
      member_->local_scopes_, member_->use_cuda_, build_strategy,
      member_->nccl_ctxs_.get());
S
sneaxiy 已提交
206 207 208 209 210 211 212 213 214 215 216 217

  auto max_memory_size = GetEagerDeletionThreshold();
  if (max_memory_size >= 0) {
    for (auto &place : member_->places_) {
      if (!platform::is_gpu_place(place)) continue;
      auto gpu_place = boost::get<platform::CUDAPlace>(place);
      if (gcs_[gpu_place.device] == nullptr) {
        ref_cnts_[gpu_place.device].reset(new details::ReferenceCountMap());
        cur_ref_cnts_[gpu_place.device].reset(
            new details::AtomicReferenceCountMap());
        gcs_[gpu_place.device].reset(
            new StreamGarbageCollector<Tensor>(gpu_place, max_memory_size));
S
sneaxiy 已提交
218 219
      }
    }
S
sneaxiy 已提交
220 221 222 223 224 225 226 227 228 229
    if (!gcs_.empty()) {
      auto ref_cnt_pass =
          ir::PassRegistry::Instance().Get("reference_count_pass");
      ref_cnt_pass->SetNotOwned(details::kGlobalReferenceCount, &ref_cnts_);
      ref_cnt_pass->SetNotOwned(details::kCurReferenceCount, &cur_ref_cnts_);
      ref_cnt_pass->SetNotOwned(details::kGarbageCollector, &gcs_);
      graph = ref_cnt_pass->Apply(std::move(graph));
      graph->SetNotOwned("garbage_collector", &gcs_);
    }
  }
C
chengduoZH 已提交
230
#else
X
Xin Pan 已提交
231 232 233
  std::unique_ptr<ir::Graph> graph = ApplyParallelExecutorPass(
      main_program, member_->places_, loss_var_name, params,
      member_->local_scopes_, member_->use_cuda_, build_strategy);
Y
Yu Yang 已提交
234
#endif
X
Xin Pan 已提交
235

Y
yuyang18 已提交
236 237 238 239 240 241
  if (exec_strategy.type_ == ExecutionStrategy::kDefault) {
    member_->executor_.reset(new details::ThreadedSSAGraphExecutor(
        exec_strategy, member_->local_scopes_, places, std::move(graph)));
  } else {
    member_->executor_.reset(new details::FastThreadedSSAGraphExecutor(
        exec_strategy, member_->local_scopes_, places, std::move(graph)));
C
chengduoZH 已提交
242
  }
243 244 245 246

  member_->executor_.reset(new details::ScopeBufferedSSAGraphExecutor(
      exec_strategy, member_->local_scopes_, std::move(var_infos),
      member_->places_, std::move(member_->executor_)));
247 248
}

Y
Yancey1989 已提交
249
void ParallelExecutor::BCastParamsToDevices(
250
    const std::unordered_set<std::string> &vars) const {
251
  // the initializing bcast, all vars would be bcast from device(0).
252
  for (auto &var : vars) {
253
    framework::Variable *main_var = member_->local_scopes_[0]->FindVar(var);
J
JiayiFeng 已提交
254
    if (main_var == nullptr || !main_var->IsType<LoDTensor>()) {
255 256 257 258 259 260
      continue;
    }

    auto &main_tensor = main_var->Get<LoDTensor>();
    auto &dims = main_tensor.dims();
    if (paddle::platform::is_gpu_place(main_tensor.place())) {
C
chengduoZH 已提交
261
#ifdef PADDLE_WITH_CUDA
262
      std::vector<void *> buffers;
263 264 265 266 267
      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;
268

269
        if (i == 0) {
270 271
          buffer = const_cast<void *>(main_tensor.data<void>());
        } else {
Y
Yu Yang 已提交
272
          auto local_scope = member_->local_scopes_[i];
273
          auto *t = local_scope->Var(var)->GetMutable<LoDTensor>();
Y
Update  
Yu Yang 已提交
274
          t->Resize(dims);
275
          buffer = t->mutable_data(place, main_tensor.type());
Y
Update  
Yu Yang 已提交
276
        }
277
        buffers.push_back(buffer);
278
      }
279

280 281 282 283 284 285
      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]);
286 287
          platform::dynload::ncclBcast(buffers[i], numel, data_type, 0,
                                       nccl_ctx.comm_, nccl_ctx.stream());
288
        }
289
        member_->nccl_ctxs_->WaitAll();
290
      }
C
chengduoZH 已提交
291 292 293
#else
      PADDLE_THROW("Not compiled with CUDA");
#endif
294 295
    } else {
      platform::CPUPlace cpu;
Y
Yancey1989 已提交
296
      for (size_t i = 0; i < member_->places_.size(); ++i) {
297
        if (i == 0) continue;
Y
Yancey1989 已提交
298

299 300
        auto local_scope = member_->local_scopes_[i];
        auto *t = local_scope->Var(var)->GetMutable<LoDTensor>();
301 302 303 304

        // 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@") {
305 306 307 308 309 310
          t->Resize(dims);
          t->mutable_data(cpu, main_tensor.type());
          paddle::framework::TensorCopy(main_tensor, cpu, t);
        } else {
          t->ShareDataWith(main_tensor);
        }
311
      }
Y
Stash  
Yu Yang 已提交
312 313
    }
  }
Y
Yu Yang 已提交
314
}
315

316 317
void ParallelExecutor::Run(const std::vector<std::string> &fetch_tensors,
                           const std::string &fetched_var_name) {
318
  platform::RecordBlock b(0);
S
sneaxiy 已提交
319 320 321
#ifdef PADDLE_WITH_CUDA
  if (!gcs_.empty()) {
    ResetReferenceCount();
S
sneaxiy 已提交
322 323 324 325 326 327 328
    for (auto &pair : cur_ref_cnts_) {
      auto &name_map = *(pair.second);
      for (auto &fetch_name : fetch_tensors) {
        name_map.erase(fetch_name);
      }
      name_map.erase(fetched_var_name);
    }
S
sneaxiy 已提交
329 330
  }
#endif
331 332 333
  auto fetch_data = member_->executor_->Run(fetch_tensors);
  *member_->global_scope_->Var(fetched_var_name)->GetMutable<FeedFetchList>() =
      fetch_data;
334
}
335

336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354
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_);
355 356 357 358 359
    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 已提交
360 361
    for (size_t j = 0; j < member_->places_.size(); ++j) {
      // TODO(panxy0718): Do I need to delete this var?
362
      auto t =
363
          member_->local_scopes_[j]->Var(pair.first)->GetMutable<LoDTensor>();
364 365
      t->ShareDataWith(lod_tensors[j]);
      t->set_lod(lod_tensors[j].lod());
X
Xin Pan 已提交
366 367 368 369
    }
  }
}

370
ParallelExecutor::~ParallelExecutor() {
C
chengduoZH 已提交
371
  if (member_->own_local_scope_) {
372
    for (size_t i = 1; i < member_->local_scopes_.size(); ++i) {
373 374 375 376
      Scope *local_scope = member_->local_scopes_[i];
      if (member_->global_scope_->HasKid(local_scope)) {
        member_->global_scope_->DeleteScope(local_scope);
      }
377 378 379 380
    }
  }
}

381
}  // namespace framework
Y
Yang Yang 已提交
382
}  // namespace paddle
S
sneaxiy 已提交
383

384
USE_PASS(fuse_elewise_add_act_pass);
S
sneaxiy 已提交
385 386 387 388 389 390 391
USE_PASS(graph_viz_pass);
USE_PASS(multi_devices_pass);
USE_PASS(multi_devices_check_pass);
USE_PASS(multi_devices_print_pass);
#ifdef PADDLE_WITH_CUDA
USE_PASS(reference_count_pass);
#endif
新手
引导
客服 返回
顶部