Test ConfigLoader on-demand daemon poll ingress (#1504)

libkineto/src/ConfigLoader.cpp

@@ -120,6 +120,27 @@ void ConfigLoader::stopUpdateThread() {
   stopThread();
 }
 
+void ConfigLoader::resetDaemonConfigLoaderForTesting() {
+  daemonConfigLoader_.reset();
+}
+
+bool ConfigLoader::waitForUpdateThreadLoopCountForTesting(
+    uint64_t target,
+    std::chrono::milliseconds timeout) {
+  std::unique_lock<std::mutex> lock(loopCountMutex_);
+  return loopCountCondVar_.wait_for(lock, timeout, [this, target] {
+    return updateThreadLoopCount_.load(std::memory_order_acquire) >= target;
+  });
+}
+
+std::chrono::seconds ConfigLoader::onDemandConfigUpdateIntervalForTesting() {
+  std::scoped_lock lock(configLock_);
+  // config_ is the authoritative source; updateConfigThread caches its value
+  // into onDemandConfigUpdateIntervalSecs_ on each base-config refresh.
+  return config_ ? config_->onDemandConfigUpdateIntervalSecs()
+                 : onDemandConfigUpdateIntervalSecs_;
+}
+
 ConfigLoader::~ConfigLoader() {
   stopThread();
 #if !USE_GOOGLE_LOG
@@ -267,6 +288,13 @@ void ConfigLoader::updateConfigThread() {
           onDemandConfig->verboseLogLevel(),
           onDemandConfig->verboseLogModules());
     }
+    // Mark one completed iteration and wake any test waiting for deterministic
+    // progression of the real poll thread.
+    {
+      std::scoped_lock lock(loopCountMutex_);
+      updateThreadLoopCount_.fetch_add(1, std::memory_order_release);
+    }
+    loopCountCondVar_.notify_all();
   }
 }
 

libkineto/src/ConfigLoader.h

@@ -12,6 +12,7 @@
 #include <atomic>
 #include <chrono>
 #include <condition_variable>
+#include <cstdint>
 #include <functional>
 #include <mutex>
 #include <string>
@@ -96,6 +97,34 @@ class ConfigLoader {
   // leaving the join to run during static destruction.
   void stopUpdateThread();
 
+  // Test-only. Returns how many iterations the background poll thread has
+  // completed. A test that installs a fake daemon config loader can start the
+  // thread and wait for this count to advance, then deterministically observe
+  // the effects of a known number of real poll iterations. Loaded with acquire
+  // ordering so that observing an advanced count also makes that iteration's
+  // writes visible to the observer.
+  [[nodiscard]] uint64_t updateThreadLoopCountForTesting() const {
+    return updateThreadLoopCount_.load(std::memory_order_acquire);
+  }
+
+  // Test-only. Blocks until the poll thread's iteration count reaches target,
+  // or the timeout elapses; returns true if the count was reached.
+  bool waitForUpdateThreadLoopCountForTesting(
+      uint64_t target,
+      std::chrono::milliseconds timeout);
+
+  // Test-only. Drops the cached daemon config loader so the next poll rebuilds
+  // it from the currently registered factory. The loader is a member of this
+  // process-wide singleton, so a test that injected a fake via the factory must
+  // clear it (after stopping the thread) or a later test would reuse a loader
+  // pointing at destroyed test state.
+  void resetDaemonConfigLoaderForTesting();
+
+  // Test-only. Returns the on-demand poll interval the background thread is
+  // currently using, taken from the loaded base config. Lets a test size a
+  // timeout to the live cadence instead of assuming the default.
+  std::chrono::seconds onDemandConfigUpdateIntervalForTesting();
+
  private:
   ConfigLoader();
   ~ConfigLoader();
@@ -128,6 +157,13 @@ class ConfigLoader {
   std::condition_variable updateThreadCondVar_;
   std::mutex updateThreadMutex_;
   std::atomic_bool stopFlag_{false};
+
+  // Incremented at the end of each updateConfigThread() iteration. Test-only
+  // observation point; see updateThreadLoopCountForTesting(). loopCountCondVar_
+  // is notified on each increment so a test can block until a target count.
+  std::atomic<uint64_t> updateThreadLoopCount_{0};
+  std::mutex loopCountMutex_;
+  std::condition_variable loopCountCondVar_;
 };
 
 } // namespace KINETO_NAMESPACE

libkineto/test/CMakeLists.txt

@@ -62,6 +62,24 @@ if(NOT WIN32)
         nlohmann_json::nlohmann_json
         ${XPU_XPUPTI_LIBRARY})
     gtest_discover_tests(AsyncActivityProfilerHandlerTest)
+
+    # SyncActivityProfilerHandlerTest
+    add_executable(SyncActivityProfilerHandlerTest
+        SyncActivityProfilerHandlerTest.cpp)
+    target_link_libraries(SyncActivityProfilerHandlerTest PRIVATE
+        gtest_main
+        kineto_base kineto_api
+        ${XPU_XPUPTI_LIBRARY})
+    gtest_discover_tests(SyncActivityProfilerHandlerTest)
+
+    # GenericActivityProfilerTeardownTest
+    add_executable(GenericActivityProfilerTeardownTest
+        GenericActivityProfilerTeardownTest.cpp)
+    target_link_libraries(GenericActivityProfilerTeardownTest PRIVATE
+        gtest_main
+        kineto_base kineto_api
+        ${XPU_XPUPTI_LIBRARY})
+    gtest_discover_tests(GenericActivityProfilerTeardownTest)
 endif()
 
 if(KINETO_BACKEND STREQUAL "cuda")
@@ -96,6 +114,13 @@ target_link_libraries(CuptiStringsTest PRIVATE
     gtest_main
     kineto_base kineto_api)
 gtest_discover_tests(CuptiStringsTest)
+
+# DevicePropertiesTest
+add_executable(DevicePropertiesTest DevicePropertiesTest.cpp)
+target_link_libraries(DevicePropertiesTest PRIVATE
+    gtest_main
+    kineto_base kineto_api)
+gtest_discover_tests(DevicePropertiesTest)
 endif()
 
 if(KINETO_BACKEND STREQUAL "rocm")

libkineto/test/ConfigLoaderTest.cpp

@@ -8,61 +8,135 @@
 
 #include <gtest/gtest.h>
 
+#include <chrono>
+#include <cstdint>
+#include <memory>
+#include <string>
 #include <utility>
 #include <vector>
 
 #include "include/Config.h"
 #include "src/ConfigLoader.h"
+#include "src/DaemonConfigLoader.h"
 
 using namespace KINETO_NAMESPACE;
 
 namespace {
 
 // Records how ConfigLoader dispatches to a handler and lets a test control
-// whether canAcceptConfig() accepts, so the fan-out logic can be asserted
-// without a real profiler or the daemon poll thread.
+// whether canAcceptConfig() accepts. When the real poll thread is running,
+// set canAcceptResult before starting it and do not mutate these fields while
+// it runs.
 struct RecordingConfigHandler : ConfigLoader::ConfigHandler {
   bool canAcceptResult{true};
-
   int acceptCalls{0};
   const Config* lastAcceptedConfig{nullptr};
 
+  // Copied at accept time. The daemon-poll path dispatches a Config that lives
+  // only for the poll iteration, so lastAcceptedConfig dangles afterward; a
+  // value copy of a parsed field stays valid for assertions.
+  std::string lastRequestTraceID;
+
   bool canAcceptConfig() override {
     return canAcceptResult;
   }
 
   bool acceptConfig(const Config& cfg) override {
     ++acceptCalls;
     lastAcceptedConfig = &cfg;
+    lastRequestTraceID = cfg.requestTraceID();
     return true;
   }
 };
 
-// Drives the ConfigLoader singleton's handler-dispatch API directly. The daemon
-// config loader factory is left unset, so the polling thread never reads an
-// on-demand config and never calls the registered handlers -- every callback a
-// test observes comes from its own notifyHandlers()/canHandlerAcceptConfig()
-// call. Handlers are unregistered in TearDown because the singleton persists
-// across tests in the binary.
+// Canned daemon responses and recorded queries. Owned by the test; the fake
+// below holds a reference to it. The poll thread writes the recorded fields, so
+// a test reads them only after stopping (joining) the thread.
+struct DaemonPollProbe {
+  std::string onDemandConfig;
+  int readOnDemandCalls{0};
+  bool lastActivitiesRequested{false};
+};
+
+// Stands in for the dynolog IPC config source, so the real background poll
+// thread runs against canned configs with no daemon.
+class FakeDaemonConfigLoader : public IDaemonConfigLoader {
+ public:
+  explicit FakeDaemonConfigLoader(DaemonPollProbe& probe) : probe_(probe) {}
+
+  std::string readBaseConfig() override {
+    return "";
+  }
+
+  std::string readOnDemandConfig(bool activities) override {
+    ++probe_.readOnDemandCalls;
+    probe_.lastActivitiesRequested = activities;
+    return probe_.onDemandConfig;
+  }
+
+  void setCommunicationFabric(bool /*enabled*/) override {}
+
+ private:
+  DaemonPollProbe& probe_;
+};
+
+// Drives the ConfigLoader singleton. Two styles of test live here:
+//   * Handler fan-out tests call notifyHandlers()/canHandlerAcceptConfig()
+//     directly and install no daemon factory, so the background poll thread
+//     (started by addHandler) reads nothing and never calls the handlers.
+//   * Daemon-poll tests install a FakeDaemonConfigLoader factory, start the
+//     real poll thread, and wait on the thread's iteration counter to observe
+//     real poll iterations deterministically.
+// The singleton persists across tests, so TearDown stops the thread, drops the
+// injected loader, clears the factory, and unregisters handlers.
 class ConfigLoaderTest : public ::testing::Test {
  protected:
   static ConfigLoader& loader() {
     return ConfigLoader::instance();
   }
 
+  // Installs a fake daemon config source. Call before starting the poll thread
+  // (before registering a handler) so the thread's first iteration uses it.
+  static void installFakeDaemon(DaemonPollProbe& probe) {
+    ConfigLoader::setDaemonConfigLoaderFactory(
+        [&probe]() { return std::make_unique<FakeDaemonConfigLoader>(probe); });
+  }
+
+  // Blocks until the thread's iteration counter reaches target or the timeout
+  // elapses. Returns false on timeout.
+  [[nodiscard]] static bool waitForLoopCount(
+      uint64_t target,
+      std::chrono::milliseconds timeout = std::chrono::seconds(5)) {
+    return loader().waitForUpdateThreadLoopCountForTesting(target, timeout);
+  }
+
   void registerHandler(
       ConfigLoader::ConfigKind kind,
       ConfigLoader::ConfigHandler* handler) {
     loader().addHandler(kind, handler);
     registered_.emplace_back(kind, handler);
   }
 
+  // Registers handler as an ActivityProfiler handler (which starts the poll
+  // thread) and blocks until the thread has completed loops iterations.
+  // Returns false if that count is not reached before the timeout.
+  [[nodiscard]] bool startPollingAndWait(
+      RecordingConfigHandler& handler,
+      uint64_t loops) {
+    const uint64_t base = loader().updateThreadLoopCountForTesting();
+    registerHandler(ConfigLoader::ConfigKind::ActivityProfiler, &handler);
+    return waitForLoopCount(base + loops);
+  }
+
   void TearDown() override {
-    // Join the background config-update thread before this test process exits.
-    // addHandler() starts it; leaving the join to static destruction races the
-    // thread against teardown and can abort with a mutex lock on a destroyed
-    // mutex.
+    // Join the poll thread first, so nothing touches the loader afterward.
     loader().stopUpdateThread();
+
+    // Drop the injected loader and factory: both capture this test's probe,
+    // which does not outlive the test.
+    loader().resetDaemonConfigLoaderForTesting();
+    ConfigLoader::setDaemonConfigLoaderFactory(nullptr);
+
     // removeHandler is a no-op for a handler already removed by the test, so
     // double removal is safe.
     for (const auto& [kind, handler] : registered_) {
@@ -76,6 +150,8 @@ class ConfigLoaderTest : public ::testing::Test {
       registered_;
 };
 
+// ---- Handler fan-out ----
+
 // notifyHandlers() forwards the config to every registered handler across all
 // config kinds, passing through the same config object.
 TEST_F(ConfigLoaderTest, NotifyHandlersForwardsConfigToAllRegisteredHandlers) {
@@ -131,4 +207,100 @@ TEST_F(ConfigLoaderTest, CanHandlerAcceptConfigVacuouslyTrueWithNoHandlers) {
   EXPECT_TRUE(loader().canHandlerAcceptConfig(kind));
 }
 
+// ---- Real daemon poll thread ----
+//
+// These tests run the actual updateConfigThread() against a fake daemon and use
+// the iteration counter to synchronize. They tolerate a pre-existing local
+// config file (/etc/libkineto.conf or $KINETO_CONFIG) on the test system.
+
+// The poll thread reads the on-demand config from the daemon, parses it, and
+// dispatches it to registered handlers, requesting activities while the handler
+// can accept.
+TEST_F(ConfigLoaderTest, ThreadPollsDaemonAndDispatchesOnDemandConfig) {
+  DaemonPollProbe probe;
+  probe.onDemandConfig = "REQUEST_TRACE_ID=daemon-trace-42\n";
+  installFakeDaemon(probe);
+
+  RecordingConfigHandler handler;
+  ASSERT_TRUE(startPollingAndWait(handler, /*loops=*/1));
+  loader().stopUpdateThread();
+
+  EXPECT_GE(probe.readOnDemandCalls, 1);
+  EXPECT_TRUE(probe.lastActivitiesRequested);
+  EXPECT_GE(handler.acceptCalls, 1);
+  EXPECT_EQ(handler.lastRequestTraceID, "daemon-trace-42");
+}
+
+// The thread re-polls the on-demand config on its update cadence, so over a
+// couple of intervals it reads the daemon more than once.
+//
+// This runs at the real cadence, so it takes a few seconds. It cannot be sped
+// up by injecting a smaller ON_DEMAND_CONFIG_UPDATE_INTERVAL_SECS via the fake
+// daemon's base config: updateBaseConfig reads the local config file first and
+// uses the daemon base config only when the local read is empty, so a present
+// local config (as on the test hosts) shadows the fake. If that local-first
+// ordering is ever inverted (see the TODO in updateBaseConfig), injecting a
+// faster interval would let this test finish without the wait.
+TEST_F(ConfigLoaderTest, ThreadRepeatedlyPollsOnDemandConfig) {
+  DaemonPollProbe probe;
+  probe.onDemandConfig = "REQUEST_TRACE_ID=x\n";
+  installFakeDaemon(probe);
+
+  RecordingConfigHandler handler;
+  const uint64_t base = loader().updateThreadLoopCountForTesting();
+  registerHandler(ConfigLoader::ConfigKind::ActivityProfiler, &handler);
+
+  // The first poll is immediate; after it, the thread's on-demand interval
+  // reflects the loaded base config. Size the wait for the second poll to that
+  // live interval (with slack) so a host that configured a larger interval does
+  // not cause a spurious timeout.
+  ASSERT_TRUE(waitForLoopCount(base + 1));
+  const auto timeout = std::chrono::duration_cast<std::chrono::milliseconds>(
+      loader().onDemandConfigUpdateIntervalForTesting() * 3 +
+      std::chrono::seconds(5));
+  ASSERT_TRUE(waitForLoopCount(base + 2, timeout));
+  loader().stopUpdateThread();
+
+  EXPECT_GE(probe.readOnDemandCalls, 2);
+}
+
+// An empty on-demand response (no config posted) dispatches nothing.
+TEST_F(ConfigLoaderTest, ThreadDropsEmptyOnDemandConfig) {
+  DaemonPollProbe probe; // onDemandConfig defaults empty
+  installFakeDaemon(probe);
+
+  RecordingConfigHandler handler;
+  ASSERT_TRUE(startPollingAndWait(handler, /*loops=*/1));
+  loader().stopUpdateThread();
+
+  EXPECT_GE(probe.readOnDemandCalls, 1);
+  EXPECT_EQ(handler.acceptCalls, 0);
+}
+
+// The thread requests an activities config only while its handlers can accept
+// one; a busy handler suppresses the activities request.
+TEST_F(ConfigLoaderTest, ThreadSuppressesActivitiesRequestWhenHandlerBusy) {
+  DaemonPollProbe probe;
+  probe.onDemandConfig = "REQUEST_TRACE_ID=x\n";
+  installFakeDaemon(probe);
+
+  RecordingConfigHandler handler;
+  handler.canAcceptResult = false; // set before the thread starts
+  ASSERT_TRUE(startPollingAndWait(handler, /*loops=*/1));
+  loader().stopUpdateThread();
+
+  EXPECT_GE(probe.readOnDemandCalls, 1);
+  EXPECT_FALSE(probe.lastActivitiesRequested);
+}
+
+// With no daemon factory installed (the default on non-daemon hosts), the poll
+// thread reads no on-demand config and dispatches nothing.
+TEST_F(ConfigLoaderTest, ThreadWithoutDaemonFactoryDispatchesNothing) {
+  RecordingConfigHandler handler;
+  ASSERT_TRUE(startPollingAndWait(handler, /*loops=*/1));
+  loader().stopUpdateThread();
+
+  EXPECT_EQ(handler.acceptCalls, 0);
+}
+
 } // namespace