MB-33332: Keep committed prepare in HashTable - Ephemeral

We need to keep the completed prepare in the HashTable so that any
subsequent prepares can re-use (or chain if there is a range read)
the same OSV to ensure that we have snapshot with unique keys.

Change-Id: Idcdc3711fc4b183c28bab112e95285869d1041f1
Reviewed-on: http://review.couchbase.org/109841
Reviewed-by: Dave Rigby <[email protected]>
Tested-by: Dave Rigby <[email protected]>

Author: BenHuddleston

engines/ep/src/ephemeral_vb.cc

@@ -392,8 +392,11 @@ EphemeralVBucket::updateStoredValue(const HashTable::HashBucketLock& hbl,
                                     const Item& itm,
                                     const VBQueueItemCtx& queueItmCtx,
                                     bool justTouch) {
-    // Don't make any update if we know that this is a pending sync write. We
-    // need to abort before we make any modification to the seqList.
+    // Don't make any update if we know that this is a pending sync write that
+    // has not yet been completed. We need to return early before we make any
+    // modification to the seqList. This is typically handled in the EP case by
+    // the HashTable functions, but for ephemeral we update the seqList first so
+    // we need an earlier check.
     if (v.isPending()) {
         return std::make_tuple(
                 nullptr, MutationStatus::IsPendingSyncWrite, VBNotifyCtx{});
@@ -403,14 +406,10 @@ EphemeralVBucket::updateStoredValue(const HashTable::HashBucketLock& hbl,
     StoredValue* newSv = nullptr;
     MutationStatus status(MutationStatus::WasClean);
 
-    // If this is a new SyncWrite then we don't actually want to update an
-    // existing StoredValue, we just want to add the new pending. What about
-    // if we already have an existing (committed) pending I hear you ask? It
-    // should be stale and the StaleItemDeleter should remove it when it next
-    // runs. It has no affect on operations so we can ignore it. Do this before
-    // we take any locks as addNewStoredValue will attempt to acquire the same
-    // locks.
-    if (itm.isPending()) {
+    // If this is a new SyncWrite then we should just add a new prepare. If we
+    // have a prepare already (that has been completed) we can replace the
+    // existing one.
+    if (itm.isPending() && !v.isCompleted()) {
         std::tie(newSv, notifyCtx) =
                 addNewStoredValue(hbl, itm, queueItmCtx, GenerateRevSeqno::No);
         return {newSv, status, notifyCtx};
@@ -669,11 +668,9 @@ VBNotifyCtx EphemeralVBucket::commitStoredValue(
     // 1a. If an existing Committed exists, update that with the data from
     //     the Prepare.
     // 1b. If no existing Committed OSV, then create a new Committed and
-    // populate
+    //     populate
     //     with data from the Prepare.
-    // 2. Mark the Prepare as stale in the seqList and remove it from the
-    //    HashTable - this means that the key is no longer locked against
-    //    mutations (ESYNC_WRITE_IN_PROGRESS).
+    // 2. Mark the prepare as completed
 
     // Look for an existing Committed OSV under this key.
     StoredValue* newCommitted;
@@ -712,20 +709,7 @@ VBNotifyCtx EphemeralVBucket::commitStoredValue(
                                   GenerateRevSeqno::No);
     }
 
-    // Set the OSV to stale so that we can:
-    // a) Allow the StaleItemPurger to delete the item
-    // b) Ensure that we delete the OSV on destruction of the seqList as it does
-    //    not exist in the HashTable
-    std::lock_guard<std::mutex> listWriteLg(seqList->getListWriteLock());
-    // Remove the prepare from the hash table so that we can now allow mutations
-    // against the same key
-    auto release = ht.unlocked_release(values.pending.getHBL(),
-                                       values.pending.getSV());
-    seqList->markItemStale(listWriteLg, std::move(release), nullptr);
-    // Manually release this ptr (not reset) as we don't want to destruct the
-    // item in the seqList.
-    release.release();
-
+    values.pending.setCommitted(CommittedState::PrepareCommitted);
     return notifyCtx;
 }
 

engines/ep/src/hash_table.cc

@@ -17,6 +17,7 @@
 
 #include "hash_table.h"
 
+#include "ep_time.h"
 #include "item.h"
 #include "stats.h"
 #include "stored_value_factories.h"
@@ -98,6 +99,11 @@ HashTable::StoredValueProxy::~StoredValueProxy() {
     valueStats.epilogue(pre, value);
 }
 
+void HashTable::StoredValueProxy::setCommitted(CommittedState state) {
+    value->setCommitted(state);
+    value->setCompletedOrDeletedTime(ep_current_time());
+}
+
 HashTable::HashTable(EPStats& st,
                      std::unique_ptr<AbstractStoredValueFactory> svFactory,
                      size_t initialSize,
@@ -286,7 +292,7 @@ HashTable::FindInnerResult HashTable::findInner(const DocKey& key) {
     for (StoredValue* v = values[hbl.getBucketNum()].get().get(); v;
          v = v->getNext().get().get()) {
         if (v->hasKey(key)) {
-            if (v->isPending()) {
+            if (v->isPending() || v->isCompleted()) {
                 Expects(!foundPend);
                 foundPend = v;
             } else {
@@ -340,42 +346,33 @@ HashTable::UpdateResult HashTable::unlocked_updateStoredValue(
                 "call on a non-active HT object");
     }
 
-    switch (v.getCommitted()) {
-    case CommittedState::Pending:
-    case CommittedState::PreparedMaybeVisible:
-        // Cannot update a SV if it's a Pending item.
+    if (v.isPending()) {
         return {MutationStatus::IsPendingSyncWrite, nullptr};
-    case CommittedState::PrepareAborted:
-    case CommittedState::PrepareCommitted:
-        // We shouldn't be trying to use PrepareCompleted states yet
-        throw std::logic_error(
-                "HashTable::unlocked_updateStoredValue"
-                " attempting to update a completed prepare");
-    case CommittedState::CommittedViaMutation:
-    case CommittedState::CommittedViaPrepare:
-        // Logically /can/ update a non-Pending StoredValue with a Pending Item;
-        // however internally this is implemented as a separate (new)
-        // StoredValue object for the Pending item.
-        if (itm.isPending()) {
-            auto* sv = HashTable::unlocked_addNewStoredValue(hbl, itm);
-            return {MutationStatus::WasClean, sv};
-        }
+    }
 
-        // item is not Pending; can directly replace the existing SV.
-        MutationStatus status = v.isDirty() ? MutationStatus::WasDirty
-                                            : MutationStatus::WasClean;
+    // Logically /can/ update a non-Pending StoredValue with a Pending Item;
+    // however internally this is implemented as a separate (new)
+    // StoredValue object for the Pending item. We can replace a completed
+    // prepare with a new prepare though (so just drop through and update
+    // normally).
+    if (itm.isPending() && !v.isCompleted()) {
+        auto* sv = HashTable::unlocked_addNewStoredValue(hbl, itm);
+        return {MutationStatus::WasClean, sv};
+    }
 
-        const auto preProps = valueStats.prologue(&v);
+    // Can directly replace the existing SV.
+    MutationStatus status =
+            v.isDirty() ? MutationStatus::WasDirty : MutationStatus::WasClean;
 
-        /* setValue() will mark v as undeleted if required */
-        v.setValue(itm);
-        updateFreqCounter(v);
+    const auto preProps = valueStats.prologue(&v);
 
-        valueStats.epilogue(preProps, &v);
+    /* setValue() will mark v as undeleted if required */
+    v.setValue(itm);
+    updateFreqCounter(v);
 
-        return {status, &v};
-    }
-    folly::assume_unreachable();
+    valueStats.epilogue(preProps, &v);
+
+    return {status, &v};
 }
 
 StoredValue* HashTable::unlocked_addNewStoredValue(const HashBucketLock& hbl,
@@ -420,7 +417,7 @@ HashTable::Statistics::StoredValueProperties::StoredValueProperties(
     isDeleted = sv->isDeleted();
     isTempItem = sv->isTempItem();
     isSystemItem = sv->getKey().getCollectionID().isSystem();
-    isPreparedSyncWrite = sv->isPending();
+    isPreparedSyncWrite = sv->isPending() || sv->isCompleted();
 }
 
 HashTable::Statistics::StoredValueProperties HashTable::Statistics::prologue(
@@ -634,28 +631,53 @@ HashTable::FindResult HashTable::findForWrite(const DocKey& key,
                                               WantsDeleted wantsDeleted) {
     auto result = findInner(key);
 
-    /// Writing using the Pending StoredValue (if found), else committed.
-    auto* sv = result.pendingSV ? result.pendingSV : result.committedSV;
+    // We found a prepare. It may have been completed (Ephemeral) though. If it
+    // has been completed we will return the committed StoredValue.
+    if (result.pendingSV && !result.pendingSV->isCompleted()) {
+        // Early return if we found a prepare. We should always return prepares
+        // regardless of whether or not they are deleted or the caller has asked
+        // for deleted SVs. For example, consider searching for a SyncDelete, we
+        // should always return the deleted prepare.
+        return {result.pendingSV, std::move(result.lock)};
+    }
 
-    if (!sv) {
+    /// Writing using the Pending StoredValue (if found), else committed.
+    if (!result.committedSV) {
         // No item found - return null.
         return {nullptr, std::move(result.lock)};
     }
 
-    // Early return if we found a prepare. We should always return prepares
-    // regardless of whether or not they are deleted or the caller has asked for
-    // deleted SVs. For example, consider searching for a SyncDelete, we should
-    // always return the deleted prepare.
+    if (result.committedSV->isDeleted() && wantsDeleted == WantsDeleted::No) {
+        // Deleted items should only be returned if caller asked for them -
+        // otherwise return null.
+        return {nullptr, std::move(result.lock)};
+    }
+    return {result.committedSV, std::move(result.lock)};
+}
+
+HashTable::FindResult HashTable::findForSyncWrite(const DocKey& key) {
+    auto result = findInner(key);
+
     if (result.pendingSV) {
-        return {sv, std::move(result.lock)};
+        // Early return if we found a prepare. We should always return
+        // prepares regardless of whether or not they are deleted or the caller
+        // has asked for deleted SVs. For example, consider searching for a
+        // SyncDelete, we should always return the deleted prepare. Also,
+        // we always return completed prepares.
+        return {result.pendingSV, std::move(result.lock)};
     }
 
-    if (sv->isDeleted() && wantsDeleted == WantsDeleted::No) {
+    if (!result.committedSV) {
+        // No item found - return null.
+        return {nullptr, std::move(result.lock)};
+    }
+
+    if (result.committedSV->isDeleted()) {
         // Deleted items should only be returned if caller asked for them -
         // otherwise return null.
         return {nullptr, std::move(result.lock)};
     }
-    return {sv, std::move(result.lock)};
+    return {result.committedSV, std::move(result.lock)};
 }
 
 HashTable::StoredValueProxy HashTable::findForWrite(StoredValueProxy::RetSVPTag,

engines/ep/src/hash_table.h

@@ -714,9 +714,7 @@ class HashTable {
             value->setBySeqno(newSeqno);
         }
 
-        void setCommitted(CommittedState state) {
-            value->setCommitted(state);
-        }
+        void setCommitted(CommittedState state);
 
     private:
         HashBucketLock lock;
@@ -746,8 +744,9 @@ class HashTable {
      * change shouldn't affect the (old) reference count; the reference count
      * will get updated later part of actually changing the item.
      *
-     * Returns the Prepare StoredValue for the key if one exists, otherwise
-     * returns the Committed StoredValue, or nullptr neither exist.
+     * Returns the Prepare StoredValue for the key if one exists that has not
+     * been completed, otherwise returns the Committed StoredValue, or nullptr
+     * if neither exist.
      *
      * @param key The key of the item to find
      * @param wantsDeleted whether a deleted value should be returned
@@ -770,6 +769,28 @@ class HashTable {
             const DocKey& key,
             WantsDeleted wantsDeleted = WantsDeleted::Yes);
 
+    /**
+     * Find an item with the specified key for write access.
+     *
+     * Does not modify referenced status, as typically the lookup of a SV to
+     * change shouldn't affect the (old) reference count; the reference count
+     * will get updated later part of actually changing the item.
+     *
+     * Returns the Prepare StoredValue for the key if one exists, otherwise
+     * returns the Committed StoredValue, or nullptr neither exist. Will return
+     * completed prepares.
+     *
+     * @param key The key of the item to find
+     * @return A FindResult consisting of:
+     *         - a pointer to a StoredValue -- NULL if not found
+     *         - a (locked) HashBucketLock for the key's hash bucket. Note
+     *         this always returns a locked object; even if the requested key
+     *           doesn't exist. This is to facilitate use-cases where the caller
+     *         subsequently needs to insert a StoredValue for this key, to
+     *         avoid unlocking and re-locking the mutex.
+     */
+    FindResult findForSyncWrite(const DocKey& key);
+
     /**
      * @return A pair of StoredValues (pending and prepare) so that we can
      *         update accordingly when committing a SyncWrite.

engines/ep/src/stored-value.cc

@@ -233,6 +233,8 @@ std::unique_ptr<Item> StoredValue::toItem(
     switch (getCommitted()) {
     case CommittedState::Pending:
     case CommittedState::PreparedMaybeVisible:
+    case CommittedState::PrepareAborted:
+    case CommittedState::PrepareCommitted:
         if (!durabilityReqs) {
             throw std::logic_error(
                     "StoredValue::toItemImpl: attempted to create Item from "
@@ -249,13 +251,6 @@ std::unique_ptr<Item> StoredValue::toItem(
     case CommittedState::CommittedViaMutation:
         // nothing do to.
         break;
-    case CommittedState::PrepareAborted:
-    case CommittedState::PrepareCommitted:
-        // We shouldn't be trying to create Item's out of these (for now)
-        throw std::logic_error(
-                "StoredValue::toItemImpl: attempt to create"
-                "Item from completed prepare");
-        break;
     }
 
     return item;
@@ -471,6 +466,15 @@ void StoredValue::incrementAge() {
     }
 }
 
+void StoredValue::setCompletedOrDeletedTime(rel_time_t time) {
+    if (isOrdered()) {
+        return static_cast<OrderedStoredValue*>(this)
+                ->setCompletedOrDeletedTime(time);
+    } else {
+        // Do nothing, only applicable to OSV
+    }
+}
+
 std::ostream& operator<<(std::ostream& os, const StoredValue& sv) {
 
     // type, address
@@ -600,7 +604,7 @@ void OrderedStoredValue::setValueImpl(const Item& itm) {
 }
 
 void OrderedStoredValue::setCompletedOrDeletedTime(rel_time_t time) {
-    if (!isDeleted() && !isPending()) {
+    if (!(isDeleted() || isCompleted())) {
         throw std::logic_error(
                 "OrderedStoredValue::setCompletedOrDeletedTime: Called on "
                 "Alive item");

engines/ep/src/stored-value.h

@@ -540,7 +540,7 @@ class StoredValue {
      * @return true if the item is locked
      */
     bool isLocked(rel_time_t curtime) const {
-        if (isDeleted()) {
+        if (isDeleted() || isCompleted()) {
             // Deleted items cannot be locked.
             return false;
         }
@@ -827,6 +827,21 @@ class StoredValue {
         return !isPending();
     }
 
+    /**
+     * Returns true if the stored value is Completed (by Abort or Commit).
+     */
+    bool isCompleted() const {
+        return (getCommitted() == CommittedState::PrepareAborted) ||
+               (getCommitted() == CommittedState::PrepareCommitted);
+    }
+
+    /**
+     * Set the time the item was completed or deleted at to the specified time.
+     *
+     * Only applicable for an OSV so we do nothing for a normal SV.
+     */
+    void setCompletedOrDeletedTime(rel_time_t time);
+
 protected:
     /**
      * Constructor - protected as allocation needs to be done via