MB-34946: Use HashTable::FindCommitResult in processSet

In a following change, we need to make use of the FindCommitResult
in VBucket::processSet as it contains both the pending and
committed StoredValues. To keep a single processSet interface,
update all callers to pass a FindCommitResult instead of a single
StoredValue*.

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

Author: BenHuddleston

engines/ep/src/hash_table.cc

@@ -96,14 +96,42 @@ HashTable::StoredValueProxy::StoredValueProxy(HashBucketLock&& hbl,
 }
 
 HashTable::StoredValueProxy::~StoredValueProxy() {
-    valueStats.get().epilogue(pre, value);
+    if (value) {
+        valueStats.get().epilogue(pre, value);
+    }
 }
 
 void HashTable::StoredValueProxy::setCommitted(CommittedState state) {
     value->setCommitted(state);
     value->setCompletedOrDeletedTime(ep_current_time());
 }
 
+StoredValue* HashTable::StoredValueProxy::release() {
+    auto* tmp = value;
+    value = nullptr;
+    return tmp;
+}
+
+StoredValue* HashTable::FindCommitResult::selectSVToModify(bool durability) {
+    if (durability) {
+        if (pending) {
+            return pending.getSV();
+        } else {
+            return committed;
+        }
+    } else {
+        if (pending && !pending->isCompleted()) {
+            return pending.getSV();
+        } else {
+            return committed;
+        }
+    }
+}
+
+StoredValue* HashTable::FindCommitResult::selectSVToModify(const Item& itm) {
+    return selectSVToModify(itm.isPending());
+}
+
 HashTable::HashTable(EPStats& st,
                      std::unique_ptr<AbstractStoredValueFactory> svFactory,
                      size_t initialSize,

engines/ep/src/hash_table.h

@@ -723,6 +723,15 @@ class HashTable {
 
         void setCommitted(CommittedState state);
 
+        /**
+         * Release this handle to the StoredValue* to the caller. This will
+         * allow us to do things such as call a HashTable delete function using
+         * the StoredValueProxy without worrying about double stats calls.
+         *
+         * @return The StoredValue* for value
+         */
+        StoredValue* release();
+
     private:
         HashBucketLock lock;
         StoredValue* value;
@@ -741,6 +750,28 @@ class HashTable {
      * returned (in the pending StoredValueProxy).
      */
     struct FindCommitResult {
+        /**
+         * Return the StoredValue that should generally be used based on the
+         * item. Callers of the findForWrite and findForSyncWrite functions
+         * typically choose one or the other based on the item being updated
+         * (itm.isPending() ? findForSyncWrite(...) : findForWrite(...)). This
+         * function allows the caller to do the same with the FindCommitResult.
+         *
+         * @param bool is the current operation a durable one
+         * @return The prepare if the item is pending and the prepare exists,
+         *         otherwise the commit. The prepare if the item is not pending
+         *         and the prepare is not completed, otherwise the commit. If
+         *         neither exist, nullptr is returned.
+         */
+        StoredValue* selectSVToModify(bool durability);
+
+        /// Overload of above selectSVToModify that takes an Item.
+        StoredValue* selectSVToModify(const Item& itm);
+
+        HashBucketLock& getHBL() {
+            return pending.getHBL();
+        }
+
         StoredValueProxy pending;
         StoredValue* committed;
     };

engines/ep/src/vbucket.cc

@@ -1493,10 +1493,9 @@ ENGINE_ERROR_CODE VBucket::set(
     bool cas_op = (itm.getCas() != 0);
 
     { // HashBucketLock scope
-        auto htRes = itm.isPending() ? ht.findForSyncWrite(itm.getKey())
-                                     : ht.findForWrite(itm.getKey());
-        auto* v = htRes.storedValue;
-        auto& hbl = htRes.lock;
+        auto htRes = ht.findForCommit(itm.getKey());
+        auto* v = htRes.selectSVToModify(itm);
+        auto& hbl = htRes.getHBL();
 
         cb::StoreIfStatus storeIfStatus = cb::StoreIfStatus::Continue;
         if (predicate && (storeIfStatus = callPredicate(predicate, v)) ==
@@ -1533,7 +1532,7 @@ ENGINE_ERROR_CODE VBucket::set(
         queueItmCtx.preLinkDocumentContext = &preLinkDocumentContext;
         MutationStatus status;
         boost::optional<VBNotifyCtx> notifyCtx;
-        std::tie(status, notifyCtx) = processSet(hbl,
+        std::tie(status, notifyCtx) = processSet(htRes,
                                                  v,
                                                  itm,
                                                  itm.getCas(),
@@ -1608,10 +1607,13 @@ ENGINE_ERROR_CODE VBucket::replace(
     }
 
     { // HashBucketLock scope
-        auto htRes = itm.isPending() ? ht.findForSyncReplace(itm.getKey())
-                                     : ht.findForWrite(itm.getKey());
-        auto* v = htRes.storedValue;
-        auto& hbl = htRes.lock;
+        auto htRes = ht.findForCommit(itm.getKey());
+        auto* v = htRes.selectSVToModify(itm);
+        auto& hbl = htRes.getHBL();
+
+        if (v && v->isCompleted()) {
+            v = nullptr;
+        }
 
         cb::StoreIfStatus storeIfStatus = cb::StoreIfStatus::Continue;
         if (predicate && (storeIfStatus = callPredicate(predicate, v)) ==
@@ -1638,7 +1640,7 @@ ENGINE_ERROR_CODE VBucket::replace(
                     queueItmCtx.durability =
                             DurabilityItemCtx{itm.getDurabilityReqs(), cookie};
                 }
-                std::tie(mtype, notifyCtx) = processSet(hbl,
+                std::tie(mtype, notifyCtx) = processSet(htRes,
                                                         v,
                                                         itm,
                                                         0,
@@ -1706,9 +1708,9 @@ ENGINE_ERROR_CODE VBucket::replace(
 ENGINE_ERROR_CODE VBucket::addBackfillItem(
         Item& itm,
         const Collections::VB::Manifest::CachingReadHandle& cHandle) {
-    auto htRes = ht.findForWrite(itm.getKey());
-    auto* v = htRes.storedValue;
-    auto& hbl = htRes.lock;
+    auto htRes = ht.findForCommit(itm.getKey());
+    auto* v = htRes.selectSVToModify(itm);
+    auto& hbl = htRes.getHBL();
 
     // Note that this function is only called on replica or pending vbuckets.
     if (v && v->isLocked(ep_current_time())) {
@@ -1725,7 +1727,7 @@ ENGINE_ERROR_CODE VBucket::addBackfillItem(
             nullptr /* No pre link should happen */};
     MutationStatus status;
     boost::optional<VBNotifyCtx> notifyCtx;
-    std::tie(status, notifyCtx) = processSet(hbl,
+    std::tie(status, notifyCtx) = processSet(htRes,
                                              v,
                                              itm,
                                              0,
@@ -1798,9 +1800,9 @@ ENGINE_ERROR_CODE VBucket::prepare(
         GenerateBySeqno genBySeqno,
         GenerateCas genCas,
         const Collections::VB::Manifest::CachingReadHandle& cHandle) {
-    auto htRes = ht.findOnlyPrepared(itm.getKey());
-    auto* v = htRes.storedValue;
-    auto& hbl = htRes.lock;
+    auto htRes = ht.findForCommit(itm.getKey());
+    auto* v = htRes.pending.getSV();
+    auto& hbl = htRes.getHBL();
     bool maybeKeyExists = true;
     MutationStatus status;
     boost::optional<VBNotifyCtx> notifyCtx;
@@ -1853,7 +1855,7 @@ ENGINE_ERROR_CODE VBucket::prepare(
     } else {
         // Not a valid duplicate prepare, call processSet and hit the SyncWrite
         // checks.
-        std::tie(status, notifyCtx) = processSet(hbl,
+        std::tie(status, notifyCtx) = processSet(htRes,
                                                  v,
                                                  itm,
                                                  cas,
@@ -1924,10 +1926,9 @@ ENGINE_ERROR_CODE VBucket::setWithMeta(
         GenerateBySeqno genBySeqno,
         GenerateCas genCas,
         const Collections::VB::Manifest::CachingReadHandle& cHandle) {
-    auto htRes = itm.isPending() ? ht.findForSyncWrite(itm.getKey())
-                                 : ht.findForWrite(itm.getKey());
-    auto* v = htRes.storedValue;
-    auto& hbl = htRes.lock;
+    auto htRes = ht.findForCommit(itm.getKey());
+    auto* v = htRes.selectSVToModify(itm);
+    auto& hbl = htRes.getHBL();
     bool maybeKeyExists = true;
 
     // Effectively ignore logically deleted keys, they cannot stop the op
@@ -1994,7 +1995,7 @@ ENGINE_ERROR_CODE VBucket::setWithMeta(
 
     MutationStatus status;
     boost::optional<VBNotifyCtx> notifyCtx;
-    std::tie(status, notifyCtx) = processSet(hbl,
+    std::tie(status, notifyCtx) = processSet(htRes,
                                              v,
                                              itm,
                                              cas,
@@ -3124,7 +3125,7 @@ void VBucket::decrDirtyQueuePendingWrites(size_t decrementBy)
 }
 
 std::pair<MutationStatus, boost::optional<VBNotifyCtx>> VBucket::processSet(
-        const HashTable::HashBucketLock& hbl,
+        HashTable::FindCommitResult& htRes,
         StoredValue*& v,
         Item& itm,
         uint64_t cas,
@@ -3151,6 +3152,12 @@ std::pair<MutationStatus, boost::optional<VBNotifyCtx>> VBucket::processSet(
             Expects(itm.isCommitted());
             getPassiveDM().completeSyncWrite(
                     itm.getKey(), PassiveDurabilityMonitor::Resolution::Commit);
+
+            // @TODO we must remove the prepare and overwrite the mutation if we
+            // are replacing a prepare with a mutation
+            // Release the pending SV from the SVP that is holding it to prevent
+            // a double stat update that would cause a stat underflow exception.
+            htRes.pending.release();
         }
 
         // This is a new SyncWrite, we just want to add a new prepare unless we
@@ -3161,12 +3168,12 @@ std::pair<MutationStatus, boost::optional<VBNotifyCtx>> VBucket::processSet(
             // We have to modify the StoredValue pointer passed in or we do not
             // return the correct cas to the client.
             std::tie(v, notifyCtx) = addNewStoredValue(
-                    hbl, itm, queueItmCtx, GenerateRevSeqno::No);
+                    htRes.getHBL(), itm, queueItmCtx, GenerateRevSeqno::No);
             return {MutationStatus::WasClean, notifyCtx};
         }
     }
 
-    return processSetInner(hbl,
+    return processSetInner(htRes.getHBL(),
                            v,
                            itm,
                            cas,

engines/ep/src/vbucket.h

@@ -1766,10 +1766,10 @@ class VBucket : public std::enable_shared_from_this<VBucket> {
      * Prevents operations on in-flight SyncWrites.
      * Redirects the addition of new prepares to addNewStoredValue.
      *
-     * @param hbl Hash table bucket lock that must be held
-     * @param v Reference to the ptr of StoredValue. This can be changed if a
-     *          new StoredValue is added or just its contents is changed if the
-     *          exisiting StoredValue is updated
+     * @param htRes Committed and Pending StoredValues, include HBL.
+     * @param v Reference to the ptr of StoredValue to modify. This can be
+     *          changed if a new StoredValue is added or just its contents is
+     *          changes if the existing StoredValue is updated.
      * @param itm Item to be added/updated. On success, its revSeqno is updated
      * @param cas value to match
      * @param allowExisting set to false if you want set to fail if the
@@ -1785,7 +1785,7 @@ class VBucket : public std::enable_shared_from_this<VBucket> {
      *                info (if operation was successful).
      */
     std::pair<MutationStatus, boost::optional<VBNotifyCtx>> processSet(
-            const HashTable::HashBucketLock& hbl,
+            HashTable::FindCommitResult& htRes,
             StoredValue*& v,
             Item& itm,
             uint64_t cas,

engines/ep/tests/module_tests/dcp_stream_test.cc

@@ -1401,11 +1401,11 @@ void SingleThreadedActiveStreamTest::setupProducer(
 
 MutationStatus SingleThreadedActiveStreamTest::public_processSet(
         VBucket& vb, Item& item, const VBQueueItemCtx& ctx) {
-    auto htRes = ctx.durability ? vb.ht.findForSyncWrite(item.getKey())
-                                : vb.ht.findForWrite(item.getKey());
+    auto htRes = vb.ht.findForCommit(item.getKey());
+    auto* v = htRes.selectSVToModify(item);
     return vb
-            .processSet(htRes.lock,
-                        htRes.storedValue,
+            .processSet(htRes,
+                        v,
                         item,
                         0 /*cas*/,
                         true /*allowExisting*/,

engines/ep/tests/module_tests/durability_monitor_test.cc

@@ -153,14 +153,17 @@ void DurabilityMonitorTest::addSyncWrites(const std::vector<int64_t>& seqnos,
 }
 
 MutationStatus DurabilityMonitorTest::processSet(Item& item) {
-    auto htRes = vb->ht.findForWrite(item.getKey());
     VBQueueItemCtx ctx;
     ctx.genBySeqno = GenerateBySeqno::No;
     ctx.durability =
             DurabilityItemCtx{item.getDurabilityReqs(), /*cookie*/ nullptr};
+
+    auto htRes = vb->ht.findForCommit(item.getKey());
+    auto* v = htRes.selectSVToModify(item);
+
     return vb
-            ->processSet(htRes.lock,
-                         htRes.storedValue,
+            ->processSet(htRes,
+                         v,
                          item,
                          item.getCas(),
                          true /*allow_existing*/,

engines/ep/tests/module_tests/vbucket_test.cc

@@ -199,16 +199,13 @@ MutationStatus VBucketTestBase::public_processSet(Item& itm,
                                                   const VBQueueItemCtx& ctx) {
     // Need to take the collections read handle before the hbl
     auto cHandle = vbucket->lockCollections(itm.getKey());
-    auto hbl_sv = lockAndFind(itm.getKey(), ctx);
+
+    auto htRes = vbucket->ht.findForCommit(itm.getKey());
+    auto* v = htRes.selectSVToModify(itm);
+
     return vbucket
-            ->processSet(hbl_sv.first,
-                         hbl_sv.second,
-                         itm,
-                         cas,
-                         true,
-                         false,
-                         ctx,
-                         {/*no predicate*/})
+            ->processSet(
+                    htRes, v, itm, cas, true, false, ctx, {/*no predicate*/})
             .first;
 }