XrdApps::JCache: add 'stats=<seconds>" configuration or XRD_JCACHE_ST…

…ATS=<sec> to enable regular dumping of the cache summary output (useful when we run inside a long living proxy)
cern-eos · Jun 13, 2024 · acd8096 · acd8096
1 parent 8a73b1e
commit acd8096
Show file tree

Hide file tree

Showing 5 changed files with 104 additions and 3 deletions.
diff --git a/src/XrdApps/XrdClJCachePlugin/file/CacheStats.hh b/src/XrdApps/XrdClJCachePlugin/file/CacheStats.hh
@@ -32,6 +32,7 @@
 #include <map>
 #include <mutex>
 #include <sstream>
+#include <thread>
 
 #include "file/XrdClJCacheFile.hh"
 #include <set>
@@ -55,6 +56,9 @@ struct CacheStats {
   }
 
   ~CacheStats() {
+    if (dumperInterval.load()) {
+      stop();
+    }
     if (dumponexit.load() && totaldatasize) {
       using namespace std::chrono;
       std::string jsonpath = XrdCl::JCacheFile::sJsonPath + "jcache.";
@@ -96,6 +100,18 @@ struct CacheStats {
     }
   }
 
+  void Reset() {
+    bytesRead = 0 ;
+    bytesReadV = 0;
+    bytesCached = 0;
+    bytesCachedV = 0;
+    readOps = 0;
+    readVOps = 0;
+    readVreadOps = 0;
+    nreadfiles = 0;
+    peakrate = 0;
+    bench.Reset();
+  }
   static std::string bytesToHumanReadable(double bytes) {
     const char *suffixes[] = {"B",  "KB", "MB", "GB", "TB",
                               "PB", "EB", "ZB", "YB"};
@@ -248,6 +264,7 @@ struct CacheStats {
 
   static std::string GlobalStats(CacheStats &sStats) {
     std::ostringstream oss;
+
     oss << "# "
            "-------------------------------------------------------------------"
            "---- #"
@@ -260,6 +277,7 @@ struct CacheStats {
            "---- #"
         << std::endl;
 
+
     oss << "# JCache : cache combined hit rate  : " << std::fixed
         << std::setprecision(2) << sStats.CombinedHitRate() << " %"
         << std::endl;
@@ -324,8 +342,8 @@ struct CacheStats {
     oss << "# JCache : app readrate             : " << std::fixed
         << std::setprecision(2)
         << sStats.bytesToHumanReadable((sStats.ReadBytes() / sStats.realTime))
-        << "/s" << " [ peak (1s) "
-        << sStats.bytesToHumanReadable(sStats.peakrate) << "/s ]" << std::endl;
+        << "/s" << " [ peak (1s) ";
+    oss << sStats.bytesToHumanReadable(sStats.peakrate) << "/s ]" << std::endl;
     oss << "# "
            "-------------------------------------------------------------------"
            "---- #"
@@ -353,5 +371,59 @@ struct CacheStats {
   JCache::TimeBench bench;
   std::vector<uint64_t> bytes_per_second;
   std::atomic<double> peakrate;
+  // Thread-related variables
+  std::atomic<uint64_t> dumperInterval;
+  std::thread dumperThread;
+  std::atomic<bool> stopFlag;
+
+
+  // Loop to regulary dump the statistics and to reset global counters
+  static void dumper(CacheStats* stats) {
+    while (!stats->stopFlag.load()) {
+      if (stats->dumperInterval) {
+        std::this_thread::sleep_for(std::chrono::seconds(stats->dumperInterval));
+      } else {
+        break;
+      }
+      std::cerr << "dumper global stats" << std::endl;
+      if (XrdCl::JCacheFile::sEnableSummary) {
+        XrdCl::JCacheFile::sStats.GetTimes();
+        XrdCl::JCacheFile::sStats.bytes_per_second =
+          XrdCl::JCacheFile::sStats.bench.GetBins((int)(XrdCl::JCacheFile::sStats.realTime));
+        XrdCl::JCacheFile::sStats.peakrate =
+          *(std::max_element(XrdCl::JCacheFile::sStats.bytes_per_second.begin(),
+                             XrdCl::JCacheFile::sStats.bytes_per_second.end()));
+        if (XrdCl::JCacheFile::sStats.realTime < 1) {
+          XrdCl::JCacheFile::sStats.peakrate = XrdCl::JCacheFile::sStats.ReadBytes() / XrdCl::JCacheFile::sStats.realTime;
+        }
+        std::string st = CacheStats::GlobalStats(XrdCl::JCacheFile::sStats);
+        std::cerr << st << std::endl;
+      }
+    }
+  }
+
+  // Method to start the dumper process in a separate thread
+  void run() {
+    stopFlag = false;
+    dumperThread = std::thread(&dumper, this);
+    dumperThread.detach();
+  }
+
+  // Method to stop the dumper process and join the thread
+  void stop() {
+    stopFlag = true;
+    if (dumperThread.joinable()) {
+      dumperThread.join();
+    }
+  }
+
+  // Configure thread to dump statistics every 'interval' seconds
+  void SetInterval(uint64_t interval) {
+    dumperInterval = interval;
+    stop();
+    if (interval) {
+      run();
+    }
+  }
 }; // class CacheStats
 } // namespace JCache
diff --git a/src/XrdApps/XrdClJCachePlugin/file/TimeBench.hh b/src/XrdApps/XrdClJCachePlugin/file/TimeBench.hh
@@ -47,6 +47,13 @@ private:
 public:
   TimeBench() : totalBytes(0), nbins(10) {}
 
+  void Reset() {
+    std::lock_guard<std::mutex> guard(mtx);
+    measurements.clear();
+    bins.clear();
+    totalBytes = 0;
+  }
+
   void AddMeasurement(uint64_t bytes) {
     std::lock_guard<std::mutex> guard(mtx);
     auto now = Clock::now();

diff --git a/src/XrdApps/XrdClJCachePlugin/file/XrdClJCacheFile.cc b/src/XrdApps/XrdClJCachePlugin/file/XrdClJCacheFile.cc
@@ -33,6 +33,7 @@ std::string XrdCl::JCacheFile::sJsonPath = "";
 bool XrdCl::JCacheFile::sEnableJournalCache = true;
 bool XrdCl::JCacheFile::sEnableVectorCache = false;
 bool XrdCl::JCacheFile::sEnableSummary = true;
+
 JCache::CacheStats XrdCl::JCacheFile::sStats(true);
 JCache::Cleaner XrdCl::JCacheFile::sCleaner;
 JournalManager XrdCl::JCacheFile::sJournalManager;
@@ -500,4 +501,15 @@ void JCacheFile::LogStats() {
       pStats->HitRateV(), pStats->bytesRead.load(), pStats->bytesReadV.load(),
       pStats->bytesCached.load(), pStats->bytesCachedV.load());
 }
+
+//----------------------------------------------------------------------------
+//! @brief set stats interval in CachStats class
+//----------------------------------------------------------------------------
+void
+JCacheFile::SetStatsInterval(uint64_t interval)
+{
+  sStats.SetInterval(interval);
+}
+
+
 } // namespace XrdCl
diff --git a/src/XrdApps/XrdClJCachePlugin/file/XrdClJCacheFile.hh b/src/XrdApps/XrdClJCachePlugin/file/XrdClJCacheFile.hh
@@ -233,6 +233,9 @@ public:
   static bool sEnableSummary;
   static JournalManager sJournalManager;
 
+  //! @brief set stats interval in seconds
+  static void SetStatsInterval(uint64_t interval);
+
   //----------------------------------------------------------------------------
   //! @brief log cache hit statistics
   //----------------------------------------------------------------------------
@@ -241,7 +244,7 @@ public:
   //! @brief global plugin cache hit statistics
   static JCache::CacheStats sStats;
 
-  //! @brief cleaner instance 
+  //! @brief cleaner instance
   static JCache::Cleaner sCleaner;
 
 private:

diff --git a/src/XrdApps/XrdClJCachePlugin/plugin/XrdClJCachePlugin.hh b/src/XrdApps/XrdClJCachePlugin/plugin/XrdClJCachePlugin.hh
@@ -65,6 +65,9 @@ public:
       auto its = config->find("summary");
       JCacheFile::SetSummary(its != config->end() ? its->second == "true"
                                                   : true);
+      auto itsi = config->find("stats");
+      JCacheFile::SetStatsInterval(itsi != config->end() ? std::stoll(std::string(itsi->second),0,10) : 0);
+
 
       if (const char *v = getenv("XRD_JCACHE_CACHE")) {
         JCacheFile::SetCache((std::string(v).length()) ? std::string(v) : "");
@@ -91,6 +94,10 @@ public:
                                                           : "");
       }
 
+      if (const char *v = getenv("XRD_JCACHE_STATS")) {
+        JCacheFile::SetStatsInterval((std::string(v).length()) ? std::stoll(std::string(v),0,10) : 0);
+      }
+
       Log *log = DefaultEnv::GetLog();
       log->Info(1, "JCache : cache directory: %s",
                 JCacheFile::sCachePath.c_str());