Make per-request RNGs deterministic by default (reproducible QPS)

Differential Revision: D115671541

Author: excelle08

packages/feedsim/third_party/src/workloads/ranking/DriverNodeRank.cc

@@ -13,6 +13,8 @@
 // limitations under the License.
 
 #include <algorithm>
+#include <atomic>
+#include <cstdlib>
 #include <cstring>
 #include <iostream>
 #include <memory>
@@ -67,11 +69,40 @@ static std::shared_ptr<folly::CPUThreadPoolExecutor> g_session_pool;
 const int kMaxRequestSize = 8192;
 const int kRecomputeQPSPeriod = 1;  // Reduced from 5 to 1 second for faster feedback
 
+// Deterministic per-thread RNG seed. The driver's per-thread RNGs used to be
+// seeded from std::random_device, making the generated request stream (sizes,
+// content) non-reproducible run-to-run. Seed from a fixed base so each thread
+// keeps an independent but reproducible sequence. Set FEEDSIM_RNG_RANDOM=1 to
+// restore the old non-deterministic seed.
+static bool feedsimRngRandom() {
+  static const bool kRandom = [] {
+    const char* e = std::getenv("FEEDSIM_RNG_RANDOM");
+    return e != nullptr && e[0] == '1';
+  }();
+  return kRandom;
+}
+
+static unsigned detRngSeed(unsigned base) {
+  if (feedsimRngRandom()) {
+    return std::random_device{}();
+  }
+  static std::atomic<unsigned> ctr{0};
+  return base + ctr.fetch_add(1) * 2654435761u;
+}
+
+static unsigned detRngSeedTid(unsigned base, int thread_id) {
+  if (feedsimRngRandom()) {
+    return static_cast<unsigned>(std::random_device{}()) +
+        static_cast<unsigned>(thread_id);
+  }
+  return base + static_cast<unsigned>(thread_id);
+}
+
 // Simple random string generator (replaces oldisim/Util.h RandomString)
 static std::string RandomString(size_t length) {
   static const char charset[] =
       "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz";
-  thread_local std::mt19937 rng(std::random_device{}());
+  thread_local std::mt19937 rng(detRngSeed(0xD5117A2Du));
   std::uniform_int_distribution<size_t> dist(0, sizeof(charset) - 2);
   std::string str(length, 0);
   for (size_t i = 0; i < length; ++i) {
@@ -202,15 +233,13 @@ void ThreadStartup(int thread_id,
 
   // Initialize Silesia RNG per thread
   if (g_silesia_loader && g_silesia_loader->isLoaded()) {
-    this_thread.silesia_rng.seed(
-        static_cast<unsigned>(std::random_device{}()) + thread_id);
+    this_thread.silesia_rng.seed(detRngSeedTid(0x51E51A00u, thread_id));
   }
 
   // Initialize request size sampler RNG per thread
   if ((g_req_size_sampler && g_req_size_sampler->isLoaded()) ||
       g_rpc_dist_registry) {
-    this_thread.req_size_rng.seed(
-        static_cast<unsigned>(std::random_device{}()) + 0xDEADBEEF + thread_id);
+    this_thread.req_size_rng.seed(detRngSeedTid(0xDEADBEEFu, thread_id));
   }
 
   // If user gave QPS target, initialize QPS modulation

packages/feedsim/third_party/src/workloads/ranking/LeafNodeRank.cc

@@ -18,6 +18,7 @@
 #include <cmath>
 #include <cstdint>
 #include <cstdio>
+#include <cstdlib>
 #include <functional>
 #include <memory>
 #include <mutex>
@@ -311,10 +312,57 @@ static std::vector<ranking::RankingResponse> g_response_templates;
 static std::vector<size_t> g_response_template_sizes;
 static std::atomic<size_t> g_response_template_idx{0};
 
+// Deterministic per-thread RNG seed. These per-request RNGs used to be seeded
+// from std::random_device, which made benchmark QPS non-reproducible run-to-run
+// (per-request response-size / fanout sampling shifted the SLA operating point;
+// see the T1_TRN ~4% QPS-CoV investigation). Seed from a fixed base mixed with
+// a per-thread counter so each thread keeps an independent but reproducible
+// sequence. Set FEEDSIM_RNG_RANDOM=1 to restore the old non-deterministic seed.
+static bool feedsimRngRandom() {
+  static const bool kRandom = [] {
+    const char* e = std::getenv("FEEDSIM_RNG_RANDOM");
+    return e != nullptr && e[0] == '1';
+  }();
+  return kRandom;
+}
+
+static unsigned detRngSeed(unsigned base) {
+  if (feedsimRngRandom()) {
+    return std::random_device{}();
+  }
+  static std::atomic<unsigned> ctr{0};
+  return base + ctr.fetch_add(1) * 2654435761u;
+}
+
+// Deterministic per-thread seed variant for call sites that already have a
+// stable thread index.
+static unsigned detRngSeedTid(unsigned base, int thread_id) {
+  if (feedsimRngRandom()) {
+    return std::random_device{}() ^ static_cast<unsigned>(thread_id + 1);
+  }
+  return base ^ static_cast<unsigned>(thread_id + 1);
+}
+
+// Default deterministic seed for the logical workload RNGs (node/page/pointer).
+static constexpr unsigned kFeedsimDefaultSeed = 42u;
+
+// Resolve a logical workload RNG seed. When the option is not passed we use a
+// fixed default (reproducible runs); an explicit negative value opts back into
+// a time-based seed for callers that want non-determinism.
+static unsigned resolveSeed(bool given, long arg, unsigned deflt) {
+  if (!given) {
+    return deflt;
+  }
+  if (arg < 0) {
+    return static_cast<unsigned>(
+        std::chrono::system_clock::now().time_since_epoch().count());
+  }
+  return static_cast<unsigned>(arg);
+}
+
 // Pool of per-thread RNGs for response-size sampling (per-thread to avoid
 // shared-state contention; std::mt19937 is not thread-safe).
-static thread_local std::mt19937 g_response_size_rng{
-    std::random_device{}()};
+static thread_local std::mt19937 g_response_size_rng{detRngSeed(0xF00D5127u)};
 
 // Helper: measure the serialized size of a RankingResponse via CompactProto.
 // Used at pool init to build the (size → template) lookup table.
@@ -476,8 +524,7 @@ void ThreadStartup(
   // used elsewhere in this struct.
   this_thread.rpc_registry = g_rpc_registry.get();
   this_thread.rpc_silesia = g_rpc_silesia;
-  this_thread.rpc_rng.seed(
-      std::random_device{}() ^ static_cast<unsigned>(thread_id + 1));
+  this_thread.rpc_rng.seed(detRngSeedTid(0x2C519A00u, thread_id));
   if (this_thread.rpc_registry != nullptr && srEventBasePool != nullptr) {
     auto evbs = srEventBasePool->getAllEventBases();
     this_thread.mock_clients.reserve(evbs.size());
@@ -507,29 +554,18 @@ void ThreadStartup(
   // Store shared DLRM ranker
   this_thread.dlrm_ranker = shared_dlrm_ranker;
 
-  unsigned noderank_seed;
-  if (args.node_rank_seed_given) {
-    noderank_seed = static_cast<unsigned>(args.node_rank_seed_arg);
-  } else {
-    noderank_seed = std::chrono::system_clock::now().time_since_epoch().count();
-  }
+  unsigned noderank_seed = resolveSeed(
+      args.node_rank_seed_given, args.node_rank_seed_arg, kFeedsimDefaultSeed);
 
-  unsigned pointer_chase_seed;
-  if (args.pointer_chase_seed_given) {
-    pointer_chase_seed = static_cast<unsigned>(args.pointer_chase_seed_arg);
-  } else {
-    pointer_chase_seed =
-        std::chrono::system_clock::now().time_since_epoch().count();
-  }
+  unsigned pointer_chase_seed = resolveSeed(
+      args.pointer_chase_seed_given,
+      args.pointer_chase_seed_arg,
+      kFeedsimDefaultSeed);
 
   // Only initialize PageRank if we're using it
   if (g_workload_type == WorkloadType::PAGERANK) {
-    unsigned page_rank_seed;
-    if (args.page_rank_seed_given) {
-      page_rank_seed = static_cast<unsigned>(args.page_rank_seed_arg);
-    } else {
-      page_rank_seed = std::chrono::system_clock::now().time_since_epoch().count();
-    }
+    unsigned page_rank_seed = resolveSeed(
+        args.page_rank_seed_given, args.page_rank_seed_arg, kFeedsimDefaultSeed);
     auto graph = params.makeGraphCopy(g_shared_graph);
     this_thread.page_ranker = std::make_unique<ranking::dwarfs::PageRank>(
         std::move(graph), args.cpu_threads_arg, page_rank_seed);
@@ -637,8 +673,7 @@ void ThreadStartup(
   // used elsewhere in this struct.
   this_thread.rpc_registry = g_rpc_registry.get();
   this_thread.rpc_silesia = g_rpc_silesia;
-  this_thread.rpc_rng.seed(
-      std::random_device{}() ^ static_cast<unsigned>(thread_id + 1));
+  this_thread.rpc_rng.seed(detRngSeedTid(0x2C519A00u, thread_id));
   if (this_thread.rpc_registry != nullptr && srEventBasePool != nullptr) {
     auto evbs = srEventBasePool->getAllEventBases();
     this_thread.mock_clients.reserve(evbs.size());
@@ -664,27 +699,16 @@ void ThreadStartup(
       }
     }
   }
-  unsigned noderank_seed;
-  if (args.node_rank_seed_given) {
-    noderank_seed = static_cast<unsigned>(args.node_rank_seed_arg);
-  } else {
-    noderank_seed = std::chrono::system_clock::now().time_since_epoch().count();
-  }
+  unsigned noderank_seed = resolveSeed(
+      args.node_rank_seed_given, args.node_rank_seed_arg, kFeedsimDefaultSeed);
 
-  unsigned page_rank_seed;
-  if (args.page_rank_seed_given) {
-    page_rank_seed = static_cast<unsigned>(args.page_rank_seed_arg);
-  } else {
-    page_rank_seed = std::chrono::system_clock::now().time_since_epoch().count();
-  }
+  unsigned page_rank_seed = resolveSeed(
+      args.page_rank_seed_given, args.page_rank_seed_arg, kFeedsimDefaultSeed);
 
-  unsigned pointer_chase_seed;
-  if (args.pointer_chase_seed_given) {
-    pointer_chase_seed = static_cast<unsigned>(args.pointer_chase_seed_arg);
-  } else {
-    pointer_chase_seed =
-        std::chrono::system_clock::now().time_since_epoch().count();
-  }
+  unsigned pointer_chase_seed = resolveSeed(
+      args.pointer_chase_seed_given,
+      args.pointer_chase_seed_arg,
+      kFeedsimDefaultSeed);
 
   this_thread.page_ranker = std::make_unique<ranking::dwarfs::PageRank>(
       std::move(graph), args.cpu_threads_arg, page_rank_seed);

packages/feedsim/third_party/src/workloads/ranking/LeafNodeRankCmdline.ggo

@@ -35,9 +35,9 @@ option "max_response_size" - "Maximum response size in bytes returned by the lea
 option "compression_data_size" - "Number of bytes to compress per request." int default="131072"
 option "rank_trials_per_thread" - "Number of iterations each CPU thread executes of rank work." int default="1"
 option "min_icache_iterations" - "At least this number of icache busting iteration will be executed." int default="0"
-option "node_rank_seed" - "Seed for random number generator. If not provided, current time will be used." long optional
-option "page_rank_seed" - "Seed for PageRank random number generator. If not provided, current time will be used." long optional
-option "pointer_chase_seed" - "Seed for PointerChase random number generator. If not provided, current time will be used." long optional
+option "node_rank_seed" - "Seed for random number generator. Default 42 for deterministic runs; pass -1 for a time-based (non-deterministic) seed." long optional
+option "page_rank_seed" - "Seed for PageRank random number generator. Default 42 for deterministic runs; pass -1 for a time-based (non-deterministic) seed." long optional
+option "pointer_chase_seed" - "Seed for PointerChase random number generator. Default 42 for deterministic runs; pass -1 for a time-based (non-deterministic) seed." long optional
 option "chase_iterations" - "Number of chases to execute on handler thread." int default="5120"
 option "io_chase_iterations" - "Number of chases to execute on IO threads." int default="5120"
 option "io_time_ms" - "Milliseconds to sleep emualting I/O offcpu." int default="200"

packages/feedsim/third_party/src/workloads/ranking/mock_services/MockServiceHandler.cc

@@ -15,7 +15,9 @@
 #include "MockServiceHandler.h"
 
 #include <algorithm>
+#include <atomic>
 #include <chrono>
+#include <cstdlib>
 #include <cstring>
 #include <random>
 #include <thread>
@@ -84,7 +86,24 @@ constexpr int32_t kSpinThresholdUs = 200;
 // or buggy client from triggering bad_alloc on small-RAM hosts.
 constexpr uint32_t kMaxResponseSize = 16 * 1024 * 1024;
 
-thread_local std::mt19937 tlRng{std::random_device{}()};
+// Deterministic per-thread RNG seed. This thread_local RNG (mock response-body
+// generation) used to be seeded from std::random_device, making benchmark
+// output non-reproducible run-to-run. Seed from a fixed base mixed with a
+// per-thread counter so each thread keeps an independent but reproducible
+// sequence. Set FEEDSIM_RNG_RANDOM=1 to restore the old non-deterministic seed.
+static unsigned detRngSeed(unsigned base) {
+  static const bool kRandom = [] {
+    const char* e = std::getenv("FEEDSIM_RNG_RANDOM");
+    return e != nullptr && e[0] == '1';
+  }();
+  if (kRandom) {
+    return std::random_device{}();
+  }
+  static std::atomic<unsigned> ctr{0};
+  return base + ctr.fetch_add(1) * 2654435761u;
+}
+
+thread_local std::mt19937 tlRng{detRngSeed(0x5EED3110u)};
 
 // Free function so async continuations can use it without capturing `this`.
 // Takes a shared_ptr by value so the SilesiaLoader outlives the continuation