/**
 * Copyright (c) 2025 Huawei Technologies Co., Ltd.
 * This program is free software, you can redistribute it and/or modify it under the terms and conditions of
 * CANN Open Software License Agreement Version 2.0 (the "License").
 * Please refer to the License for details. You may not use this file except in compliance with the License.
 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED,
 * INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY, OR FITNESS FOR A PARTICULAR PURPOSE.
 * See LICENSE in the root of the software repository for the full text of the License.
 */

#include "jit_executor.h"

#include <checker.h>
#include "framework/runtime/gert_api.h"
#include "common/memory/tensor_trans_utils.h"
#include "graph/ge_context.h"
#include "common/model/external_allocator_manager.h"
#include "graph/load/model_manager/model_manager.h"
#include "graph/optimize/symbolic/infer_symbolic_shape/symbolic_infer_util.h"
#include "graph/utils/op_type_utils.h"
#include "graph/debug/ge_attr_define.h"
#include "acl/acl_rt.h"
#include <set>

#define JIT_ASSERT(exp, tsk, ...)                \
  do {                                           \
    bool tmp_ret = (exp);                        \
    if (!tmp_ret) {                              \
      std::vector<gert::Tensor> error_outputs;   \
      if ((tsk.callback) != nullptr) {           \
        tsk.callback(ge::FAILED, error_outputs); \
      }                                          \
      GE_ASSERT_TRUE(tmp_ret, __VA_ARGS__);      \
      return ::ErrorResult();                    \
    }                                            \
  } while (false)

#define JIT_ASSERT_NOTNULL(v, t, ...) JIT_ASSERT(((v) != nullptr), (t), __VA_ARGS__)
#define JIT_ASSERT_SUCCESS(v, t, ...) JIT_ASSERT(((v) == ge::SUCCESS), (t), __VA_ARGS__)
#define JIT_ASSERT_RT_OK(v, t, ...) JIT_ASSERT(((v) == 0), (t), __VA_ARGS__)

namespace ge {
namespace {
const int32_t kJitInvalidDeviceId = -1;

static int32_t &GetJitThreadDeviceId() {
  thread_local int32_t tl_device_id = kJitInvalidDeviceId;
  return tl_device_id;
}

static aclError SetDeviceCached(int32_t device_id) {
  auto &tl_id = GetJitThreadDeviceId();
  if (tl_id == device_id) {
    return ACL_SUCCESS;
  }
  const aclError ret = aclrtSetDevice(device_id);
  if (ret == ACL_SUCCESS) {
    tl_id = device_id;
  }
  return ret;
}

void PrepareOutputs(const ExecutionPoint &ep, std::vector<gert::Tensor> &outputs,
                    std::vector<GeTensor> &output_ge_tensors, bool need_ge_tensors = true) {
  const size_t out_num = ep.GetEpOutNum();
  outputs.resize(out_num);
  if (need_ge_tensors) {
    output_ge_tensors.resize(out_num);
  }
  for (auto &tensor : outputs) {
    tensor.SetOriginFormat(FORMAT_RESERVED);
    tensor.SetData(gert::TensorData());
  }
}

bool IsEnableBatchCpy(const std::vector<gert::Tensor> &inputs) {
  std::string input_batch_cpy_str;
  (void)GetThreadLocalContext().GetOption(configure_option::INPUT_BATCH_CPY, input_batch_cpy_str);
  GELOGI("Get input_batch_cpy_str=%s, size of inputs=%zu", input_batch_cpy_str.c_str(), inputs.size());
  return (!input_batch_cpy_str.empty() && input_batch_cpy_str == "1" && inputs.size() > 1);
}

// todo if host exec option, remember to handle
Status CopyHostInputsToDevice(UserGraphExecution &execution_task, Allocator *const allocator,
                              std::vector<gert::Tensor> &device_gert_tensors,
                              const std::set<size_t> &keep_on_host_idxs = {}) {
  const auto *external_rt_inputs = execution_task.external_rt_inputs;
  auto &inputs_memblocks = execution_task.inputs_memblocks;
  bool enable_input_batch_cpy = IsEnableBatchCpy(*external_rt_inputs);
  if (keep_on_host_idxs.empty()) {
    GE_ASSERT_SUCCESS(TensorTransUtils::TransHostGertTensorsToDevice(
        allocator, *external_rt_inputs, device_gert_tensors, inputs_memblocks, enable_input_batch_cpy));
    return SUCCESS;
  }

  std::vector<gert::Tensor> to_device_src;
  std::vector<gert::Tensor> host_vec;
  to_device_src.reserve(external_rt_inputs->size());
  host_vec.reserve(keep_on_host_idxs.size());
  GELOGI("copy host inputs to device, keep_on_host_idxs size %zu, external inputs size %zu.", keep_on_host_idxs.size(),
         external_rt_inputs->size());
  for (size_t i = 0U; i < external_rt_inputs->size(); ++i) {
    gert::Tensor src((*external_rt_inputs)[i].GetShape(), (*external_rt_inputs)[i].GetFormat(),
                     (*external_rt_inputs)[i].GetDataType());
    src.MutableOriginShape() = (*external_rt_inputs)[i].GetOriginShape();
    src.MutableStorageShape() = (*external_rt_inputs)[i].GetStorageShape();
    src.MutableTensorData().ShareFrom((*external_rt_inputs)[i].GetTensorData());
    if (keep_on_host_idxs.count(i) > 0U) {
      GELOGI("input[%zu] keep on host, not transfer to device.", i);
      host_vec.emplace_back(std::move(src));
    } else {
      to_device_src.emplace_back(std::move(src));
    }
  }

  std::vector<gert::Tensor> to_device_dst;
  std::vector<MemBlock *> to_device_blocks(to_device_src.size(), nullptr);
  if (!to_device_src.empty()) {
    GE_ASSERT_SUCCESS(TensorTransUtils::TransHostGertTensorsToDevice(allocator, to_device_src, to_device_dst,
                                                                     to_device_blocks, enable_input_batch_cpy));
  }

  device_gert_tensors.resize(external_rt_inputs->size());
  inputs_memblocks.resize(external_rt_inputs->size(), nullptr);
  size_t host_pos = 0U;
  size_t device_pos = 0U;
  for (size_t i = 0U; i < external_rt_inputs->size(); ++i) {
    if (keep_on_host_idxs.count(i) > 0U) {
      device_gert_tensors[i] = std::move(host_vec[host_pos++]);
    } else {
      device_gert_tensors[i] = std::move(to_device_dst[device_pos]);
      inputs_memblocks[i] = to_device_blocks[device_pos];
      ++device_pos;
    }
  }
  return SUCCESS;
}

Status FreeInputsAllocByJit(std::vector<MemBlock *> &input_blocks) {
  for (auto &mem_block : input_blocks) {
    if (mem_block == nullptr) {
      continue;
    }
    mem_block->Free();
  }
  return SUCCESS;
}

Status GetAllCondInputData(const ComputeGraphPtr &graph, std::set<size_t> &data_idx,
                           std::map<ComputeGraph *, std::set<size_t>> &cache) {
  GE_ASSERT_NOTNULL(graph);
  auto it = cache.find(graph.get());
  if (it != cache.end()) {
    data_idx = it->second;
    return SUCCESS;
  }
  for (const auto &node : graph->GetAllNodes()) {
    auto cond_input = SymbolicInferUtil::GetCondInput(node);
    if (cond_input == nullptr) {
      continue;
    }
    if (!OpTypeUtils::IsDataNode(cond_input->GetType())) {
      continue;
    }
    int32_t data_index = -1;
    GE_ASSERT_TRUE(AttrUtils::GetInt(cond_input->GetOpDesc(), "index", data_index), "get data node %s index failed",
                   cond_input->GetNamePtr());
    data_idx.insert(static_cast<size_t>(data_index));
  }
  cache[graph.get()] = data_idx;
  return SUCCESS;
}

Status BuildCompileInputs(const std::vector<gert::Tensor> &ori_inputs, const ComputeGraphPtr &graph,
                          std::vector<gert::Tensor> &compile_inputs,
                          std::map<ComputeGraph *, std::set<size_t>> &cond_input_cache, std::mutex &cache_mutex) {
  std::set<size_t> need_host_data_idx;
  {
    std::lock_guard<std::mutex> locker(cache_mutex);
    GE_ASSERT_SUCCESS(GetAllCondInputData(graph, need_host_data_idx, cond_input_cache));
    GE_ASSERT_SUCCESS(SymbolicInferUtil::GetValueDependentInputIdxs(graph, need_host_data_idx));
  }

  compile_inputs = TensorTransUtils::ShareFromGertTenosrs(ori_inputs);
  for (size_t data_idx : need_host_data_idx) {
    GE_ASSERT_TRUE(data_idx < compile_inputs.size());
    if (gert::TensorPlacementUtils::IsOnHost(compile_inputs[data_idx].GetPlacement())) {
      GELOGI("input[%zu] already on host, skip copy.", data_idx);
      continue;
    }
    GELOGI("input[%zu] need copy data to host.", data_idx);
    gert::Tensor host_tensor;
    GE_ASSERT_SUCCESS(TensorTransUtils::TransGertTensorToHost(compile_inputs[data_idx], host_tensor));
    compile_inputs[data_idx] = std::move(host_tensor);
  }
  return SUCCESS;
}

}  // namespace
JitExecutor::JitExecutor(GraphManager &graph_manager, UserGraphExecutionQueue &task_queue, ExecutionOrder &order,
                         CompileContext &compile_context, CompiledModelCache &cmc, std::mutex &mutex,
                         const std::map<MemoryType, std::pair<const void *, size_t>> *fixed_feature_memory_settings)
    : graph_manager_(graph_manager),
      task_queue_(task_queue),
      order_(order),
      compile_context_(compile_context),
      cmc_(cmc),
      mutex_(mutex),
      fixed_feature_memory_settings_(fixed_feature_memory_settings) {}

std::vector<JitExecutor::DataNodeInfo> JitExecutor::GetOrCreateDataNodeInfos(const ComputeGraphPtr &graph) {
  std::lock_guard<std::mutex> locker(guarded_execution_cache_mutex_);
  auto [it, inserted] = data_node_cache_.try_emplace(graph.get());
  if (!inserted) {
    return it->second;
  }
  for (const auto &node : graph->GetDirectNode()) {
    if (!OpTypeUtils::IsDataNode(node->GetType())) {
      continue;
    }
    int32_t data_index = -1;
    (void)AttrUtils::GetInt(node->GetOpDesc(), ATTR_NAME_INDEX, data_index);
    if (data_index < 0) {
      continue;
    }
    it->second.push_back({node->GetOpDesc(), data_index});
  }
  return it->second;
}

void JitExecutor::MarkHostTensorOnDataNodes(const std::vector<gert::Tensor> &inputs, const ComputeGraphPtr &graph) {
  if (graph == nullptr) {
    return;
  }
  const auto data_nodes = GetOrCreateDataNodeInfos(graph);
  for (const auto &data_node : data_nodes) {
    const int32_t data_index = data_node.input_index;
    if (static_cast<size_t>(data_index) >= inputs.size()) {
      continue;
    }
    if (!gert::TensorPlacementUtils::IsOnHost(inputs[data_index].GetPlacement())) {
      continue;
    }
    (void)AttrUtils::SetBool(data_node.op_desc, ATTR_NAME_HOST_TENSOR_AS_MODEL_INPUT, true);
    GELOGI("mark data node input index %d as host tensor.", data_index);
  }
}

std::unique_ptr<JitExecutor> JitExecutor::Create(
    GraphManager &graph_manager, UserGraphExecutionQueue &task_queue, ExecutionOrder &order,
    CompileContext &compile_context, CompiledModelCache &cmc, std::mutex &mutex,
    const std::map<MemoryType, std::pair<const void *, size_t>> *fixed_feature_memory_settings) {
  auto jit = std::unique_ptr<JitExecutor>(
      new JitExecutor(graph_manager, task_queue, order, compile_context, cmc, mutex, fixed_feature_memory_settings));
  GE_ASSERT_NOTNULL(jit);

  // add rt context before create jix executor
  jit.get()->device_id_ = static_cast<int32_t>(GetContext().DeviceId());
  GE_ASSERT_RT_OK(aclrtSetDevice(jit.get()->device_id_));
  GELOGI("Set device, device id:%u.", GetContext().DeviceId());
  GE_ASSERT_RT_OK(aclrtCreateStream(&(jit.get()->stream_)));
  GE_ASSERT_RT_OK(aclrtSetStreamFailureMode(jit.get()->stream_, ACL_STOP_ON_FAILURE));
  // prepare allocator
  auto device_allocator = gert::AllocatorFactory::Create("usergraph", gert::kOnDeviceHbm);
  GE_ASSERT_NOTNULL(device_allocator);
  jit.get()->device_allocator_ = std::move(device_allocator);
  GE_ASSERT_SUCCESS(graph_manager.RegisterExternalAllocator(jit.get()->stream_, jit.get()->device_allocator_));
  // 对于Execute接口，子图间的output是jit内部给的，静态图场景且没有外置allocator时需要手动申请内存，上面的device_allocator是针对tf场景不考虑用户会外置allocator
  auto jit_allocator = gert::AllocatorFactory::Create("usergraph", gert::kOnDeviceHbm);
  GE_ASSERT_NOTNULL(jit_allocator);
  jit.get()->external_allocator_ = std::move(jit_allocator);
  return jit;
}

Status JitExecutor::Finalize() {
  // remove每张子图的时候需要降序进行remove，因为前一张图中的内存会给下一张内存复用，例如输出tensor，如果前一张先释放，后一张释放的时候会出现heap-use-after-free
  // 通过给jit挂外置allocator的方法必须在load
  // graph之前，因为LoadGraph接口中会默认create一个allocator，而在execute阶段会优先使用默认create的allocator 在load
  // graph之前外置allocator会导致load过程中去申请const、feature等内存，这就强制要求这个外置allocator的生命周期大于整个GE的生命周期，否则会在remove的时候释放const内存失败。所以不在JIT中外置allocator
  {
    std::lock_guard<std::mutex> locker(guarded_execution_cache_mutex_);
    guarded_execution_cache_.clear();
    cond_input_data_cache_.clear();
    data_node_cache_.clear();
  }
  auto sorted_geps_to_inner_graph_id = SortMapByValue(geps_to_inner_ge_graph_id_, false);
  for (const auto &gep_2_id : sorted_geps_to_inner_graph_id) {
    GELOGI("[Jit]RemoveGraph %u", gep_2_id.second);
    GE_ASSERT_SUCCESS(graph_manager_.RemoveGraph(gep_2_id.second));
  }
  geps_to_inner_ge_graph_id_.clear();
  compiled_ge_graph_id_.clear();
  GE_ASSERT_RT_OK(aclrtSetDevice(device_id_));
  GE_ASSERT_SUCCESS(graph_manager_.UnregisterExternalAllocator(stream_));
  device_allocator_ = nullptr;
  external_allocator_ = nullptr;
  GE_ASSERT_RT_OK(aclrtDestroyStream(stream_));
  GE_ASSERT_RT_OK(aclrtResetDevice(device_id_));
  return SUCCESS;
}

Status JitExecutor::CompileGraph(UserGraphExecution &task, uint64_t session_id) {
  ExecutionPoint *ep;
  GE_ASSERT_RT_OK(aclrtSetDevice(device_id_));
  std::vector<GeTensor> ge_tensors;
  GE_ASSERT_SUCCESS(TensorTransUtils::GertTensors2GeTensors(*task.external_rt_inputs, ge_tensors));
  GE_ASSERT_SUCCESS(order_.FirstPoint(ge_tensors, ep));
  GELOGD("Get EP[%ld] of USER_GRAPH[%u] for CompileGraph", ep->GetId(), task.user_graph_id);

  auto gep = ep->FindOrCreateGuarded(*task.external_rt_inputs);
  GE_ASSERT_NOTNULL(gep);
  GELOGD("Get GEP[compiled_graph_id:%u] [compiled? %d] of EP[%ld] USER_GRAPH[%u].", gep->GetCompiledGraphId(),
         gep->Compiled(), ep->GetId(), task.user_graph_id);
  std::vector<Tensor> tensors;
  for (auto ge_tensor : ge_tensors) {
    tensors.emplace_back(TensorAdapter::AsTensor(ge_tensor));
  }
  GE_ASSERT_SUCCESS(Compile(tensors, gep, session_id));
  if (!ep->IsLast()) {
    GELOGD("Get EP[%ld] of USER_GRAPH[%u] is not last, need compile whole graph", ep->GetId(), task.user_graph_id);
    return ge::GE_GRAPH_NOT_BUILT;
  }
  return SUCCESS;
}

Status JitExecutor::LoadGraph(UserGraphExecution &task) {
  ExecutionPoint *ep;
  GE_ASSERT_RT_OK(aclrtSetDevice(device_id_));
  aclrtStream const stream = (task.stream == nullptr) ? stream_ : task.stream;
  std::vector<GeTensor> ge_tensors;
  GE_ASSERT_SUCCESS(TensorTransUtils::GertTensors2GeTensors(*task.external_rt_inputs, ge_tensors));
  GE_ASSERT_SUCCESS(order_.FirstPoint(ge_tensors, ep));
  GELOGD("Get EP[%ld] of USER_GRAPH[%u] for LoadGraph", ep->GetId(), task.user_graph_id);

  auto gep = ep->FindGuarded(*task.external_rt_inputs);
  if (gep == nullptr || !gep->Compiled()) {
    GELOGE(ge::FAILED, "Guard does not exist or Compiled EP[%ld], USER_GRAPH[%u]", ep->GetId(), task.user_graph_id);
    return FAILED;
  }
  GELOGD("Get GEP[compiled_graph_id:%u] [compiled? %d] of EP[%ld] USER_GRAPH[%u].", gep->GetCompiledGraphId(),
         gep->Compiled(), ep->GetId(), task.user_graph_id);
  auto iter = geps_to_inner_ge_graph_id_.find(gep);
  GE_ASSERT_TRUE(iter != geps_to_inner_ge_graph_id_.end());

  GE_ASSERT_SUCCESS(compile_context_.Load(iter->second, task.load_options, stream));
  return SUCCESS;
}

Status JitExecutor::RunWithCallback(UserGraphExecution &&task) {
  ExecutionPoint *ep;
  GE_ASSERT_RT_OK(SetDeviceCached(device_id_));
  JIT_ASSERT_NOTNULL(task.external_rt_inputs, task);

  std::vector<GeTensor> ge_tensors;
  ep = order_.GetFirstPoint();
  if (ep == nullptr) {
    JIT_ASSERT_SUCCESS(TensorTransUtils::GertTensors2GeTensors(*task.external_rt_inputs, ge_tensors), task);
    JIT_ASSERT_SUCCESS(order_.FirstPoint(ge_tensors, ep), task);
  }
  GELOGD("Get EP[%ld] of USER_GRAPH[%u]", ep->GetId(), task.user_graph_id);
  std::vector<gert::Tensor> tensors0;
  GE_MAKE_GUARD(free_input_mem, [&task]() { (void)FreeInputsAllocByJit(task.inputs_memblocks); });
  std::set<size_t> keep_on_host_idxs;
  if (ep != nullptr && ep->GetSlicedGraph() != nullptr) {
    std::lock_guard<std::mutex> locker(guarded_execution_cache_mutex_);
    JIT_ASSERT_SUCCESS(SymbolicInferUtil::GetValueDependentInputIdxs(ep->GetSlicedGraph(), keep_on_host_idxs), task);
  }
  JIT_ASSERT_SUCCESS(CopyHostInputsToDevice(task, device_allocator_.get(), tensors0, keep_on_host_idxs), task);
  if (ep != nullptr && ep->GetSlicedGraph() != nullptr) {
    MarkHostTensorOnDataNodes(tensors0, ep->GetSlicedGraph());
  }

  std::vector<gert::Tensor> tensors1;
  auto inputs = &tensors0;
  auto outputs = &tensors1;

  while (ep != nullptr) {
    const bool refresh_ge_tensors = !order_.HasNextPoint(*ep);
    PrepareOutputs(*ep, *outputs, ge_tensors, refresh_ge_tensors);
    GE_ASSERT_SUCCESS(ProcessAndExecuteGraphAsync(task, stream_, *inputs, *outputs, ep));
    if (refresh_ge_tensors) {
      for (size_t i = 0U; i < ge_tensors.size(); ++i) {
        JIT_ASSERT_SUCCESS(TensorTransUtils::TransRtTensorToGeTensor((*outputs)[i], ge_tensors[i]), task);
      }
    }
    JIT_ASSERT_SUCCESS(order_.NextPoint(*ep, ge_tensors, ep), task);
    if (ep != nullptr) {
      std::swap(inputs, outputs);
      MarkHostTensorOnDataNodes(*inputs, ep->GetSlicedGraph());
    }
  }
  JIT_ASSERT_RT_OK(aclrtSynchronizeStream(stream_), task);
  GE_CHECK_NOTNULL(task.callback);
  std::vector<gert::Tensor> host_tensors;
  GE_ASSERT_SUCCESS(TensorTransUtils::TransGertTensorsToHost(*outputs, host_tensors));
  task.callback(SUCCESS, host_tensors);
  return SUCCESS;
}

Status JitExecutor::Execute(UserGraphExecution &&task) {
  const auto ret = TryExecuteWithoutProcess(task);
  if (ret != ge::UNSUPPORTED) {
    return ret;
  }
  ExecutionPoint *ep;
  GE_ASSERT_RT_OK(SetDeviceCached(device_id_));
  rtStream_t const stream = (task.stream == nullptr) ? stream_ : task.stream;
  const bool has_allocator = (ExternalAllocatorManager::GetExternalAllocator(stream) != nullptr);

  const auto first_ep = order_.GetFirstPoint();
  std::vector<GeTensor> ge_tensors;
  if (first_ep == nullptr) {
    for (const auto &rt_tensor : *task.external_rt_inputs) {
      GeTensor ge_tensor;
      GE_ASSERT_SUCCESS(TensorTransUtils::TransRtTensorToGeTensor(rt_tensor, ge_tensor));
      ge_tensors.emplace_back(std::move(ge_tensor));
    }
    GE_ASSERT_SUCCESS(order_.FirstPoint(ge_tensors, ep));
  } else {
    ep = first_ep;
  }
  GELOGD("Get EP[%ld] of USER_GRAPH[%u]", ep->GetId(), task.user_graph_id);

  std::vector<gert::Tensor> tensors0;
  std::vector<gert::Tensor> tensors1;
  auto inputs = task.external_rt_inputs;
  auto outputs = &tensors0;

  while (ep != nullptr) {
    const bool refresh_ge_tensors = !order_.HasNextPoint(*ep);
    PrepareOutputs(*ep, *outputs, ge_tensors, refresh_ge_tensors);
    if (ep->IsLast()) {
      outputs = task.rt_outputs;
    }
    const bool need_malloc = (!has_allocator && !ep->IsLast());
    GE_ASSERT_SUCCESS(ProcessAndExecuteGraphAsync(task, stream, *inputs, *outputs, ep, need_malloc));
    if (refresh_ge_tensors) {
      for (size_t i = 0U; i < ge_tensors.size(); ++i) {
        GE_ASSERT_SUCCESS(TensorTransUtils::TransRtTensorToGeTensor((*outputs)[i], ge_tensors[i]));
      }
    }
    task.load_options.clear();
    GE_ASSERT_SUCCESS(order_.NextPoint(*ep, ge_tensors, ep));
    if (ep != nullptr) {
      inputs = outputs;
      outputs = (outputs == &tensors0) ? &tensors1 : &tensors0;
    }
  }
  return SUCCESS;
}

Status JitExecutor::MallocOutputsForStatic(uint32_t guarded_ep_instance_id, const GuardedExecutionPoint *gep,
                                           std::vector<gert::Tensor> &outputs) {
  CompiledGraphSummaryPtr summary{nullptr};
  GE_ASSERT_SUCCESS(graph_manager_.GetCompiledGraphSummary(guarded_ep_instance_id, summary));
  if (summary->IsStatic()) {
    // 只有静态的slice graph需要手动申请output内存，动态的ge内部会申请
    GraphNodePtr graph_node = make_shared<GraphNode>(guarded_ep_instance_id);
    graph_node->SetComputeGraph(gep->GetGraph());
    GE_ASSERT_SUCCESS(ModelManager::GetInstance().MallocOutputsMemory(guarded_ep_instance_id, graph_node,
                                                                      guarded_ep_instance_id, stream_, outputs));
  }
  return SUCCESS;
}

Status JitExecutor::GetOrCompileGuardedExecutionPoint(UserGraphExecution &task,
                                                      const std::vector<gert::Tensor> &compile_inputs,
                                                      ExecutionPoint *ep, rtStream_t stream,
                                                      GuardedExecutionInfo &execution_info) {
  GuardedExecutionCacheEntry cache_entry;
  {
    std::lock_guard<std::mutex> locker(guarded_execution_cache_mutex_);
    const auto cache_it = guarded_execution_cache_.find(ep);
    if (cache_it != guarded_execution_cache_.end()) {
      cache_entry = cache_it->second;
    }
  }

  if (cache_entry.valid) {
    std::lock_guard<std::mutex> locker(mutex_);
    execution_info.gep = ep->FindGuardedByCompiledGraphId(cache_entry.compiled_graph_id, compile_inputs);
    if (execution_info.gep != nullptr && execution_info.gep->Compiled()) {
      execution_info.instance_id = cache_entry.instance_id;
      return SUCCESS;
    }
  }

  std::lock_guard<std::mutex> locker(mutex_);
  execution_info.gep = ep->FindOrCreateGuarded(compile_inputs);
  JIT_ASSERT_NOTNULL(execution_info.gep, task);
  GELOGD("Get GEP[compiled_graph_id:%u] [compiled? %d] of EP[%ld] USER_GRAPH[%u], session_id:%llu.",
         execution_info.gep->GetCompiledGraphId(), execution_info.gep->Compiled(), ep->GetId(), task.user_graph_id,
         task.session_id);

  JIT_ASSERT_SUCCESS(CompileAndLoad(compile_inputs, execution_info.gep, execution_info.instance_id, stream,
                                    task.load_options, task.session_id),
                     task);
  std::lock_guard<std::mutex> cache_locker(guarded_execution_cache_mutex_);
  guarded_execution_cache_[ep] = {true, execution_info.gep->GetCompiledGraphId(), execution_info.instance_id};
  return SUCCESS;
}

Status JitExecutor::ProcessAndExecuteGraphAsync(UserGraphExecution &task, const rtStream_t stream,
                                                const std::vector<gert::Tensor> &inputs,
                                                std::vector<gert::Tensor> &outputs, ExecutionPoint *ep,
                                                bool need_malloc_output) {
  std::vector<gert::Tensor> compile_inputs;
  GE_ASSERT_SUCCESS(BuildCompileInputs(inputs, ep->GetSlicedGraph(), compile_inputs, cond_input_data_cache_,
                                       guarded_execution_cache_mutex_));
  // 标记必须与编译/执行同用一份 compile_inputs：value-dependent/cond 输入经 D2H 后为 host placement，
  // 其 Data 节点的 ATTR_NAME_HOST_TENSOR_AS_MODEL_INPUT 标记随 sliced graph 进入 GEP 编译，
  // 供静态图(DavinciModel)加载时将对应 input index 并入随路拷贝集合；
  // 标记基于原始 inputs 时看不到 D2H 生成的 host 副本，静态场景会退化为零拷贝校验失败
  MarkHostTensorOnDataNodes(compile_inputs, ep->GetSlicedGraph());
  GuardedExecutionInfo execution_info;
  GE_ASSERT_SUCCESS(GetOrCompileGuardedExecutionPoint(task, compile_inputs, ep, stream, execution_info));
  GELOGD("ExecuteGraphWithStreamAsync GEP[ins_id:%u] of EP[%ld] USER_GRAPH[%u].", execution_info.instance_id,
         execution_info.gep->GetOwnerEp()->GetId(), task.user_graph_id);
  GE_ASSERT_RT_OK(aclrtSetDevice(device_id_));

  if (need_malloc_output) {
    GE_ASSERT_SUCCESS(MallocOutputsForStatic(execution_info.instance_id, execution_info.gep, outputs));
  }
  // value-dependent/cond 输入在 BuildCompileInputs 中已被 D2H 为 host tensor，guard 匹配与 GEP 编译
  // 均建立在该 host 假设上（符号化推导从 host 数据取值）。执行阶段必须传入同一份 compile_inputs，
  // 由执行器按 host placement 消化（RT2 动态图走 alloc+H2D，静态图走随路拷贝）。
  // 若此处传入原始 device inputs，会形成"编译假设 host、执行传入 device"的 placement 不一致：
  // 真机环境下执行链路按 host 语义消费该输入时解引用 device 地址，导致主进程 coredump。
  JIT_ASSERT_SUCCESS(
      graph_manager_.ExecuteGraphWithStreamAsync(execution_info.instance_id, stream, compile_inputs, outputs), task);
  return SUCCESS;
}

Status JitExecutor::TryExecuteWithoutProcess(UserGraphExecution &task) {
  const auto first_ep = order_.GetFirstPoint();
  if (first_ep == nullptr || !first_ep->IsLast()) {
    return ge::UNSUPPORTED;
  }
  GELOGD("Get EP[%ld] of USER_GRAPH[%u] for LoadGraph", first_ep->GetId(), task.user_graph_id);
  const auto gep = first_ep->FindGuarded(*(task.external_rt_inputs));
  if (gep == nullptr || !gep->Compiled()) {
    return ge::UNSUPPORTED;
  }
  GELOGD("Get GEP[compiled_graph_id:%u] [compiled? %d] of EP[%ld] USER_GRAPH[%u].", gep->GetCompiledGraphId(),
         gep->Compiled(), first_ep->GetId(), task.user_graph_id);
  const auto iter = geps_to_inner_ge_graph_id_.find(gep);
  if (iter == geps_to_inner_ge_graph_id_.end()) {
    return ge::UNSUPPORTED;
  }
  GELOGD("Graph id:%u No need Execute with Jit process.", iter->second);
  rtStream_t const stream = (task.stream == nullptr) ? stream_ : task.stream;
  GE_ASSERT_SUCCESS(
      graph_manager_.ExecuteGraphWithStreamAsync(iter->second, stream, *(task.external_rt_inputs), *(task.rt_outputs)));
  return SUCCESS;
}

Status JitExecutor::Compile(const std::vector<ge::Tensor> &inputs, GuardedExecutionPoint *gep, uint64_t session_id) {
  std::lock_guard<std::mutex> locker(mutex_);
  if (!gep->Compiled()) {
    auto instance_id = compile_context_.GenNewGraphId();
    GELOGI("Start to compile GEP[%u] for EP[%ld].", instance_id, gep->GetOwnerEp()->GetId());
    GE_ASSERT_TRUE(geps_to_inner_ge_graph_id_.emplace(gep, instance_id).second);

    GE_ASSERT_SUCCESS(compile_context_.Compile(instance_id, gep->GetGraph(), inputs,
                                               gep->GetOwnerEp()->GetEpGraphOptions(), session_id));
    GE_ASSERT_RT_OK(SetDeviceCached(device_id_));
    compiled_ge_graph_id_.emplace_back(instance_id);
    GE_ASSERT_TRUE(gep->SetCompiled(instance_id, gep->GetGraph()));
  }
  return SUCCESS;
}

Status JitExecutor::CompileAndLoad(const std::vector<gert::Tensor> &inputs, GuardedExecutionPoint *gep,
                                   uint32_t &instance_id, const rtStream_t stream,
                                   const std::map<AscendString, AscendString> &load_options, uint64_t session_id) {
  /*
   * | epm status    | instance exists | instance does not exist |
   * |---------------|-----------------|--------------------|
   * | not compiled  | ERROR           | compile + load     |
   * | compiled      | load            | fork + load        |
   * | loaded        | do nothing      | ERROR              |
   *
   *  NOTE: The "Compiled & Instance Exists" scenario represents a cache aged case,
   *        which has not been implemented yet.
   */
  // todo handle mutex here
  // compile and load need to lock mutex_
  // find instance just read mutex
  if (!gep->Compiled()) {
    instance_id = compile_context_.GenNewGraphId();
    GELOGI("Start to compile GEP[%u] for EP[%ld], session_id: %llu.", instance_id, gep->GetOwnerEp()->GetId(),
           session_id);
    GE_ASSERT_TRUE(geps_to_inner_ge_graph_id_.emplace(gep, instance_id).second);

    std::map<std::string, std::string> options;
    GE_ASSERT_SUCCESS(cmc_.CreateKeyOptionForGuardedExecutionPoint(gep, options));
    GE_ASSERT_SUCCESS(compile_context_.Compile(instance_id, gep->GetGraph(), inputs, options, session_id),
                      "GEP:%u, EP:%ld, session_id:%llu", instance_id, gep->GetOwnerEp()->GetId(), session_id);
    GE_ASSERT_RT_OK(SetDeviceCached(device_id_));
    // todo 编译失败的时候，需要处理死锁问题
    compiled_ge_graph_id_.emplace_back(instance_id);
    GE_ASSERT_TRUE(gep->SetCompiled(instance_id, gep->GetGraph()));
    GE_ASSERT_SUCCESS(ApplyFixedFeatureMemory(instance_id));
    GE_ASSERT_SUCCESS(compile_context_.Load(instance_id, load_options, stream));
  } else {
    auto iter = geps_to_inner_ge_graph_id_.find(gep);
    if (iter == geps_to_inner_ge_graph_id_.end()) {
      instance_id = compile_context_.GenNewGraphId();
      GE_ASSERT_SUCCESS(compile_context_.Fork(gep->GetCompiledGraphId(), instance_id));
      GE_ASSERT_RT_OK(SetDeviceCached(device_id_));
      GE_ASSERT_SUCCESS(ApplyFixedFeatureMemory(instance_id));
      GE_ASSERT_SUCCESS(compile_context_.Load(instance_id, load_options, stream));
      GE_ASSERT_TRUE(geps_to_inner_ge_graph_id_.emplace(gep, instance_id).second);
      gep->SetForked(instance_id);
    } else {
      instance_id = iter->second;
    }
  }
  return SUCCESS;
}

Status JitExecutor::ApplyFixedFeatureMemory(uint32_t instance_id) const {
  if (fixed_feature_memory_settings_ == nullptr) {
    return SUCCESS;
  }
  for (const auto &setting : *fixed_feature_memory_settings_) {
    GE_CHK_STATUS_RET(graph_manager_.SetFixedFeatureMemoryBase(instance_id, setting.first, setting.second.first,
                                                               setting.second.second),
                      "Apply fixed feature memory base failed, instance_id:%u, type:%d", instance_id,
                      static_cast<int32_t>(setting.first));
    GELOGI("Apply fixed feature memory base success, instance_id:%u, type:%d, memory:%p, size:%zu", instance_id,
           static_cast<int32_t>(setting.first), setting.second.first, setting.second.second);
  }
  return SUCCESS;
}

bool JitExecutor::IsUserGraphNeedRebuild() {
  return std::any_of(compiled_ge_graph_id_.cbegin(), compiled_ge_graph_id_.cend(), [this](uint32_t graph_id) {
    const auto is_graph_need_rebuild = compile_context_.IsGraphNeedRebuild(graph_id);
    GELOGI("Graph instance id %u need rebuild : %d", graph_id, is_graph_need_rebuild);
    return is_graph_need_rebuild;
  });
}
}  // namespace ge