text: /**
   * Copyright 2021 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/engine/perf/cpu_sampling.h"
  
  #if !defined(_WIN32) && !defined(_WIN64) && !defined(__ANDROID__) && !defined(ANDROID) && !defined(__APPLE__)
  #include <sys/syscall.h>
  #endif
  #include <cmath>
  #include <cstdio>
  #include <algorithm>
  #include <fstream>
  #include <memory>
  #include <string>
  
  #include "minddata/dataset/api/python/pybind_conversion.h"
  #include "minddata/dataset/core/config_manager.h"
  #include "minddata/dataset/engine/execution_tree.h"
  #include "minddata/dataset/util/path.h"
  
  using json = nlohmann::json;
  namespace mindspore {
  namespace dataset {
  bool BaseCpu::fetched_all_process_shared_ = false;
  std::unordered_map<int32_t, std::vector<pid_t>> BaseCpu::op_process_shared_ = {};
  
  #if !defined(_WIN32) && !defined(_WIN64) && !defined(__ANDROID__) && !defined(ANDROID) && !defined(__APPLE__)
  #define USING_LINUX
  #endif
  
  BaseCpu::BaseCpu() {
    pre_cpu_stat_.user_stat_ = 0;
    pre_cpu_stat_.sys_stat_ = 0;
    pre_cpu_stat_.io_stat_ = 0;
    pre_cpu_stat_.idle_stat_ = 0;
    pre_cpu_stat_.total_stat_ = 0;
    fetched_all_process_ = false;
    pre_fetched_state_ = false;
    cpu_processor_num_ = 0;
  }
  
  Status BaseCpu::GetTotalCpuTime(uint64_t *total_stat) {
    std::ifstream file("/proc/stat");
    if (!file.is_open()) {
      MS_LOG(INFO) << "Open CPU file failed when collect CPU information";
      return Status::OK();
    }
    std::string str;
    getline(file, str);
    uint64_t user = 0, sys = 0, idle = 0, iowait = 0, nice = 0, irq = 0, softirq = 0;
    if (sscanf_s(str.c_str(), "%*s %lu %lu %lu %lu %lu %lu %lu", &user, &nice, &sys, &idle, &iowait, &irq, &softirq) ==
        EOF) {
      file.close();
      return Status(StatusCode::kMDUnexpectedError, "Get device CPU failed.");
    }
    file.close();
    *total_stat = user + nice + sys + idle + iowait + irq + softirq;
  
    return Status::OK();
  }
  
  Status DeviceCpu::ParseCpuInfo(const std::string &str) {
    CpuStat cpu_stat;
    uint64_t nice = 0;
    uint64_t irq = 0;
    uint64_t softirq = 0;
    if (sscanf_s(str.c_str(), "%*s %lu %lu %lu %lu %lu %lu %lu", &cpu_stat.user_stat_, &nice, &cpu_stat.sys_stat_,
                 &cpu_stat.idle_stat_, &cpu_stat.io_stat_, &irq, &softirq) == EOF) {
      return Status(StatusCode::kMDUnexpectedError, "Get device CPU failed.");
    }
  
    cpu_stat.total_stat_ =
      cpu_stat.user_stat_ + nice + cpu_stat.sys_stat_ + cpu_stat.idle_stat_ + cpu_stat.io_stat_ + irq + softirq;
    // Calculate the utilization from the second sampling
    if (!first_collect_) {
      CpuUtil info;
      info.user_utilization_ = round((cpu_stat.user_stat_ - pre_cpu_stat_.user_stat_) * 1.0 /
                                     (cpu_stat.total_stat_ - pre_cpu_stat_.total_stat_) * 100);
      info.sys_utilization_ = round((cpu_stat.sys_stat_ - pre_cpu_stat_.sys_stat_) * 1.0 /
                                    (cpu_stat.total_stat_ - pre_cpu_stat_.total_stat_) * 100);
      info.io_utilization_ = round((cpu_stat.io_stat_ - pre_cpu_stat_.io_stat_) * 1.0 /
                                   (cpu_stat.total_stat_ - pre_cpu_stat_.total_stat_) * 100);
      info.idle_utilization_ = round((cpu_stat.idle_stat_ - pre_cpu_stat_.idle_stat_) * 1.0 /
                                     (cpu_stat.total_stat_ - pre_cpu_stat_.total_stat_) * 100);
      cpu_util_.emplace_back(info);
    }
    pre_cpu_stat_.user_stat_ = cpu_stat.user_stat_;
    pre_cpu_stat_.sys_stat_ = cpu_stat.sys_stat_;
    pre_cpu_stat_.io_stat_ = cpu_stat.io_stat_;
    pre_cpu_stat_.idle_stat_ = cpu_stat.idle_stat_;
    pre_cpu_stat_.total_stat_ = cpu_stat.total_stat_;
  
    return Status::OK();
  }
  
  Status DeviceCpu::ParseCtxt(const std::string &str) {
    uint64_t ctxt;
    if (sscanf_s(str.c_str(), "%*s %lu", &ctxt) == EOF) {
      return Status(StatusCode::kMDUnexpectedError, "Get context switch count failed.");
    }
    // Calculate the utilization from the second sampling
    if (!first_collect_) {
      context_switch_count_.push_back(ctxt - pre_context_switch_count_);
    }
    pre_context_switch_count_ = ctxt;
    return Status::OK();
  }
  
  Status DeviceCpu::ParseRunningProcess(const std::string &str) {
    uint32_t running_process;
    if (sscanf_s(str.c_str(), "%*s %ud", &running_process) == EOF) {
      return Status(StatusCode::kMDUnexpectedError, "Get context switch count failed.");
    }
    // Drop the first value in order to collect same amount of CPU utilization
    if (!first_collect_) {
      running_process_.push_back(running_process);
    }
  
    return Status::OK();
  }
  
  Status DeviceCpu::Collect(const ExecutionTree *tree) {
    std::ifstream file("/proc/stat");
    if (!file.is_open()) {
      MS_LOG(INFO) << "Open CPU file failed when collect CPU information";
      return Status::OK();
    }
    bool first_line = true;
    std::string line;
    while (getline(file, line)) {
      if (first_line) {
        first_line = false;
        RETURN_IF_NOT_OK(ParseCpuInfo(line));
      }
      if (line.find("ctxt") != std::string::npos) {
        RETURN_IF_NOT_OK(ParseCtxt(line));
      }
      if (line.find("procs_running") != std::string::npos) {
        RETURN_IF_NOT_OK(ParseRunningProcess(line));
      }
    }
    file.close();
  
    first_collect_ = false;
    return Status::OK();
  }
  Status DeviceCpu::Analyze(std::string *name, double *utilization, std::string *extra_message) {
    RETURN_UNEXPECTED_IF_NULL(name);
    name->clear();
    name->append("device_info");
    int total_samples = cpu_util_.size();
    int sum = 0;
    // Only analyze the middle half of the samples
    // Starting and ending may be impacted by startup or ending pipeline activities
    int start_analyze = total_samples / 4;
    int end_analyze = total_samples - start_analyze;
  
    for (int i = start_analyze; i < end_analyze; i++) {
      sum += cpu_util_[i].user_utilization_;
      sum += cpu_util_[i].sys_utilization_;
    }
  
    // Note device utilization is already in range of 0-1, so don't
    // need to divide by number of CPUS
    if ((end_analyze - start_analyze) > 0) {
      *utilization = sum / (end_analyze - start_analyze);
    }
    return Status::OK();
  }
  
  Status DeviceCpu::SaveToFile(const std::string &file_path) {
    Path path = Path(file_path);
    json output;
    if (path.Exists()) {
      MS_LOG(DEBUG) << file_path << " exists already";
      try {
        std::ifstream file(file_path);
        file >> output;
      } catch (const std::exception &err) {
        RETURN_STATUS_UNEXPECTED("Invalid file, failed to open json file: " + file_path +
                                 ", please delete it and try again!");
      }
    } else {
      output["sampling_interval"] = GlobalContext::config_manager()->monitor_sampling_interval();
    }
  
    std::vector<int8_t> user_util;
    std::transform(cpu_util_.begin(), cpu_util_.end(), std::back_inserter(user_util),
                   [&](const CpuUtil &info) { return info.user_utilization_; });
    std::vector<int8_t> sys_util;
    std::transform(cpu_util_.begin(), cpu_util_.end(), std::back_inserter(sys_util),
                   [&](const CpuUtil &info) { return info.sys_utilization_; });
    std::vector<int8_t> io_util;
    std::transform(cpu_util_.begin(), cpu_util_.end(), std::back_inserter(io_util),
                   [&](const CpuUtil &info) { return info.io_utilization_; });
    std::vector<int8_t> idle_util;
    std::transform(cpu_util_.begin(), cpu_util_.end(), std::back_inserter(idle_util),
                   [&](const CpuUtil &info) { return info.idle_utilization_; });
  
    output["device_info"] = {{"user_utilization", user_util},
                             {"sys_utilization", sys_util},
                             {"io_utilization", io_util},
                             {"idle_utilization", idle_util},
                             {"runable_processes", running_process_},
                             {"context_switch_count", context_switch_count_}};
  
    // Discard the content of the file when opening.
    std::ofstream os(file_path, std::ios::trunc);
    os << output;
    os.close();
  
    MS_LOG(INFO) << "Save device CPU success.";
    return Status::OK();
  }
  
  Status OperatorCpu::ParseCpuInfo(int32_t op_id, int64_t thread_id,
                                   std::unordered_map<int32_t, std::unordered_map<int64_t, CpuOpStat>> *op_stat) {
    RETURN_UNEXPECTED_IF_NULL(op_stat);
    pid_t pid = 0;
  #if defined(USING_LINUX)
    pid = syscall(SYS_getpid);
  #endif
    std::string stat_path = "/proc/" + std::to_string(pid) + "/task/" + std::to_string(thread_id) + "/stat";
  
    // Judge whether file exist first
    Path temp_path(stat_path);
    if (!temp_path.Exists()) {
      (*op_stat)[op_id][thread_id].user_stat_ = 0;
      (*op_stat)[op_id][thread_id].sys_stat_ = 0;
      return Status(StatusCode::kMDFileNotExist);
    }
  
    std::ifstream file(stat_path);
    if (!file.is_open()) {
      MS_LOG(INFO) << "Open CPU file failed when collect CPU information";
      return Status::OK();
    }
    std::string str;
    getline(file, str);
    uint64_t utime;
    uint64_t stime;
    if (sscanf_s(str.c_str(), "%*d %*s %*s %*lu %*lu %*lu %*lu %*lu %*lu %*lu %*lu %*lu %*lu %lu %lu", &utime, &stime) ==
        EOF) {
      file.close();
      return Status(StatusCode::kMDUnexpectedError, "Get device CPU failed.");
    }
    file.close();
    (*op_stat)[op_id][thread_id].user_stat_ = utime;
    (*op_stat)[op_id][thread_id].sys_stat_ = stime;
  
    return Status::OK();
  }
  
  Status OperatorCpu::Collect(const ExecutionTree *tree) {
    RETURN_UNEXPECTED_IF_NULL(tree);
    if (first_collect_) {
      for (auto iter = tree->begin(); iter != tree->end(); ++iter) {
        id_count_++;
        op_name_[iter->id()] = iter->NameWithID();
        op_parallel_workers_[iter->id()] = iter->NumWorkers();
      }
  #if defined(USING_LINUX)
      cpu_processor_num_ = get_nprocs_conf();
  #endif
    }
  
    // Obtain the op and thread mapping
    op_thread_.clear();
    List<Task> allTasks = tree->AllTasks()->GetTask();
    for (auto &task1 : allTasks) {
      int32_t op_id = task1.get_operator_id();
      op_thread_[op_id].emplace_back(task1.get_linux_id());
    }
  
    // add process id into op_thread
    if (!fetched_all_process_) {
      {
        py::gil_scoped_acquire gil_acquire;
        py::module ds = py::module::import("mindspore.dataset.engine.datasets");
        py::tuple process_info = ds.attr("_get_operator_process")();
        py::dict sub_process = py::reinterpret_borrow<py::dict>(process_info[0]);
        fetched_all_process_ = py::reinterpret_borrow<py::bool_>(process_info[1]);
        // parse dict value
        op_process_ = toIntMap(sub_process);
        BaseCpu::op_process_shared_ = op_process_;
        BaseCpu::fetched_all_process_shared_ = fetched_all_process_;
      }
  
      // judge whether there is device_que operator, if so operator id may need increase by one, temp use directly
      for (auto item : op_process_) {
        if (!item.second.empty()) {
          if (op_thread_.find(item.first) != op_thread_.end()) {
            op_thread_[item.first].insert(op_thread_[item.first].end(), item.second.begin(), item.second.end());
          } else {
            op_thread_[item.first] = item.second;
          }
        }
      }
    }
  
    uint64_t total_stat_;
    RETURN_IF_NOT_OK(GetTotalCpuTime(&total_stat_));
    std::vector<CpuOpUtil> cpu_step_util_;
    std::unordered_map<int32_t, std::unordered_map<int64_t, CpuOpStat>> op_stat_;
  
    if (!first_collect_) {
      // obtain all the op id in current tasks
      std::vector<int32_t> total_op_id;
      (void)std::transform(op_thread_.begin(), op_thread_.end(), std::back_inserter(total_op_id),
                           [](const auto &iter) { return iter.first; });
  
      // iter all the op, and obtain the CPU utilization of each operator
      for (auto op_id = -1; op_id < id_count_; op_id++) {
        float user_util = 0, sys_util = 0;
        auto iter = std::find(total_op_id.begin(), total_op_id.end(), op_id);
        if (iter != total_op_id.end()) {
          for (auto thread_id : op_thread_[op_id]) {
            if (ParseCpuInfo(op_id, thread_id, &op_stat_) == Status::OK()) {
              user_util += (op_stat_[op_id][thread_id].user_stat_ - pre_op_stat_[op_id][thread_id].user_stat_) * 1.0 /
                           (total_stat_ - pre_total_stat_) * 100;
              sys_util += (op_stat_[op_id][thread_id].sys_stat_ - pre_op_stat_[op_id][thread_id].sys_stat_) * 1.0 /
                          (total_stat_ - pre_total_stat_) * 100;
            }
          }
        }
        CpuOpUtil info;
        info.op_id_ = op_id;
        info.sys_utilization_ = sys_util;
        info.user_utilization_ = user_util;
        cpu_step_util_.emplace_back(info);
      }
      cpu_op_util_.emplace_back(cpu_step_util_);
    } else {
      // mainly obtain the init CPU execute time in first collect
      for (const auto &iter : op_thread_) {
        int32_t op_id = iter.first;
        for (auto thread_id_ : iter.second) {
          // ParseCpuInfo may execute failed for cpu data not ready, but we still get next thread cpu info
          (void)ParseCpuInfo(op_id, thread_id_, &op_stat_);
        }
      }
    }
  
    // copy current op_stat into pre_op_stat
    pre_op_stat_ = op_stat_;
    pre_total_stat_ = total_stat_;
  
    first_collect_ = false;
    return Status::OK();
  }
  
  Status OperatorCpu::Analyze(std::string *name, double *utilization, std::string *extra_message) {
    RETURN_UNEXPECTED_IF_NULL(name);
    RETURN_UNEXPECTED_IF_NULL(extra_message);
    int total_samples = cpu_op_util_.size();
  
    // Only analyze the middle half of the samples
    // Starting and ending may be impacted by startup or ending pipeline activities
    constexpr int64_t sample_sections = 4;
    int64 start_analyze = total_samples / sample_sections;
    int64 end_analyze = total_samples - start_analyze;
    double op_util = 0;
    *utilization = 0;
  
    // start loop from 0 was as don't want to analyze op -1
    for (auto op_id = 0; op_id < id_count_; op_id++) {
      int64 sum = 0;
      int64 index = op_id + 1;
      for (int i = start_analyze; i < end_analyze; i++) {
        sum += cpu_op_util_[i][index].user_utilization_;
        sum += cpu_op_util_[i][index].sys_utilization_;
      }
      if ((end_analyze - start_analyze) > 0) {
        op_util = 1.0 * sum * cpu_processor_num_ / (op_parallel_workers_[op_id] * (end_analyze - start_analyze));
      }
      if (op_util > *utilization) {
        *utilization = op_util;
        name->clear();
        (void)name->append(op_name_[op_id]);
      }
      (void)extra_message->append(op_name_[op_id] + " utilization per thread: " + std::to_string(op_util) + "% (" +
                                  std::to_string(op_parallel_workers_[op_id]) + " parallel_workers); ");
    }
    return Status::OK();
  }
  
  Status OperatorCpu::SaveToFile(const std::string &file_path) {
    Path path = Path(file_path);
    json output;
    if (path.Exists()) {
      MS_LOG(DEBUG) << file_path << "already exist.";
      try {
        std::ifstream file(file_path);
        file >> output;
      } catch (const std::exception &err) {
        RETURN_STATUS_UNEXPECTED("Invalid file, failed to open json file: " + file_path +
                                 ", please delete it and try again!");
      }
    }
  
    uint8_t index = 0;
    json OpWriter;
    for (auto op_id = -1; op_id < id_count_; op_id++) {
      std::vector<uint16_t> user_util;
      std::vector<uint16_t> sys_util;
      std::transform(
        cpu_op_util_.begin(), cpu_op_util_.end(), std::back_inserter(user_util),
        [&](const std::vector<CpuOpUtil> &info) { return int16_t(info[index].user_utilization_ * cpu_processor_num_); });
      std::transform(
        cpu_op_util_.begin(), cpu_op_util_.end(), std::back_inserter(sys_util),
        [&](const std::vector<CpuOpUtil> &info) { return int16_t(info[index].sys_utilization_ * cpu_processor_num_); });
  
      json per_op_info = {{"metrics", {{"user_utilization", user_util}, {"sys_utilization", sys_util}}},
                          {"op_id", op_id}};
      OpWriter.emplace_back(per_op_info);
      index++;
    }
    output["op_info"] = OpWriter;
  
    // Discard the content of the file when opening.
    std::ofstream os(file_path, std::ios::trunc);
    os << output;
    os.close();
  
    MS_LOG(INFO) << "Save device CPU success.";
    return Status::OK();
  }
  
  Status ProcessCpu::ParseCpuInfo() {
    uint64_t total_stat_;
    RETURN_IF_NOT_OK(GetTotalCpuTime(&total_stat_));
  
    if (!pre_fetched_state_) {
      process_id_.clear();
      pid_t main_pid = 0;
  #if defined(USING_LINUX)
      main_pid = syscall(SYS_getpid);
  #endif
      process_id_.emplace_back(main_pid);
      op_process_ = BaseCpu::op_process_shared_;
      fetched_all_process_ = BaseCpu::fetched_all_process_shared_;
      for (const auto &item : op_process_) {
        for (const auto &id : item.second) {
          process_id_.emplace_back(id);
        }
      }
    }
  
    float user_util = 0, sys_util = 0;
    for (const auto &pid : process_id_) {
      std::string stat_path = "/proc/" + std::to_string(pid) + "/stat";
  
      std::ifstream file(stat_path);
      if (!file.is_open()) {
        MS_LOG(INFO) << "Open CPU file failed when collect CPU information";
        continue;
      }
      std::string str;
      getline(file, str);
      uint64_t user = 0, sys = 0;
      if (sscanf_s(str.c_str(), "%*d %*s %*s %*lu %*lu %*lu %*lu %*lu %*lu %*lu %*lu %*lu %*lu %lu %lu", &user, &sys) ==
          EOF) {
        file.close();
        return Status(StatusCode::kMDUnexpectedError, "Get device CPU failed.");
      }
      file.close();
  
      // Calculate the utilization from the second sampling
      if (!first_collect_ && (pre_process_stat_.find(pid) != pre_process_stat_.end())) {
        user_util += (user - pre_process_stat_[pid].user_stat_) * 1.0 / (total_stat_ - pre_total_stat_) * 100;
        sys_util += (sys - pre_process_stat_[pid].sys_stat_) * 1.0 / (total_stat_ - pre_total_stat_) * 100;
      }
      pre_process_stat_[pid].user_stat_ = user;
      pre_process_stat_[pid].sys_stat_ = sys;
    }
    if (!first_collect_) {
      CpuProcessUtil info;
      info.user_utilization_ = user_util;
      info.sys_utilization_ = sys_util;
      process_util_.emplace_back(info);
    }
    pre_total_stat_ = total_stat_;
    first_collect_ = false;
    pre_fetched_state_ = fetched_all_process_;
    return Status::OK();
  }
  
  Status ProcessCpu::Collect(const ExecutionTree *tree) {
    RETURN_UNEXPECTED_IF_NULL(tree);
    if (first_collect_) {
  #if defined(USING_LINUX)
      cpu_processor_num_ = get_nprocs_conf();
  #endif
    }
    RETURN_IF_NOT_OK(ParseCpuInfo());
  
    return Status::OK();
  }
  
  Status ProcessCpu::Analyze(std::string *name, double *utilization, std::string *extra_message) {
    RETURN_UNEXPECTED_IF_NULL(name);
    RETURN_UNEXPECTED_IF_NULL(utilization);
    RETURN_UNEXPECTED_IF_NULL(extra_message);
    name->clear();
    name->append("process_info");
    int total_samples = process_util_.size();
    int64 sum = 0;
    // Only analyze the middle half of the samples
    // Starting and ending may be impacted by startup or ending pipeline activities
    constexpr int64_t sample_sections = 4;
    int64 start_analyze = total_samples / sample_sections;
    int64 end_analyze = total_samples - start_analyze;
  
    for (int i = start_analyze; i < end_analyze; i++) {
      sum += process_util_[i].user_utilization_;
      sum += process_util_[i].sys_utilization_;
    }
  
    if ((end_analyze - start_analyze) > 0) {
      *utilization = sum / (end_analyze - start_analyze);
    }
    return Status::OK();
  }
  
  Status ProcessCpu::SaveToFile(const std::string &file_path) {
    Path path = Path(file_path);
    json output;
    if (path.Exists()) {
      MS_LOG(DEBUG) << file_path << "already exist.";
      try {
        std::ifstream file(file_path);
        file >> output;
      } catch (const std::exception &err) {
        RETURN_STATUS_UNEXPECTED("Invalid file, failed to open json file: " + file_path +
                                 ", please delete it and try again!");
      }
    } else {
      output["sampling_interval"] = GlobalContext::config_manager()->monitor_sampling_interval();
    }
  
    std::vector<int16_t> user_util;
    std::transform(process_util_.begin(), process_util_.end(), std::back_inserter(user_util),
                   [&](const CpuProcessUtil &info) { return uint16_t(info.user_utilization_ * cpu_processor_num_); });
    std::vector<int16_t> sys_util;
    std::transform(process_util_.begin(), process_util_.end(), std::back_inserter(sys_util),
                   [&](const CpuProcessUtil &info) { return uint16_t(info.sys_utilization_ * cpu_processor_num_); });
  
    output["process_info"] = {{"user_utilization", user_util}, {"sys_utilization", sys_util}};
    output["cpu_processor_num"] = cpu_processor_num_;
  
    // Discard the content of the file when opening.
    std::ofstream os(file_path, std::ios::trunc);
    os << output;
    os.close();
  
    MS_LOG(INFO) << "Save process CPU success.";
    return Status::OK();
  }
  
  Status CpuSampling::CollectTimeStamp() {
    time_stamp_.emplace_back(ProfilingTime::GetCurMilliSecond());
    return Status::OK();
  }
  
  // Sample action
  Status CpuSampling::Sample() {
    // Collect cpu information
    for (auto cpu : cpu_) {
      RETURN_IF_NOT_OK(cpu->Collect(this->tree_));
    }
  
    // Collect time stamp
    RETURN_IF_NOT_OK(CollectTimeStamp());
    return Status::OK();
  }
  
  Status CpuSampling::SaveTimeStampToFile() {
    // Save time stamp to json file
    // If the file is already exist, simply add the data to corresponding field.
    Path path = Path(file_path_);
    json output;
    if (path.Exists()) {
      try {
        std::ifstream file(file_path_);
        file >> output;
      } catch (const std::exception &err) {
        RETURN_STATUS_UNEXPECTED("Invalid file, failed to open json file: " + file_path_ +
                                 ", please delete it and try again!");
      }
    }
    output["time_stamp"] = time_stamp_;
    std::ofstream os(file_path_, std::ios::trunc);
    os << output;
    os.close();
  
    return Status::OK();
  }
  
  Status CpuSampling::SaveSamplingItervalToFile() {
    // If the file is already exist, simply add the data to corresponding field.
    Path path = Path(file_path_);
    json output;
    if (path.Exists()) {
      try {
        std::ifstream file(file_path_);
        file >> output;
      } catch (const std::exception &err) {
        RETURN_STATUS_UNEXPECTED("Invalid file, failed to open json file: " + file_path_ +
                                 ", please delete it and try again!");
      }
    }
    output["sampling_interval"] = GlobalContext::config_manager()->monitor_sampling_interval();
    std::ofstream os(file_path_, std::ios::trunc);
    os << output;
    os.close();
  
    return Status::OK();
  }
  
  // Analyze profiling data and output warning messages
  Status CpuSampling::Analyze() {
    std::string name;
    double utilization = 0;
    constexpr double total_cpu_thold = 90;
    constexpr double op_cpu_thold = 80;
    // Keep track of specific information returned by differentn CPU sampling types
    double total_utilization = 0;
    double max_op_utilization = 0;
    std::string max_op_name;
    std::string detailed_op_cpu_message;
  
    // Save cpu information to json file
    for (auto cpu : cpu_) {
      std::string extra_message;
      RETURN_IF_NOT_OK(cpu->Analyze(&name, &utilization, &extra_message));
      if (name == "device_info") {
        total_utilization = utilization;
      } else if (name != "process_info") {
        max_op_utilization = utilization;
        max_op_name = name;
        detailed_op_cpu_message = extra_message;
      }
    }
    if ((total_utilization < total_cpu_thold) && (max_op_utilization > op_cpu_thold)) {
      MS_LOG(WARNING) << "Operator " << max_op_name << " is using " << max_op_utilization << "% CPU per thread.  "
                      << "This operator may benefit from increasing num_parallel_workers."
                      << "Full Operator CPU utiliization for all operators: " << detailed_op_cpu_message << std::endl;
    }
    return Status::OK();
  }
  
  // Save profiling data to file
  Status CpuSampling::SaveToFile() {
    // Save time stamp to json file
    RETURN_IF_NOT_OK(SaveTimeStampToFile());
  
    // Save time stamp to json file
    RETURN_IF_NOT_OK(SaveSamplingItervalToFile());
  
    // Save cpu information to json file
    for (auto cpu : cpu_) {
      RETURN_IF_NOT_OK(cpu->SaveToFile(file_path_));
    }
  
    return Status::OK();
  }
  
  Status CpuSampling::Init(const std::string &dir_path, const std::string &device_id) {
    file_path_ = (Path(dir_path) / Path("minddata_cpu_utilization_" + device_id + ".json")).ToString();
    std::shared_ptr<DeviceCpu> device_cpu = std::make_shared<DeviceCpu>();
    std::shared_ptr<OperatorCpu> operator_cpu = std::make_shared<OperatorCpu>();
    std::shared_ptr<ProcessCpu> process_cpu = std::make_shared<ProcessCpu>();
    cpu_.push_back(device_cpu);
    cpu_.push_back(operator_cpu);
    cpu_.push_back(process_cpu);
    return Status::OK();
  }
  
  Status CpuSampling::ChangeFileMode() {
    if (chmod(common::SafeCStr(file_path_), S_IRUSR | S_IWUSR) == -1) {
      std::string err_str = "Change file mode failed," + file_path_;
      return Status(StatusCode::kMDUnexpectedError, err_str);
    }
    return Status::OK();
  }
  }  // namespace dataset
  }  // namespace mindspore
  