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 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 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
bool IsCommunicationOp(const Instruction& instr) {
  const std::set<std::string> special_comm_op_set = {
      "send",
      "recv",
      "send_v2",
      "recv_v2",
  };
  const std::string& op_name = instr.OpBase()->Type();
  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;
}

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++;
186 187 188
  }
}

W
wanghuancoder 已提交
189 190 191 192 193 194 195 196 197 198 199 200 201 202 203
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 已提交
204 205
GetUnusedVars(const BlockDesc& block,
              const std::vector<std::shared_ptr<OperatorBase>>& ops) {
W
wanghuancoder 已提交
206 207 208
  std::unordered_map<std::string, size_t> var_op_idx_map;

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

    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 已提交
220
          info.Build(op.get());
W
wanghuancoder 已提交
221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247
        }

        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 已提交
248 249 250
    auto op = ops[op_idx].get();
    result[op].emplace_back(name);
    VLOG(4) << op->Type() << " " << name;
W
wanghuancoder 已提交
251
  }
252
  VLOG(4) << "gc map size:" << result.size();
W
wanghuancoder 已提交
253 254 255
  return result;
}

L
Leo Chen 已提交
256 257 258
void BuildVariableScope(const framework::BlockDesc& block,
                        VariableScope* var_scope,
                        bool use_local_scope) {
259 260 261 262 263 264 265 266
  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();

267
  for (auto& var_desc : block.AllVars()) {
268
    auto var_name = var_desc->Name();
X
xiongkun 已提交
269 270 271
    // 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.
272
    if (var_name == framework::kEmptyVarName) {
W
wanghuancoder 已提交
273 274
      continue;
    }
275

276
    if (var_desc->Persistable()) {
277 278 279 280 281 282 283 284
      // 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 已提交
285

286
      VLOG(3) << "Initialize Variable " << var_name;
287 288
      // NOTE(zhiqiu): if var exists in scope and the type is right,
      // InitializeVariable will not create a new variable.
289 290 291
      InitializeVariable(ptr, var_desc->GetType());
      VLOG(3) << "Create Variable " << var_name << " global, which pointer is "
              << ptr << " type is " << static_cast<int>(var_desc->GetType());
292
    } else {
293 294 295 296 297
      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 已提交
298
    }
299
    var_scope->AddVar(var_name, var_desc);
W
wanghuancoder 已提交
300 301 302
  }
}

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

309
    auto& info = OpInfoMap::Instance().Get(op_type);
W
wanghuancoder 已提交
310 311 312

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

W
wanghuancoder 已提交
314
    AttributeMap op_attr_map = op->GetAttrMap();
315
    AttributeMap op_runtime_attr_map = op->GetRuntimeAttrMap();
W
wanghuancoder 已提交
316 317 318 319

    if (info.Checker() != nullptr) {
      info.Checker()->Check(&op_attr_map);
    }
320 321 322 323 324 325 326

    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 已提交
327
    auto op_base =
328 329
        info.Creator()(op_type, inputs_names, outputs_names, op_attr_map);
    op_base->SetRuntimeAttributeMap(op_runtime_attr_map);
330 331 332 333 334 335 336 337 338 339

#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 已提交
340
    ops->emplace_back(std::unique_ptr<OperatorBase>(op_base));
W
wanghuancoder 已提交
341
  }
342 343
}

344
std::tuple<VariableValueMap, VariableIdMap> BuildVariableMap(
345 346
    const VariableNameMap& var_name_map,
    VariableScope* var_scope,
347
    Scope* local_scope,
348
    bool find_var_recursively = false,
349
    bool allow_var_not_in_scope = false) {
350 351 352 353 354 355 356 357
  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) {
358 359
      auto* var = local_scope->FindVar(var_name);

360
      if (!var_scope->HasVar(var_name)) {
361
        if (find_var_recursively && var) {
362
          VLOG(3) << "Add " << var_name << " to var_scope";
363
          var_scope->AddVar(var_name, nullptr);
364
        } else if (allow_var_not_in_scope) {
365 366 367
          VLOG(4) << var_name << " don't exist in variable scope, skip it!";
          continue;
        }
368
      }
369
      auto var_id = var_scope->VarId(var_name);
370
      vars.push_back(var);
371 372 373 374 375 376 377
      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 已提交
378

L
Leo Chen 已提交
379 380 381
void ApplyDeviceGuard(const OperatorBase* op_base,
                      const platform::Place& place,
                      OpKernelType* expected_kernel_key) {
382 383 384 385 386 387 388 389 390
  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 &&
391 392 393 394
               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.
395 396 397 398 399 400 401 402 403 404
      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.";
405 406 407 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
    } 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.";
435 436 437 438 439 440 441
    } else {
      PADDLE_THROW(
          platform::errors::Fatal("Unsupported current place %s", op_device));
    }
  }
}

L
Leo Chen 已提交
442
void HandleOperatorBase(const platform::Place& place,
L
Leo Chen 已提交
443 444
                        const VariableScope* var_scope,
                        std::shared_ptr<OperatorBase> op_base,
445 446
                        OpFuncNode* op_func_node,
                        Scope* local_scope) {
447 448 449 450
  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 已提交
451
  if (IsSupportedHeterPlace(place)) {
452 453 454 455 456 457 458
    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));
  }
459
  op_func_node->kernel_func_ = nullptr;
460
  op_base->Run(*local_scope, place);  // Run without data transformer.
461 462 463 464 465 466 467 468 469 470 471
  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 已提交
472 473 474 475 476
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,
477 478
                     const ExecutionConfig& execution_config,
                     bool use_local_scope) {
479 480
  Scope* local_scope = use_local_scope ? var_scope->GetMutableLocalScope()
                                       : var_scope->GetMutableScope();
X
xiongkun 已提交
481 482 483
  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 已提交
484
  CreateAllOps(block, &ops_unique);
485

486
  if (!execution_config.used_for_jit) {
487 488 489 490 491 492 493 494 495
    // 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 已提交
496

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

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

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

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

521 522 523 524 525 526 527 528 529 530 531 532 533 534 535 536
    // 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);

537 538 539
    // 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.

540
    framework::VariableNameMap& input_name_map = op->Inputs();
W
wanghuancoder 已提交
541
    VariableValueMap ins_map;
542
    VariableIdMap ins_name2id;
543 544 545 546 547 548
    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 已提交
549

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

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

566 567 568 569 570 571
    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();
    }

572 573
    SingleStreamGuard single_stream_guard(ops[i]);

574
    VLOG(4) << "Start run " << place << " " << op->DebugStringEx(local_scope);
575

576 577 578 579 580 581 582 583 584 585
#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

586 587
    try {
      if (dynamic_cast<framework::OperatorWithKernel*>(op) == nullptr) {
L
Leo Chen 已提交
588
        VLOG(4) << "HandleOperatorBase";
589
        // op is not a operatorwithkernel, so direcly run OperatorBase::Run()
L
Leo Chen 已提交
590
        HandleOperatorBase(
591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609 610 611 612 613
            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;
        }
614

615 616 617 618 619 620
        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);
621 622 623 624 625 626
#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
627
        VLOG(4) << "expected_kernel_key : " << expected_kernel_key;
628
        // change device by the device_guard()
L
Leo Chen 已提交
629
        ApplyDeviceGuard(op, place, &expected_kernel_key);
630 631 632 633 634 635 636 637 638 639 640

        // 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 {
641 642
            if (!op_with_kernel->SupportsKernelType(expected_kernel_key,
                                                    exec_ctx)) {
643 644 645 646 647 648 649 650 651 652 653 654 655 656
              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;
              }
657 658 659
            }
          }
        }
660 661 662 663 664 665 666 667 668 669 670 671
        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 已提交
672
        if (IsSupportedHeterPlace(kernel_type.place_)) {
673 674 675 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
          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);
        }
705

706 707 708 709 710 711 712 713 714 715 716
        // 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));
        }
717

718 719 720 721 722 723
        // 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,
724
                                                   output_name_map,
725 726 727 728 729 730 731 732 733 734 735 736 737 738 739 740 741 742 743 744 745
                                                   &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);
          }
746
        }
747

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

769 770 771
    VLOG(4) << "End run " << place << " "
            << op_func_node.operator_base_->DebugStringEx(local_scope);

L
Leo Chen 已提交
772
    vec_func_list->emplace_back(op_func_node);
773

L
Leo Chen 已提交
774
    // gc---------------------------------------------
L
Leo Chen 已提交
775
    auto iter = unused_var_map.find(op);
W
wanghuancoder 已提交
776
    if (iter == unused_var_map.end()) {
777
      interpreter::LogDeviceMemoryStats(place);
W
wanghuancoder 已提交
778 779 780 781 782 783 784 785
      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) {
786
      auto* var = local_scope->FindVar(var_name);
787
      if (var == nullptr || skip_gc_vars.find(var_name) != skip_gc_vars.end()) {
W
wanghuancoder 已提交
788 789 790
        continue;
      }

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

    interpreter::LogDeviceMemoryStats(place);
W
wanghuancoder 已提交
800
  }
801 802 803 804 805 806

  // 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 已提交
807 808
}

809 810 811 812 813 814 815 816 817 818 819 820
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";
821 822 823
  }
}

824
}  // namespace interpreter
W
wanghuancoder 已提交
825 826
}  // namespace framework
}  // namespace paddle