Reduce coupling between vbucket.h and other classes

Move a number of VBucket inline methods to vbucket.cc; so clients of
vbucket don't need to know the details of the impementation (and hence
include those header files). Do a similar thing for KVBucket and
EPEngine.

Note: This is mostly a preparation patch; subsequent patches will
break the dependancy between vbucket.h and checkpoint.h.

Change-Id: I1080d5a51666c8bb82791c417504a5a830724f2f
Reviewed-on: http://review.couchbase.org/82693
Reviewed-by: James Harrison <[email protected]>
Tested-by: Build Bot <[email protected]>

Author: daverigby

engines/ep/src/access_scanner.cc

@@ -27,6 +27,7 @@
 #include "hash_table.h"
 #include "kv_bucket.h"
 #include "mutation_log.h"
+#include "stats.h"
 #include "vb_count_visitor.h"
 
 #include <numeric>

engines/ep/src/checkpoint.cc

@@ -1858,7 +1858,7 @@ void CheckpointManager::itemsPersisted() {
     }
 }
 
-size_t CheckpointManager::getMemoryUsage_UNLOCKED() {
+size_t CheckpointManager::getMemoryUsage_UNLOCKED() const {
     if (checkpointList.empty()) {
         return 0;
     }
@@ -1870,12 +1870,12 @@ size_t CheckpointManager::getMemoryUsage_UNLOCKED() {
     return memUsage;
 }
 
-size_t CheckpointManager::getMemoryUsage() {
+size_t CheckpointManager::getMemoryUsage() const {
     LockHolder lh(queueLock);
     return getMemoryUsage_UNLOCKED();
 }
 
-size_t CheckpointManager::getMemoryUsageOfUnrefCheckpoints() {
+size_t CheckpointManager::getMemoryUsageOfUnrefCheckpoints() const {
     LockHolder lh(queueLock);
 
     if (checkpointList.empty()) {

engines/ep/src/checkpoint.h

@@ -71,11 +71,6 @@ struct index_entry {
     int64_t mutation_id;
 };
 
-typedef struct {
-    uint64_t start;
-    snapshot_range_t range;
-} snapshot_info_t;
-
 /**
  * Flag indicating that we must send checkpoint end meta item for the cursor
  */
@@ -551,7 +546,7 @@ class Checkpoint {
      * Returns the memory held by all the queued items which includes
      * key, metadata and the blob.
      */
-    size_t getMemConsumption() {
+    size_t getMemConsumption() const {
         return effectiveMemUsage;
     }
 
@@ -795,14 +790,14 @@ class CheckpointManager {
     /**
      * Return memory consumption of all the checkpoints managed
      */
-    size_t getMemoryUsage_UNLOCKED();
+    size_t getMemoryUsage_UNLOCKED() const;
 
-    size_t getMemoryUsage();
+    size_t getMemoryUsage() const;
 
     /**
      * Return memory consumption of unreferenced checkpoints
      */
-    size_t getMemoryUsageOfUnrefCheckpoints();
+    size_t getMemoryUsageOfUnrefCheckpoints() const;
 
     /**
      * Function returns a list of cursors to drop so as to unreference

engines/ep/src/ep_engine.cc

@@ -2004,6 +2004,10 @@ ENGINE_ERROR_CODE EventuallyPersistentEngine::itemAllocate(
     }
 }
 
+void EventuallyPersistentEngine::itemRelease(const void* cookie, item* itm) {
+    delete reinterpret_cast<Item*>(itm);
+}
+
 ENGINE_ERROR_CODE EventuallyPersistentEngine::flush(const void *cookie){
     if (!deleteAllEnabled) {
         return ENGINE_ENOTSUP;
@@ -3418,6 +3422,38 @@ void EventuallyPersistentEngine::addSeqnoVbStats(const void *cookie,
     }
 }
 
+void EventuallyPersistentEngine::addLookupResult(const void* cookie,
+                                                 std::unique_ptr<Item> result) {
+    LockHolder lh(lookupMutex);
+    auto it = lookups.find(cookie);
+    if (it != lookups.end()) {
+        if (it->second != NULL) {
+            LOG(EXTENSION_LOG_DEBUG,
+                "Cleaning up old lookup result for '%s'",
+                it->second->getKey().data());
+        } else {
+            LOG(EXTENSION_LOG_DEBUG, "Cleaning up old null lookup result");
+        }
+        lookups.erase(it);
+    }
+    lookups[cookie] = std::move(result);
+}
+
+bool EventuallyPersistentEngine::fetchLookupResult(const void* cookie,
+                                                   std::unique_ptr<Item>& itm) {
+    // This will return *and erase* the lookup result for a connection.
+    // You look it up, you own it.
+    LockHolder lh(lookupMutex);
+    auto it = lookups.find(cookie);
+    if (it != lookups.end()) {
+        itm = std::move(it->second);
+        lookups.erase(it);
+        return true;
+    } else {
+        return false;
+    }
+}
+
 ENGINE_ERROR_CODE EventuallyPersistentEngine::doSeqnoStats(const void *cookie,
                                                           ADD_STAT add_stat,
                                                           const char* stat_key,

engines/ep/src/ep_engine.h

@@ -23,6 +23,7 @@
 #include "connhandler.h"
 #include "kv_bucket.h"
 #include "storeddockey.h"
+#include "stats.h"
 #include "taskable.h"
 #include "vb_visitors.h"
 
@@ -142,12 +143,7 @@ class EventuallyPersistentEngine : public ENGINE_HANDLE_V1 {
         return ret;
     }
 
-
-    void itemRelease(const void* cookie, item *itm)
-    {
-        (void)cookie;
-        delete (Item*)itm;
-    }
+    void itemRelease(const void* cookie, item *itm);
 
     ENGINE_ERROR_CODE get(const void* cookie,
                           item** itm,
@@ -706,35 +702,9 @@ class EventuallyPersistentEngine : public ENGINE_HANDLE_V1 {
     void addSeqnoVbStats(const void *cookie, ADD_STAT add_stat,
                                   const VBucketPtr &vb);
 
-    void addLookupResult(const void* cookie, std::unique_ptr<Item> result) {
-        LockHolder lh(lookupMutex);
-        auto it = lookups.find(cookie);
-        if (it != lookups.end()) {
-            if (it->second != NULL) {
-                LOG(EXTENSION_LOG_DEBUG,
-                    "Cleaning up old lookup result for '%s'",
-                    it->second->getKey().data());
-            } else {
-                LOG(EXTENSION_LOG_DEBUG, "Cleaning up old null lookup result");
-            }
-            lookups.erase(it);
-        }
-        lookups[cookie] = std::move(result);
-    }
-
-    bool fetchLookupResult(const void* cookie, std::unique_ptr<Item>& itm) {
-        // This will return *and erase* the lookup result for a connection.
-        // You look it up, you own it.
-        LockHolder lh(lookupMutex);
-        auto it = lookups.find(cookie);
-        if (it != lookups.end()) {
-            itm = std::move(it->second);
-            lookups.erase(it);
-            return true;
-        } else {
-            return false;
-        }
-    }
+    void addLookupResult(const void* cookie, std::unique_ptr<Item> result);
+
+    bool fetchLookupResult(const void* cookie, std::unique_ptr<Item>& itm);
 
     // Initialize all required callbacks of this engine with the underlying
     // server.

engines/ep/src/ep_types.h

@@ -51,6 +51,11 @@ typedef struct {
     uint64_t end;
 } snapshot_range_t;
 
+typedef struct {
+    uint64_t start;
+    snapshot_range_t range;
+} snapshot_info_t;
+
 /**
  * The following options can be specified
  * for retrieving an item for get calls

engines/ep/src/kv_bucket.cc

@@ -501,6 +501,21 @@ bool KVBucket::isMetaDataResident(VBucketPtr &vb, const DocKey& key) {
     }
 }
 
+void KVBucket::logQTime(TaskId taskType, const ProcessClock::duration enqTime) {
+    const auto ns_count =
+            std::chrono::duration_cast<std::chrono::microseconds>(enqTime)
+                    .count();
+    stats.schedulingHisto[static_cast<int>(taskType)].add(ns_count);
+}
+
+void KVBucket::logRunTime(TaskId taskType,
+                          const ProcessClock::duration runTime) {
+    const auto ns_count =
+            std::chrono::duration_cast<std::chrono::microseconds>(runTime)
+                    .count();
+    stats.taskRuntimeHisto[static_cast<int>(taskType)].add(ns_count);
+}
+
 ENGINE_ERROR_CODE KVBucket::set(Item& itm,
                                 const void* cookie,
                                 cb::StoreIfPredicate predicate) {
@@ -2637,6 +2652,24 @@ void KVBucket::runVbStatePersistTask(int vbid) {
     scheduleVBStatePersist(vbid);
 }
 
+bool KVBucket::compactionCanExpireItems() {
+    // Process expired items only if memory usage is lesser than
+    // compaction_exp_mem_threshold and disk queue is small
+    // enough (marked by replication_throttle_queue_cap)
+
+    bool isMemoryUsageOk =
+            (stats.getTotalMemoryUsed() <
+             (stats.getMaxDataSize() * compactionExpMemThreshold));
+
+    size_t queueSize = stats.diskQueueSize.load();
+    bool isQueueSizeOk =
+            ((stats.replicationThrottleWriteQueueCap == -1) ||
+             (queueSize <
+              static_cast<size_t>(stats.replicationThrottleWriteQueueCap)));
+
+    return (isMemoryUsageOk && isQueueSizeOk);
+}
+
 void KVBucket::setCursorDroppingLowerUpperThresholds(size_t maxSize) {
     Configuration &config = engine.getConfiguration();
     stats.cursorDroppingLThreshold.store(static_cast<size_t>(maxSize *

engines/ep/src/kv_bucket.h

@@ -589,17 +589,9 @@ class KVBucket : public KVBucketIface {
 
     bool isMetaDataResident(VBucketPtr &vb, const DocKey& key);
 
-    void logQTime(TaskId taskType, const ProcessClock::duration enqTime) {
-        const auto ns_count = std::chrono::duration_cast
-                <std::chrono::microseconds>(enqTime).count();
-        stats.schedulingHisto[static_cast<int>(taskType)].add(ns_count);
-    }
+    void logQTime(TaskId taskType, const ProcessClock::duration enqTime);
 
-    void logRunTime(TaskId taskType, const ProcessClock::duration runTime) {
-        const auto ns_count = std::chrono::duration_cast
-                <std::chrono::microseconds>(runTime).count();
-        stats.taskRuntimeHisto[static_cast<int>(taskType)].add(ns_count);
-    }
+    void logRunTime(TaskId taskType, const ProcessClock::duration runTime);
 
     bool multiBGFetchEnabled() {
         StorageProperties storeProp = getStorageProperties();
@@ -707,20 +699,7 @@ class KVBucket : public KVBucketIface {
         compactionExpMemThreshold = static_cast<double>(to) / 100.0;
     }
 
-    bool compactionCanExpireItems() {
-        // Process expired items only if memory usage is lesser than
-        // compaction_exp_mem_threshold and disk queue is small
-        // enough (marked by replication_throttle_queue_cap)
-
-        bool isMemoryUsageOk = (stats.getTotalMemoryUsed() <
-                          (stats.getMaxDataSize() * compactionExpMemThreshold));
-
-        size_t queueSize = stats.diskQueueSize.load();
-        bool isQueueSizeOk = ((stats.replicationThrottleWriteQueueCap == -1) ||
-             (queueSize < static_cast<size_t>(stats.replicationThrottleWriteQueueCap)));
-
-        return (isMemoryUsageOk && isQueueSizeOk);
-    }
+    bool compactionCanExpireItems();
 
     void setCursorDroppingLowerUpperThresholds(size_t maxSize);
 

engines/ep/src/vbucket.cc

@@ -247,6 +247,18 @@ VBucket::~VBucket() {
     LOG(EXTENSION_LOG_INFO, "Destroying vbucket %d\n", id);
 }
 
+int64_t VBucket::getHighSeqno() const {
+    return checkpointManager.getHighSeqno();
+}
+
+size_t VBucket::getChkMgrMemUsage() const {
+    return checkpointManager.getMemoryUsage();
+}
+
+size_t VBucket::getChkMgrMemUsageOfUnrefCheckpoints() const {
+    return checkpointManager.getMemoryUsageOfUnrefCheckpoints();
+}
+
 void VBucket::fireAllOps(EventuallyPersistentEngine &engine,
                          ENGINE_ERROR_CODE code) {
     std::unique_lock<std::mutex> lh(pendingOpLock);
@@ -292,6 +304,17 @@ void VBucket::fireAllOps(EventuallyPersistentEngine &engine) {
     }
 }
 
+void VBucket::getBackfillItems(std::vector<queued_item>& items) {
+    LockHolder lh(backfill.mutex);
+    size_t num_items = backfill.items.size();
+    while (!backfill.items.empty()) {
+        items.push_back(backfill.items.front());
+        backfill.items.pop();
+    }
+    stats.vbBackfillQueueSize.fetch_sub(num_items);
+    stats.memOverhead->fetch_sub(num_items * sizeof(queued_item));
+}
+
 void VBucket::setState(vbucket_state_t to) {
     WriterLockHolder wlh(stateLock);
     setState_UNLOCKED(to, wlh);
@@ -431,6 +454,22 @@ void VBucket::handlePreExpiry(StoredValue& v) {
     }
 }
 
+bool VBucket::addPendingOp(const void* cookie) {
+    LockHolder lh(pendingOpLock);
+    if (state != vbucket_state_pending) {
+        // State transitioned while we were waiting.
+        return false;
+    }
+    // Start a timer when enqueuing the first client.
+    if (pendingOps.empty()) {
+        pendingOpsStart = gethrtime();
+    }
+    pendingOps.push_back(cookie);
+    ++stats.pendingOps;
+    ++stats.pendingOpsTotal;
+    return true;
+}
+
 size_t VBucket::getNumNonResidentItems() const {
     if (eviction == VALUE_ONLY) {
         return ht.getNumInMemoryNonResItems();