OpenHarmony-v6.0-Release/s

/**
 * Copyright 2020-2022 Huawei Technologies Co., Ltd
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#include "minddata/dataset/include/dataset/execute.h"

#include <algorithm>
#include <fstream>
#include <locale>
#include <string>

#include "minddata/dataset/core/de_tensor.h"
#include "minddata/dataset/core/tensor_row.h"
#include "minddata/dataset/core/tensor.h"
#include "minddata/dataset/core/type_id.h"
#include "minddata/dataset/kernels/data/compose_op.h"
#include "minddata/dataset/kernels/ir/tensor_operation.h"
#include "minddata/dataset/kernels/tensor_op.h"
#include "minddata/dataset/util/log_adapter.h"
#include "minddata/dataset/core/ascend_resource.h"
#include "minddata/dataset/kernels/image/dvpp/utils/CommonDataType.h"
#include "minddata/dataset/kernels/ir/vision/ascend_vision_ir.h"
#if !defined(BUILD_LITE) && defined(ENABLE_D)
#include "minddata/dataset/kernels/image/image_utils.h"
#endif
#ifndef BUILD_LITE
#include "mindspore/core/utils/file_utils.h"
namespace platform = mindspore;
#else
#include "mindspore/lite/src/common/file_utils.h"
namespace platform = mindspore::lite;
#endif
#if !defined(BUILD_LITE) && defined(ENABLE_D)
#include "minddata/dataset/kernels/image/dvpp/acl_adapter.h"
#include "minddata/dataset/kernels/image/dvpp/utils/ErrorCode.h"
#endif

namespace mindspore {
namespace dataset {
using json = nlohmann::json;

struct Execute::ExtraInfo {
  std::multimap<std::string, std::vector<uint32_t>> aipp_cfg_;
  bool init_with_shared_ptr_ = true;  // Initial execute object with shared_ptr as default
#if defined(WITH_BACKEND) || defined(ENABLE_ACL)
  std::multimap<std::string, std::string> op2para_map_ = {{vision::kDvppCropJpegOperation, "size"},
                                                          {vision::kDvppDecodeResizeOperation, "size"},
                                                          {vision::kDvppDecodeResizeCropOperation, "crop_size"},
                                                          {vision::kDvppDecodeResizeCropOperation, "resize_size"},
                                                          {vision::kDvppNormalizeOperation, "mean"},
                                                          {vision::kDvppNormalizeOperation, "std"},
                                                          {vision::kDvppResizeJpegOperation, "size"}};
#endif
};

Status Execute::InitResource(MapTargetDevice device_type, uint32_t device_id) {
  if (device_type == MapTargetDevice::kAscend310) {
    MS_LOG(INFO) << "InitResource for Ascend310";
#if defined(WITH_BACKEND) || defined(ENABLE_ACL)
    if (device_resource_ == nullptr) {
      device_resource_ = std::make_shared<AscendResource>();
      Status rc = device_resource_->InitResource(device_id);
      if (!rc.IsOk()) {
        device_resource_ = nullptr;
        std::string err_msg = "Initialize Ascend310 resource fail";
        MS_LOG(ERROR) << err_msg;
        RETURN_STATUS_UNEXPECTED(err_msg);
      }
    } else {
      MS_LOG(INFO) << "Ascend310 context resource had been initialized.";
    }
#endif
    device_type_ = device_type;
#if !defined(BUILD_LITE) && defined(ENABLE_D)
  } else if (device_type == MapTargetDevice::kAscend910B) {
    MS_LOG(INFO) << "InitResource for Ascend910B";
    if (device_context_ == nullptr) {
      auto ms_context = MsContext::GetInstance();
      RETURN_UNEXPECTED_IF_NULL(ms_context);
      device_context_ = device::DeviceContextManager::GetInstance().GetOrCreateDeviceContext(
        {ms_context->get_param<std::string>(MS_CTX_DEVICE_TARGET), ms_context->get_param<uint32_t>(MS_CTX_DEVICE_ID)});
      RETURN_UNEXPECTED_IF_NULL(device_context_);
      device_context_->Initialize();
      RETURN_UNEXPECTED_IF_NULL(device_context_->device_res_manager_);

      std::string soc_version;
      auto ret = AclAdapter::GetInstance().GetSocName(&soc_version);
      if (ret != APP_ERR_OK) {
        std::string error = "Get Soc Version failed.";
        RETURN_STATUS_UNEXPECTED(error);
      }
      if (soc_version.find("Ascend910B") == std::string::npos && soc_version.find("Ascend910C") == std::string::npos) {
        std::string err_msg = "The SoC: " + soc_version + " is not Ascend910B / Ascend910C";
        MS_LOG(ERROR) << err_msg;
        // reset the device_context_, because the Soc is not support yet
        device_context_ = nullptr;
        RETURN_STATUS_UNEXPECTED(err_msg);
      }

      if (!device_context_->device_res_manager_->CreateStream(&stream_id_)) {
        std::string err_msg = "Create new stream failed on Ascend910B platform.";
        MS_LOG(ERROR) << err_msg;
        RETURN_STATUS_UNEXPECTED(err_msg);
      }
      MS_LOG(INFO) << "Create new stream id: " << std::to_string(stream_id_);
      device_type_ = device_type;
    } else {
      MS_LOG(INFO) << "Ascend910B context resource had been initialized.";
    }
#endif
  }
  return Status::OK();
}

Execute::Execute(const std::shared_ptr<TensorOperation> &op, MapTargetDevice device_type, uint32_t device_id) {
  (void)ops_.emplace_back(op);
  device_type_ = device_type;
  info_ = std::make_shared<ExtraInfo>();
#if !defined(BUILD_LITE) && defined(ENABLE_D)
  device_context_ = nullptr;
#endif

  // Ascend910B
  if (op->Type() == MapTargetDevice::kAscend910B) {
    // Update the device_type
    device_type = MapTargetDevice::kAscend910B;
    MS_LOG(INFO) << "Update device type: " << std::to_string(static_cast<int>(device_type));
  }

  (void)InitResource(device_type, device_id);
}

Execute::Execute(const std::shared_ptr<TensorTransform> &op, MapTargetDevice device_type, uint32_t device_id) {
  // Initialize the op and other context
  (void)transforms_.emplace_back(op);

  info_ = std::make_shared<ExtraInfo>();
  device_type_ = device_type;
  (void)InitResource(device_type, device_id);
}

Execute::Execute(const std::reference_wrapper<TensorTransform> &op, MapTargetDevice device_type, uint32_t device_id) {
  // Initialize the transforms_ and other context
  std::shared_ptr<TensorOperation> operation = op.get().Parse();
  (void)ops_.emplace_back(std::move(operation));

  info_ = std::make_shared<ExtraInfo>();
  info_->init_with_shared_ptr_ = false;
  device_type_ = device_type;
  (void)InitResource(device_type, device_id);
}

// Execute function for the example case: auto decode(new vision::Decode());
Execute::Execute(TensorTransform *op, MapTargetDevice device_type, uint32_t device_id) {
  // Initialize the transforms_ and other context
  (void)ops_.emplace_back(op->Parse());

  info_ = std::make_shared<ExtraInfo>();
  info_->init_with_shared_ptr_ = false;
  device_type_ = device_type;
  (void)InitResource(device_type, device_id);
}

Execute::Execute(const std::vector<std::shared_ptr<TensorOperation>> &ops, MapTargetDevice device_type,
                 uint32_t device_id)
    : ops_(ops), device_type_(device_type) {
  info_ = std::make_shared<ExtraInfo>();
  (void)InitResource(device_type, device_id);
}

Execute::Execute(const std::vector<std::shared_ptr<TensorTransform>> &ops, MapTargetDevice device_type,
                 uint32_t device_id) {
  // Initialize the transforms_ and other context
  transforms_ = ops;

  info_ = std::make_shared<ExtraInfo>();
  device_type_ = device_type;
  (void)InitResource(device_type, device_id);
}

Execute::Execute(const std::vector<std::reference_wrapper<TensorTransform>> &ops, MapTargetDevice device_type,
                 uint32_t device_id) {
  // Initialize the transforms_ and other context
  if (device_type == MapTargetDevice::kCpu) {
    (void)std::transform(
      ops.begin(), ops.end(), std::back_inserter(ops_),
      [](TensorTransform &operation) -> std::shared_ptr<TensorOperation> { return operation.Parse(); });
  } else {
    for (auto &op : ops) {
      (void)ops_.emplace_back(op.get().Parse(device_type));
    }
  }

  info_ = std::make_shared<ExtraInfo>();
  info_->init_with_shared_ptr_ = false;
  device_type_ = device_type;
  (void)InitResource(device_type, device_id);
}

// Execute function for the example vector case: auto decode(new vision::Decode());
Execute::Execute(const std::vector<TensorTransform *> &ops, MapTargetDevice device_type, uint32_t device_id) {
  // Initialize the transforms_ and other context
  (void)std::transform(
    ops.begin(), ops.end(), std::back_inserter(ops_),
    [](TensorTransform *operation) -> std::shared_ptr<TensorOperation> { return operation->Parse(); });

  info_ = std::make_shared<ExtraInfo>();
  info_->init_with_shared_ptr_ = false;
  device_type_ = device_type;
  (void)InitResource(device_type, device_id);
}

Execute::~Execute() {
  if (device_type_ == MapTargetDevice::kAscend310) {
    if (device_resource_) {
      auto rc = device_resource_->FinalizeResource();
      if (rc.IsError()) {
        MS_LOG(ERROR) << "Device resource release failed, error msg is " << rc;
      }
    } else {
      MS_LOG(ERROR) << "Device resource is nullptr which is illegal under case Ascend310";
    }
  }
}

Status Execute::UpdateOperation(const std::shared_ptr<TensorOperation> &op) {
  // clear the ops_ first
  ops_.clear();
  (void)ops_.emplace_back(op);
  return Status::OK();
}

Status Execute::BuildTransforms(std::vector<std::shared_ptr<TensorOp>> *transforms_rt) {
  // Parse TensorTransform transforms_ into TensorOperation ops_
  if (info_->init_with_shared_ptr_) {
    RETURN_IF_NOT_OK(ParseTransforms());
    info_->init_with_shared_ptr_ = false;
  }
  CHECK_FAIL_RETURN_UNEXPECTED(!ops_.empty(), "Input TensorOperation should be provided.");

  std::map<MapTargetDevice, std::string> env_list = {
    {MapTargetDevice::kCpu, "kCpu"}, {MapTargetDevice::kGpu, "kGpu"}, {MapTargetDevice::kAscend310, "kAscend310"}};

  // Validate and build runtime ops
  for (size_t i = 0; i < ops_.size(); i++) {
    if (ops_[i] == nullptr) {
      std::string err_msg = "Input TensorOperation[" + std::to_string(i) +
                            "] is unsupported on your input device:" + env_list.at(device_type_);
      MS_LOG(ERROR) << err_msg;
      RETURN_STATUS_UNEXPECTED(err_msg);
    }
    RETURN_IF_NOT_OK(ops_[i]->ValidateParams());
    (void)transforms_rt->emplace_back(ops_[i]->Build());
  }

  // if transforms_rt[0] is ComposeOp and all dvpp, extract all the ops from ComposeOp
  if ((*transforms_rt)[0]->IsDvppOp() && (*transforms_rt)[0]->Name() == kComposeOp) {
    if (dynamic_cast<ComposeOp *>((*transforms_rt)[0].get())->IsMixedOps()) {
      std::string err_msg = "Currently, it is not supported to mix DVPP transforms with CPU transforms in Compose.";
      MS_LOG(ERROR) << err_msg;
      RETURN_STATUS_UNEXPECTED(err_msg);
    }
    *transforms_rt = std::move(dynamic_cast<ComposeOp *>((*transforms_rt)[0].get())->GetOps());
    MS_LOG(INFO) << "Extract the ComposeOp to " << std::to_string((*transforms_rt).size()) << " ops.";
  }

  return Status::OK();
}

Status Execute::operator()(const mindspore::MSTensor &input, mindspore::MSTensor *output) {
  // Validate input tensor
  RETURN_UNEXPECTED_IF_NULL(output);
  CHECK_FAIL_RETURN_UNEXPECTED(input.DataSize() > 0, "Input Tensor has no data.");
  CHECK_FAIL_RETURN_UNEXPECTED(ValidateDevice(), "Device Type should be 'Ascend310' or 'CPU'.");
  std::vector<std::shared_ptr<TensorOp>> transforms_rt;
  CHECK_FAIL_RETURN_UNEXPECTED(BuildTransforms(&transforms_rt), "Building Transform ops failed!");

  if (device_type_ == MapTargetDevice::kCpu) {
    // Convert mindspore::Tensor to dataset::Tensor
    std::shared_ptr<dataset::Tensor> de_tensor;
    Status rc = dataset::Tensor::CreateFromMemory(dataset::TensorShape(input.Shape()),
                                                  MSTypeToDEType(static_cast<TypeId>(input.DataType())),
                                                  (const uchar *)(input.Data().get()), input.DataSize(), &de_tensor);
    if (rc.IsError()) {
      MS_LOG(ERROR) << rc;
      return rc;
    }

    // Apply transforms on tensor
    for (auto &t : transforms_rt) {
      TensorRow de_tensor_row;
      TensorRow de_output_row;
      de_tensor_row.push_back(de_tensor);
      de_output_row.resize(1);
      Status rc_ = t->Compute(de_tensor_row, &de_output_row);
      if (rc_.IsError()) {
        MS_LOG(ERROR) << rc_;
        return rc_;
      }

      // For next transform
      de_tensor = std::move(de_output_row[0]);
    }

    // Convert dataset::Tensor to mindspore::Tensor
    if (!de_tensor->HasData()) {
      std::stringstream ss;
      ss << "Transformation returned an empty tensor with shape " << de_tensor->shape();
      RETURN_STATUS_UNEXPECTED(ss.str());
    }
    *output = mindspore::MSTensor(std::make_shared<DETensor>(de_tensor));
  } else if (device_type_ ==
             MapTargetDevice::kAscend310) {  // Ascend310 case, where we must set Ascend resource on each operations
#if defined(WITH_BACKEND) || defined(ENABLE_ACL)
    CHECK_FAIL_RETURN_UNEXPECTED(device_resource_, "Device resource is nullptr which is illegal under case Ascend310.");
    // Sink data from host into device
    std::shared_ptr<mindspore::dataset::DeviceTensor> device_input;
    RETURN_IF_NOT_OK(device_resource_->Sink(input, &device_input));

    for (auto &t : transforms_rt) {
      // Initialize AscendResource for each operations
      std::shared_ptr<DeviceTensor> device_output;
      RETURN_IF_NOT_OK(t->SetAscendResource(device_resource_));

      RETURN_IF_NOT_OK(t->Compute(device_input, &device_output));

      // For next transform
      device_input = std::move(device_output);
    }
    CHECK_FAIL_RETURN_UNEXPECTED(device_input->HasDeviceData(), "Apply transform failed, output tensor has no data.");

    *output = mindspore::MSTensor(std::make_shared<DETensor>(device_input, true));
#endif
  } else {
    std::string err_msg = "Your input device is not supported. (Option: CPU or Ascend310)";
    MS_LOG(ERROR) << err_msg;
    RETURN_STATUS_UNEXPECTED(err_msg);
  }
  return Status::OK();
}

Status Execute::operator()(const std::vector<MSTensor> &input_tensor_list, std::vector<MSTensor> *output_tensor_list) {
  // Validate input tensor
  RETURN_UNEXPECTED_IF_NULL(output_tensor_list);
  CHECK_FAIL_RETURN_UNEXPECTED(!input_tensor_list.empty(), "Input Tensor is not valid.");
  output_tensor_list->clear();
  for (auto &tensor : input_tensor_list) {
    CHECK_FAIL_RETURN_UNEXPECTED(tensor.DataSize() > 0, "Input Tensor has no data.");
  }
  CHECK_FAIL_RETURN_UNEXPECTED(ValidateDevice(), "Device Type should be 'Ascend310' or 'CPU'.");
  std::vector<std::shared_ptr<TensorOp>> transforms_rt;
  CHECK_FAIL_RETURN_UNEXPECTED(BuildTransforms(&transforms_rt), "Building Transform ops failed!");

  if (device_type_ == MapTargetDevice::kCpu) {  // Case CPU
    TensorRow de_tensor_list;
    for (auto &tensor : input_tensor_list) {
      std::shared_ptr<dataset::Tensor> de_tensor;
      Status rc = dataset::Tensor::CreateFromMemory(
        dataset::TensorShape(tensor.Shape()), MSTypeToDEType(static_cast<TypeId>(tensor.DataType())),
        (const uchar *)(tensor.Data().get()), tensor.DataSize(), &de_tensor);
      if (rc.IsError()) {
        MS_LOG(ERROR) << rc;
        RETURN_IF_NOT_OK(rc);
      }
      (void)de_tensor_list.emplace_back(std::move(de_tensor));
    }
    // Apply transforms on tensor
    for (auto &t : transforms_rt) {
      TensorRow de_output_list;
      RETURN_IF_NOT_OK(t->Compute(de_tensor_list, &de_output_list));
      // For next transform
      de_tensor_list = std::move(de_output_list);
    }
    int32_t idx = 0;
    for (auto &tensor : de_tensor_list) {
      if (!tensor->HasData()) {
        std::stringstream ss;
        ss << "Transformation returned an empty tensor at location " << idx << ". ";
        ss << "The shape of the tensor is " << tensor->shape();
        MS_LOG(WARNING) << ss.str();
      }
      auto ms_tensor = mindspore::MSTensor(std::make_shared<DETensor>(tensor));
      (void)output_tensor_list->emplace_back(ms_tensor);
      ++idx;
    }
    CHECK_FAIL_RETURN_UNEXPECTED(!output_tensor_list->empty(), "Output Tensor is not valid.");
  } else if (device_type_ ==
             MapTargetDevice::kAscend310) {  // Ascend310 case, where we must set Ascend resource on each operations
    CHECK_FAIL_RETURN_UNEXPECTED(device_resource_, "Device resource is nullptr which is illegal under case Ascend310.");
    for (auto &input_tensor : input_tensor_list) {
      // Sink each data from host into device
      std::shared_ptr<dataset::DeviceTensor> device_input;
      RETURN_IF_NOT_OK(device_resource_->Sink(input_tensor, &device_input));

      for (auto &t : transforms_rt) {
        std::shared_ptr<DeviceTensor> device_output;
        RETURN_IF_NOT_OK(t->SetAscendResource(device_resource_));

        RETURN_IF_NOT_OK(t->Compute(device_input, &device_output));

        // For next transform
        device_input = std::move(device_output);
      }
      CHECK_FAIL_RETURN_UNEXPECTED(device_input->HasDeviceData(), "Apply transform failed, output tensor has no data");
      // Due to the limitation of Ascend310 memory, we have to pop every data onto host memory
      // So the speed of this batch method is slower than solo mode
      std::shared_ptr<mindspore::dataset::Tensor> host_output;
      RETURN_IF_NOT_OK(device_resource_->Pop(device_input, &host_output));

      auto ms_tensor = mindspore::MSTensor(std::make_shared<DETensor>(host_output));
      (void)output_tensor_list->emplace_back(ms_tensor);
      // Release the data on the device because we have copied one piece onto host
      RETURN_IF_NOT_OK(device_resource_->DeviceDataRelease());
    }
    CHECK_FAIL_RETURN_UNEXPECTED(!output_tensor_list->empty(), "Output Tensor vector is empty.");
  } else {
    std::string err_msg = "Your input device is not supported. (Option: CPU or Ascend310)";
    MS_LOG(ERROR) << err_msg;
    RETURN_STATUS_UNEXPECTED(err_msg);
  }
  return Status::OK();
}

Status PyExecute::operator()(const std::vector<std::shared_ptr<Tensor>> &input_tensor_list,
                             std::vector<std::shared_ptr<Tensor>> *out) {
  // Validate input tensor
  CHECK_FAIL_RETURN_UNEXPECTED(!input_tensor_list.empty(), "Input Tensor is not valid.");
  RETURN_UNEXPECTED_IF_NULL(out);
  out->clear();
  for (auto &tensor : input_tensor_list) {
    CHECK_FAIL_RETURN_UNEXPECTED(tensor->Size() > 0, "Input Tensor has no data.");
  }
  CHECK_FAIL_RETURN_UNEXPECTED(ValidateDevice(), "Device Type should be 'CPU'.");

  std::vector<std::shared_ptr<TensorOp>> transforms_rt;
  CHECK_FAIL_RETURN_UNEXPECTED(BuildTransforms(&transforms_rt), "Building Transform ops failed!");

  if (!transforms_rt[0]->IsDvppOp()) {
    TensorRow de_tensor_list(input_tensor_list);

    // Apply transforms on tensor
    for (auto &t : transforms_rt) {
      TensorRow de_output_list;
      RETURN_IF_NOT_OK(t->Compute(de_tensor_list, &de_output_list));
      // For next transform
      de_tensor_list = std::move(de_output_list);
    }
    *out = std::move(de_tensor_list.getRow());
    CHECK_FAIL_RETURN_UNEXPECTED(!out->empty(), "Output Tensor is not valid.");
#if !defined(BUILD_LITE) && defined(ENABLE_D)
  } else if (transforms_rt[0]->IsDvppOp() == true) {
    (void)InitResource(MapTargetDevice::kAscend910B);

    // construct the device tensor list by host tensor
    std::vector<std::shared_ptr<DeviceTensorAscend910B>> device_tensor_list;
    for (auto &item : input_tensor_list) {
      std::shared_ptr<DeviceTensorAscend910B> device_tensor = nullptr;
      // here we use the first op's IsHWC() to create device tensor
      if (transforms_rt[0]->Name() == kDvppConvertColorOp) {
        std::vector<int> channels = {1, 3, 4};
        RETURN_IF_NOT_OK(DeviceTensorAscend910B::CreateDeviceTensor(item, device_context_, stream_id_, &device_tensor,
                                                                    transforms_rt[0]->IsHWC(), channels));
      } else {
        RETURN_IF_NOT_OK(DeviceTensorAscend910B::CreateDeviceTensor(item, device_context_, stream_id_, &device_tensor,
                                                                    transforms_rt[0]->IsHWC()));
      }
      device_tensor_list.push_back(std::move(device_tensor));
    }

    // hold all the inputs and release them when the executor finish
    std::vector<std::vector<std::shared_ptr<DeviceTensorAscend910B>>> hold_input_list;

    for (auto &t : transforms_rt) {
      MS_LOG(INFO) << "Execute " << t->Name() << " transform.";
      std::vector<std::shared_ptr<DeviceTensorAscend910B>> device_output_list;

      // if the op is Decode, we should get the height and width form JPEG header and create the output tensor first
      int img_width = 0;
      int img_height = 0;
      if (t->Name() == kDvppDecodeOp) {
        for (int32_t k = 0; k < input_tensor_list.size(); k++) {
          CHECK_FAIL_RETURN_UNEXPECTED(input_tensor_list[k]->shape().Rank() == 1,
                                       "Invalid data shape. Currently only support 1D. Its rank is: " +
                                         std::to_string(input_tensor_list[k]->shape().Rank()));
          CHECK_FAIL_RETURN_UNEXPECTED(IsNonEmptyJPEG(input_tensor_list[k]) == true,
                                       "Invalid image type. Currently only support JPG. Its shape is: " +
                                         input_tensor_list[k]->shape().ToString());
          CHECK_FAIL_RETURN_UNEXPECTED(
            input_tensor_list[k]->type() == DataType::DE_UINT8,
            "Invalid data type. Currently only support uint8. Its type is: " + input_tensor_list[k]->type().ToString());
          RETURN_IF_NOT_OK(GetJpegImageInfo(input_tensor_list[k], &img_width, &img_height));
          TensorShape shape{1, img_height, img_width, 3};
          DataType type(DataType::DE_UINT8);
          std::shared_ptr<DeviceTensorAscend910B> device_tensor = nullptr;
          RETURN_IF_NOT_OK(
            DeviceTensorAscend910B::CreateDeviceTensor(shape, type, device_tensor_list[k]->GetDeviceContext(),
                                                       device_tensor_list[k]->GetStreamID(), &device_tensor));
          device_output_list.push_back(std::move(device_tensor));
        }
      }

      RETURN_IF_NOT_OK(t->Compute(device_tensor_list, &device_output_list));

      // move the input to the hold_input_list first
      hold_input_list.push_back(std::move(device_tensor_list));

      // For next transform
      device_tensor_list = std::move(device_output_list);
    }

    // the transforms all done, do SyncStream
    if (!device_context_->device_res_manager_->SyncStream(stream_id_)) {
      std::string err_msg = "SyncStream stream id: " + std::to_string(stream_id_) + " failed.";
      RETURN_STATUS_UNEXPECTED(err_msg);
    }

    // copy the data from device tensor to host tensor
    std::vector<std::shared_ptr<Tensor>> host_out(device_tensor_list.size());
    int32_t i = 0;
    for (auto &item : device_tensor_list) {
      CHECK_FAIL_RETURN_UNEXPECTED(item->ToHostTensor(&host_out[i]), "Copy tensor from device to host failed.");
      i += 1;
    }
    *out = std::move(host_out);
    CHECK_FAIL_RETURN_UNEXPECTED(!out->empty(), "Output Tensor is not valid.");

    // release all the inputs' device memory and workspace memory
    for (auto &tensor_list : hold_input_list) {
      for (auto &item : tensor_list) {
        if (!item->ReleaseDeviceMemory()) {
          std::string err_msg = "Release the device memory failed after the dvpp ops executed.";
          MS_LOG(ERROR) << err_msg;
          RETURN_STATUS_UNEXPECTED(err_msg);
        }
      }
    }
#endif
  } else {
    std::string err_msg = "Your input device is not supported. (Option: CPU or Ascend910B)";
    MS_LOG(ERROR) << err_msg;
    RETURN_STATUS_UNEXPECTED(err_msg);
  }
  return Status::OK();
}

std::vector<uint32_t> AippSizeFilter(const std::vector<uint32_t> &resize_para, const std::vector<uint32_t> &crop_para) {
  std::vector<uint32_t> aipp_size;

  // Special condition where (no Crop and no Resize) or (no Crop and resize with fixed ratio) will lead to dynamic input
  if ((resize_para.empty() || resize_para.size() == 1) && crop_para.empty()) {
    aipp_size = {0, 0};
    MS_LOG(WARNING) << "Dynamic input shape is not supported, incomplete aipp config file will be generated. Please "
                       "checkout your TensorTransform input, both src_image_size_h and src_image_size will be 0.";
    return aipp_size;
  }

  if (resize_para.empty()) {  // If only Crop operation exists
    aipp_size = crop_para;
  } else if (crop_para.empty()) {  // If only Resize operation with 2 parameters exists
    aipp_size = resize_para;
  } else {  // If both of them exist
    if (resize_para.size() == 1) {
      aipp_size = crop_para;
    } else {
      aipp_size =
        *min_element(resize_para.begin(), resize_para.end()) < *min_element(crop_para.begin(), crop_para.end())
          ? resize_para
          : crop_para;
    }
  }

  aipp_size[0] = DVPP_ALIGN_UP(aipp_size[0], VPC_HEIGHT_ALIGN);  // H
  aipp_size[1] = DVPP_ALIGN_UP(aipp_size[1], VPC_WIDTH_ALIGN);   // W
  return aipp_size;
}

std::vector<uint32_t> AippMeanFilter(const std::vector<uint32_t> &normalize_para) {
  std::vector<uint32_t> aipp_mean;
  if (normalize_para.size() == 6) {  // If Normalize operation exist
    std::transform(normalize_para.begin(), normalize_para.begin() + 3, std::back_inserter(aipp_mean),
                   [](uint32_t i) { return static_cast<uint32_t>(i / 10000); });
  } else {
    aipp_mean = {0, 0, 0};
  }
  return aipp_mean;
}

std::vector<float> AippStdFilter(const std::vector<uint32_t> &normalize_para) {
  std::vector<float> aipp_std;
  if (normalize_para.size() == 6) {  // If Normalize operation exist
    auto zeros = std::find(std::begin(normalize_para), std::end(normalize_para), 0);
    if (zeros == std::end(normalize_para)) {
      if (std::any_of(normalize_para.begin() + 3, normalize_para.end(), [](uint32_t i) { return i == 0; })) {
        MS_LOG(ERROR) << "value in normalize para got 0.";
        return {};
      }
      std::transform(normalize_para.begin() + 3, normalize_para.end(), std::back_inserter(aipp_std),
                     [](uint32_t i) { return 10000 / static_cast<float>(i); });
    } else {  // If 0 occurs in std vector
      MS_LOG(WARNING) << "Detect 0 in std vector, please verify your input.";
      aipp_std = {1.0, 1.0, 1.0};
    }
  } else {
    aipp_std = {1.0, 1.0, 1.0};
  }
  return aipp_std;
}

Status AippInfoCollection(std::map<std::string, std::string> *aipp_options, const std::vector<uint32_t> &aipp_size,
                          const std::vector<uint32_t> &aipp_mean, const std::vector<float> &aipp_std) {
  RETURN_UNEXPECTED_IF_NULL(aipp_options);
  // Several aipp config parameters
  aipp_options->insert(std::make_pair("related_input_rank", "0"));
  aipp_options->insert(std::make_pair("src_image_size_w", std::to_string(aipp_size[1])));
  aipp_options->insert(std::make_pair("src_image_size_h", std::to_string(aipp_size[0])));
  aipp_options->insert(std::make_pair("crop", "false"));
  aipp_options->insert(std::make_pair("input_format", "YUV420SP_U8"));
  aipp_options->insert(std::make_pair("aipp_mode", "static"));
  aipp_options->insert(std::make_pair("csc_switch", "true"));
  aipp_options->insert(std::make_pair("rbuv_swap_switch", "false"));
  // Y = AX + b, this part is A
  std::vector<int32_t> color_space_matrix = {256, 0, 359, 256, -88, -183, 256, 454, 0};
  int count = 0;
  for (int i = 0; i < 3; i++) {
    for (int j = 0; j < 3; j++) {
      std::string key_word = "matrix_r" + std::to_string(i) + "c" + std::to_string(j);
      aipp_options->insert(std::make_pair(key_word, std::to_string(color_space_matrix[count])));
      ++count;
    }
  }
  // This part is b
  std::vector<uint32_t> color_space_bias = {0, 128, 128};
  for (int i = 0; i < 3; i++) {
    std::string key_word = "input_bias_" + std::to_string(i);
    aipp_options->insert(std::make_pair(key_word, std::to_string(color_space_bias[i])));
  }
  // Y = (X - mean - min) * [std^(-1)], this part is mean
  for (int i = 0; i < aipp_mean.size(); i++) {
    std::string key_word = "mean_chn_" + std::to_string(i);
    aipp_options->insert(std::make_pair(key_word, std::to_string(aipp_mean[i])));
  }
  // This part is min
  for (int i = 0; i < aipp_mean.size(); i++) {
    std::string key_word = "min_chn_" + std::to_string(i);
    aipp_options->insert(std::make_pair(key_word, "0.0"));
  }
  // This part is std^(-1)
  for (int i = 0; i < aipp_std.size(); i++) {
    std::string key_word = "var_reci_chn_" + std::to_string(i);
    aipp_options->insert(std::make_pair(key_word, std::to_string(aipp_std[i])));
  }
  return Status::OK();
}

std::string Execute::AippCfgGenerator() {
  std::string config_location = "./aipp.cfg";
  if (info_ == nullptr) {
    MS_LOG(ERROR) << "info_ is null";
    return "";
  }
#if defined(WITH_BACKEND) || defined(ENABLE_ACL)
  if (info_->init_with_shared_ptr_) {
    auto rc = ParseTransforms();
    RETURN_SECOND_IF_ERROR(rc, "");
    info_->init_with_shared_ptr_ = false;
  }
  std::vector<uint32_t> paras;  // Record the parameters value of each Ascend operations
  for (int32_t i = 0; i < ops_.size(); i++) {
    // Validate operation ir
    json ir_info;
    if (ops_[i] == nullptr) {
      MS_LOG(ERROR) << "Input TensorOperation[" + std::to_string(i) + "] is null.";
      return "";
    }

    // Define map between operation name and parameter name
    auto rc = ops_[i]->to_json(&ir_info);
    if (rc.IsError()) {
      MS_LOG(ERROR) << "IR information serialize to json failed, error msg is " << rc;
      return "";
    }

    // Collect the information of operations
    for (auto pos = info_->op2para_map_.equal_range(ops_[i]->Name()); pos.first != pos.second; ++pos.first) {
      auto paras_key_word = pos.first->second;
      paras = ir_info[paras_key_word].get<std::vector<uint32_t>>();
      info_->aipp_cfg_.insert(std::make_pair(ops_[i]->Name(), paras));
    }
  }

  std::ofstream outfile;
  outfile.open(config_location, std::ofstream::out);

  if (!outfile.is_open()) {
    MS_LOG(ERROR) << "Fail to open Aipp config file, please verify your system config(including authority)."
                  << "We will return empty string which represent the location of Aipp config file in this case.";
    return "";
  }

  if (device_type_ == MapTargetDevice::kAscend310) {
    // Process resize parameters and crop parameters to find out the final size of input data
    std::vector<uint32_t> resize_paras;
    std::vector<uint32_t> crop_paras;

    // Find resize parameters
    std::map<std::string, std::vector<uint32_t>>::iterator iter;
    if (info_->aipp_cfg_.find(vision::kDvppResizeJpegOperation) != info_->aipp_cfg_.end()) {
      iter = info_->aipp_cfg_.find(vision::kDvppResizeJpegOperation);
      resize_paras = iter->second;
    } else if (info_->aipp_cfg_.find(vision::kDvppDecodeResizeOperation) != info_->aipp_cfg_.end()) {
      iter = info_->aipp_cfg_.find(vision::kDvppDecodeResizeOperation);
      resize_paras = iter->second;
    }

    // Find crop parameters
    if (info_->aipp_cfg_.find(vision::kDvppCropJpegOperation) != info_->aipp_cfg_.end()) {
      iter = info_->aipp_cfg_.find(vision::kDvppCropJpegOperation);
      crop_paras = iter->second;
    } else if (info_->aipp_cfg_.find(vision::kDvppDecodeResizeCropOperation) != info_->aipp_cfg_.end()) {
      iter = info_->aipp_cfg_.find(vision::kDvppDecodeResizeCropOperation);
      crop_paras = iter->second;
    }
    if (crop_paras.size() == 1) {
      (void)crop_paras.emplace_back(crop_paras[0]);
    }

    std::vector<uint32_t> aipp_size = AippSizeFilter(resize_paras, crop_paras);

    // Process Normalization parameters to find out the final Normalization parameters for Aipp module
    std::vector<uint32_t> normalize_paras;
    if (info_->aipp_cfg_.find(vision::kDvppNormalizeOperation) != info_->aipp_cfg_.end()) {
      for (auto pos = info_->aipp_cfg_.equal_range(vision::kDvppNormalizeOperation); pos.first != pos.second;
           ++pos.first) {
        auto mean_or_std = pos.first->second;
        normalize_paras.insert(normalize_paras.end(), mean_or_std.begin(), mean_or_std.end());
      }
    }

    std::vector<uint32_t> aipp_mean = AippMeanFilter(normalize_paras);
    std::vector<float> aipp_std = AippStdFilter(normalize_paras);

    std::map<std::string, std::string> aipp_options;
    auto rc = AippInfoCollection(&aipp_options, aipp_size, aipp_mean, aipp_std);
    if (rc.IsError()) {
      MS_LOG(ERROR) << "Aipp information initialization failed, error msg is " << rc;
      outfile.close();
      return "";
    }

    std::string tab_char(4, ' ');
    outfile << "aipp_op {" << std::endl;
    for (auto &option : aipp_options) {
      outfile << tab_char << option.first << " : " << option.second << std::endl;
    }
    outfile << "}";
    outfile.close();
  } else {  // For case GPU or CPU
    outfile << "aipp_op {" << std::endl << "}";
    outfile.close();
    MS_LOG(WARNING) << "Your runtime environment is not Ascend310, this config file will lead to undefined behavior on "
                       "computing result. Please check that.";
  }

  platform::ChangeFileMode(config_location, S_IRUSR | S_IWUSR);
#endif
  return config_location;
}

bool IsEmptyPtr(const std::shared_ptr<TensorTransform> &api_ptr) { return api_ptr == nullptr; }

Status Execute::ParseTransforms() {
  auto iter = std::find_if(transforms_.begin(), transforms_.end(), IsEmptyPtr);
  if (iter != transforms_.end()) {
    std::string err_msg = "Your input TensorTransforms contain at least one nullptr, please check your input.";
    MS_LOG(ERROR) << err_msg;
    RETURN_STATUS_UNEXPECTED(err_msg);
  }

  if (device_type_ == MapTargetDevice::kCpu) {
    (void)std::transform(transforms_.begin(), transforms_.end(), std::back_inserter(ops_),
                         [](const std::shared_ptr<TensorTransform> &operation) -> std::shared_ptr<TensorOperation> {
                           return operation->Parse();
                         });
  } else if (device_type_ == MapTargetDevice::kAscend310) {
    for (auto &transform_ : transforms_) {
      (void)ops_.emplace_back(transform_->Parse(device_type_));
    }
  } else if (device_type_ == MapTargetDevice::kAscend910B) {
    for (auto &transform_ : transforms_) {
      (void)ops_.emplace_back(transform_->Parse(device_type_));
    }
  } else {
    std::string err_msg = "Your input device is not supported. (Option: CPU or Ascend310)";
    MS_LOG(ERROR) << err_msg;
    RETURN_STATUS_UNEXPECTED(err_msg);
  }

  return Status::OK();
}

Status Execute::ValidateDevice() {
  if (device_type_ != MapTargetDevice::kCpu && device_type_ != MapTargetDevice::kAscend310 &&
      device_type_ != MapTargetDevice::kGpu && device_type_ != MapTargetDevice::kAscend910B) {
    std::string err_msg = "Your input device is not supported. (Option: CPU, GPU, Ascend310 or Ascend910B).";
    MS_LOG(ERROR) << err_msg;
    RETURN_STATUS_UNEXPECTED(err_msg);
  }
  return Status::OK();
}

Status Execute::DeviceMemoryRelease() {
  CHECK_FAIL_RETURN_UNEXPECTED(device_resource_, "Device resource is nullptr which is illegal under case Ascend310.");
  Status rc = device_resource_->DeviceDataRelease();
  if (rc.IsError()) {
    std::string err_msg = "Error in device data release";
    MS_LOG(ERROR) << err_msg;
    RETURN_STATUS_UNEXPECTED(err_msg);
  }
  return Status::OK();
}

Status Execute::Run(const std::vector<std::shared_ptr<dataset::Execute>> &data_graph,
                    const std::vector<mindspore::MSTensor> &inputs, std::vector<mindspore::MSTensor> *outputs) {
  RETURN_UNEXPECTED_IF_NULL(outputs);
  std::vector<MSTensor> transform_inputs = inputs;
  std::vector<MSTensor> transform_outputs;
  if (!data_graph.empty()) {
    for (const auto &exes : data_graph) {
      CHECK_FAIL_RETURN_UNEXPECTED(exes != nullptr, "Given execute object is null.");
      Status ret = exes->operator()(transform_inputs, &transform_outputs);
      if (ret != kSuccess) {
        MS_LOG(ERROR) << "Run preprocess failed:" << ret.GetErrDescription();
        return ret;
      }
      MS_LOG(DEBUG) << "transform_outputs[0].Shape: " << transform_outputs[0].Shape();
      transform_inputs = transform_outputs;
    }
    *outputs = std::move(transform_outputs);
  } else {
    std::string msg = "The set of Executors can not be empty.";
    MS_LOG(ERROR) << msg;
    RETURN_STATUS_UNEXPECTED(msg);
  }
  return Status::OK();
}

// In the current stage, there is a cyclic dependency between libmindspore.so and c_dataengine.so,
// we make a C function here and dlopen by libminspore.so to avoid linking explicitly,
// will be fix after decouling libminspore.so into multi submodules
extern "C" {
// ExecuteRun_C has C-linkage specified, but returns user-defined type 'mindspore::Status' which is incompatible with C
void ExecuteRun_C(const std::vector<std::shared_ptr<dataset::Execute>> &data_graph,
                  const std::vector<mindspore::MSTensor> &inputs, std::vector<mindspore::MSTensor> *outputs,
                  Status *s) {
  Status ret = Execute::Run(data_graph, inputs, outputs);
  if (s == nullptr) {
    return;
  }
  *s = Status(ret);
}
}
}  // namespace dataset
}  // namespace mindspore