Invert ConfigLoader -> profiler dependency (take 2) (#271)

libkineto/include/libkineto.h

@@ -36,6 +36,7 @@ extern "C" {
 namespace libkineto {
 
 class Config;
+class ConfigLoader;
 
 struct CpuTraceBuffer {
   TraceSpan span{0, 0, "none"};
@@ -46,6 +47,9 @@ struct CpuTraceBuffer {
 class LibkinetoApi {
  public:
 
+  LibkinetoApi(ConfigLoader& configLoader) : configLoader_(configLoader) {
+  }
+
   // Called by client that supports tracing API.
   // libkineto can still function without this.
   void registerClient(ClientInterface* client);
@@ -92,11 +96,17 @@ class LibkinetoApi {
     suppressLibkinetoLogMessages();
   }
 
+  // Provides access to profier configuration manaegement
+  ConfigLoader& configLoader() {
+    return configLoader_;
+  }
+
  private:
 
   // Client is initialized once both it and libkineto has registered
   void initClientIfRegistered();
 
+  ConfigLoader& configLoader_;
   std::unique_ptr<ActivityProfilerInterface> activityProfiler_{};
   ClientInterface* client_{};
   int32_t clientRegisterThread_{0};

libkineto/src/ActivityProfilerController.cpp

@@ -25,11 +25,18 @@ namespace KINETO_NAMESPACE {
 
 constexpr milliseconds kProfilerIntervalMsecs(1000);
 
-ActivityProfilerController::ActivityProfilerController(bool cpuOnly) {
-  profiler_ = std::make_unique<ActivityProfiler>(CuptiActivityInterface::singleton(), cpuOnly);
+ActivityProfilerController::ActivityProfilerController(
+    ConfigLoader& configLoader, bool cpuOnly)
+    : configLoader_(configLoader) {
+  profiler_ = std::make_unique<ActivityProfiler>(
+      CuptiActivityInterface::singleton(), cpuOnly);
+  configLoader_.addHandler(ConfigLoader::ConfigKind::ActivityProfiler, this);
 }
 
 ActivityProfilerController::~ActivityProfilerController() {
+  LOG(INFO) << "ActivityProfilerController::~ActivityProfilerController()";
+  configLoader_.removeHandler(
+      ConfigLoader::ConfigKind::ActivityProfiler, this);
   if (profilerThread_) {
     // signaling termination of the profiler loop
     stopRunloop_ = true;
@@ -65,6 +72,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 +125,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,21 +9,23 @@
 
 #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);
+  explicit ActivityProfilerController(ConfigLoader& configLoader, bool cpuOnly);
   ActivityProfilerController(const ActivityProfilerController&) = delete;
   ActivityProfilerController& operator=(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);
@@ -72,6 +77,7 @@ class ActivityProfilerController {
   std::unique_ptr<ActivityLogger> logger_;
   std::thread* profilerThread_{nullptr};
   std::atomic_bool stopRunloop_{false};
+  ConfigLoader& configLoader_;
 };
 
 } // namespace KINETO_NAMESPACE

libkineto/src/ActivityProfilerProxy.cpp

@@ -13,16 +13,21 @@
 
 namespace KINETO_NAMESPACE {
 
-void ActivityProfilerProxy::init() {
-  if (!controller_) {
-    controller_ = new ActivityProfilerController(cpuOnly_);
-  }
+ActivityProfilerProxy::ActivityProfilerProxy(
+    bool cpuOnly, ConfigLoader& configLoader)
+  : cpuOnly_(cpuOnly), configLoader_(configLoader) {
 }
 
 ActivityProfilerProxy::~ActivityProfilerProxy() {
   delete controller_;
 };
 
+void ActivityProfilerProxy::init() {
+  if (!controller_) {
+    controller_ = new ActivityProfilerController(configLoader_, cpuOnly_);
+  }
+}
+
 void ActivityProfilerProxy::scheduleTrace(const std::string& configStr) {
   Config config;
   config.parse(configStr);

libkineto/src/ActivityProfilerProxy.h

@@ -26,11 +26,12 @@ using namespace libkineto;
 
 class ActivityProfilerController;
 class Config;
+class ConfigLoader;
 
 class ActivityProfilerProxy : public ActivityProfilerInterface {
 
  public:
-  ActivityProfilerProxy(bool cpuOnly) : cpuOnly_(cpuOnly) {}
+  ActivityProfilerProxy(bool cpuOnly, ConfigLoader& configLoader);
   ~ActivityProfilerProxy() override;
 
   void init() override;
@@ -64,6 +65,7 @@ class ActivityProfilerProxy : public ActivityProfilerInterface {
 
  private:
   bool cpuOnly_{true};
+  ConfigLoader& configLoader_;
   ActivityProfilerController* controller_{nullptr};
 };
 

libkineto/src/Config.cpp

@@ -148,6 +148,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,30 +141,29 @@ 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) {
-  configFileName_ = getenv(kConfigFileEnvVar.data());
-  if (configFileName_ == nullptr) {
-    configFileName_ = kConfigFile.data();
-  }
-  if (daemonConfigLoaderFactory()) {
-    daemonConfigLoader_ = daemonConfigLoaderFactory()();
-  }
-  updateBaseConfig();
-  SET_LOG_VERBOSITY_LEVEL(config_.verboseLogLevel(), config_.verboseLogModules());
-  if (daemonConfigLoader_) {
-    daemonConfigLoader_->setCommunicationFabric(config_.ipcFabricEnabled());
+}
+
+void ConfigLoader::startThread() {
+  if (!updateThread_) {
+    initBaseConfig();
+    if (daemonConfigLoaderFactory()) {
+      daemonConfigLoader_ = daemonConfigLoaderFactory()();
+      daemonConfigLoader_->setCommunicationFabric(config_.ipcFabricEnabled());
+    }
+    updateThread_ =
+        std::make_unique<std::thread>(&ConfigLoader::updateConfigThread, this);
+
+    setupSignalHandler(config_.sigUsr2Enabled());
   }
-  updateThread_ =
-      std::make_unique<std::thread>(&ConfigLoader::updateConfigThread, this);
 }
 
 ConfigLoader::~ConfigLoader() {
+  LOG(INFO) << "ConfigLoader::~ConfigLoader()";
   if (updateThread_) {
     stopFlag_ = true;
     {
@@ -186,6 +182,16 @@ void ConfigLoader::handleOnDemandSignal() {
   }
 }
 
+void ConfigLoader::initBaseConfig() {
+  if (!configFileName_) {
+    configFileName_ = getenv(kConfigFileEnvVar.data());
+    if (configFileName_ == nullptr) {
+      configFileName_ = kConfigFile.data();
+    }
+    updateBaseConfig();
+  }
+}
+
 void ConfigLoader::updateBaseConfig() {
   // First try reading local config file
   // If that fails, read from daemon
@@ -220,51 +226,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() {
@@ -325,10 +303,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