parallel_executor.cc 10.7 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>
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

Y
yuyang18 已提交
159 160 161 162 163 164
  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 已提交
165
  }
Y
yuyang18 已提交
166 167 168 169

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

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

    auto &main_tensor = main_var->Get<LoDTensor>();
    auto &dims = main_tensor.dims();
    if (paddle::platform::is_gpu_place(main_tensor.place())) {
C
chengduoZH 已提交
184
#ifdef PADDLE_WITH_CUDA
185
      std::vector<void *> buffers;
186 187 188 189 190
      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;
191

X
Xin Pan 已提交
192
        if (i == 0) {
193 194
          buffer = const_cast<void *>(main_tensor.data<void>());
        } else {
Y
Yu Yang 已提交
195
          auto local_scope = member_->local_scopes_[i];
196
          auto *t = local_scope->Var(var)->GetMutable<LoDTensor>();
Y
Update  
Yu Yang 已提交
197
          t->Resize(dims);
198
          buffer = t->mutable_data(place, main_tensor.type());
Y
Update  
Yu Yang 已提交
199
        }
200
        buffers.push_back(buffer);
201
      }
202

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

222 223
        auto local_scope = member_->local_scopes_[i];
        auto *t = local_scope->Var(var)->GetMutable<LoDTensor>();
C
chengduo 已提交
224 225 226 227

        // 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@") {
228 229 230 231 232 233
          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 已提交
234
      }
Y
Stash  
Yu Yang 已提交
235 236
    }
  }
Y
Yu Yang 已提交
237
}
Y
Yu Yang 已提交
238

Y
Yu Yang 已提交
239 240
void ParallelExecutor::Run(const std::vector<std::string> &fetch_tensors,
                           const std::string &fetched_var_name) {
X
Xin Pan 已提交
241
  platform::RecordBlock b(0);
S
sneaxiy 已提交
242 243 244 245 246
#ifdef PADDLE_WITH_CUDA
  if (!gcs_.empty()) {
    ResetReferenceCount();
  }
#endif
Y
Yu Yang 已提交
247 248 249
  auto fetch_data = member_->executor_->Run(fetch_tensors);
  *member_->global_scope_->Var(fetched_var_name)->GetMutable<FeedFetchList>() =
      fetch_data;
Y
Yu Yang 已提交
250
}
Y
Yu Yang 已提交
251

Y
Yu Yang 已提交
252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270
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_);
271 272 273 274 275
    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 已提交
276 277
    for (size_t j = 0; j < member_->places_.size(); ++j) {
      // TODO(panxy0718): Do I need to delete this var?
278
      auto t =
Y
Yu Yang 已提交
279
          member_->local_scopes_[j]->Var(pair.first)->GetMutable<LoDTensor>();
280 281
      t->ShareDataWith(lod_tensors[j]);
      t->set_lod(lod_tensors[j].lod());
X
Xin Pan 已提交
282 283 284 285
    }
  }
}

286
ParallelExecutor::~ParallelExecutor() {
C
chengduoZH 已提交
287
  if (member_->own_local_scope_) {
288
    for (size_t i = 1; i < member_->local_scopes_.size(); ++i) {
M
minqiyang 已提交
289 290 291 292
      Scope *local_scope = member_->local_scopes_[i];
      if (member_->global_scope_->HasKid(local_scope)) {
        member_->global_scope_->DeleteScope(local_scope);
      }
293 294 295 296
    }
  }
}

Y
Yu Yang 已提交
297
}  // namespace framework
Y
Yang Yang 已提交
298
}  // namespace paddle
S
sneaxiy 已提交
299 300 301
#ifdef PADDLE_WITH_CUDA
USE_PASS(reference_count_pass);
#endif