Merge "Merge branch 'neo' into 'master'"

Author: Gerrit Code Review

daemon/mc_time.cc

@@ -236,10 +236,12 @@ void mc_time_clock_tick() {
             if (cb::logger::get() != nullptr) {
                 /* log all variables used in time calculations */
                 LOG_WARNING(
-                        "system clock changed? difference = {}, "
+                        "system clock changed? Expected delta of {}s ±1 since "
+                        "last check, actual difference = {}s, "
                         "memcached_epoch = {}, "
                         "memcached_uptime = {}, new memcached_epoch = {}, "
                         "next check {}",
+                        memcached_check_system_time,
                         difference,
                         memcached_epoch.load(),
                         memcached_uptime.load(),

engines/ep/configuration.json

@@ -766,7 +766,7 @@
         },
         "max_checkpoints": {
             "default": "10",
-            "descr": "Desired/expected max number of checkpoints in each VBucket for balanced systems where the load is approximately evenly spread across vbuckets. That is NOT a hard limit (see also max_checkpoints_hard_limit_multiplier). Directly used in the auto-computation for the actual checkpoint_max_size.",
+            "descr": "Max number of checkpoints in each VBucket.",
             "dynamic": true,
             "type": "size_t",
             "validator": {
@@ -775,17 +775,6 @@
                 }
             }
         },
-        "max_checkpoints_hard_limit_multiplier" : {
-            "default": "100",
-            "descr": "Defines the actual max checkpoints (per vbucket) hard limit as (max_checkpoints * multiplier). The multiplier doesn't affect the auto checkpoint_max_size computation.",
-            "dynamic": true,
-            "type": "size_t",
-            "validator": {
-                "range": {
-                    "min": 1
-                }
-            }
-        },
         "max_failover_entries": {
             "default": "25",
             "descr": "maximum number of failover log entries",

engines/ep/management/cbepctl

@@ -198,8 +198,7 @@ Available params for "set":
     item_num_based_new_chk       - true if a new checkpoint can be created based
                                    on.
                                    the number of items in the open checkpoint.
-    max_checkpoints              - Desired/expected max number of checkpoints in each VBucket for balanced systems where the load is approximately evenly spread across vbuckets. That is NOT a hard limit (see also max_checkpoints_hard_limit_multiplier). Directly used in the auto-computation for the actual checkpoint_max_size.
-    max_checkpoints_hard_limit_multiplier - Defines the actual max checkpoints (per vbucket) hard limit as (max_checkpoints * multiplier). The multiplier doesn't affect the auto checkpoint_max_size computation.
+    max_checkpoints              - Max number of checkpoints allowed per vbucket.
     checkpoint_memory_ratio      - Max ratio of the bucket quota that can be allocated in checkpoints.
     checkpoint_memory_recovery_upper_mark - Fraction of the checkpoint quota (as computed by checkpoint_memory_ratio) that triggers attempt of memory releasing from checkpoint.
     checkpoint_memory_recovery_lower_mark - Fraction of the checkpoint quota (as computed by checkpoint_memory_ratio) that represents the target of checkpoint memory recovery. Memory recovery yields when reached.

engines/ep/src/checkpoint_config.cc

@@ -44,42 +44,33 @@ std::ostream& operator<<(std::ostream& os, const CheckpointRemoval& mode) {
  */
 class CheckpointConfig::ChangeListener : public ValueChangedListener {
 public:
-    explicit ChangeListener(const Configuration& config,
-                            CheckpointConfig& ckptConfig)
-        : config(config), checkpointConfig(ckptConfig) {
+    explicit ChangeListener(CheckpointConfig& c) : config(c) {
     }
 
     void sizeValueChanged(const std::string& key, size_t value) override {
-        if (key == "chk_period") {
-            checkpointConfig.setCheckpointPeriod(value);
-        } else if (key == "chk_max_items") {
-            checkpointConfig.setCheckpointMaxItems(value);
-        } else if (key == "max_checkpoints") {
-            const auto newMaxCheckpoints =
-                    value * config.getMaxCheckpointsHardLimitMultiplier();
-            checkpointConfig.setMaxCheckpoints(newMaxCheckpoints);
-        } else if (key == "max_checkpoints_hard_limit_multiplier") {
-            const auto newMaxCheckpoints = config.getMaxCheckpoints() * value;
-            checkpointConfig.setMaxCheckpoints(newMaxCheckpoints);
+        if (key.compare("chk_period") == 0) {
+            config.setCheckpointPeriod(value);
+        } else if (key.compare("chk_max_items") == 0) {
+            config.setCheckpointMaxItems(value);
+        } else if (key.compare("max_checkpoints") == 0) {
+            config.setMaxCheckpoints(value);
         }
     }
 
     void booleanValueChanged(const std::string& key, bool value) override {
         if (key.compare("item_num_based_new_chk") == 0) {
-            checkpointConfig.allowItemNumBasedNewCheckpoint(value);
+            config.allowItemNumBasedNewCheckpoint(value);
         }
     }
 
 private:
-    const Configuration& config;
-    CheckpointConfig& checkpointConfig;
+    CheckpointConfig& config;
 };
 
 CheckpointConfig::CheckpointConfig(Configuration& config)
     : checkpointPeriod(config.getChkPeriod()),
       checkpointMaxItems(config.getChkMaxItems()),
-      maxCheckpoints(config.getMaxCheckpoints() *
-                     config.getMaxCheckpointsHardLimitMultiplier()),
+      maxCheckpoints(config.getMaxCheckpoints()),
       itemNumBasedNewCheckpoint(config.isItemNumBasedNewChk()),
       persistenceEnabled(config.getBucketType() == "persistent"),
       checkpointRemovalMode(
@@ -88,22 +79,19 @@ CheckpointConfig::CheckpointConfig(Configuration& config)
 
 void CheckpointConfig::addConfigChangeListener(
         EventuallyPersistentEngine& engine) {
-    auto& config = engine.getConfiguration();
-    auto& ckptConfig = engine.getCheckpointConfig();
-    config.addValueChangedListener(
-            "chk_period", std::make_unique<ChangeListener>(config, ckptConfig));
-    config.addValueChangedListener(
+    Configuration& configuration = engine.getConfiguration();
+    configuration.addValueChangedListener(
+            "chk_period",
+            std::make_unique<ChangeListener>(engine.getCheckpointConfig()));
+    configuration.addValueChangedListener(
             "chk_max_items",
-            std::make_unique<ChangeListener>(config, ckptConfig));
-    config.addValueChangedListener(
+            std::make_unique<ChangeListener>(engine.getCheckpointConfig()));
+    configuration.addValueChangedListener(
             "max_checkpoints",
-            std::make_unique<ChangeListener>(config, ckptConfig));
-    config.addValueChangedListener(
-            "max_checkpoints_hard_limit_multiplier",
-            std::make_unique<ChangeListener>(config, ckptConfig));
-    config.addValueChangedListener(
+            std::make_unique<ChangeListener>(engine.getCheckpointConfig()));
+    configuration.addValueChangedListener(
             "item_num_based_new_chk",
-            std::make_unique<ChangeListener>(config, ckptConfig));
+            std::make_unique<ChangeListener>(engine.getCheckpointConfig()));
 }
 
 void CheckpointConfig::setCheckpointPeriod(size_t value) {

engines/ep/src/checkpoint_config.h

@@ -55,6 +55,11 @@ class CheckpointConfig {
         return checkpointRemovalMode;
     }
 
+    // @todo: Test only. Remove as soon as param made dynamic in EPConfig.
+    void setCheckpointRemovalMode(CheckpointRemoval mode) {
+        checkpointRemovalMode = mode;
+    }
+
 protected:
     friend class EventuallyPersistentEngine;
     friend class SynchronousEPEngine;

engines/ep/src/checkpoint_manager.cc

@@ -971,10 +971,13 @@ CheckpointManager::ItemsForCursor CheckpointManager::getItemsForCursor(
         if (enteredNewCp) {
             const auto& checkpoint = **cursor.getCheckpoint();
             result.checkpointType = checkpoint.getCheckpointType();
-            result.ranges.push_back({{checkpoint.getSnapshotStartSeqno(),
-                                      checkpoint.getSnapshotEndSeqno()},
-                                     checkpoint.getHighCompletedSeqno(),
-                                     checkpoint.getHighPreparedSeqno()});
+            result.ranges.push_back(
+                    {{checkpoint.getSnapshotStartSeqno(),
+                      checkpoint.getSnapshotEndSeqno()},
+                     checkpoint.getHighCompletedSeqno(),
+                     checkpoint.getHighPreparedSeqno(),
+                     checkpoint.getState() ==
+                             checkpoint_state::CHECKPOINT_CLOSED});
             enteredNewCp = false;
 
             // As we cross into new checkpoints, update the maxDeletedRevSeqno

engines/ep/src/checkpoint_manager.h

@@ -60,6 +60,8 @@ struct CheckpointSnapshotRange {
     // HPS that should be flushed when the entire range has been persisted.
     // This is the seqno of the latest prepare in this checkpoint.
     std::optional<uint64_t> highPreparedSeqno = {};
+    // Is the checkpoint closed
+    bool isClosed = false;
 };
 
 /**

engines/ep/src/checkpoint_remover.cc

@@ -183,6 +183,10 @@ CheckpointMemRecoveryTask::attemptCursorDropping() {
             case CheckpointRemoval::Eager: {
 #if CB_DEVELOPMENT_ASSERTS
                 Expects(manager.removeClosedUnrefCheckpoints().count == 0);
+#else
+                // MB-51408: We need to make the call for keeping executing the
+                // inner checkpoint creation logic - minimal fix for Neo.
+                manager.removeClosedUnrefCheckpoints();
 #endif
                 break;
             }
@@ -244,6 +248,10 @@ bool CheckpointMemRecoveryTask::runInner() {
         // This is not cheap to verify, as it requires scanning every
         // vbucket, so is only checked if dev asserts are on.
         Expects(attemptCheckpointRemoval().second == 0);
+#else
+        // MB-51408: We need to make the call for keeping executing the inner
+        // checkpoint creation logic - minimal fix for Neo.
+        attemptCheckpointRemoval();
 #endif
         break;
     }

engines/ep/src/dcp/active_stream.cc

@@ -1177,6 +1177,32 @@ void ActiveStream::processItems(
          */
         std::optional<uint64_t> highNonVisibleSeqno;
         for (auto& qi : outstandingItemsResult.items) {
+            if (qi->getOperation() == queue_op::checkpoint_end) {
+                // At the end of each checkpoint remove its snapshot range, so
+                // we don't use it to set nextSnapStart for the next checkpoint.
+                // If statement ensures that we have indeed completed the
+                // snapshot range by checking that checkpoint_end seqno is
+                // greater than the end of the range. Also account for an edge
+                // case where we have closed checkpoint who's checkpoint_end is
+                // equal to snapshot end seqno. This can happen when snapshots
+                // partially received by a replica which is then promoted to
+                // active
+                auto rangeItr = outstandingItemsResult.ranges.begin();
+                auto snapEnd = static_cast<int64_t>(rangeItr->getEnd());
+                if (snapEnd < qi->getBySeqno() ||
+                    (rangeItr->isClosed && snapEnd == qi->getBySeqno())) {
+                    outstandingItemsResult.ranges.erase(rangeItr);
+                } else {
+                    log(spdlog::level::level_enum::err,
+                        "ActiveStream::processItems checkpoint_end:{} "
+                        "should not be in the current snapshot range "
+                        "s:{}->e:{}",
+                        qi->getBySeqno(),
+                        rangeItr->getStart(),
+                        rangeItr->getEnd());
+                }
+            }
+
             if (qi->getOperation() == queue_op::checkpoint_start) {
                 /* if there are already other mutations, then they belong to the
                    previous checkpoint and hence we must create a snapshot and
@@ -1205,8 +1231,6 @@ void ActiveStream::processItems(
 
                 nextSnapStart =
                         outstandingItemsResult.ranges.begin()->getStart();
-                outstandingItemsResult.ranges.erase(
-                        outstandingItemsResult.ranges.begin());
 
                 continue;
             }