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

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

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

Y
yuyang18 已提交
27
#include "paddle/fluid/framework/details/fast_threaded_ssa_graph_executor.h"
28
#include "paddle/fluid/framework/details/multi_devices_helper.h"
Y
yuyang18 已提交
29
#include "paddle/fluid/framework/details/scope_buffered_ssa_graph_executor.h"
Y
Yu Yang 已提交
30
#include "paddle/fluid/framework/details/threaded_ssa_graph_executor.h"
31
#include "paddle/fluid/platform/profiler.h"
Y
Yu Yang 已提交
32

Y
Yang Yang 已提交
33
namespace paddle {
Y
Yu Yang 已提交
34 35
namespace framework {

Y
Yu Yang 已提交
36 37 38
class ParallelExecutorPrivate {
 public:
  explicit ParallelExecutorPrivate(const std::vector<platform::Place> &places)
Y
Yu Yang 已提交
39
      : places_(places) {}
Y
Yu Yang 已提交
40 41 42 43

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

Y
Yu Yang 已提交
46
#ifdef PADDLE_WITH_CUDA
Y
Yu Yang 已提交
47
  std::unique_ptr<platform::NCCLContextMap> nccl_ctxs_;
Y
Yu Yang 已提交
48
#endif
C
chengduoZH 已提交
49 50
  bool own_local_scope_;
  bool use_cuda_;
51
  bool use_all_reduce_;
Y
Yu Yang 已提交
52 53
};

54 55 56 57
std::vector<Scope *> &ParallelExecutor::GetLocalScopes() {
  return member_->local_scopes_;
}

Y
Yu Yang 已提交
58
ParallelExecutor::ParallelExecutor(
59
    const std::vector<platform::Place> &places,
Y
Yu Yang 已提交
60
    const std::unordered_set<std::string> &params,
61 62
    const std::unordered_set<std::string> &bcast_vars,
    const ProgramDesc &main_program, const std::string &loss_var_name,
Y
yuyang18 已提交
63
    Scope *scope, const std::vector<Scope *> &local_scopes,
64
    const ExecutionStrategy &exec_strategy, const BuildStrategy &build_strategy,
65
    size_t num_trainers, size_t trainer_id)
Y
Yu Yang 已提交
66
    : member_(new ParallelExecutorPrivate(places)) {
Y
Yu Yang 已提交
67
  member_->global_scope_ = scope;
68
  member_->use_cuda_ = exec_strategy.use_cuda_;
69 70 71 72 73 74 75 76
  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.");
  }
Y
Yu Yang 已提交
77

78
  // Step 1. Bcast the params to devs.
Y
Yu Yang 已提交
79
  // Create local scopes
80
  if (local_scopes.empty()) {
C
chengduoZH 已提交
81
    member_->own_local_scope_ = true;
Y
Yu Yang 已提交
82 83
    member_->local_scopes_.emplace_back(member_->global_scope_);
    for (size_t i = 1; i < member_->places_.size(); ++i) {
Y
Debug  
Yu Yang 已提交
84
      member_->local_scopes_.emplace_back(&scope->NewScope());
85 86
    }
  } else {
C
chengduoZH 已提交
87
    member_->own_local_scope_ = false;
88 89
    PADDLE_ENFORCE_EQ(member_->places_.size(), local_scopes.size());
    for (size_t i = 0; i < member_->places_.size(); ++i) {
90
      member_->local_scopes_.emplace_back(&local_scopes[i]->NewScope());
91
    }
Y
Yu Yang 已提交
92 93
  }

C
chengduoZH 已提交
94
  if (member_->use_cuda_) {
Y
Yu Yang 已提交
95 96
// Bcast Parameters to all GPUs
#ifdef PADDLE_WITH_CUDA
C
chengduoZH 已提交
97 98 99 100 101 102 103 104 105
    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 已提交
106
#endif
C
chengduoZH 已提交
107 108 109
  }

  if (member_->local_scopes_.size() != 1 && local_scopes.empty()) {
Y
Yancey1989 已提交
110
    BCastParamsToDevices(bcast_vars);
Y
Yu Yang 已提交
111
  }
Y
yuyang18 已提交
112 113 114 115 116 117 118 119 120 121
  // 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();
  }
Y
Yu Yang 已提交
122

X
Xin Pan 已提交
123 124
// Step 3. Convert main_program to SSA form and dependency graph. Also, insert
// ncclOp
Y
yuyang18 已提交
125
#ifdef PADDLE_WITH_CUDA
126
  std::unique_ptr<ir::Graph> graph = build_strategy.Apply(
X
Xin Pan 已提交
127
      main_program, member_->places_, loss_var_name, params,
128
      member_->local_scopes_, member_->use_cuda_, member_->nccl_ctxs_.get());
S
sneaxiy 已提交
129 130 131 132 133 134 135 136 137 138 139 140

  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 已提交
141 142
      }
    }
S
sneaxiy 已提交
143 144 145 146 147 148 149 150 151 152
    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 已提交
153
#else
154 155 156
  std::unique_ptr<ir::Graph> graph =
      build_strategy.Apply(main_program, member_->places_, loss_var_name,
                           params, member_->local_scopes_, member_->use_cuda_);
Y
Yu Yang 已提交
157
#endif
X
Xin Pan 已提交
158

C
chengduo 已提交
159 160 161 162 163 164
  if (VLOG_IS_ON(5)) {
    // If the loss_var_name is given, the number of graph should be only one.
    if (loss_var_name.size()) {
      PADDLE_ENFORCE_EQ(ir::GraphNum(*graph), 1,
                        "The number of graph should be only one");
    }
C
chengduo 已提交
165 166
  }

Y
yuyang18 已提交
167 168 169 170 171 172
  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 已提交
173
  }
Y
yuyang18 已提交
174 175 176 177

  member_->executor_.reset(new details::ScopeBufferedSSAGraphExecutor(
      exec_strategy, member_->local_scopes_, std::move(var_infos),
      member_->places_, std::move(member_->executor_)));
Y
Yu Yang 已提交
178 179
}

Y
Yancey1989 已提交
180
void ParallelExecutor::BCastParamsToDevices(
181
    const std::unordered_set<std::string> &vars) const {
X
Xin Pan 已提交
182
  // the initializing bcast, all vars would be bcast from device(0).
183
  for (auto &var : vars) {
X
Xin Pan 已提交
184
    framework::Variable *main_var = member_->local_scopes_[0]->FindVar(var);
J
JiayiFeng 已提交
185
    if (main_var == nullptr || !main_var->IsType<LoDTensor>()) {
186 187 188 189 190 191
      continue;
    }

    auto &main_tensor = main_var->Get<LoDTensor>();
    auto &dims = main_tensor.dims();
    if (paddle::platform::is_gpu_place(main_tensor.place())) {
C
chengduoZH 已提交
192
#ifdef PADDLE_WITH_CUDA
193
      std::vector<void *> buffers;
194 195 196 197 198
      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;
199

X
Xin Pan 已提交
200
        if (i == 0) {
201 202
          buffer = const_cast<void *>(main_tensor.data<void>());
        } else {
Y
Yu Yang 已提交
203
          auto local_scope = member_->local_scopes_[i];
204
          auto *t = local_scope->Var(var)->GetMutable<LoDTensor>();
Y
Update  
Yu Yang 已提交
205
          t->Resize(dims);
206
          buffer = t->mutable_data(place, main_tensor.type());
Y
Update  
Yu Yang 已提交
207
        }
208
        buffers.push_back(buffer);
209
      }
210

211 212 213 214 215 216
      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 已提交
217 218
          platform::dynload::ncclBcast(buffers[i], numel, data_type, 0,
                                       nccl_ctx.comm_, nccl_ctx.stream());
219
        }
220
        member_->nccl_ctxs_->WaitAll();
221
      }
C
chengduoZH 已提交
222 223 224
#else
      PADDLE_THROW("Not compiled with CUDA");
#endif
225 226
    } else {
      platform::CPUPlace cpu;
Y
Yancey1989 已提交
227
      for (size_t i = 0; i < member_->places_.size(); ++i) {
X
Xin Pan 已提交
228
        if (i == 0) continue;
Y
Yancey1989 已提交
229

230 231
        auto local_scope = member_->local_scopes_[i];
        auto *t = local_scope->Var(var)->GetMutable<LoDTensor>();
C
chengduo 已提交
232 233 234 235

        // 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@") {
236 237 238 239 240 241
          t->Resize(dims);
          t->mutable_data(cpu, main_tensor.type());
          paddle::framework::TensorCopy(main_tensor, cpu, t);
        } else {
          t->ShareDataWith(main_tensor);
        }
Y
Yu Yang 已提交
242
      }
Y
Stash  
Yu Yang 已提交
243 244
    }
  }
Y
Yu Yang 已提交
245
}
Y
Yu Yang 已提交
246

Y
Yu Yang 已提交
247 248
void ParallelExecutor::Run(const std::vector<std::string> &fetch_tensors,
                           const std::string &fetched_var_name) {
X
Xin Pan 已提交
249
  platform::RecordBlock b(0);
S
sneaxiy 已提交
250 251 252
#ifdef PADDLE_WITH_CUDA
  if (!gcs_.empty()) {
    ResetReferenceCount();
S
sneaxiy 已提交
253 254 255 256 257 258 259
    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 已提交
260 261
  }
#endif
S
sneaxiy 已提交
262 263 264
  auto fetch_data = member_->executor_->Run(fetch_tensors);
  *member_->global_scope_->Var(fetched_var_name)->GetMutable<FeedFetchList>() =
      fetch_data;
Y
Yu Yang 已提交
265
}
Y
Yu Yang 已提交
266

Y
Yu Yang 已提交
267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285
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_);
286 287 288 289 290
    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 已提交
291 292
    for (size_t j = 0; j < member_->places_.size(); ++j) {
      // TODO(panxy0718): Do I need to delete this var?
293
      auto t =
Y
Yu Yang 已提交
294
          member_->local_scopes_[j]->Var(pair.first)->GetMutable<LoDTensor>();
295 296
      t->ShareDataWith(lod_tensors[j]);
      t->set_lod(lod_tensors[j].lod());
X
Xin Pan 已提交
297 298 299 300
    }
  }
}

301
ParallelExecutor::~ParallelExecutor() {
C
chengduoZH 已提交
302
  if (member_->own_local_scope_) {
303
    for (size_t i = 1; i < member_->local_scopes_.size(); ++i) {
M
minqiyang 已提交
304 305 306 307
      Scope *local_scope = member_->local_scopes_[i];
      if (member_->global_scope_->HasKid(local_scope)) {
        member_->global_scope_->DeleteScope(local_scope);
      }
308 309
    }
  }
S
sneaxiy 已提交
310

S
sneaxiy 已提交
311 312
  // member_ must be destructed before gcs_ since the destructor of
  // ReferenceCountOpHandle use raw pointers of gcs_ inside.
S
sneaxiy 已提交
313
  member_.reset();
314 315
}

Y
Yu Yang 已提交
316
}  // namespace framework
Y
Yang Yang 已提交
317
}  // namespace paddle
S
sneaxiy 已提交
318 319 320
#ifdef PADDLE_WITH_CUDA
USE_PASS(reference_count_pass);
#endif