!5596 add async profling feature

Merge pull request !5596 from gukecai/profiling

CMakeLists.txt

@@ -48,6 +48,7 @@ endif()
 
 include(${CMAKE_SOURCE_DIR}/cmake/mind_expression.cmake)
 include_directories(${CMAKE_CURRENT_SOURCE_DIR})
+include_directories(${CMAKE_CURRENT_SOURCE_DIR}/third_party/securec/include)
 include_directories(${CMAKE_CURRENT_SOURCE_DIR}/third_party/flatbuffers/include)
 include_directories(${CMAKE_CURRENT_SOURCE_DIR}/third_party/flatbuffers/include/flatbuffers)
 

mindspore/ccsrc/profiler/CMakeLists.txt

@@ -2,4 +2,10 @@ if (ENABLE_GPU)
     file(GLOB_RECURSE PROFILER_SRC_LIST RELATIVE ${CMAKE_CURRENT_SOURCE_DIR} "device/gpu/*.cc")
     set_property(SOURCE ${PROFILER_SRC_LIST} PROPERTY COMPILE_DEFINITIONS SUBMODULE_ID=mindspore::SubModuleId::SM_PROFILER)
     add_library(_mindspore_profiler_obj OBJECT ${PROFILER_SRC_LIST})
+endif ()
+
+if (ENABLE_D)
+    file(GLOB_RECURSE PROFILER_SRC_LIST RELATIVE ${CMAKE_CURRENT_SOURCE_DIR} "device/ascend/*.cc")
+    set_property(SOURCE ${PROFILER_SRC_LIST} PROPERTY COMPILE_DEFINITIONS SUBMODULE_ID=mindspore::SubModuleId::SM_PROFILER)
+    add_library(_mindspore_profiler_obj OBJECT ${PROFILER_SRC_LIST})
 endif ()

mindspore/ccsrc/profiler/device/ascend/ascend_profiling.cc

@@ -0,0 +1,83 @@
+/**
+ * Copyright 2019-2020 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 "profiler/device/ascend/ascend_profiling.h"
+#include <cstdarg>
+#include <iomanip>
+#include <utility>
+#include "utils/log_adapter.h"
+#include "./securec.h"
+namespace mindspore {
+namespace profiler {
+namespace ascend {
+const int kMaxEvents = 10000;
+const int kEventDescMax = 256;
+const int kMaxEventTypes = 8;
+const int kIndent = 8;
+
+AscendProfiler::AscendProfiler() : counter_(0) { Reset(); }
+
+void AscendProfiler::RecordEvent(EventType event_type, const char *fmt, ...) {
+  va_list args;
+  va_start(args, fmt);
+
+  char buf[kEventDescMax];
+  if (vsnprintf_s(buf, kEventDescMax, kEventDescMax - 1, fmt, args) == -1) {
+    MS_LOG(ERROR) << "format failed:" << fmt;
+    va_end(args);
+    return;
+  }
+
+  va_end(args);
+  std::string event = buf;
+  auto index = counter_++;
+  auto &evt = events_[index];
+  evt.timestamp = std::chrono::system_clock::now();
+  evt.desc = std::move(event);
+  evt.event_type = event_type;
+}
+
+void AscendProfiler::Dump(std::ostream &output_stream) {
+  MS_LOG(INFO) << "start dump async profiling info";
+  if (events_.empty()) {
+    return;
+  }
+
+  auto first_evt = events_[0];
+  auto start = first_evt.timestamp;
+  std::vector<decltype(start)> prev_timestamps;
+  prev_timestamps.resize(kMaxEventTypes, start);
+
+  for (int i = 0; i < counter_; ++i) {
+    auto &evt = events_[i];
+    auto elapsed = std::chrono::duration_cast<std::chrono::microseconds>(evt.timestamp - start).count();
+    auto &prev_ts = prev_timestamps[evt.event_type];
+    auto cost = std::chrono::duration_cast<std::chrono::microseconds>(evt.timestamp - prev_ts).count();
+    prev_ts = evt.timestamp;
+    output_stream << std::setw(kIndent) << elapsed << "\t\t" << cost << "\t\t" << evt.desc << std::endl;
+  }
+
+  events_.clear();
+  MS_LOG(INFO) << "end";
+}
+
+void AscendProfiler::Reset() {
+  counter_ = 0;
+  events_.clear();
+  events_.resize(kMaxEvents);
+}
+}  // namespace ascend
+}  // namespace profiler
+}  // namespace mindspore

mindspore/ccsrc/profiler/device/ascend/ascend_profiling.h

@@ -0,0 +1,60 @@
+/**
+ * Copyright 2020 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.
+ */
+
+#ifndef MINDSPORE_ASCEND_PROFILING_H_
+#define MINDSPORE_ASCEND_PROFILING_H_
+#include <atomic>
+#include <chrono>
+#include <mutex>
+#include <ostream>
+#include <string>
+#include <vector>
+using std::string;
+
+namespace mindspore {
+namespace profiler {
+namespace ascend {
+enum EventType { kGeneral = 0, kCompiler, kExecution, kCallback };
+struct Event {
+  std::chrono::system_clock::time_point timestamp;
+  EventType event_type;
+  std::string desc;
+};
+class AscendProfiler {
+ public:
+  AscendProfiler();
+  ~AscendProfiler() = default;
+
+  static AscendProfiler &GetInstance() {
+    static AscendProfiler instance;
+    return instance;
+  }
+
+  void RecordEvent(EventType event_type, const char *fmt, ...);
+
+  void Reset();
+
+  void Dump(std::ostream &os);
+
+ private:
+  std::vector<Event> events_;
+  std::atomic_int counter_;
+};
+
+}  // namespace ascend
+}  // namespace profiler
+}  // namespace mindspore
+#endif  // MINDSPORE_ASCEND_PROFILING_H_

mindspore/ccsrc/profiler/device/ascend/blocking_queue.h

@@ -0,0 +1,147 @@
+/**
+ * Copyright 2019-2020 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.
+ */
+
+#ifndef MINDSPORE_BLOCKING_QUEUE_H_
+#define MINDSPORE_BLOCKING_QUEUE_H_
+
+#include <stdint.h>
+#include <condition_variable>
+#include <list>
+#include <mutex>
+#include <utility>
+
+static const int kDefaultMaxQueueSize = 2048;
+namespace mindspore {
+namespace profiler {
+namespace ascend {
+template <typename T>
+class BlockingQueue {
+ public:
+  explicit BlockingQueue(uint32_t max_size = kDefaultMaxQueueSize) : max_size_(max_size), is_stoped_(false) {}
+
+  ~BlockingQueue() {}
+
+  bool Pop(T *item) {
+    std::unique_lock<std::mutex> lock(mutex_);
+
+    while (queue_.empty() && !is_stoped_) {
+      empty_cond_.wait(lock);
+    }
+
+    if (is_stoped_) {
+      return false;
+    }
+
+    *item = std::move(queue_.front());
+    queue_.pop_front();
+
+    full_cond_.notify_one();
+
+    return true;
+  }
+
+  bool Push(const T &item, bool is_wait = true) {
+    std::unique_lock<std::mutex> lock(mutex_);
+
+    while (queue_.size() >= max_size_ && !is_stoped_) {
+      if (!is_wait) {
+        return false;
+      }
+      full_cond_.wait(lock);
+    }
+
+    if (is_stoped_) {
+      return false;
+    }
+
+    queue_.push_back(item);
+
+    empty_cond_.notify_one();
+
+    return true;
+  }
+
+  bool Push(T &&item, bool is_wait = true) {
+    std::unique_lock<std::mutex> lock(mutex_);
+
+    while (queue_.size() >= max_size_ && !is_stoped_) {
+      if (!is_wait) {
+        return false;
+      }
+      full_cond_.wait(lock);
+    }
+
+    if (is_stoped_) {
+      return false;
+    }
+
+    queue_.emplace_back(std::move(item));
+
+    empty_cond_.notify_one();
+
+    return true;
+  }
+
+  void Stop() {
+    {
+      std::unique_lock<std::mutex> lock(mutex_);
+      is_stoped_ = true;
+    }
+
+    full_cond_.notify_all();
+    empty_cond_.notify_all();
+  }
+
+  void Restart() {
+    std::unique_lock<std::mutex> lock(mutex_);
+    is_stoped_ = false;
+  }
+
+  // if the queue is stoped ,need call this function to release the unprocessed items
+  std::list<T> GetRemainItems() {
+    std::unique_lock<std::mutex> lock(mutex_);
+
+    if (!is_stoped_) {
+      return std::list<T>();
+    }
+
+    return queue_;
+  }
+
+  bool IsFull() {
+    std::unique_lock<std::mutex> lock(mutex_);
+    return queue_.size() >= max_size_;
+  }
+
+  void Clear() {
+    std::unique_lock<std::mutex> lock(mutex_);
+    queue_.clear();
+  }
+
+ private:
+  std::list<T> queue_;
+  std::mutex mutex_;
+  std::condition_variable empty_cond_;
+  std::condition_variable full_cond_;
+  uint32_t max_size_;
+
+  bool is_stoped_;
+};
+}  // namespace ascend
+}  // namespace profiler
+}  // namespace mindspore
+
+#endif  // MINDSPORE_BLOCKING_QUEUE_H_

mindspore/ccsrc/profiler/device/ascend/profiling_context.cc

@@ -0,0 +1,23 @@
+/**
+ * Copyright 2019-2020 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 "profiler/device/ascend/profiling_context.h"
+
+namespace mindspore {
+namespace profiler {
+namespace ascend {}
+}  // namespace profiler
+}  // namespace mindspore

mindspore/ccsrc/profiler/device/ascend/profiling_context.h

@@ -0,0 +1,60 @@
+/**
+ * Copyright 2019-2020 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.
+ */
+
+#ifndef MINDSPORE_PROFILING_CONTEXT_H_
+#define MINDSPORE_PROFILING_CONTEXT_H_
+
+#include <sys/syscall.h>
+#include <unistd.h>
+#include <atomic>
+#include <cstdint>
+#include <unordered_map>
+#include "profiler/device/ascend/ascend_profiling.h"
+namespace mindspore {
+namespace profiler {
+namespace ascend {
+
+inline pid_t GetTid() {
+  thread_local static pid_t tid = syscall(__NR_gettid);
+  return tid;
+}
+
+#define RECORD_PROFILING_EVENT(profiler, evt_type, fmt, category, node_name, ...)                                \
+  do {                                                                                                           \
+    if (profiler != nullptr) {                                                                                   \
+      if (node_name != nullptr) {                                                                                \
+        profiler->RecordEvent(evt_type, "tid:%lu [%s] [%s] " fmt, GetTid(), node_name, category, ##__VA_ARGS__); \
+      } else {                                                                                                   \
+        profiler->RecordEvent(evt_type, "tid:%lu [%s] " fmt, GetTid(), category, ##__VA_ARGS__);                 \
+      }                                                                                                          \
+    }                                                                                                            \
+  } while (0)
+
+#define RECORD_MODEL_EXECUTION_EVENT(profiler, fmt, ...) \
+  RECORD_PROFILING_EVENT((profiler), kGeneral, fmt, "ModelExecutor", nullptr, ##__VA_ARGS__)
+
+#define RECORD_COMPILE_EVENT(profiler, name, fmt, ...) \
+  RECORD_PROFILING_EVENT((profiler), kCompiler, fmt, "Compilation", name, ##__VA_ARGS__)
+
+#define RECORD_EXECUTION_EVENT(profiler, name, fmt, ...) \
+  RECORD_PROFILING_EVENT((profiler), kExecution, fmt, "Execution", name, ##__VA_ARGS__)
+
+#define RECORD_CALLBACK_EVENT(profiler, name, fmt, ...) \
+  RECORD_PROFILING_EVENT((profiler), kCallback, fmt, "Callback", name, ##__VA_ARGS__)
+}  // namespace ascend
+}  // namespace profiler
+}  // namespace mindspore
+#endif  // MINDSPORE_PROFILING_CONTEXT_H_

mindspore/ccsrc/profiler/device/ascend/rt_callback_manager.cc

@@ -0,0 +1,130 @@
+/**
+ * Copyright 2019-2020 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 "profiler/device/ascend/rt_callback_manager.h"
+#include "utils/log_adapter.h"
+#include "runtime/event.h"
+
+namespace mindspore {
+namespace profiler {
+namespace ascend {
+
+CallbackManager::CallbackManager(rtStream_t stream) : stream_(stream) {}
+
+Status CallbackManager::Init() {
+  MS_LOG(INFO) << "CallbackManager init, Start to async process event";
+  ret_future_ = std::async([&] { return CallbackProcess(); });
+  if (!ret_future_.valid()) {
+    MS_LOG(ERROR) << "Failed to init callback manager.";
+    return kFail;
+  }
+
+  return kSuccess;
+}
+
+Status CallbackManager::CallbackProcess() {
+  std::pair<rtEvent_t, std::pair<rtCallback_t, void *>> entry;
+  while (true) {
+    if (!callback_queue_.Pop(&entry)) {
+      MS_LOG(INFO) << "CallbackManager stopped";
+      return kFail;
+    }
+
+    auto event = entry.first;
+    if (event == nullptr) {
+      return kSuccess;
+    }
+
+    auto rt_err = rtEventSynchronize(event);
+    if (rt_err != RT_ERROR_NONE) {
+      MS_LOG(ERROR) << "rtEventSynchronize failed. ret:" << rt_err;
+      auto ret = rtEventDestroy(event);
+      if (ret != RT_ERROR_NONE) {
+        MS_LOG(ERROR) << "rtEventDestroy failed";
+      }
+      return kFail;
+    }
+
+    auto ret = rtEventDestroy(event);
+    if (ret != RT_ERROR_NONE) {
+      MS_LOG(ERROR) << "rtEventDestroy failed";
+    }
+
+    auto cb_func = entry.second.first;
+    auto cb_args = entry.second.second;
+    cb_func(cb_args);
+  }
+}
+
+Status CallbackManager::Destroy() {
+  MS_LOG(INFO) << "To destroy callback manager.";
+  if (!ret_future_.valid()) {
+    MS_LOG(INFO) << "CallbackManager not initialized.";
+    return kSuccess;
+  }
+
+  std::pair<rtEvent_t, std::pair<rtCallback_t, void *>> eof_entry;
+  eof_entry.first = nullptr;
+  callback_queue_.Push(eof_entry);
+
+  auto ret = ret_future_.get();
+  MS_LOG(INFO) << "Callback manager ended. ret:" << ret;
+  return ret;
+}
+
+Status CallbackManager::RegisterCallback(rtCallback_t callback, void *user_data) {
+  MS_LOG(INFO) << "To register callback";
+  rtEvent_t event = nullptr;
+  auto ret = rtEventCreate(&event);
+  if (ret != RT_ERROR_NONE) {
+    MS_LOG(ERROR) << "Create event failed";
+    return kFail;
+  }
+
+  ret = rtEventRecord(event, stream_);
+  if (ret != RT_ERROR_NONE) {
+    MS_LOG(ERROR) << "Record event failed";
+    return kFail;
+  }
+  auto cb = std::pair<rtCallback_t, void *>(callback, user_data);
+  auto entry = std::pair<rtEvent_t, std::pair<rtCallback_t, void *>>(event, std::move(cb));
+  if (!callback_queue_.Push(entry)) {
+    return kFail;
+  }
+
+  MS_LOG(INFO) << "Registering callback successfully";
+  return kSuccess;
+}
+
+void CallbackManager::RtCallbackFunc(void *data) {
+  MS_LOG(INFO) << "To invoke callback function";
+  auto callback_func = reinterpret_cast<std::function<void()> *>(data);
+  (*callback_func)();
+  delete callback_func;
+}
+
+Status CallbackManager::RegisterCallback(const std::function<void()> &callback) {
+  auto func = std::unique_ptr<std::function<void()>>(new (std::nothrow) std::function<void()>(callback));
+  if (func == nullptr) {
+    MS_LOG(ERROR) << "callback is nullptr";
+    return kInvalidParam;
+  }
+  MS_LOG(INFO) << "Callback registered";
+  return RegisterCallback(RtCallbackFunc, func.release());
+}
+}  // namespace ascend
+}  // namespace profiler
+}  // namespace mindspore

mindspore/ccsrc/profiler/device/ascend/rt_callback_manager.h

@@ -0,0 +1,63 @@
+/**
+ * Copyright 2019-2020 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.
+ */
+
+#ifndef MINDSPORE_RT_CALLBACK_MANAGER_H_
+#define MINDSPORE_RT_CALLBACK_MANAGER_H_
+
+#include <condition_variable>
+#include <functional>
+#include <future>
+#include <memory>
+#include <utility>
+#include "profiler/device/ascend/blocking_queue.h"
+#include "runtime/base.h"
+
+namespace mindspore {
+namespace profiler {
+namespace ascend {
+using rtCallback_t = std::function<void(void *)>;
+enum Status { kSuccess = 0, kFail, kInvalidParam };
+class CallbackManager {
+ public:
+  static CallbackManager &GetInstance(rtStream_t stream) {
+    static CallbackManager instance(stream);
+    return instance;
+  }
+
+  explicit CallbackManager(rtStream_t stream);
+
+  ~CallbackManager() = default;
+
+  Status Init();
+
+  Status Destroy();
+
+  Status RegisterCallback(rtCallback_t callback, void *user_data);
+  Status RegisterCallback(const std::function<void()> &callback);
+
+ private:
+  Status CallbackProcess();
+  static void RtCallbackFunc(void *data);
+
+  BlockingQueue<std::pair<rtEvent_t, std::pair<rtCallback_t, void *>>> callback_queue_;
+  rtStream_t stream_;
+  std::future<Status> ret_future_;
+};
+}  // namespace ascend
+}  // namespace profiler
+}  // namespace mindspore
+
+#endif  // MINDSPORE_RT_CALLBACK_MANAGER_H_