MB-70281: Add fusion max pending upload bytes to memcached config

This patch adds the following two fusion configuration parameters to
memcached settings along with a stat to expose each one:
    * fusion_max_pending_upload_bytes
    * fusion_max_pending_upload_bytes_lwm_ratio

Also, expose fusion_pending_upload_bytes as a memcached global stat, which is retrieved from magma's GetFusionPendingUploadBytes() API.

Change-Id: If55c029c6f81b4d16f410cde7983343b24e6e628
Reviewed-on: https://review.couchbase.org/c/kv_engine/+/239483
Tested-by: Build Bot <build@couchbase.com>
Reviewed-by: Paolo Cocchi <paolo.cocchi@couchbase.com>

Author: mohammadzaeem

daemon/memcached.cc

@@ -373,6 +373,20 @@ static void settings_init() {
                 magma::Magma::SetNumThreads(magma::Magma::FusionMigrator,
                                             s.getFusionNumMigratorThreads());
             });
+    settings.addChangeListener(
+            "fusion_max_pending_upload_bytes",
+            [&settings](const auto&, auto& s) {
+                magma::Magma::SetFusionPendingUploadThresholds(
+                        s.getFusionMaxPendingUploadBytes(),
+                        settings.getFusionMaxPendingUploadBytesLwmRatio());
+            });
+    settings.addChangeListener(
+            "fusion_max_pending_upload_bytes_lwm_ratio",
+            [&settings](const auto&, auto& s) {
+                magma::Magma::SetFusionPendingUploadThresholds(
+                        settings.getFusionMaxPendingUploadBytes(),
+                        s.getFusionMaxPendingUploadBytesLwmRatio());
+            });
 }
 
 /**

daemon/settings.cc

@@ -420,6 +420,23 @@ void Settings::reconfigure(const nlohmann::json& json) {
                                 {"value", value.dump()},
                                 {"reason", "Fusion support is not enabled"});
             }
+        } else if (key == "fusion_max_pending_upload_bytes") {
+            if (isFusionSupportEnabled()) {
+                setFusionMaxPendingUploadBytes(value.get<size_t>());
+            } else {
+                LOG_WARNING_CTX("Ignore fusion_max_pending_upload_bytes",
+                                {"value", value.dump()},
+                                {"reason", "Fusion support is not enabled"});
+            }
+        } else if (key == "fusion_max_pending_upload_bytes_lwm_ratio") {
+            if (isFusionSupportEnabled()) {
+                setFusionMaxPendingUploadBytesLwmRatio(value.get<double>());
+            } else {
+                LOG_WARNING_CTX(
+                        "Ignore fusion_max_pending_upload_bytes_lwm_ratio",
+                        {"value", value.dump()},
+                        {"reason", "Fusion support is not enabled"});
+            }
         } else if (key == "phosphor_config"sv) {
             auto config = value.get<std::string>();
             // throw an exception if the config is invalid
@@ -1117,6 +1134,29 @@ void Settings::updateSettings(const Settings& other, bool apply) {
         }
     }
 
+    if (other.has.fusion_max_pending_upload_bytes) {
+        if (other.getFusionMaxPendingUploadBytes() !=
+            getFusionMaxPendingUploadBytes()) {
+            LOG_INFO_CTX("Change fusion max pending upload bytes",
+                         {"from", getFusionMaxPendingUploadBytes()},
+                         {"to", other.getFusionMaxPendingUploadBytes()});
+            setFusionMaxPendingUploadBytes(
+                    other.getFusionMaxPendingUploadBytes());
+        }
+    }
+
+    if (other.has.fusion_max_pending_upload_bytes_lwm_ratio) {
+        if (other.getFusionMaxPendingUploadBytesLwmRatio() !=
+            getFusionMaxPendingUploadBytesLwmRatio()) {
+            LOG_INFO_CTX(
+                    "Change fusion max pending upload bytes LWM ratio",
+                    {"from", getFusionMaxPendingUploadBytesLwmRatio()},
+                    {"to", other.getFusionMaxPendingUploadBytesLwmRatio()});
+            setFusionMaxPendingUploadBytesLwmRatio(
+                    other.getFusionMaxPendingUploadBytesLwmRatio());
+        }
+    }
+
     if (other.has.num_reader_threads &&
         other.getNumReaderThreads() != getNumReaderThreads()) {
         LOG_INFO_CTX("Change number of reader threads",

daemon/settings.h

@@ -363,6 +363,28 @@ class Settings {
         notify_changed("fusion_num_migrator_threads");
     }
 
+    size_t getFusionMaxPendingUploadBytes() const {
+        return fusion_max_pending_upload_bytes.load(std::memory_order_acquire);
+    }
+
+    void setFusionMaxPendingUploadBytes(size_t value) {
+        fusion_max_pending_upload_bytes.store(value, std::memory_order_release);
+        has.fusion_max_pending_upload_bytes = true;
+        notify_changed("fusion_max_pending_upload_bytes");
+    }
+
+    double getFusionMaxPendingUploadBytesLwmRatio() const {
+        return fusion_max_pending_upload_bytes_lwm_ratio.load(
+                std::memory_order_acquire);
+    }
+
+    void setFusionMaxPendingUploadBytesLwmRatio(double value) {
+        fusion_max_pending_upload_bytes_lwm_ratio.store(
+                value, std::memory_order_release);
+        has.fusion_max_pending_upload_bytes_lwm_ratio = true;
+        notify_changed("fusion_max_pending_upload_bytes_lwm_ratio");
+    }
+
     size_t getMaxUserConnections() const {
         return getMaxConnections() - getSystemConnections();
     }
@@ -1435,6 +1457,14 @@ class Settings {
     // The number of Fusion Migrator threads
     std::atomic<size_t> fusion_num_migrator_threads{4};
 
+    // The maximum number of pending upload bytes to be synced across all
+    // volumes
+    std::atomic<size_t> fusion_max_pending_upload_bytes{0};
+
+    // The proportion of max_pending_upload_bytes beyond which syncs for volumes
+    // with the highest pending bytes are only allowed.
+    std::atomic<double> fusion_max_pending_upload_bytes_lwm_ratio{0.6};
+
     /**
      * Note that it is not safe to add new listeners after we've spun up
      * new threads as we don't try to lock the object.
@@ -1704,6 +1734,8 @@ class Settings {
         bool fusion_sync_rate_limit = false;
         bool fusion_num_uploader_threads = false;
         bool fusion_num_migrator_threads = false;
+        bool fusion_max_pending_upload_bytes = false;
+        bool fusion_max_pending_upload_bytes_lwm_ratio = false;
         bool num_reader_threads = false;
         bool num_writer_threads = false;
         bool num_auxio_threads = false;

daemon/settings_test.cc

@@ -1014,6 +1014,32 @@ TEST(SettingsUpdateTest, FusionNumMigratorThreads) {
     EXPECT_EQ(4, settings.getFusionNumMigratorThreads());
 }
 
+TEST(SettingsUpdateTest, FusionMaxPendingUploadBytes) {
+    Settings updated;
+    Settings settings;
+    settings.setFusionMaxPendingUploadBytes(10_MiB);
+    // Setting to the same value succeeds
+    updated.setFusionMaxPendingUploadBytes(10_MiB);
+    settings.updateSettings(updated, false);
+    // Setting to a different value succeeds
+    updated.setFusionMaxPendingUploadBytes(50_MiB);
+    settings.updateSettings(updated, true);
+    EXPECT_EQ(50_MiB, settings.getFusionMaxPendingUploadBytes());
+}
+
+TEST(SettingsUpdateTest, FusionMaxPendingUploadBytesLwmRatio) {
+    Settings updated;
+    Settings settings;
+    settings.setFusionMaxPendingUploadBytesLwmRatio(0.7);
+    // Setting to the same value succeeds
+    updated.setFusionMaxPendingUploadBytesLwmRatio(0.7);
+    settings.updateSettings(updated, false);
+    // Setting to a different value succeeds
+    updated.setFusionMaxPendingUploadBytesLwmRatio(0.5);
+    settings.updateSettings(updated, true);
+    EXPECT_EQ(0.5, settings.getFusionMaxPendingUploadBytesLwmRatio());
+}
+
 TEST(SettingsUpdateTest, DefaultReqIsDynamic) {
     Settings updated;
     Settings settings;

daemon/stats.cc

@@ -81,6 +81,17 @@ static void server_global_stats(const StatCollector& collector) {
             collector.addStat(
                     Key::fusion_num_migrator_threads,
                     magma::Magma::GetNumThreads(magma::Magma::FusionMigrator));
+            const auto [maxPendingUploadBytes, lwmRatio] =
+                    magma::Magma::GetFusionPendingUploadThresholds();
+            collector.addStat(Key::fusion_max_pending_upload_bytes,
+                              maxPendingUploadBytes);
+
+            // The lwm ratio will return 0 if maxPendingUploadBytes is 0,
+            // else the ratio configured
+            collector.addStat(Key::fusion_max_pending_upload_bytes_lwm_ratio,
+                              lwmRatio);
+            collector.addStat(Key::fusion_pending_upload_bytes,
+                              magma::Magma::GetFusionPendingUploadBytes());
         }
 
         auto sdks = SdkConnectionManager::instance().getConnectedSdks();

statistics/stat_definitions.json

@@ -2095,6 +2095,24 @@
         "added": "8.0.0"
 
     },
+    {
+        "key": "fusion_max_pending_upload_bytes",
+        "unit": "bytes",
+        "description": "The maximum number of pending upload bytes to be synced across all volumes.",
+        "added": "8.1.0"
+    },
+    {
+        "key": "fusion_max_pending_upload_bytes_lwm_ratio",
+        "unit": "none",
+        "description": "The proportion of max_pending_upload_bytes beyond which syncs for volumes with highest pending bytes are only allowed.",
+        "added": "8.1.0"
+    },
+    {
+        "key": "fusion_pending_upload_bytes",
+        "unit": "bytes",
+        "description": "The total pending upload bytes across all fusion volumes.",
+        "added": "8.1.0"
+    },
     /* EPBucket::getFileStats */
     {
         "key": "ep_db_data_size",

tests/testapp/testapp.cc

@@ -560,6 +560,8 @@ nlohmann::json TestappTest::generate_config() {
         ret["fusion_sync_rate_limit"] = 75_MiB;
         ret["fusion_num_uploader_threads"] = 1;
         ret["fusion_num_migrator_threads"] = 1;
+        ret["fusion_max_pending_upload_bytes"] = 0;
+        ret["fusion_max_pending_upload_bytes_lwm_ratio"] = 0.6;
     }
 
     return ret;