MB-15009: 3/3 Defragment HashTable - StoredValue

Building on the existing value defragmenter, we now
defragment StoredValues for persistent buckets.

The StoredValue carries an age (hidden in the value tag)
which when it reaches a new threshold, triggers the
defragger to ask the HashTable to reallocate the stored
value.

Change-Id: I4b50ebe4a7abc50e15389bc1d7dea4301543a174
Reviewed-on: http://review.couchbase.org/106450
Well-Formed: Build Bot <[email protected]>
Reviewed-by: Dave Rigby <[email protected]>
Tested-by: Build Bot <[email protected]>

Author: jimwwalker

engines/ep/benchmarks/defragmenter_bench.cc

@@ -103,7 +103,8 @@ class DefragmentBench : public benchmark::Fixture {
         // the given age.
         DefragmentVisitor visitor(age_threshold,
                                   DefragmenterTask::getMaxValueSize(
-                                          get_mock_server_api()->alloc_hooks));
+                                          get_mock_server_api()->alloc_hooks),
+                                  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/src/defragmenter.cc

@@ -40,9 +40,22 @@ 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)));
+            boost::optional<uint8_t> storedValueAge;
+
+            // Only defragment persistent buckets because the HashTable defrag
+            // method doesn't yet know how to maintain the ephemeral seqno
+            // linked-list
+            if (engine->getConfiguration().getBucketType() == "persistent") {
+                storedValueAge = getStoredValueAgeThreshold();
+            }
+
+            auto visitor = std::make_unique<DefragmentVisitor>(
+                    getAgeThreshold(),
+                    getMaxValueSize(alloc_hooks),
+                    storedValueAge);
+
+            prAdapter =
+                    std::make_unique<PauseResumeVBAdapter>(std::move(visitor));
             epstore_position = engine->getKVBucket()->startPosition();
         }
 
@@ -80,9 +93,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 +164,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

@@ -20,12 +20,14 @@
 // DegragmentVisitor implementation ///////////////////////////////////////////
 
 DefragmentVisitor::DefragmentVisitor(uint8_t age_threshold_,
-                                     size_t max_size_class)
+                                     size_t max_size_class,
+                                     boost::optional<uint8_t> sv_age_threshold)
     : max_size_class(max_size_class),
       age_threshold(age_threshold_),
       defrag_count(0),
       visited_count(0),
-      currentVb(nullptr) {
+      currentVb(nullptr),
+      sv_age_threshold(sv_age_threshold) {
 }
 
 DefragmentVisitor::~DefragmentVisitor() {
@@ -57,6 +59,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 +78,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 +89,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,7 +31,9 @@
  */
 class DefragmentVisitor : public VBucketAwareHTVisitor {
 public:
-    DefragmentVisitor(uint8_t age_threshold_, size_t max_size_class);
+    DefragmentVisitor(uint8_t age_threshold_,
+                      size_t max_size_class,
+                      boost::optional<uint8_t> sv_age_threshold);
 
     ~DefragmentVisitor();
 
@@ -51,9 +53,15 @@ 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.
@@ -72,7 +80,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

@@ -2923,6 +2923,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;
 }

engines/ep/src/hash_table.h

@@ -798,6 +798,19 @@ class HashTable {
                                     bool keyMetaDataOnly,
                                     item_eviction_policy_t evictionPolicy);
 
+    /**
+     * 'Defragment' the StoredValue, this really means reallocate the object
+     * which has the effect of repacking the underlying allocators memory pool.
+     *
+     * As the object is reallocated the input StoredValue is 'consumed', i.e.
+     * it will be copied into a new allocate and the original freed, the
+     * input is deleted on successful reallocation.
+     *
+     * @param v The rvalue reference to the StoredValue to be reallocated.
+     * @return true if v was found and reallocated
+     */
+    bool reallocateStoredValue(StoredValue&& v);
+
     /**
      * Dump a representation of the HashTable to stderr.
      */