Benchmark test for serving VOQ stats subscription

Summary:
Adding a benchmark test for fsdb server memory and cpu usage when serving path
subscription for 64K VOQs stats.
Running test with 100 stats publish and subscription serve iterations after
initial pub/sub streams are established and initial sync is completed.

Differential Revision: D66718104

fbshipit-source-id: 4f0773eaa086c7381041500d16d252db8d4e35a1

fboss/fsdb/benchmarks/BUCK

@@ -43,6 +43,7 @@ cpp_benchmark(
         "//folly/json:dynamic",
         "//folly/logging:init",
         "//folly/logging:logging",
+        "//folly/synchronization:baton",
     ],
     external_deps = [
         "gflags",

fboss/fsdb/benchmarks/FsdbBenchmarkTestHelper.cpp

@@ -9,13 +9,16 @@ const thriftpath::RootThriftPath<facebook::fboss::fsdb::FsdbOperStateRoot>
 const thriftpath::RootThriftPath<facebook::fboss::fsdb::FsdbOperStatsRoot>
     statsRoot;
 const std::vector<std::string> kPublishRoot{"agent"};
+const uint32_t kStateServeIntervalMs = 50;
+const uint32_t kStatsServeIntervalMs = 50;
 
 } // anonymous namespace
 
 namespace facebook::fboss::fsdb::test {
 
 void FsdbBenchmarkTestHelper::setup() {
-  fsdbTestServer_ = std::make_unique<fsdb::test::FsdbTestServer>();
+  fsdbTestServer_ = std::make_unique<fsdb::test::FsdbTestServer>(
+      0, kStateServeIntervalMs, kStatsServeIntervalMs);
   FLAGS_fsdbPort = fsdbTestServer_->getFsdbPort();
   const std::string clientId = "agent";
   pubsubMgr_ = std::make_unique<fsdb::FsdbPubSubManager>(clientId);
@@ -78,4 +81,29 @@ std::vector<std::string> FsdbBenchmarkTestHelper::getAgentStatsPath() {
   return statsRoot.agent().tokens();
 }
 
+void FsdbBenchmarkTestHelper::addSubscription(
+    const fsdb::FsdbStateSubscriber::FsdbOperStateUpdateCb& stateUpdateCb) {
+  auto stateChangeCb = [this](
+                           fsdb::SubscriptionState /*oldState*/,
+                           fsdb::SubscriptionState newState,
+                           std::optional<bool> /*initialSyncHasData*/) {
+    if (newState == fsdb::SubscriptionState::CONNECTED) {
+      subscriptionConnected_.store(true);
+    } else {
+      subscriptionConnected_.store(false);
+    }
+  };
+  // agent is Path publisher for stats
+  pubsubMgr_->addStatPathSubscription(
+      getAgentStatsPath(), stateChangeCb, stateUpdateCb);
+}
+
+bool FsdbBenchmarkTestHelper::isSubscriptionConnected() {
+  return subscriptionConnected_.load();
+}
+
+void FsdbBenchmarkTestHelper::removeSubscription() {
+  pubsubMgr_->removeStatPathSubscription(getAgentStatsPath());
+}
+
 } // namespace facebook::fboss::fsdb::test

fboss/fsdb/benchmarks/FsdbBenchmarkTestHelper.h

@@ -25,6 +25,10 @@ class FsdbBenchmarkTestHelper {
   fsdb::test::FsdbTestServer& testServer() {
     return *fsdbTestServer_;
   }
+  void addSubscription(
+      const fsdb::FsdbStateSubscriber::FsdbOperStateUpdateCb& stateUpdateCb);
+  void removeSubscription();
+  bool isSubscriptionConnected();
 
  private:
   std::vector<std::string> getAgentStatsPath();
@@ -33,6 +37,7 @@ class FsdbBenchmarkTestHelper {
   std::unique_ptr<fsdb::FsdbPubSubManager> pubsubMgr_;
   folly::Baton<> publisherConnected_;
   std::atomic_bool readyForPublishing_ = false;
+  std::atomic_bool subscriptionConnected_ = false;
 };
 
 } // namespace facebook::fboss::fsdb::test

fboss/fsdb/benchmarks/VoqCounterScaleBench.cpp

@@ -4,12 +4,17 @@
 #include <folly/logging/xlog.h>
 #include <gflags/gflags.h>
 
+#include <folly/synchronization/Baton.h>
 #include "fboss/fsdb/benchmarks/FsdbBenchmarkTestHelper.h"
 #include "fboss/fsdb/benchmarks/StateGenerator.h"
 
 DEFINE_int32(n_sysports, 16000, "number of SysPorts");
 DEFINE_int32(n_voqs_per_sysport, 4, "number of VOQs per SysPort");
 DEFINE_int32(n_publish_iters, 1000, "number of iterations for stats publish");
+DEFINE_int32(
+    n_pubsub_iters,
+    100,
+    "number of iterations for stats publish and serve");
 
 namespace facebook::fboss::fsdb::test {
 
@@ -55,4 +60,52 @@ BENCHMARK(FsdbPublishVoqStats) {
   helper.TearDown();
 }
 
+BENCHMARK(FsdbSubscribeVoqStats) {
+  auto n_iterations = FLAGS_n_pubsub_iters;
+
+  folly::BenchmarkSuspender suspender;
+
+  // setup FsdbTestServer
+  FsdbBenchmarkTestHelper helper;
+  helper.setup();
+
+  // start publisher and publish initial empty stats
+  helper.startPublisher();
+
+  auto stats = std::make_shared<AgentStats>();
+  helper.publishPath(*stats, 0);
+  helper.waitForPublisherConnected();
+
+  // setup subscription and signal when updates are received
+  folly::Baton<> updateReceived;
+
+  fsdb::FsdbStateSubscriber::FsdbOperStateUpdateCb subscriptionCb =
+      [&](fsdb::OperState&& /*operState*/) { updateReceived.post(); };
+  helper.addSubscription(subscriptionCb);
+
+  // complete initial sync of full stats before starting benchmark loop
+  StateGenerator::fillVoqStats(
+      stats.get(), FLAGS_n_sysports, FLAGS_n_voqs_per_sysport);
+  uint64_t lastPublishedStamp = 0;
+  helper.publishPath(*stats, ++lastPublishedStamp);
+
+  updateReceived.wait();
+  updateReceived.reset();
+
+  // start benchmark for incremental publish and subscription serve
+  suspender.dismiss();
+
+  for (int round = 0; round < n_iterations; round++) {
+    StateGenerator::updateVoqStats(stats.get());
+    helper.publishPath(*stats, ++lastPublishedStamp);
+    updateReceived.wait();
+    updateReceived.reset();
+  }
+
+  suspender.rehire();
+
+  helper.removeSubscription();
+  helper.TearDown();
+}
+
 } // namespace facebook::fboss::fsdb::test