Refactor queue size as parameter

Priyank Warkhede · meta-codesync[bot] · commit 5836d547e4dd · 2025-11-03T10:12:53.000-08:00
Summary:
For ease of review, separating this no-op refactoring into separate diff.

Refactor subscription serve queue size to be passed as a parameter while
creating Subscription object.
Also adding queue sizes for Path and default (non-path) subscriptions in
StorageParams.

In next diff we will use this to choose different queue size for serving
stats subscriptions.

Differential Revision: D85965178

fbshipit-source-id: a87c2cfe1efbd8a6498e2677f21264ec3a2bff21
diff --git a/fboss/fsdb/oper/NaivePeriodicSubscribableStorageBase.cpp b/fboss/fsdb/oper/NaivePeriodicSubscribableStorageBase.cpp
@@ -30,6 +30,21 @@ DEFINE_bool(
     serveHeartbeats,
     false,
     "Whether or not to serve hearbeats in subscription streams");
+// default queue size for subscription serve updates
+// is chosen to be large enough so that in prod we
+// should not see any dropped updates. This value will
+// be tuned to lower value after monitoring and max-scale
+// testing.
+// Rationale for choice of this specific large value:
+// * minimum enqueue interval is state subscription serve interval (50msec).
+// * Thrift streams holds ~100 updates before pipe queue starts building up.
+// So with 1024 default queue size, pipe can get full
+// only in the exceptional scenario where subscriber does not
+// read any update for > 1 min.
+DEFINE_int32(
+    subscriptionServeQueueSize,
+    1024,
+    "Max subscription serve updates to queue, default 1024");
 
 namespace facebook::fboss::fsdb {
 
@@ -437,7 +452,8 @@ NaivePeriodicSubscribableStorageBase::subscribe_encoded_impl(
       protocol,
       getPublisherRoot(path.begin(), path.end()),
       heartbeatThread_ ? heartbeatThread_->getEventBase() : nullptr,
-      heartbeatInterval);
+      heartbeatInterval,
+      params_.pathSubscriptionServeQueueSize_);
   subMgr().registerSubscription(std::move(subscription));
   return std::move(gen);
 }
@@ -462,7 +478,8 @@ NaivePeriodicSubscribableStorageBase::subscribe_delta_impl(
       protocol,
       getPublisherRoot(path.begin(), path.end()),
       heartbeatThread_ ? heartbeatThread_->getEventBase() : nullptr,
-      heartbeatInterval);
+      heartbeatInterval,
+      params_.defaultSubscriptionServeQueueSize_);
   subMgr().registerSubscription(std::move(subscription));
   return std::move(gen);
 }
@@ -486,7 +503,8 @@ NaivePeriodicSubscribableStorageBase::subscribe_encoded_extended_impl(
       std::move(publisherRoot),
       protocol,
       heartbeatThread_ ? heartbeatThread_->getEventBase() : nullptr,
-      heartbeatInterval);
+      heartbeatInterval,
+      params_.pathSubscriptionServeQueueSize_);
   subMgr().registerExtendedSubscription(std::move(subscription));
   return std::move(gen);
 }
@@ -510,7 +528,8 @@ NaivePeriodicSubscribableStorageBase::subscribe_delta_extended_impl(
       std::move(publisherRoot),
       protocol,
       heartbeatThread_ ? heartbeatThread_->getEventBase() : nullptr,
-      heartbeatInterval);
+      heartbeatInterval,
+      params_.defaultSubscriptionServeQueueSize_);
   subMgr().registerExtendedSubscription(std::move(subscription));
   return std::move(gen);
 }
@@ -536,7 +555,8 @@ NaivePeriodicSubscribableStorageBase::subscribe_patch_impl(
       patchOperProtocol_,
       std::move(root),
       heartbeatThread_ ? heartbeatThread_->getEventBase() : nullptr,
-      heartbeatInterval);
+      heartbeatInterval,
+      params_.defaultSubscriptionServeQueueSize_);
   subMgr().registerExtendedSubscription(std::move(subscription));
   return std::move(gen);
 }
@@ -562,7 +582,8 @@ NaivePeriodicSubscribableStorageBase::subscribe_patch_extended_impl(
       patchOperProtocol_,
       std::move(root),
       heartbeatThread_ ? heartbeatThread_->getEventBase() : nullptr,
-      heartbeatInterval);
+      heartbeatInterval,
+      params_.defaultSubscriptionServeQueueSize_);
   subMgr().registerExtendedSubscription(std::move(subscription));
   return std::move(gen);
 }
diff --git a/fboss/fsdb/oper/NaivePeriodicSubscribableStorageBase.h b/fboss/fsdb/oper/NaivePeriodicSubscribableStorageBase.h
@@ -17,6 +17,7 @@
 
 DECLARE_int32(storage_thread_heartbeat_ms);
 DECLARE_bool(serveHeartbeats);
+DECLARE_int32(subscriptionServeQueueSize);
 
 namespace facebook::fboss::fsdb {
 
@@ -55,7 +56,11 @@ class NaivePeriodicSubscribableStorageBase {
         bool convertToIDPaths = false,
         bool requireResponseOnInitialSync = false,
         bool exportPerSubscriberMetrics = false,
-        bool serveGetRequestsWithLastPublishedState = true)
+        bool serveGetRequestsWithLastPublishedState = true,
+        int32_t pathSubscriptionServeQueueSize =
+            FLAGS_subscriptionServeQueueSize,
+        int32_t defaultSubscriptionServeQueueSize =
+            FLAGS_subscriptionServeQueueSize)
         : subscriptionServeInterval_(subscriptionServeInterval),
           subscriptionHeartbeatInterval_(subscriptionHeartbeatInterval),
           trackMetadata_(trackMetadata),
@@ -64,7 +69,10 @@ class NaivePeriodicSubscribableStorageBase {
           requireResponseOnInitialSync_(requireResponseOnInitialSync),
           exportPerSubscriberMetrics_(exportPerSubscriberMetrics),
           serveGetRequestsWithLastPublishedState_(
-              serveGetRequestsWithLastPublishedState) {}
+              serveGetRequestsWithLastPublishedState),
+          pathSubscriptionServeQueueSize_(pathSubscriptionServeQueueSize),
+          defaultSubscriptionServeQueueSize_(
+              defaultSubscriptionServeQueueSize) {}
 
     StorageParams& setServeGetRequestsWithLastPublishedState(bool val) {
       serveGetRequestsWithLastPublishedState_ = val;
@@ -79,6 +87,8 @@ class NaivePeriodicSubscribableStorageBase {
     const bool requireResponseOnInitialSync_;
     const bool exportPerSubscriberMetrics_;
     bool serveGetRequestsWithLastPublishedState_;
+    const int32_t pathSubscriptionServeQueueSize_;
+    const int32_t defaultSubscriptionServeQueueSize_;
   };
 
   explicit NaivePeriodicSubscribableStorageBase(
diff --git a/fboss/fsdb/oper/Subscription.cpp b/fboss/fsdb/oper/Subscription.cpp
@@ -7,22 +7,6 @@
 #include <folly/coro/Sleep.h>
 #include <optional>
 
-// default queue size for subscription serve updates
-// is chosen to be large enough so that in prod we
-// should not see any dropped updates. This value will
-// be tuned to lower value after monitoring and max-scale
-// testing.
-// Rationale for choice of this specific large value:
-// * minimum enqueue interval is state subscription serve interval (50msec).
-// * Thrift streams holds ~100 updates before pipe queue starts building up.
-// So with 1024 default queue size, pipe can get full
-// only in the exceptional scenario where subscriber does not
-// read any update for > 1 min.
-DEFINE_int32(
-    subscriptionServeQueueSize,
-    1024,
-    "Max subscription serve updates to queue, default 1024");
-
 DEFINE_bool(
     forceCloseSlowSubscriber,
     true,
@@ -129,9 +113,10 @@ DeltaSubscription::create(
     OperProtocol protocol,
     std::optional<std::string> publisherRoot,
     folly::EventBase* heartbeatEvb,
-    std::chrono::milliseconds heartbeatInterval) {
-  auto [generator, pipe] = folly::coro::BoundedAsyncPipe<OperDelta>::create(
-      FLAGS_subscriptionServeQueueSize);
+    std::chrono::milliseconds heartbeatInterval,
+    int32_t pipeCapacity) {
+  auto [generator, pipe] =
+      folly::coro::BoundedAsyncPipe<OperDelta>::create(pipeCapacity);
   std::vector<std::string> path(begin, end);
   auto subscription = std::make_unique<DeltaSubscription>(
       std::move(subscriber),
@@ -302,9 +287,10 @@ ExtendedPathSubscription::create(
     std::optional<std::string> publisherRoot,
     OperProtocol protocol,
     folly::EventBase* heartbeatEvb,
-    std::chrono::milliseconds heartbeatInterval) {
-  auto [generator, pipe] = folly::coro::BoundedAsyncPipe<gen_type>::create(
-      FLAGS_subscriptionServeQueueSize);
+    std::chrono::milliseconds heartbeatInterval,
+    int32_t pipeCapacity) {
+  auto [generator, pipe] =
+      folly::coro::BoundedAsyncPipe<gen_type>::create(pipeCapacity);
   auto subscription = std::make_shared<ExtendedPathSubscription>(
       std::move(subscriber),
       makeSimplePathMap(paths),
@@ -365,9 +351,10 @@ ExtendedDeltaSubscription::create(
     std::optional<std::string> publisherRoot,
     OperProtocol protocol,
     folly::EventBase* heartbeatEvb,
-    std::chrono::milliseconds heartbeatInterval) {
-  auto [generator, pipe] = folly::coro::BoundedAsyncPipe<gen_type>::create(
-      FLAGS_subscriptionServeQueueSize);
+    std::chrono::milliseconds heartbeatInterval,
+    int32_t pipeCapacity) {
+  auto [generator, pipe] =
+      folly::coro::BoundedAsyncPipe<gen_type>::create(pipeCapacity);
   auto subscription = std::make_shared<ExtendedDeltaSubscription>(
       std::move(subscriber),
       makeSimplePathMap(paths),
@@ -480,7 +467,8 @@ ExtendedPatchSubscription::create(
     OperProtocol protocol,
     std::optional<std::string> publisherRoot,
     folly::EventBase* heartbeatEvb,
-    std::chrono::milliseconds heartbeatInterval) {
+    std::chrono::milliseconds heartbeatInterval,
+    int32_t pipeCapacity) {
   RawOperPath p;
   p.path() = std::move(path);
   return create(
@@ -489,7 +477,8 @@ ExtendedPatchSubscription::create(
       std::move(protocol),
       std::move(publisherRoot),
       std::move(heartbeatEvb),
-      std::move(heartbeatInterval));
+      std::move(heartbeatInterval),
+      pipeCapacity);
 }
 
 std::pair<
@@ -501,7 +490,8 @@ ExtendedPatchSubscription::create(
     OperProtocol protocol,
     std::optional<std::string> publisherRoot,
     folly::EventBase* heartbeatEvb,
-    std::chrono::milliseconds heartbeatInterval) {
+    std::chrono::milliseconds heartbeatInterval,
+    int32_t pipeCapacity) {
   std::map<SubscriptionKey, ExtendedOperPath> extendedPaths;
   for (auto& [key, path] : paths) {
     std::vector<OperPathElem> extendedPath;
@@ -517,7 +507,8 @@ ExtendedPatchSubscription::create(
       std::move(protocol),
       std::move(publisherRoot),
       std::move(heartbeatEvb),
-      std::move(heartbeatInterval));
+      std::move(heartbeatInterval),
+      pipeCapacity);
 }
 
 std::pair<
@@ -529,9 +520,10 @@ ExtendedPatchSubscription::create(
     OperProtocol protocol,
     std::optional<std::string> publisherRoot,
     folly::EventBase* heartbeatEvb,
-    std::chrono::milliseconds heartbeatInterval) {
-  auto [generator, pipe] = folly::coro::BoundedAsyncPipe<gen_type>::create(
-      FLAGS_subscriptionServeQueueSize);
+    std::chrono::milliseconds heartbeatInterval,
+    int32_t pipeCapacity) {
+  auto [generator, pipe] =
+      folly::coro::BoundedAsyncPipe<gen_type>::create(pipeCapacity);
   auto subscription = std::make_unique<ExtendedPatchSubscription>(
       std::move(subscriber),
       std::move(paths),
diff --git a/fboss/fsdb/oper/Subscription.h b/fboss/fsdb/oper/Subscription.h
@@ -17,7 +17,6 @@
 #include <folly/io/async/EventBase.h>
 #include <folly/json/dynamic.h>
 
-DECLARE_int32(subscriptionServeQueueSize);
 DECLARE_bool(forceCloseSlowSubscriber);
 
 namespace facebook::fboss::fsdb {
@@ -307,9 +306,10 @@ class PathSubscription : public BasePathSubscription,
       OperProtocol protocol,
       std::optional<std::string> publisherRoot,
       folly::EventBase* heartbeatEvb,
-      std::chrono::milliseconds heartbeatInterval) {
-    auto [generator, pipe] = folly::coro::BoundedAsyncPipe<value_type>::create(
-        FLAGS_subscriptionServeQueueSize);
+      std::chrono::milliseconds heartbeatInterval,
+      int32_t pipeCapacity) {
+    auto [generator, pipe] =
+        folly::coro::BoundedAsyncPipe<value_type>::create(pipeCapacity);
     std::vector<std::string> path(begin, end);
     auto subscription = std::make_unique<PathSubscription>(
         std::move(subscriber),
@@ -435,7 +435,8 @@ class DeltaSubscription : public BaseDeltaSubscription,
       OperProtocol protocol,
       std::optional<std::string> publisherRoot,
       folly::EventBase* heartbeatEvb,
-      std::chrono::milliseconds heartbeatInterval);
+      std::chrono::milliseconds heartbeatInterval,
+      int32_t pipeCapacity);
 
   DeltaSubscription(
       SubscriptionIdentifier&& subscriber,
@@ -528,7 +529,8 @@ class ExtendedPathSubscription : public ExtendedSubscription,
       std::optional<std::string> publisherRoot,
       OperProtocol protocol,
       folly::EventBase* heartbeatEvb,
-      std::chrono::milliseconds heartbeatInterval);
+      std::chrono::milliseconds heartbeatInterval,
+      int32_t pipeCapacity);
 
   bool shouldConvertToDynamic() const override {
     return false;
@@ -620,7 +622,8 @@ class ExtendedDeltaSubscription : public ExtendedSubscription,
       std::optional<std::string> publisherRoot,
       OperProtocol protocol,
       folly::EventBase* heartbeatEvb,
-      std::chrono::milliseconds heartbeatInterval);
+      std::chrono::milliseconds heartbeatInterval,
+      int32_t pipeCapacity);
 
   void buffer(TaggedOperDelta&& newVal);
 
@@ -718,7 +721,8 @@ class ExtendedPatchSubscription : public ExtendedSubscription,
       OperProtocol protocol,
       std::optional<std::string> publisherRoot,
       folly::EventBase* heartbeatEvb,
-      std::chrono::milliseconds heartbeatInterval);
+      std::chrono::milliseconds heartbeatInterval,
+      int32_t pipeCapacity);
 
   // Multipath
   static std::pair<
@@ -730,7 +734,8 @@ class ExtendedPatchSubscription : public ExtendedSubscription,
       OperProtocol protocol,
       std::optional<std::string> publisherRoot,
       folly::EventBase* heartbeatEvb,
-      std::chrono::milliseconds heartbeatInterval);
+      std::chrono::milliseconds heartbeatInterval,
+      int32_t pipeCapacity);
 
   // Extended paths
   static std::pair<
@@ -742,7 +747,8 @@ class ExtendedPatchSubscription : public ExtendedSubscription,
       OperProtocol protocol,
       std::optional<std::string> publisherRoot,
       folly::EventBase* heartbeatEvb,
-      std::chrono::milliseconds heartbeatInterval);
+      std::chrono::milliseconds heartbeatInterval,
+      int32_t pipeCapacity);
 
   ExtendedPatchSubscription(
       SubscriptionIdentifier&& subscriber,
diff --git a/fboss/fsdb/oper/tests/SubscriptionTests.cpp b/fboss/fsdb/oper/tests/SubscriptionTests.cpp
@@ -9,6 +9,8 @@
 
 namespace facebook::fboss::fsdb::test {
 
+constexpr int32_t kSubscriptionServeQueueSize = 100;
+
 namespace {
 template <typename Gen>
 folly::coro::Task<typename Gen::value_type> consumeOne(Gen& generator) {
@@ -35,7 +37,8 @@ class SubscriptionTests : public ::testing::Test {
           OperProtocol::BINARY,
           std::nullopt,
           heartbeatThread_->getEventBase(),
-          std::chrono::milliseconds(100));
+          std::chrono::milliseconds(100),
+          kSubscriptionServeQueueSize);
     } else {
       return SubscriptionT::create(
           SubscriptionIdentifier("test-sub"),
@@ -44,7 +47,8 @@ class SubscriptionTests : public ::testing::Test {
           OperProtocol::BINARY,
           std::nullopt,
           heartbeatThread_->getEventBase(),
-          std::chrono::milliseconds(100));
+          std::chrono::milliseconds(100),
+          kSubscriptionServeQueueSize);
     }
   }
 
