MB-15009: Merge branch 'couchbase/vulcan' into 'couchbase/alice'

* couchbase/vulcan:
  MB-15009: Allow stored value age to be changed by cbepctl
  MB-15009: 3/3 Defragment HashTable - StoredValue
  MB-15009: 2/3 Refactor VBucketTest
  MB-15009: 1/3 Make frequency counter 8-bit

Change-Id: I096d84e05d08e693b4ab961d3e9d98e74625d544

Author: jimwwalker

engines/ep/benchmarks/defragmenter_bench.cc

@@ -101,9 +101,12 @@ class DefragmentBench : public benchmark::Fixture {
             std::chrono::milliseconds chunk_duration) {
         // Create and run visitor for the specified number of iterations, with
         // the given age.
-        DefragmentVisitor visitor(age_threshold,
-                                  DefragmenterTask::getMaxValueSize(
-                                          get_mock_server_api()->alloc_hooks));
+        DefragmentVisitor visitor(DefragmenterTask::getMaxValueSize(
+                get_mock_server_api()->alloc_hooks));
+
+        visitor.setBlobAgeThreshold(age_threshold);
+        visitor.setStoredValueAgeThreshold(age_threshold);
+
         // Need to run 10 passes; so we allow the deframenter to defrag at
         // least once (given the age_threshold may be up to 10).
         const size_t passes = 10;

engines/ep/configuration.json

@@ -289,7 +289,7 @@
             "dynamic": false,
             "type": "std::string"
         },
-	"dcp_noop_mandatory_for_v5_features": {
+        "dcp_noop_mandatory_for_v5_features": {
             "default": "true",
             "descr": "Forces clients to enable noop for v5 features",
             "type": "bool"
@@ -307,7 +307,12 @@
         },
         "defragmenter_age_threshold": {
             "default": "10",
-            "descr": "How old (measured in number of defragmenter passes) must a document be to be considered for degragmentation.",
+            "descr": "How old (measured in number of DefragmenterVisitor passes) must a document be to be considered for defragmentation.",
+            "type": "size_t"
+        },
+        "defragmenter_stored_value_age_threshold": {
+            "default": "10",
+            "descr": "How old (measured in number of DefragmenterVisitor passes) must a StoredValue be to be considered for defragmentation.",
             "type": "size_t"
         },
         "defragmenter_chunk_duration": {

engines/ep/docs/stats.org

@@ -300,6 +300,8 @@ For introductory information on stats within Couchbase, start with the
 |                                    | run (in seconds).                      |
 | ep_defragmenter_num_moved          | Number of items moved by the           |
 |                                    | defragmentater task.                   |
+| ep_defragmenter_sv_num_moved       | Number of StoredValues moved by the    |
+|                                    | defragmentater task.                   |
 | ep_defragmenter_num_visited        | Number of items visited (considered    |
 |                                    | for defragmentation) by the            |
 |                                    | defragmenter task.                     |

engines/ep/management/cbepctl

@@ -188,6 +188,9 @@ Available params for "set":
     defragmenter_age_threshold   - How old (measured in number of defragmenter
                                    passes) must a document be to be considered
                                    for defragmentation.
+    defragmenter_stored_value_age_threshold   - How old (measured in number of defragmenter
+                                   passes) must a StoredValue (key + meta) be to be considered
+                                   for defragmentation.
     defragmenter_chunk_duration  - Maximum time (in ms) defragmentation task
                                    will run for before being paused (and
                                    resumed at the next defragmenter_interval).

engines/ep/src/defragmenter.cc

@@ -40,9 +40,11 @@ bool DefragmenterTask::run(void) {
         // then resume from where we last were, otherwise create a new visitor
         // starting from the beginning.
         if (!prAdapter) {
-            prAdapter = std::make_unique<PauseResumeVBAdapter>(
-                    std::make_unique<DefragmentVisitor>(
-                            getAgeThreshold(), getMaxValueSize(alloc_hooks)));
+            auto visitor = std::make_unique<DefragmentVisitor>(
+                    getMaxValueSize(alloc_hooks));
+
+            prAdapter =
+                    std::make_unique<PauseResumeVBAdapter>(std::move(visitor));
             epstore_position = engine->getKVBucket()->startPosition();
         }
 
@@ -70,6 +72,13 @@ bool DefragmenterTask::run(void) {
         const auto start = ProcessClock::now();
         const auto deadline = start + getChunkDuration();
         visitor.setDeadline(deadline);
+        visitor.setBlobAgeThreshold(getAgeThreshold());
+        // Only defragment StoredValues of persistent buckets because the
+        // HashTable defrag method doesn't yet know how to maintain the
+        // ephemeral seqno linked-list
+        if (engine->getConfiguration().getBucketType() == "persistent") {
+            visitor.setStoredValueAgeThreshold(getStoredValueAgeThreshold());
+        }
         visitor.clearStats();
 
         // Do it - set off the visitor.
@@ -80,9 +89,7 @@ bool DefragmenterTask::run(void) {
         // Defrag complete. Restore thread caching.
         alloc_hooks->enable_thread_cache(old_tcache);
 
-        // Update stats
-        stats.defragNumMoved.fetch_add(visitor.getDefragCount());
-        stats.defragNumVisited.fetch_add(visitor.getVisitedCount());
+        updateStats(visitor);
 
         // Release any free memory we now have in the allocator back to the OS.
         // TODO: Benchmark this - is it necessary? How much of a slowdown does it
@@ -153,6 +160,17 @@ size_t DefragmenterTask::getAgeThreshold() const {
     return engine->getConfiguration().getDefragmenterAgeThreshold();
 }
 
+size_t DefragmenterTask::getStoredValueAgeThreshold() const {
+    return engine->getConfiguration().getDefragmenterStoredValueAgeThreshold();
+}
+
+void DefragmenterTask::updateStats(DefragmentVisitor& visitor) {
+    stats.defragNumMoved.fetch_add(visitor.getDefragCount());
+    stats.defragStoredValueNumMoved.fetch_add(
+            visitor.getStoredValueDefragCount());
+    stats.defragNumVisited.fetch_add(visitor.getVisitedCount());
+}
+
 size_t DefragmenterTask::getMaxValueSize(ALLOCATOR_HOOKS_API* alloc_hooks) {
     size_t nbins{0};
     alloc_hooks->get_allocator_property("arenas.nbins", &nbins);

engines/ep/src/defragmenter.h

@@ -123,6 +123,10 @@ class DefragmenterTask : public GlobalTask {
     // must be to be considered for defragmentation.
     size_t getAgeThreshold() const;
 
+    // Minimum age (measured in defragmenter task passes) that a StoredValue
+    // must be to be considered for defragmentation.
+    size_t getStoredValueAgeThreshold() const;
+
     // Upper limit on how long each defragmention chunk can run for, before
     // being paused.
     std::chrono::milliseconds getChunkDuration() const;
@@ -133,6 +137,9 @@ class DefragmenterTask : public GlobalTask {
     /// Returns the underlying DefragmentVisitor instance.
     DefragmentVisitor& getDefragVisitor();
 
+    /// Update the EPStats from the visitor
+    void updateStats(DefragmentVisitor& visitor);
+
     /// Reference to EP stats, used to check on mem_used.
     EPStats &stats;
 

engines/ep/src/defragmenter_visitor.cc

@@ -19,10 +19,8 @@
 
 // DegragmentVisitor implementation ///////////////////////////////////////////
 
-DefragmentVisitor::DefragmentVisitor(uint8_t age_threshold_,
-                                     size_t max_size_class)
+DefragmentVisitor::DefragmentVisitor(size_t max_size_class)
     : max_size_class(max_size_class),
-      age_threshold(age_threshold_),
       defrag_count(0),
       visited_count(0),
       currentVb(nullptr) {
@@ -35,6 +33,14 @@ void DefragmentVisitor::setDeadline(ProcessClock::time_point deadline) {
     progressTracker.setDeadline(deadline);
 }
 
+void DefragmentVisitor::setBlobAgeThreshold(uint8_t age) {
+    age_threshold = age;
+}
+
+void DefragmentVisitor::setStoredValueAgeThreshold(uint8_t age) {
+    sv_age_threshold = age;
+}
+
 bool DefragmentVisitor::visit(const HashTable::HashBucketLock& lh,
                               StoredValue& v) {
     const size_t value_len = v.valuelen();
@@ -57,6 +63,15 @@ bool DefragmentVisitor::visit(const HashTable::HashBucketLock& lh,
             v.getValue()->incrementAge();
         }
     }
+
+    if (sv_age_threshold) {
+        if (v.getAge() >= sv_age_threshold.get()) {
+            defragmentStoredValue(v);
+        } else {
+            v.incrementAge();
+        }
+    }
+
     visited_count++;
 
     // See if we have done enough work for this chunk. If so
@@ -67,6 +82,7 @@ bool DefragmentVisitor::visit(const HashTable::HashBucketLock& lh,
 void DefragmentVisitor::clearStats() {
     defrag_count = 0;
     visited_count = 0;
+    sv_defrag_count = 0;
 }
 
 size_t DefragmentVisitor::getDefragCount() const {
@@ -77,6 +93,16 @@ size_t DefragmentVisitor::getVisitedCount() const {
     return visited_count;
 }
 
+size_t DefragmentVisitor::getStoredValueDefragCount() const {
+    return sv_defrag_count;
+}
+
 void DefragmentVisitor::setCurrentVBucket(VBucket& vb) {
     currentVb = &vb;
 }
+
+void DefragmentVisitor::defragmentStoredValue(StoredValue& v) const {
+    if (currentVb->ht.reallocateStoredValue(std::forward<StoredValue>(v))) {
+        sv_defrag_count++;
+    }
+}

engines/ep/src/defragmenter_visitor.h

@@ -31,13 +31,24 @@
  */
 class DefragmentVisitor : public VBucketAwareHTVisitor {
 public:
-    DefragmentVisitor(uint8_t age_threshold_, size_t max_size_class);
+    DefragmentVisitor(size_t max_size_class);
 
     ~DefragmentVisitor();
 
     // Set the deadline at which point the visitor will pause visiting.
     void setDeadline(ProcessClock::time_point deadline_);
 
+    /**
+     * Set the age at which Blobs are defragged 0 - 255
+     */
+    void setBlobAgeThreshold(uint8_t age);
+
+    /**
+     * Set the age at which StoredValues are defragged - by default SV
+     * defragging is off until an age is set (0-255)
+     */
+    void setStoredValueAgeThreshold(uint8_t age);
+
     // Implementation of HashTableVisitor interface:
     virtual bool visit(const HashTable::HashBucketLock& lh,
                        StoredValue& v) override;
@@ -51,16 +62,22 @@ class DefragmentVisitor : public VBucketAwareHTVisitor {
     // Returns the number of documents that have been visited.
     size_t getVisitedCount() const;
 
+    // Returns the number of StoredValues that have been defragmented.
+    size_t getStoredValueDefragCount() const;
+
     void setCurrentVBucket(VBucket& vb) override;
 
 private:
+    /// Request to reallocate the StoredValue
+    void defragmentStoredValue(StoredValue& v) const;
+
     /* Configuration parameters */
 
     // Size of the largest size class from the allocator.
     const size_t max_size_class;
 
     // How old a blob must be to consider it for defragmentation.
-    const uint8_t age_threshold;
+    uint8_t age_threshold{0};
 
     /* Runtime state */
 
@@ -72,7 +89,12 @@ class DefragmentVisitor : public VBucketAwareHTVisitor {
     size_t defrag_count;
     // How many documents have been visited.
     size_t visited_count;
+    // How many stored-values have been defrag'd
+    mutable size_t sv_defrag_count{0};
 
     // The current vbucket that is being processed
     VBucket* currentVb;
+
+    // If defined, the age at which StoredValue's are de-fragmented
+    boost::optional<uint8_t> sv_age_threshold;
 };

engines/ep/src/ep_engine.cc

@@ -592,6 +592,10 @@ protocol_binary_response_status EventuallyPersistentEngine::setFlushParam(
                     std::stoull(valz));
         } else if (strcmp(keyz, "defragmenter_age_threshold") == 0) {
             getConfiguration().setDefragmenterAgeThreshold(std::stoull(valz));
+        } else if (strcmp(keyz, "defragmenter_stored_value_age_threshold") ==
+                   0) {
+            getConfiguration().setDefragmenterStoredValueAgeThreshold(
+                    std::stoull(valz));
         } else if (strcmp(keyz, "defragmenter_chunk_duration") == 0) {
             getConfiguration().setDefragmenterChunkDuration(std::stoull(valz));
         } else if (strcmp(keyz, "defragmenter_run") == 0) {
@@ -2923,6 +2927,10 @@ ENGINE_ERROR_CODE EventuallyPersistentEngine::doEngineStats(const void *cookie,
                     add_stat, cookie);
     add_casted_stat("ep_defragmenter_num_moved", epstats.defragNumMoved,
                     add_stat, cookie);
+    add_casted_stat("ep_defragmenter_sv_num_moved",
+                    epstats.defragStoredValueNumMoved,
+                    add_stat,
+                    cookie);
 
     add_casted_stat("ep_item_compressor_num_visited",
                     epstats.compressorNumVisited,

engines/ep/src/hash_table.cc

@@ -569,6 +569,21 @@ MutationStatus HashTable::insertFromWarmup(
     return MutationStatus::NotFound;
 }
 
+bool HashTable::reallocateStoredValue(StoredValue&& sv) {
+    // Search the chain and reallocate
+    for (StoredValue::UniquePtr* curr =
+                 &values[getBucketForHash(sv.getKey().hash())];
+         curr->get().get();
+         curr = &curr->get()->getNext()) {
+        if (&sv == curr->get().get()) {
+            auto newSv = valFact->copyStoredValue(sv, std::move(sv.getNext()));
+            curr->swap(newSv);
+            return true;
+        }
+    }
+    return false;
+}
+
 void HashTable::dump() const {
     std::cerr << *this << std::endl;
 }