Invert ConfigLoader -> profiler dependency (#271)

Summary:
Pull Request resolved: #271

Change direction of dependency from ConfigLoader -> profiler to profiler -> ConfigLoader. This way, the profiler is able to lazily initialize config loader and also move towards the pluggable profiler design.

Reviewed By: xw285cornell

Differential Revision: D28802798

fbshipit-source-id: bac82cf2b90716b9e19ab8aa401a26a3d06bbe31

Author: gdankel

libkineto/src/ActivityProfilerController.cpp

@@ -26,10 +26,15 @@ namespace KINETO_NAMESPACE {
 constexpr milliseconds kProfilerIntervalMsecs(1000);
 
 ActivityProfilerController::ActivityProfilerController(bool cpuOnly) {
-  profiler_ = std::make_unique<ActivityProfiler>(CuptiActivityInterface::singleton(), cpuOnly);
+  profiler_ = std::make_unique<ActivityProfiler>(
+      CuptiActivityInterface::singleton(), cpuOnly);
+  ConfigLoader::instance().addHandler(
+      ConfigLoader::ConfigKind::ActivityProfiler, this);
 }
 
 ActivityProfilerController::~ActivityProfilerController() {
+  ConfigLoader::instance().removeHandler(
+      ConfigLoader::ConfigKind::ActivityProfiler, this);
   if (profilerThread_) {
     // signaling termination of the profiler loop
     stopRunloop_ = true;
@@ -65,6 +70,17 @@ static std::unique_ptr<ActivityLogger> makeLogger(const Config& config) {
   return loggerFactory().makeLogger(config.activitiesLogUrl());
 }
 
+bool ActivityProfilerController::canAcceptConfig() {
+  return !profiler_->isActive();
+}
+
+void ActivityProfilerController::acceptConfig(const Config& config) {
+  VLOG(1) << "acceptConfig";
+  if (config.activityProfilerEnabled()) {
+    scheduleTrace(config);
+  }
+}
+
 void ActivityProfilerController::profilerLoop() {
   setThreadName("Kineto Activity Profiler");
   VLOG(0) << "Entering activity profiler loop";
@@ -107,6 +123,11 @@ void ActivityProfilerController::profilerLoop() {
 }
 
 void ActivityProfilerController::scheduleTrace(const Config& config) {
+  VLOG(1) << "scheduleTrace";
+  if (profiler_->isActive()) {
+    LOG(ERROR) << "Ignored request - profiler busy";
+    return;
+  }
   std::lock_guard<std::mutex> lock(asyncConfigLock_);
   asyncRequestConfig_ = config.clone();
   // start a profilerLoop() thread to handle request

libkineto/src/ActivityProfilerController.h

@@ -9,19 +9,21 @@
 
 #include <atomic>
 #include <memory>
+#include <mutex>
 #include <thread>
 
 #include "ActivityLoggerFactory.h"
 #include "ActivityProfiler.h"
 #include "ActivityProfilerInterface.h"
 #include "ActivityTraceInterface.h"
+#include "ConfigLoader.h"
 #include "CuptiActivityInterface.h"
 
 namespace KINETO_NAMESPACE {
 
 class Config;
 
-class ActivityProfilerController {
+class ActivityProfilerController : public ConfigLoader::ConfigHandler {
  public:
   explicit ActivityProfilerController(bool cpuOnly);
   ActivityProfilerController(const ActivityProfilerController&) = delete;
@@ -34,6 +36,9 @@ class ActivityProfilerController {
       const std::string& protocol,
       ActivityLoggerFactory::FactoryFunc factory);
 
+  bool canAcceptConfig() override;
+  void acceptConfig(const Config& config) override;
+
   void scheduleTrace(const Config& config);
 
   void prepareTrace(const Config& config);

libkineto/src/Config.cpp

@@ -156,6 +156,7 @@ Config::Config()
           kDefaultActivitiesExternalAPINetSizeThreshold),
       activitiesExternalAPIGpuOpCountThreshold_(
           kDefaultActivitiesExternalAPIGpuOpCountThreshold),
+      activitiesOnDemandTimestamp_(milliseconds(0)),
       requestTimestamp_(milliseconds(0)),
       enableSigUsr2_(true),
       enableIpcFabric_(false) {

libkineto/src/Config.h

@@ -45,7 +45,8 @@ class Config : public AbstractConfig {
   }
 
   bool activityProfilerEnabled() const {
-    return activityProfilerEnabled_;
+    return activityProfilerEnabled_ ||
+      activitiesOnDemandTimestamp_.time_since_epoch().count() > 0;
   }
 
   // Log activitiy trace to this file

libkineto/src/ConfigLoader.cpp

@@ -15,8 +15,6 @@
 #include <chrono>
 #include <fstream>
 
-#include "libkineto.h"
-#include "ActivityProfilerProxy.h"
 #include "DaemonConfigLoader.h"
 
 #include "Logger.h"
@@ -120,7 +118,7 @@ void ConfigLoader::setDaemonConfigLoaderFactory(
 }
 
 ConfigLoader& ConfigLoader::instance() {
-  static ConfigLoader config_loader(libkineto::api());
+  static ConfigLoader config_loader;
   return config_loader;
 }
 
@@ -130,9 +128,8 @@ std::string ConfigLoader::readOnDemandConfigFromDaemon(
   if (!daemonConfigLoader_) {
     return "";
   }
-  bool events =
-      now > onDemandEventProfilerConfig_->eventProfilerOnDemandEndTime();
-  bool activities = !libkinetoApi_.activityProfiler().isActive();
+  bool events = canHandlerAcceptConfig(ConfigKind::EventProfiler);
+  bool activities = canHandlerAcceptConfig(ConfigKind::ActivityProfiler);
   return daemonConfigLoader_->readOnDemandConfig(events, activities);
 }
 
@@ -144,10 +141,8 @@ int ConfigLoader::contextCountForGpu(uint32_t device) {
   return daemonConfigLoader_->gpuContextCount(device);
 }
 
-ConfigLoader::ConfigLoader(LibkinetoApi& api)
-    : libkinetoApi_(api),
-      onDemandEventProfilerConfig_(new Config()),
-      configUpdateIntervalSecs_(kConfigUpdateIntervalSecs),
+ConfigLoader::ConfigLoader()
+    : configUpdateIntervalSecs_(kConfigUpdateIntervalSecs),
       onDemandConfigUpdateIntervalSecs_(kOnDemandConfigUpdateIntervalSecs),
       stopFlag_(false),
       onDemandSignal_(false) {
@@ -211,51 +206,23 @@ void ConfigLoader::configureFromSignal(
   if (daemonConfigLoader_) {
     daemonConfigLoader_->setCommunicationFabric(config_.ipcFabricEnabled());
   }
-  if (eventProfilerRequest(config)) {
-    if (now > onDemandEventProfilerConfig_->eventProfilerOnDemandEndTime()) {
-      LOG(INFO) << "Starting on-demand event profiling from signal";
-      std::lock_guard<std::mutex> lock(configLock_);
-      onDemandEventProfilerConfig_ = config.clone();
-    } else {
-      LOG(ERROR) << "On-demand event profiler is busy";
-    }
-  }
-  // Initiate a trace by default, even when not specified in the config.
-  // Set trace duration and iterations to 0 to suppress.
-  config.updateActivityProfilerRequestReceivedTime();
-  try {
-    auto& profiler = dynamic_cast<ActivityProfilerProxy&>(
-        libkinetoApi_.activityProfiler());
-    profiler.scheduleTrace(config);
-  } catch (const std::exception& e) {
-    LOG(ERROR) << "Failed to schedule profiler request (busy?)";
-  }
+  notifyHandlers(config);
 }
 
 void ConfigLoader::configureFromDaemon(
     time_point<system_clock> now,
     Config& config) {
   const std::string config_str = readOnDemandConfigFromDaemon(now);
-  LOG_IF(INFO, !config_str.empty()) << "Received config from dyno:\n"
-                                    << config_str;
+  if (config_str.empty()) {
+    return;
+  }
+
+  LOG(INFO) << "Received config from dyno:\n" << config_str;
   config.parse(config_str);
   if (daemonConfigLoader_) {
     daemonConfigLoader_->setCommunicationFabric(config_.ipcFabricEnabled());
   }
-  if (eventProfilerRequest(config)) {
-    std::lock_guard<std::mutex> lock(configLock_);
-    onDemandEventProfilerConfig_ = config.clone();
-  }
-  if (config_.activityProfilerEnabled() &&
-      config.activityProfilerRequestReceivedTime() > now) {
-    try {
-      auto& profiler = dynamic_cast<ActivityProfilerProxy&>(
-          libkinetoApi_.activityProfiler());
-      profiler.scheduleTrace(config);
-    } catch (const std::exception& e) {
-      LOG(ERROR) << "Failed to schedule profiler request (busy?)";
-    }
-  }
+  notifyHandlers(config);
 }
 
 void ConfigLoader::updateConfigThread() {
@@ -313,10 +280,4 @@ bool ConfigLoader::hasNewConfig(const Config& oldConfig) {
   return config_.timestamp() > oldConfig.timestamp();
 }
 
-bool ConfigLoader::hasNewEventProfilerOnDemandConfig(const Config& oldConfig) {
-  std::lock_guard<std::mutex> lock(configLock_);
-  return onDemandEventProfilerConfig_->eventProfilerOnDemandStartTime() >
-      oldConfig.eventProfilerOnDemandStartTime();
-}
-
 } // namespace KINETO_NAMESPACE

libkineto/src/ConfigLoader.h

@@ -7,6 +7,7 @@
 
 #pragma once
 
+#include <algorithm>
 #include <atomic>
 #include <chrono>
 #include <condition_variable>
@@ -27,20 +28,56 @@ class DaemonConfigLoader;
 
 class ConfigLoader {
  public:
+
   static ConfigLoader& instance();
 
-  inline std::unique_ptr<Config> getConfigCopy() {
-    std::lock_guard<std::mutex> lock(configLock_);
-    return config_.clone();
+  enum ConfigKind {
+    ActivityProfiler = 0,
+    EventProfiler,
+    NumConfigKinds
+  };
+
+  struct ConfigHandler {
+    virtual ~ConfigHandler() {}
+    virtual bool canAcceptConfig() = 0;
+    virtual void acceptConfig(const Config& cfg) = 0;
+  };
+
+  void addHandler(ConfigKind kind, ConfigHandler* handler) {
+    handlers_[kind].push_back(handler);
+  }
+
+  void removeHandler(ConfigKind kind, ConfigHandler* handler) {
+    auto it = std::find(
+        handlers_[kind].begin(), handlers_[kind].end(), handler);
+    if (it != handlers_[kind].end()) {
+      handlers_[kind].erase(it);
+    }
   }
 
-  inline std::unique_ptr<Config> getEventProfilerOnDemandConfigCopy() {
+  void notifyHandlers(const Config& cfg) {
+    for (auto& key_val : handlers_) {
+      for (ConfigHandler* handler : key_val.second) {
+        handler->acceptConfig(cfg);
+      }
+    }
+  }
+
+  bool canHandlerAcceptConfig(ConfigKind kind) {
+    for (ConfigHandler* handler : handlers_[kind]) {
+      if (!handler->canAcceptConfig()) {
+        return false;
+      }
+    }
+    return true;
+  }
+
+  inline std::unique_ptr<Config> getConfigCopy() {
     std::lock_guard<std::mutex> lock(configLock_);
-    return onDemandEventProfilerConfig_->clone();
+    return config_.clone();
   }
 
   bool hasNewConfig(const Config& oldConfig);
-  bool hasNewEventProfilerOnDemandConfig(const Config& oldConfig);
   int contextCountForGpu(uint32_t gpu);
 
   void handleOnDemandSignal();
@@ -49,7 +86,7 @@ class ConfigLoader {
       std::function<std::unique_ptr<DaemonConfigLoader>()> factory);
 
  private:
-  explicit ConfigLoader(libkineto::LibkinetoApi& api);
+  ConfigLoader();
   ~ConfigLoader();
 
   void updateConfigThread();
@@ -65,21 +102,14 @@ class ConfigLoader {
       std::chrono::time_point<std::chrono::system_clock> now,
       Config& config);
 
-  inline bool eventProfilerRequest(const Config& config) {
-    return (
-        config.eventProfilerOnDemandStartTime() >
-        onDemandEventProfilerConfig_->eventProfilerOnDemandStartTime());
-  }
-
   std::string readOnDemandConfigFromDaemon(
       std::chrono::time_point<std::chrono::system_clock> now);
 
-  LibkinetoApi& libkinetoApi_;
   std::mutex configLock_;
   const char* configFileName_;
   Config config_;
-  std::unique_ptr<Config> onDemandEventProfilerConfig_;
   std::unique_ptr<DaemonConfigLoader> daemonConfigLoader_;
+  std::map<ConfigKind, std::vector<ConfigHandler*>> handlers_;
 
   std::chrono::seconds configUpdateIntervalSecs_;
   std::chrono::seconds onDemandConfigUpdateIntervalSecs_;