Add periodic server-side stats reporting (#484)

Summary:
Pull Request resolved: #484

During long-running ucachebench warmup and benchmark phases, there was no visibility into server performance until the final results were printed. This made it difficult to monitor progress or detect issues early during multi-hour benchmark runs.

This adds a background thread that periodically reports QPS, hit ratio, and operation counts to stdout. The reporting respects phase transitions (warmup vs benchmark) and resets its interval calculations when phases change. The interval is configurable via `--stats_interval_seconds` flag (default 10s, 0 to disable).

Reviewed By: excelle08

Differential Revision: D94508507

fbshipit-source-id: 5f092b42752a628c2d844f4d5f1afc474a48e2e9

Author: charles-typ

packages/ucache_bench/server/UcacheBenchServer.cpp

@@ -9,6 +9,7 @@
 
 #include <folly/Format.h>
 #include <folly/portability/GFlags.h>
+#include <chrono>
 
 #include "cachelib/allocator/CacheAllocator.h"
 #include "cachelib/allocator/HitsPerSlabStrategy.h"
@@ -24,6 +25,10 @@ UcacheBenchServer::UcacheBenchServer(const UcacheBenchConfig& config)
   setupCacheLib();
 }
 
+UcacheBenchServer::~UcacheBenchServer() {
+  stopPeriodicStats();
+}
+
 void UcacheBenchServer::setupCacheLib() {
   // CacheLib always requires DRAM cache initialization
   // Navy (NVM/SSD cache) is optionally enabled based on navy_cache_size_mb
@@ -476,5 +481,100 @@ void UcacheBenchServer::printFinalResults(double benchmarkDurationSec) const {
   }
 }
 
+void UcacheBenchServer::startPeriodicStats(uint32_t intervalSec) {
+  if (intervalSec == 0) {
+    return;
+  }
+  statsRunning_.store(true);
+  statsThread_ =
+      std::thread([this, intervalSec]() { periodicStatsLoop(intervalSec); });
+}
+
+void UcacheBenchServer::stopPeriodicStats() {
+  if (statsRunning_.load()) {
+    statsRunning_.store(false);
+    statsCv_.notify_all();
+    if (statsThread_.joinable()) {
+      statsThread_.join();
+    }
+  }
+}
+
+void UcacheBenchServer::periodicStatsLoop(uint32_t intervalSec) {
+  // Previous snapshot for computing interval QPS
+  uint64_t prevTotalOps = 0;
+  auto prevTime = std::chrono::steady_clock::now();
+  auto phaseStartTime = prevTime;
+  TrackingPhase prevPhase = TrackingPhase::NONE;
+
+  while (statsRunning_.load()) {
+    {
+      std::unique_lock<std::mutex> lock(statsMutex_);
+      statsCv_.wait_for(lock, std::chrono::seconds(intervalSec), [this]() {
+        return !statsRunning_.load();
+      });
+    }
+
+    if (!statsRunning_.load()) {
+      break;
+    }
+
+    auto phase = currentPhase_.load();
+    if (phase == TrackingPhase::NONE) {
+      continue;
+    }
+
+    // Reset snapshot on phase transition
+    if (phase != prevPhase) {
+      prevTotalOps = 0;
+      prevTime = std::chrono::steady_clock::now();
+      phaseStartTime = prevTime;
+      prevPhase = phase;
+    }
+
+    const PhaseMetrics& metrics =
+        (phase == TrackingPhase::WARMUP) ? warmupMetrics_ : benchmarkMetrics_;
+
+    uint64_t getReqs = metrics.getRequests.load(std::memory_order_relaxed);
+    uint64_t getHits = metrics.getHits.load(std::memory_order_relaxed);
+    uint64_t setReqs = metrics.setRequests.load(std::memory_order_relaxed);
+    uint64_t deleteReqs =
+        metrics.deleteRequests.load(std::memory_order_relaxed);
+    uint64_t totalOps = getReqs + setReqs + deleteReqs;
+
+    auto now = std::chrono::steady_clock::now();
+    double intervalElapsed =
+        std::chrono::duration<double>(now - prevTime).count();
+    double phaseElapsed =
+        std::chrono::duration<double>(now - phaseStartTime).count();
+
+    double intervalQps = (intervalElapsed > 0)
+        ? (totalOps - prevTotalOps) / intervalElapsed
+        : 0.0;
+    double avgQps = (phaseElapsed > 0) ? totalOps / phaseElapsed : 0.0;
+    double hitRatio = (getReqs > 0) ? (100.0 * getHits / getReqs) : 0.0;
+
+    const char* phaseName =
+        (phase == TrackingPhase::WARMUP) ? "WARMUP" : "BENCHMARK";
+
+    printf(
+        "[Server %s] %.0fs elapsed | QPS: %.0f (avg: %.0f) | "
+        "hit_ratio: %.2f%% | ops: %lu (GET: %lu, SET: %lu, DEL: %lu)\n",
+        phaseName,
+        phaseElapsed,
+        intervalQps,
+        avgQps,
+        hitRatio,
+        totalOps,
+        getReqs,
+        setReqs,
+        deleteReqs);
+    fflush(stdout);
+
+    prevTotalOps = totalOps;
+    prevTime = now;
+  }
+}
+
 } // namespace ucachebench
 } // namespace facebook

packages/ucache_bench/server/UcacheBenchServer.h

@@ -10,8 +10,11 @@
 #include <cachelib/allocator/CacheAllocator.h>
 #include <folly/futures/Future.h>
 #include <atomic>
+#include <condition_variable>
 #include <memory>
+#include <mutex>
 #include <string>
+#include <thread>
 #ifdef OSS_BUILD
 #include "UcacheBenchMessages.h"
 #else
@@ -126,6 +129,7 @@ struct UcacheBenchConfig {
 class UcacheBenchServer {
  public:
   explicit UcacheBenchServer(const UcacheBenchConfig& config);
+  ~UcacheBenchServer();
 
   // Request handlers using Carbon protocol
   folly::SemiFuture<UcbGetReply> processUcbGet(const UcbGetRequest& req);
@@ -163,6 +167,10 @@ class UcacheBenchServer {
   // Print final results in parseable format for benchpress
   void printFinalResults(double benchmarkDurationSec) const;
 
+  // Periodic stats reporting
+  void startPeriodicStats(uint32_t intervalSec);
+  void stopPeriodicStats();
+
  private:
   void setupCacheLib();
 
@@ -171,6 +179,9 @@ class UcacheBenchServer {
   void recordSet();
   void recordDelete();
 
+  // Periodic stats thread function
+  void periodicStatsLoop(uint32_t intervalSec);
+
   UcacheBenchConfig config_;
   std::unique_ptr<CacheAllocator> cache_;
   PoolId poolId_{};
@@ -179,6 +190,12 @@ class UcacheBenchServer {
   std::atomic<TrackingPhase> currentPhase_{TrackingPhase::NONE};
   PhaseMetrics warmupMetrics_;
   PhaseMetrics benchmarkMetrics_;
+
+  // Periodic stats thread
+  std::thread statsThread_;
+  std::atomic<bool> statsRunning_{false};
+  std::mutex statsMutex_;
+  std::condition_variable statsCv_;
 };
 
 } // namespace ucachebench

packages/ucache_bench/server/main.cpp

@@ -39,6 +39,10 @@ DEFINE_uint32(
     600,
     "Timeout in seconds for waiting for clients (0 = no timeout)");
 DEFINE_bool(verbose, false, "Enable verbose logging");
+DEFINE_uint32(
+    stats_interval_seconds,
+    10,
+    "Periodic stats reporting interval in seconds during warmup/benchmark (0 = disable)");
 
 // CacheLib configuration flags
 DEFINE_uint64(
@@ -376,6 +380,9 @@ int main(int argc, char** argv) {
       });
 
       adminServer->start();
+
+      // Start periodic server-side stats reporting
+      server->startPeriodicStats(FLAGS_stats_interval_seconds);
     }
 
     // If admin server is enabled, wait for it to complete
@@ -385,6 +392,8 @@ int main(int argc, char** argv) {
       if (!completed) {
         printf("Admin server timed out or failed\n");
       }
+      // Stop periodic stats before printing final results
+      server->stopPeriodicStats();
       // Stop the admin server
       adminServer->stop();
     } else {