interpreter_util.cc 31.5 KB
Newer Older
W
wanghuancoder 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13
// Copyright (c) 2021 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.
14

15
#include "paddle/fluid/framework/new_executor/interpreter/interpreter_util.h"
16

17 18
#include <algorithm>

19
#include "paddle/fluid/distributed/auto_parallel/dist_attr.h"
20
#include "paddle/fluid/framework/details/nan_inf_utils.h"
W
wanghuancoder 已提交
21
#include "paddle/fluid/framework/executor_gc_helper.h"
22
#include "paddle/fluid/framework/new_executor/interpreter/data_transfer.h"
23
#include "paddle/fluid/framework/new_executor/interpreter/execution_config.h"
24
#include "paddle/fluid/memory/stats.h"
X
xiongkun 已提交
25 26 27
#include "paddle/fluid/operators/controlflow/conditional_block_op_helper.h"
#include "paddle/fluid/operators/controlflow/recurrent_op_helper.h"
#include "paddle/fluid/operators/controlflow/while_op_helper.h"
28
#include "paddle/fluid/operators/ops_extra_info.h"
29
#include "paddle/phi/core/kernel_context.h"
30
#include "paddle/phi/core/kernel_factory.h"
W
wanghuancoder 已提交
31

L
Leo Chen 已提交
32 33 34
#ifdef PADDLE_WITH_MKLDNN
#include "paddle/fluid/platform/mkldnn_helper.h"
#endif
35 36 37
#if defined(PADDLE_WITH_CUDA) || defined(PADDLE_WITH_HIP)
#include "paddle/fluid/platform/device/gpu/gpu_dnn.h"
#endif
L
Leo Chen 已提交
38

39
PADDLE_DEFINE_EXPORTED_bool(
40 41
    new_executor_serial_run,
    false,
42
    "Enable serial execution for standalone executor, used for debug.");
43

44 45 46 47 48
PADDLE_DEFINE_EXPORTED_bool(
    new_executor_log_memory_stats,
    false,
    "Log memory stats after each op runs, just used for debug.");

49
DECLARE_bool(use_mkldnn);
50
DECLARE_bool(check_nan_inf);
51

W
wanghuancoder 已提交
52 53
namespace paddle {
namespace framework {
54
namespace interpreter {
55

56
using VariableIdMap = std::map<std::string, std::vector<int>>;
L
liutiexing 已提交
57

58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89
// NOTE(Ruibiao): SingleStreamGuard make some multi-strem op (i.e.,
// c_allreduce_sum) run in single stream. It is dedicated to BuildOpFuncList
// which run kernel without stream synchronization.
class SingleStreamGuard {
 public:
  explicit SingleStreamGuard(std::shared_ptr<OperatorBase>& op) : op_(op) {
    if (op_->Type() == "c_allreduce_sum" &&
        op_->Attr<bool>("use_calc_stream") == false) {
      VLOG(6) << "Set c_allredce_sum's attr use_calc_stream to true";
      op_->SetAttr("use_calc_stream", true);
      is_changed = true;
    }
  }

  ~SingleStreamGuard() {
    if (!is_changed) {
      return;
    }

    if (op_->Type() == "c_allreduce_sum") {
      op_->SetAttr("use_calc_stream", false);
      VLOG(6) << "Set c_allredce_sum's attr use_calc_stream to false";
    }
  }

  DISABLE_COPY_AND_ASSIGN(SingleStreamGuard);

 private:
  bool is_changed{false};
  std::shared_ptr<OperatorBase> op_;
};

90
const std::vector<WorkQueueOptions> ConstructWorkQueueOptions(
91
    size_t host_num_threads, size_t device_num_threads, EventsWaiter* waiter) {
92 93 94 95 96 97 98 99 100 101 102 103 104
  std::vector<WorkQueueOptions> group_options;
  // for execute host Kernel
  group_options.emplace_back(/*name*/ "HostTasks",
                             /*num_threads*/ host_num_threads,
                             /*allow_spinning*/ true,
                             /*always_spinning*/ false,
                             /*track_task*/ false,
                             /*detached*/ true,
                             /*events_waiter*/ waiter);
  // for launch device Kernel
  group_options.emplace_back(/*name*/ "DeviceKernelLaunch",
                             /*num_threads*/ device_num_threads,
                             /*allow_spinning*/ true,
105
                             /*always_spinning*/ false,
106 107 108 109 110 111 112 113 114 115
                             /*track_task*/ false,
                             /*detached*/ true,
                             /*events_waiter*/ waiter);
  return group_options;
}

AsyncWorkQueue::AsyncWorkQueue(size_t host_num_threads,
                               size_t device_num_threads,
                               EventsWaiter* waiter)
    : host_num_thread_(host_num_threads) {
116 117
  queue_group_ = CreateWorkQueueGroup(
      ConstructWorkQueueOptions(host_num_threads, device_num_threads, waiter));
118 119
}

120 121
void AsyncWorkQueue::AddTask(const OpFuncType& op_func_type,
                             std::function<void()> fn) {
122
  VLOG(4) << "Add task: " << static_cast<size_t>(op_func_type) << " ";
123 124
  // NOTE(zhiqiu): use the second queue of size of, so only one thread is used.
  if (FLAGS_new_executor_serial_run) {
125 126 127 128 129 130 131
    queue_group_->AddTask(static_cast<size_t>(OpFuncType::kQueueAsync),
                          std::move(fn));
  } else {
    queue_group_->AddTask(static_cast<size_t>(op_func_type), std::move(fn));
  }
}

132
bool IsCommunicationOp(const std::string& op_name) {
133 134 135 136 137 138 139 140 141 142 143 144 145 146
  const std::set<std::string> special_comm_op_set = {
      "send",
      "recv",
      "send_v2",
      "recv_v2",
  };
  const std::string communication_op_prefix = "c_";
  if (op_name.find(communication_op_prefix) != std::string::npos ||
      special_comm_op_set.count(op_name)) {
    return true;
  }
  return false;
}

147 148 149 150
bool IsCommunicationOp(const Instruction& instr) {
  return IsCommunicationOp(instr.OpBase()->Type());
}

151 152 153 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 180 181 182 183 184 185 186 187 188
bool IsCpuOp(const Instruction& instr) {
  return platform::is_cpu_place(instr.DeviceContext().GetPlace());
}

bool IsSupportedHeterPlace(const phi::Place& place) {
  return platform::is_gpu_place(place) || platform::is_npu_place(place) ||
         platform::is_xpu_place(place) || platform::is_ipu_place(place) ||
         platform::is_custom_place(place);
}

bool IsMemcpyD2H(const Instruction& instr) {
  return instr.OpBase()->Type() == kMemcpyD2H;
}

bool IsMemcpyH2D(const Instruction& instr) {
  return instr.OpBase()->Type() == kMemcpyH2D;
}

bool IsMemcpyOp(const Instruction& instr) {
  return IsMemcpyD2H(instr) || IsMemcpyH2D(instr);
}

void AddFetch(const std::vector<std::string>& fetch_names,
              framework::BlockDesc* block) {
  auto* fetch_holder = block->Var(kFetchVarName);
  fetch_holder->SetType(proto::VarType::FETCH_LIST);
  fetch_holder->SetPersistable(true);

  int i = 0;
  for (auto& fetch_name : fetch_names) {
    // append fetch op
    auto* op = block->AppendOp();
    op->SetType("fetch_v2");
    op->SetInput("X", {fetch_name});
    op->SetOutput("Out", {kFetchVarName});
    op->SetAttr("col", {static_cast<int>(i)});
    op->CheckAttrs();
    i++;
189 190 191
  }
}

W
wanghuancoder 已提交
192 193 194 195 196 197 198 199 200 201 202 203 204 205 206
bool var_can_be_deleted(const std::string& name, const BlockDesc& block) {
  auto* var_desc = block.FindVar(name);
  if (var_desc == nullptr || var_desc->Persistable()) {
    return false;
  }

  auto type = var_desc->Proto()->type().type();

  return type == proto::VarType::LOD_TENSOR ||
         type == proto::VarType::SELECTED_ROWS ||
         type == proto::VarType::LOD_TENSOR_ARRAY;
}

std::unordered_map<const paddle::framework::OperatorBase*,
                   std::vector<std::string>>
L
Leo Chen 已提交
207 208
GetUnusedVars(const BlockDesc& block,
              const std::vector<std::shared_ptr<OperatorBase>>& ops) {
W
wanghuancoder 已提交
209 210 211
  std::unordered_map<std::string, size_t> var_op_idx_map;

  for (size_t i = 0; i < ops.size(); ++i) {
L
Leo Chen 已提交
212
    const auto& op = ops[i];
W
wanghuancoder 已提交
213 214 215 216 217 218 219 220 221 222

    OpInOutInfo info;
    for (auto& name_pair : op->Inputs()) {
      for (auto& name : name_pair.second) {
        if (!var_can_be_deleted(name, block)) {
          continue;
        }

        // var can be gc-ed
        if (!info.IsBuilt()) {
L
Leo Chen 已提交
223
          info.Build(op.get());
W
wanghuancoder 已提交
224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250
        }

        if (info.IsInArgBufferNeeded(name)) {
          // Update the last living op of variable to current op
          var_op_idx_map[name] = i;
        } else {
          VLOG(10) << "Skip reference count computing of variable "
                   << name_pair.first << "(" << name << ") in Operator "
                   << op->Type();
        }
      }
    }

    for (auto& name_pair : op->Outputs()) {
      for (auto& name : name_pair.second) {
        if (var_can_be_deleted(name, block)) {
          // Update the last living op of variable to current op
          var_op_idx_map[name] = i;
        }
      }
    }
  }

  std::unordered_map<const OperatorBase*, std::vector<std::string>> result;
  for (auto& name_op_idx_pair : var_op_idx_map) {
    auto& name = name_op_idx_pair.first;
    size_t op_idx = name_op_idx_pair.second;
L
Leo Chen 已提交
251 252 253
    auto op = ops[op_idx].get();
    result[op].emplace_back(name);
    VLOG(4) << op->Type() << " " << name;
W
wanghuancoder 已提交
254
  }
255
  VLOG(4) << "gc map size:" << result.size();
W
wanghuancoder 已提交
256 257 258
  return result;
}

L
Leo Chen 已提交
259 260 261
void BuildVariableScope(const framework::BlockDesc& block,
                        VariableScope* var_scope,
                        bool use_local_scope) {
262 263 264 265 266 267 268 269
  VLOG(3) << "Creating Variables";
  auto inner_scope = var_scope->GetMutableScope();

  // NOTE(zhiqiu): if create_local_scope_ is true, the persistable is
  // created in var_scope.scope_ , and other scope is created in local scope.
  Scope* local_scope = use_local_scope ? var_scope->GetMutableLocalScope()
                                       : var_scope->GetMutableScope();

270
  for (auto& var_desc : block.AllVars()) {
271
    auto var_name = var_desc->Name();
X
xiongkun 已提交
272 273 274
    // TODO(xiongkun): user may create a variable with name that exists before.
    // under such circumstances, we should raise a error. Currently we can't
    // get the var_desc of startup_program, so leave it later.
275
    if (var_name == framework::kEmptyVarName) {
W
wanghuancoder 已提交
276 277
      continue;
    }
278

279
    if (var_desc->Persistable()) {
280 281 282 283 284 285 286 287
      // In principle, we should put all trainable parameters in global scope,
      // which means the root of the scope tree. Some cases like quantization
      // will look up these parameters in global scope.
      const Scope* ancestor_scope = inner_scope;
      while (ancestor_scope->parent()) {
        ancestor_scope = ancestor_scope->parent();
      }
      auto* ptr = const_cast<Scope*>(ancestor_scope)->Var(var_name);
W
wanghuancoder 已提交
288

289
      VLOG(3) << "Initialize Variable " << var_name;
290 291
      // NOTE(zhiqiu): if var exists in scope and the type is right,
      // InitializeVariable will not create a new variable.
292 293 294
      InitializeVariable(ptr, var_desc->GetType());
      VLOG(3) << "Create Variable " << var_name << " global, which pointer is "
              << ptr << " type is " << static_cast<int>(var_desc->GetType());
295
    } else {
296 297 298 299 300
      auto* ptr = local_scope->Var(var_name);
      InitializeVariable(ptr, var_desc->GetType());
      VLOG(3) << "Create Variable " << var_name << " locally, which pointer is "
              << ptr << "Variable Type "
              << static_cast<int>(var_desc->GetType());
W
wanghuancoder 已提交
301
    }
302
    var_scope->AddVar(var_name, var_desc);
W
wanghuancoder 已提交
303 304 305
  }
}

L
Leo Chen 已提交
306 307
void CreateAllOps(const framework::BlockDesc& block,
                  std::vector<std::unique_ptr<OperatorBase>>* ops) {
308
  for (auto& op : block.AllOps()) {
309
    auto op_type = op->Type();
310
    VLOG(8) << "CreateOp from : " << op_type;
W
wanghuancoder 已提交
311

312
    auto& info = OpInfoMap::Instance().Get(op_type);
W
wanghuancoder 已提交
313 314 315

    const VariableNameMap& inputs_names = op->Inputs();
    const VariableNameMap& outputs_names = op->Outputs();
316

W
wanghuancoder 已提交
317
    AttributeMap op_attr_map = op->GetAttrMap();
318
    AttributeMap op_runtime_attr_map = op->GetRuntimeAttrMap();
W
wanghuancoder 已提交
319 320 321 322

    if (info.Checker() != nullptr) {
      info.Checker()->Check(&op_attr_map);
    }
323 324 325 326 327 328 329

    const auto& extra_attr_checkers =
        operators::ExtraInfoUtils::Instance().GetExtraAttrsChecker(op_type);
    for (const auto& checker : extra_attr_checkers) {
      checker(&op_runtime_attr_map, false);
    }

W
wanghuancoder 已提交
330
    auto op_base =
331 332
        info.Creator()(op_type, inputs_names, outputs_names, op_attr_map);
    op_base->SetRuntimeAttributeMap(op_runtime_attr_map);
333 334 335 336 337 338 339 340 341 342

#ifdef PADDLE_WITH_MKLDNN
    if (FLAGS_use_mkldnn) {
      if (op->HasAttr("use_mkldnn")) {
        VLOG(4) << "Set use_mkldnn=True for " << op_base->Type();
        op_base->SetAttr("use_mkldnn", true);
      }
    }
#endif

X
xiongkun 已提交
343
    ops->emplace_back(std::unique_ptr<OperatorBase>(op_base));
W
wanghuancoder 已提交
344
  }
345 346
}

347
std::tuple<VariableValueMap, VariableIdMap> BuildVariableMap(
348 349
    const VariableNameMap& var_name_map,
    VariableScope* var_scope,
350
    Scope* local_scope,
351
    bool find_var_recursively = false,
352
    bool allow_var_not_in_scope = false) {
353 354 355 356 357 358 359 360
  VariableValueMap name2var;
  VariableIdMap name2id;
  for (auto& item : var_name_map) {
    std::vector<Variable*> vars;
    std::vector<int> ids;
    vars.reserve(item.second.size());

    for (auto& var_name : item.second) {
361 362
      auto* var = local_scope->FindVar(var_name);

363
      if (!var_scope->HasVar(var_name)) {
364
        if (find_var_recursively && var) {
365
          VLOG(3) << "Add " << var_name << " to var_scope";
366
          var_scope->AddVar(var_name, nullptr);
367
        } else if (allow_var_not_in_scope) {
368 369 370
          VLOG(4) << var_name << " don't exist in variable scope, skip it!";
          continue;
        }
371
      }
372
      auto var_id = var_scope->VarId(var_name);
373
      vars.push_back(var);
374 375 376 377 378 379 380
      ids.push_back(var_id);
    }
    name2var[item.first] = std::move(vars);
    name2id[item.first] = std::move(ids);
  }
  return std::make_tuple(name2var, name2id);
}
W
wanghuancoder 已提交
381

L
Leo Chen 已提交
382 383 384
void ApplyDeviceGuard(const OperatorBase* op_base,
                      const platform::Place& place,
                      OpKernelType* expected_kernel_key) {
385 386 387 388 389 390 391 392 393
  bool need_change_place =
      (op_base->HasAttr("op_device") &&
       (op_base->Attr<std::string>("op_device").length() > 0));
  if (need_change_place) {
    auto& op_device = op_base->Attr<std::string>("op_device");
    if (op_device == "cpu" || platform::is_cpu_place(place)) {
      VLOG(3) << "Switch into CPUPlace by device_guard.";
      expected_kernel_key->place_ = platform::CPUPlace();
    } else if (op_device.find("gpu") != std::string::npos &&
394 395 396 397
               platform::is_gpu_place(place)) {
      // when the Op that does not have GPUKernel is assigned to GPU, the
      // CPUKernel will be executed and a warning will be given at the same
      // time.
398 399 400 401 402 403 404 405 406 407
      if (op_base->SupportGPU()) {
        expected_kernel_key->place_ = place;
      } else {
        expected_kernel_key->place_ = platform::CPUPlace();
        LOG_FIRST_N(WARNING, 1)
            << "Op(" << op_base->Type()
            << ") has no CUDA implementation. It will be assigned to CPUPlace.";
      }
      VLOG(3) << "Switch into " << expected_kernel_key->place_
              << " by device_guard.";
408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437
    } else if (op_device.find("npu") != std::string::npos &&
               platform::is_npu_place(place)) {
      // when the Op that does not have NPUKernel is assigned to NPU, the
      // CPUKernel will be executed and a warning will be given at the same
      // time.
      if (op_base->SupportNPU()) {
        expected_kernel_key->place_ = place;
      } else {
        expected_kernel_key->place_ = platform::CPUPlace();
        LOG_FIRST_N(WARNING, 1)
            << "Op(" << op_base->Type()
            << ") has no NPU implementation. It will be assigned to CPUPlace.";
      }
      VLOG(3) << "Switch into " << expected_kernel_key->place_
              << " by device_guard.";
    } else if (op_device.find("xpu") != std::string::npos &&
               platform::is_xpu_place(place)) {
      // when the Op that does not have XPUKernel is assigned to XPU, the
      // CPUKernel will be executed and a warning will be given at the same
      // time.
      if (op_base->SupportXPU()) {
        expected_kernel_key->place_ = place;
      } else {
        expected_kernel_key->place_ = platform::CPUPlace();
        LOG_FIRST_N(WARNING, 1)
            << "Op(" << op_base->Type()
            << ") has no XPU implementation. It will be assigned to CPUPlace.";
      }
      VLOG(3) << "Switch into " << expected_kernel_key->place_
              << " by device_guard.";
438 439 440 441 442 443 444
    } else {
      PADDLE_THROW(
          platform::errors::Fatal("Unsupported current place %s", op_device));
    }
  }
}

L
Leo Chen 已提交
445
void HandleOperatorBase(const platform::Place& place,
L
Leo Chen 已提交
446 447
                        const VariableScope* var_scope,
                        std::shared_ptr<OperatorBase> op_base,
448 449
                        OpFuncNode* op_func_node,
                        Scope* local_scope) {
450 451 452 453
  platform::DeviceContextPool& pool = platform::DeviceContextPool::Instance();
  auto* dev_ctx = pool.Get(place);
  // input, output is prepared. set the other attributes.
  op_func_node->operator_base_ = op_base;
L
Leo Chen 已提交
454
  if (IsSupportedHeterPlace(place)) {
455 456 457 458 459 460 461
    op_func_node->type_ = OpFuncType::kQueueAsync;
  } else if (platform::is_cpu_place(place)) {
    op_func_node->type_ = OpFuncType::kQueueSync;
  } else {
    PADDLE_THROW(
        platform::errors::Fatal("Unsupported current place %s", place));
  }
462
  op_func_node->kernel_func_ = nullptr;
463
  op_base->Run(*local_scope, place);  // Run without data transformer.
464 465 466 467 468 469 470 471 472 473 474
  std::unordered_set<int> no_data_transform_index;
  for (auto& it : op_func_node->input_index) {
    for (auto& id : it.second) {
      no_data_transform_index.emplace(id);
    }
  }
  op_func_node->no_data_transform_index =
      no_data_transform_index;  // all index is no-need-transform
  op_func_node->dev_ctx_ = dev_ctx;
}

L
Leo Chen 已提交
475 476 477 478 479
void BuildOpFuncList(const platform::Place& place,
                     const framework::BlockDesc& block,
                     const std::set<std::string>& skip_gc_vars,
                     std::vector<OpFuncNode>* vec_func_list,
                     VariableScope* var_scope,
480 481
                     const ExecutionConfig& execution_config,
                     bool use_local_scope) {
482 483
  Scope* local_scope = use_local_scope ? var_scope->GetMutableLocalScope()
                                       : var_scope->GetMutableScope();
X
xiongkun 已提交
484 485 486
  std::vector<std::unique_ptr<OperatorBase>>
      ops_unique;  // its elements will be moved to vec_func_list
  // Step 1: create all ops for current block.
L
Leo Chen 已提交
487
  CreateAllOps(block, &ops_unique);
488

489
  if (!execution_config.used_for_jit) {
490 491 492 493 494 495 496 497 498
    // If gc is enabled and block size > 1
    const ProgramDesc& main_program = *block.Program();
    operators::PrepareSafeEagerDeletionOnConditionalOpAndConditionalGradOp(
        main_program, block.ID(), ops_unique);
    operators::PrepareSafeEagerDeletionOnWhileOpAndWhileGradOp(
        main_program, block.ID(), ops_unique);
    operators::PrepareSafeEagerDeletionOnRecurrentOpAndRecurrentGradOp(
        main_program, block.ID(), ops_unique);
  }
X
xiongkun 已提交
499

L
Leo Chen 已提交
500 501 502
#ifdef PADDLE_WITH_MKLDNN
  platform::RegisterModelLayout(ops_unique, place);
#endif
503 504
  // its elements will be moved to vec_func_list
  std::vector<std::shared_ptr<OperatorBase>> ops;
X
xiongkun 已提交
505 506 507
  for (auto& op_unique : ops_unique) {
    ops.emplace_back(std::move(op_unique));
  }
L
Leo Chen 已提交
508
  auto unused_var_map = GetUnusedVars(block, ops);
W
wanghuancoder 已提交
509

510
  bool flag_log_is_printed = false;
L
Leo Chen 已提交
511 512
  for (size_t i = 0; i < ops.size(); ++i) {
    auto op = ops[i].get();
513 514 515
    const std::string& op_type = op->Type();

    VLOG(6) << "Build OpFuncNode from : " << op_type;
W
wanghuancoder 已提交
516

P
pangyoki 已提交
517 518
    // Print new executor log if grad op is used.
    // It's only for test and will be removed later.
519
    if (!flag_log_is_printed && op_type.find("_grad") != std::string::npos) {
520
      LOG_FIRST_N(INFO, 1) << "Standalone Executor is Used.";
P
pangyoki 已提交
521 522 523
      flag_log_is_printed = true;
    }

524 525 526 527 528 529 530 531 532 533 534 535 536 537 538 539
    // Hot fix for variables used in dataloader, like
    // 'lod_tensor_blocking_queue_0'. These variables may be created in scope,
    // and it is not existed as variable in program.
    const std::set<std::string> ops_with_var_not_in_program = {
        "create_py_reader"};
    const std::set<std::string> ops_with_var_not_in_scope = {
        "conditional_block",
        "conditional_block_grad",
        "recurrent_grad",
        "rnn_memory_helper",
        "rnn_memory_helper_grad",
        "while",
        "while_grad"};
    bool allow_var_not_in_program = ops_with_var_not_in_program.count(op_type);
    bool allow_var_not_in_scope = ops_with_var_not_in_scope.count(op_type);

540 541 542
    // ops in the control flow block may not find its inputs or outputs
    // in VariableScope of the sub-block, so we need search it in parent scope.

543
    framework::VariableNameMap& input_name_map = op->Inputs();
W
wanghuancoder 已提交
544
    VariableValueMap ins_map;
545
    VariableIdMap ins_name2id;
546 547 548 549 550 551
    std::tie(ins_map, ins_name2id) = BuildVariableMap(
        input_name_map,
        var_scope,
        local_scope,
        execution_config.used_for_control_flow_op || allow_var_not_in_program,
        allow_var_not_in_scope);
W
wanghuancoder 已提交
552

553
    framework::VariableNameMap& output_name_map = op->Outputs();
W
wanghuancoder 已提交
554
    VariableValueMap outs_map;
555
    VariableIdMap outs_name2id;
556 557 558 559
    std::tie(outs_map, outs_name2id) =
        BuildVariableMap(output_name_map,
                         var_scope,
                         local_scope,
560
                         execution_config.used_for_control_flow_op,
561
                         allow_var_not_in_scope);
W
wanghuancoder 已提交
562

563
    // step 1: build OpFuncNode
W
wanghuancoder 已提交
564
    OpFuncNode op_func_node;
565
    op_func_node.operator_base_ = ops[i];
W
wanghuancoder 已提交
566 567
    op_func_node.input_index = ins_name2id;
    op_func_node.output_index = outs_name2id;
568

569 570 571 572 573 574
    const OperatorDistAttr* dist_attr = block.Op(i)->DistAttr();
    if (dist_attr &&
        dist_attr->execution_stream() != distributed::auto_parallel::kDefault) {
      op_func_node.execution_stream_ = dist_attr->execution_stream();
    }

575 576
    SingleStreamGuard single_stream_guard(ops[i]);

577
    VLOG(4) << "Start run " << place << " " << op->DebugStringEx(local_scope);
578

579 580 581 582 583 584 585 586 587 588
#ifdef PADDLE_WITH_ASCEND_CL
    // NOTE(wangxi): nan/inf cannot be detected on NPU by checking the variable
    // values, but only through special `float_status` to checks whether
    // the operation is overflow. More about `float_status`, see:
    // https://gitee.com/ascend/modelzoo/issues/I3NF8V?from=project-issue
    if (FLAGS_check_nan_inf) {
      framework::details::NPUAllocAndClearFloatStatus(*op, *local_scope, place);
    }
#endif

589 590
    try {
      if (dynamic_cast<framework::OperatorWithKernel*>(op) == nullptr) {
L
Leo Chen 已提交
591
        VLOG(4) << "HandleOperatorBase";
592
        // op is not a operatorwithkernel, so direcly run OperatorBase::Run()
L
Leo Chen 已提交
593
        HandleOperatorBase(
594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609 610 611 612 613 614 615 616
            place, var_scope, ops[i], &op_func_node, local_scope);
      } else {
        VLOG(4) << "OP is not null";
        auto op_with_kernel = const_cast<framework::OperatorWithKernel*>(
            static_cast<const framework::OperatorWithKernel*>(op));
        VLOG(4) << "get op_with_kernel";
        // construct RuntimeContext and analysis KernelType
        RuntimeContext runtime_context({}, {});
        runtime_context.inputs.swap(ins_map);
        runtime_context.outputs.swap(outs_map);
        VLOG(4) << "get RuntimeContext";

        Scope scope, *runtime_scope = &scope;
        // NOTE(Ruibiao): We do not encourage directly using scope in OP kernel.
        // But some OPs do have such behavior (e.g., cinn_launch OP). Here
        // special treatment for them.
        if (op_with_kernel->Type() == "cinn_launch") {
          VLOG(6) << "OP(" << op_with_kernel->Type()
                  << ") use scope in kernel, "
                     "so pass a real scope to "
                     "ExecutionContext";
          runtime_scope = local_scope;
        }
617

618 619 620 621 622 623
        auto& pool = platform::DeviceContextPool::Instance();
        auto* dev_ctx = pool.Get(place);
        auto exec_ctx = ExecutionContext(
            *op_with_kernel, *runtime_scope, *dev_ctx, runtime_context);
        auto expected_kernel_key =
            op_with_kernel->GetExpectedKernelType(exec_ctx);
624 625 626 627 628 629
#if defined(PADDLE_WITH_CUDA) || defined(PADDLE_WITH_HIP)
        if (!op_with_kernel->DnnFallback() &&
            paddle::platform::CanCUDNNBeUsed(exec_ctx)) {
          expected_kernel_key.library_type_ = framework::LibraryType::kCUDNN;
        }
#endif
630
        VLOG(4) << "expected_kernel_key : " << expected_kernel_key;
631
        // change device by the device_guard()
L
Leo Chen 已提交
632
        ApplyDeviceGuard(op, place, &expected_kernel_key);
633 634 635 636 637 638 639 640 641 642 643

        // step 2. select op kernel
        auto run_phi_kernel = false;
        if (phi::KernelFactory::Instance().HasCompatiblePhiKernel(
                op_with_kernel->Type())) {
          auto phi_kernel_key = op_with_kernel->ChoosePhiKernel(exec_ctx);
          auto phi_kernel_name = op_with_kernel->PhiKernelSignature()->name;

          if (op_with_kernel->PhiKernel()->IsValid()) {
            run_phi_kernel = true;
          } else {
644 645
            if (!op_with_kernel->SupportsKernelType(expected_kernel_key,
                                                    exec_ctx)) {
646 647 648 649 650 651 652 653 654 655 656 657 658 659
              auto phi_cpu_kernel_key = FallBackToCpu(
                  expected_kernel_key, phi_kernel_key, *op_with_kernel);
              op_with_kernel->ResetPhiKernel(
                  new phi::Kernel(phi::KernelFactory::Instance().SelectKernel(
                      phi_kernel_name, phi_cpu_kernel_key)));
              if (op_with_kernel->PhiKernel()->IsValid()) {
                VLOG(6) << "Static mode PrepareImpl - kernel name: "
                        << phi_kernel_name
                        << " | kernel key: " << phi_cpu_kernel_key
                        << " | kernel: " << *(op_with_kernel->PhiKernel());
                op_with_kernel->ResetKernelType(new OpKernelType(
                    TransPhiKernelKeyToOpKernelType(phi_cpu_kernel_key)));
                run_phi_kernel = true;
              }
660 661 662
            }
          }
        }
663 664 665 666 667 668 669 670 671 672 673 674
        VLOG(4) << "if run phi kernel? : " << run_phi_kernel;
        if (!run_phi_kernel) {
          op_with_kernel->ChooseKernel(exec_ctx);
          op_func_node.kernel_func_ = *op_with_kernel->kernel_func();
        } else {
          op_func_node.phi_kernel_ = op_with_kernel->PhiKernel();
        }
        auto kernel_type = *(op_with_kernel->kernel_type());
        if (kernel_type.place_ != dev_ctx->GetPlace()) {
          dev_ctx = pool.Get(kernel_type.place_);
        }
        op_func_node.dev_ctx_ = dev_ctx;
L
Leo Chen 已提交
675
        if (IsSupportedHeterPlace(kernel_type.place_)) {
676 677 678 679 680 681 682 683 684 685 686 687 688 689 690 691 692 693 694 695 696 697 698 699 700 701 702 703 704 705 706 707
          op_func_node.type_ = OpFuncType::kQueueAsync;
        } else if (platform::is_cpu_place(kernel_type.place_)) {
          op_func_node.type_ = OpFuncType::kQueueSync;
        } else {
          PADDLE_THROW(platform::errors::Fatal("Unsupported current place %s",
                                               kernel_type.place_));
        }
        VLOG(3) << op_with_kernel->Type()
                << " : finally selected kernel_key: " << kernel_type;

        // step 3. data transform
        VariableValueMap& ins_map_temp = runtime_context.inputs;
        VariableValueMap& outs_map_temp = runtime_context.outputs;
        ApplyDataTransform(kernel_type,
                           place,
                           &ins_map_temp,
                           &outs_map_temp,
                           var_scope,
                           &op_func_node,
                           vec_func_list,
                           use_local_scope);
        VLOG(4) << "apply data transform done. ";
        // step 4. infershape, see OperatorWithKernel::RunImpl in operator.cc
        // for why.
        if (!(op->HasAttr(kAllKernelsMustComputeRuntimeShape) &&
              op->Attr<bool>(kAllKernelsMustComputeRuntimeShape))) {
          InterpretercoreInferShapeContext infer_shape_ctx(*op,
                                                           runtime_context);
          // TODO(Aurelius84): In case of control flow ops, they are NOT
          // inheritted from OperatorWithKernel.
          op_with_kernel->Info().infer_shape_(&infer_shape_ctx);
        }
708

709 710 711 712 713 714 715 716 717 718 719
        // step 5. run kernel
        if (run_phi_kernel) {
          phi::KernelContext phi_kernel_context;
          op_with_kernel->BuildPhiKernelContext(
              runtime_context, dev_ctx, &phi_kernel_context);
          (*op_func_node.phi_kernel_)(&phi_kernel_context);
        } else {
          // the place of exec_ctx maybe has changed.
          op_func_node.kernel_func_(ExecutionContext(
              *op_with_kernel, *runtime_scope, *dev_ctx, runtime_context));
        }
720

721 722 723 724 725 726
        // post-process grad_op.outputs if need cast complex grad into real
        // grad.
        // NOTE(Aurelius84): insert a transfer_dtype_op inplacely to cast it.
        if (framework::IsComplexType(kernel_type.data_type_)) {
          interpreter::HandleComplexGradToRealGrad(op_func_node,
                                                   place,
727
                                                   output_name_map,
728 729 730 731 732 733 734 735 736 737 738 739 740 741 742 743 744 745 746 747 748
                                                   &runtime_context.outputs,
                                                   var_scope,
                                                   vec_func_list,
                                                   local_scope);
        }
        if (!op_func_node.inplace_back_map.empty()) {
          auto& m = op_func_node.inplace_back_map;
          // NOTE(zhiqiu): same logic as TransferInplaceVarsBack() in
          // operator.cc
          for (auto& p : m) {
            auto* transformed_tensor =
                GetMutableLoDTensorOrSelectedRowsValueFromVar(
                    local_scope->FindVar(var_scope->GetNameById(p.first)));
            auto* original_tensor =
                GetMutableLoDTensorOrSelectedRowsValueFromVar(
                    local_scope->FindVar(var_scope->GetNameById(p.second)));
            original_tensor->ShareDataWith(*transformed_tensor);
            VLOG(4) << "Transfer inplace variable back form "
                    << var_scope->GetNameById(p.first) << " to "
                    << var_scope->GetNameById(p.second);
          }
749
        }
750

751 752 753 754 755
        // for debug nan/inf
        if (FLAGS_check_nan_inf) {
          VLOG(4) << "Check nan/inf";
          framework::details::CheckOpHasNanOrInf(*op, *runtime_scope, place);
        }
756
      }
757
    } catch (platform::EnforceNotMet& ex) {
758
      framework::InsertCallStackInfo(op_type, op->Attrs(), &ex);
759 760 761 762
      throw std::move(ex);
    } catch (platform::EOFException&) {
      std::rethrow_exception(std::current_exception());
    } catch (std::exception& ex) {
763
      LOG(WARNING) << op_type << " raises an exception "
764 765 766 767
                   << platform::demangle(typeid(ex).name()) << ", "
                   << ex.what();
      std::rethrow_exception(std::current_exception());
    } catch (...) {
768
      LOG(WARNING) << op_type << " raises an unknown exception";
769
      std::rethrow_exception(std::current_exception());
770
    }
W
wanghuancoder 已提交
771

772 773 774
    VLOG(4) << "End run " << place << " "
            << op_func_node.operator_base_->DebugStringEx(local_scope);

L
Leo Chen 已提交
775
    vec_func_list->emplace_back(op_func_node);
776

L
Leo Chen 已提交
777
    // gc---------------------------------------------
L
Leo Chen 已提交
778
    auto iter = unused_var_map.find(op);
W
wanghuancoder 已提交
779
    if (iter == unused_var_map.end()) {
780
      interpreter::LogDeviceMemoryStats(place);
W
wanghuancoder 已提交
781 782 783 784 785 786 787 788
      continue;
    }

    auto& delete_vars = iter->second;
    std::deque<std::shared_ptr<memory::Allocation>>* garbages =
        new std::deque<std::shared_ptr<memory::Allocation>>();

    for (auto& var_name : delete_vars) {
789
      auto* var = local_scope->FindVar(var_name);
790
      if (var == nullptr || skip_gc_vars.find(var_name) != skip_gc_vars.end()) {
W
wanghuancoder 已提交
791 792 793
        continue;
      }

794
      VLOG(6) << "Erase variable " << var_name;
795
      if (var->IsType<phi::DenseTensor>()) {
W
wanghuancoder 已提交
796
        garbages->emplace_back(
797
            var->GetMutable<phi::DenseTensor>()->MoveMemoryHolder());
W
wanghuancoder 已提交
798 799 800
      }
    }
    delete garbages;  // free mem
801 802

    interpreter::LogDeviceMemoryStats(place);
W
wanghuancoder 已提交
803
  }
804 805 806 807 808 809

  // NOTE(Ruibiao): Release memory cache to avoid memory fragments in Allocator.
  // It reduce about 10% memory usage for V100 8-GPU training of
  // transformer_base_bs4096_amp_fp16 and transformer_base_bs4096_pure_fp16
  // model.
  memory::Release(place);
W
wanghuancoder 已提交
810 811
}

812 813 814 815 816 817 818 819 820 821 822 823
void LogDeviceMemoryStats(const platform::Place& place) {
  if (FLAGS_new_executor_log_memory_stats && platform::is_gpu_place(place)) {
    VLOG(0) << "memory_allocated: "
            << static_cast<double>(memory::DeviceMemoryStatCurrentValue(
                   "Allocated", place.device)) /
                   1024 / 1024
            << " MB";
    VLOG(0) << "max_memory_allocated: "
            << static_cast<double>(memory::DeviceMemoryStatPeakValue(
                   "Allocated", place.device)) /
                   1024 / 1024
            << " MB";
824 825 826
  }
}

827
}  // namespace interpreter
W
wanghuancoder 已提交
828 829
}  // namespace framework
}  // namespace paddle