MB-35252: Simplify allowed duplicate prepare logic

The alowedDuplicatePrepareSeqnos set is used to determine whether an
existing prepare found in the hashtable may be replaced by a new
prepare. This is a special case, only permitted after a
disconnect/reconnect when the commit for the existing prepare may have
been deduplicated. Otherwise, a second prepare for the same key without
an intervening commit/abort is rejected as "sync write in progress".

This set was previously initialised with all seqnos between the
highCompletedSeqno and the highSeqno of the VB. Prepares before the HCS
have been completed and should not be in the hashtable. Prepares after
the highSeqno have not yet been received, and do not need to be
considered.
The number of seqnos in the set when generated in this manner can be
very large; if no SyncWrites are seen, the lower bound of the window
(the highCompletedSeqno) would be zero. This leads to a very large set
being allocated, costing time and memory.

Solution: Instead, store the current highest trackedWrite seqno.
Any pending SyncWrite in the hashtable with a seqno lower than
this value *may* be replaced by another prepare, because the expected
Commit/Abort may have been deduplicated.
If such a prepare in the hashtable is then completed (indicating the
commit/abort was *not* deduped), any further prepare for the same key
after the completion is treated like any other new SyncWrite - it does
not need special treatment.

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

Author: jameseh96

engines/ep/src/durability/passive_durability_monitor.cc

@@ -389,6 +389,15 @@ void PassiveDurabilityMonitor::postProcessRollback(
     s->highPreparedSeqno.lastWriteSeqno.reset(rollbackResult.highPreparedSeqno);
 }
 
+int64_t PassiveDurabilityMonitor::getHighestTrackedSeqno() const {
+    auto s = state.rlock();
+    if (!s->trackedWrites.empty()) {
+        return s->trackedWrites.back().getBySeqno();
+    } else {
+        return 0;
+    }
+}
+
 void PassiveDurabilityMonitor::toOStream(std::ostream& os) const {
     os << "PassiveDurabilityMonitor[" << this << "]"
        << " high_prepared_seqno:" << getHighPreparedSeqno()

engines/ep/src/durability/passive_durability_monitor.h

@@ -152,6 +152,13 @@ class PassiveDurabilityMonitor : public DurabilityMonitor {
      */
     void postProcessRollback(const RollbackResult& rollbackResult);
 
+    /**
+     * Get the highest seqno for which there is a SyncWrite in trackedWrites.
+     * Returns 0 if trackedWrites is empty.
+     *
+     */
+    int64_t getHighestTrackedSeqno() const;
+
 protected:
     void toOStream(std::ostream& os) const override;
     /**

engines/ep/src/vbucket.cc

@@ -835,9 +835,6 @@ ENGINE_ERROR_CODE VBucket::commit(
 
     Expects(prepareSeqno);
 
-    // Remove from the allowed duplicate prepare set (if it exists)
-    allowedDuplicatePrepareSeqnos.erase(res.pending->getBySeqno());
-
     // Value for Pending must never be ejected
     Expects(res.pending->isResident());
 
@@ -940,9 +937,6 @@ ENGINE_ERROR_CODE VBucket::abort(
         return ENGINE_EINVAL;
     }
 
-    // Remove from the allowed duplicate prepare set (if it exists)
-    allowedDuplicatePrepareSeqnos.erase(htRes.storedValue->getBySeqno());
-
     // If prepare seqno is not the same as our stored seqno then we should be
     // a replica and have missed a completion and a prepare due to de-dupe.
     if (prepareSeqno !=
@@ -1834,9 +1828,7 @@ ENGINE_ERROR_CODE VBucket::prepare(
             nullptr /* No pre link step needed */,
             {} /*overwritingPrepareSeqno*/};
 
-    auto itr = allowedDuplicatePrepareSeqnos.end();
-    if (v && (itr = allowedDuplicatePrepareSeqnos.find(v->getBySeqno())) !=
-                     allowedDuplicatePrepareSeqnos.end()) {
+    if (v && v->getBySeqno() <= allowedDuplicatePrepareThreshold) {
         // Valid duplicate prepare - call processSetInner and skip the
         // SyncWrite checks.
         queueItmCtx.overwritingPrepareSeqno = v->getBySeqno();
@@ -1849,23 +1841,6 @@ ENGINE_ERROR_CODE VBucket::prepare(
                                                       queueItmCtx,
                                                       {/*no predicate*/},
                                                       maybeKeyExists);
-        switch (status) {
-        case MutationStatus::WasClean:
-        case MutationStatus::WasDirty:
-            // We should not see this seqno again so remove from the set.
-            allowedDuplicatePrepareSeqnos.erase(itr);
-            break;
-        case MutationStatus::NotFound:
-        case MutationStatus::InvalidCas:
-        case MutationStatus::IsLocked:
-        case MutationStatus::NoMem:
-        case MutationStatus::NeedBgFetch:
-        case MutationStatus::IsPendingSyncWrite:
-            // The old prepare is still in the hashtable, keep tracking
-            // it in the PDM. Also. if we (e.g.) retry after NoMem we want
-            // to treat the duplicate as "allowed" still.
-            break;
-        }
     } else {
         // Not a valid duplicate prepare, call processSet and hit the SyncWrite
         // checks.
@@ -3906,22 +3881,22 @@ void VBucket::removeQueuedAckFromDM(const std::string& node) {
 }
 
 void VBucket::setDuplicateSyncWriteWindow(uint64_t highSeqno) {
-    setUpAllowedDuplicatePrepareWindow();
-}
-
-void VBucket::setUpAllowedDuplicatePrepareWindow() {
-    auto& dm = getDurabilityMonitor();
-    auto hcs = dm.getHighCompletedSeqno();
-    auto highSeqno = getHighSeqno();
-    Expects(hcs <= highSeqno);
-
-    int64_t newDuplicateCount = highSeqno - hcs;
-    allowedDuplicatePrepareSeqnos.reserve(allowedDuplicatePrepareSeqnos.size() +
-                                          newDuplicateCount);
-
-    for (int64_t dupSeqno = hcs + 1; dupSeqno <= highSeqno; dupSeqno++) {
-        allowedDuplicatePrepareSeqnos.insert(dupSeqno);
-    }
+    setUpAllowedDuplicatePrepareThreshold();
+}
+
+void VBucket::setUpAllowedDuplicatePrepareThreshold() {
+    const auto& pdm = getPassiveDM();
+    // We should only see duplicates for prepares currently in trackedWrites
+    // prepares which are not in trackedWrites are either
+    //  - Completed: A new prepare would be valid anyway, as the item is
+    //  completed
+    //  - New: Added after this setup, and should not be followed by another
+    //  prepare without an intervening Commit/Abort
+    // As a sanity check, store the current highTrackedSeqno and assert that
+    // any prepares attempting to replace an existing prepare have a seqno
+    // less than or equal to this value.
+    // If no SyncWrites are being tracked, nothing can be duplicated
+    allowedDuplicatePrepareThreshold = pdm.getHighestTrackedSeqno();
 }
 
 void VBucket::addSyncWriteForRollback(const Item& item) {

engines/ep/src/vbucket.h

@@ -1711,7 +1711,7 @@ class VBucket : public std::enable_shared_from_this<VBucket> {
      * Set the allowed duplicate prepared seqnos to the range
      * (HighCompletedSeqno and HighPreparedSeqno].
      */
-    void setUpAllowedDuplicatePrepareWindow();
+    void setUpAllowedDuplicatePrepareThreshold();
 
     std::queue<queued_item> rejectQueue;
     std::unique_ptr<FailoverTable> failovers;
@@ -2481,8 +2481,9 @@ class VBucket : public std::enable_shared_from_this<VBucket> {
 
     static double mutationMemThreshold;
 
-    // The set of prepare seqnos that we may have to overwrite.
-    std::unordered_set<int64_t> allowedDuplicatePrepareSeqnos;
+    // The seqno threshold below which we may replace a prepare with another
+    // prepare (if the associated Commit/Abort may have been deduped)
+    int64_t allowedDuplicatePrepareThreshold = 0;
 
     friend class DurabilityMonitorTest;
     friend class SingleThreadedActiveStreamTest;