MB-34910: Warmup DurabilityMonitor

Following a restart the DurabilityMonitor should be in a logically
equivalent state to it's pre-warmup state. Of interest in this patch are
the restoration of

1. High Prepared Seqno
2. High Completed Seqno
3. Highest Tracked Seqno (ActiveDurabilityMonitor only)

To achieve the restoration of 1 and 2 and with inclusion of the previous
patch warmup can just reset the variables to their pre-warmup values.

For the restoration of 3, that is just assigned the value of 1.

Further to these changes extension of the existing warmup testing showed
that following warmup, any loaded prepares lose the 'ack' count for the
current node, all prepares found on disk must have at least an 'ack'
from the node itself. This is addressed when constructing the ADM by
calling updateHighPreparedSeqno after loading all tracked writes.

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

Author: jimwwalker

engines/ep/src/durability/active_durability_monitor.cc

@@ -23,6 +23,7 @@
 #include "stats.h"
 #include "statwriter.h"
 #include "vbucket.h"
+#include "vbucket_state.h"
 
 #include <boost/algorithm/string/join.hpp>
 #include <folly/concurrency/UnboundedQueue.h>
@@ -356,8 +357,12 @@ ActiveDurabilityMonitor::ActiveDurabilityMonitor(EPStats& stats, VBucket& vb)
 ActiveDurabilityMonitor::ActiveDurabilityMonitor(
         EPStats& stats,
         VBucket& vb,
+        const vbucket_state& vbs,
         std::vector<queued_item>&& outstandingPrepares)
     : ActiveDurabilityMonitor(stats, vb) {
+    if (!vbs.replicationTopology.is_null()) {
+        setReplicationTopology(vbs.replicationTopology);
+    }
     auto s = state.wlock();
     for (auto& prepare : outstandingPrepares) {
         auto seqno = prepare->getBySeqno();
@@ -367,10 +372,18 @@ ActiveDurabilityMonitor::ActiveDurabilityMonitor(
         s->trackedWrites.emplace_back(nullptr,
                                       std::move(prepare),
                                       std::chrono::milliseconds{},
-                                      nullptr,
-                                      nullptr);
+                                      s->firstChain.get(),
+                                      s->secondChain.get());
         s->lastTrackedSeqno = seqno;
     }
+
+    // If we did load sync writes we should get them at least acked for this
+    // node, which is achieved by attempting to move the HPS
+    s->updateHighPreparedSeqno(*completedQueue);
+
+    s->lastTrackedSeqno.reset(vbs.highPreparedSeqno);
+    s->highPreparedSeqno.reset(vbs.highPreparedSeqno);
+    s->highCompletedSeqno.reset(vbs.highCompletedSeqno);
 }
 
 ActiveDurabilityMonitor::ActiveDurabilityMonitor(EPStats& stats,
@@ -386,6 +399,8 @@ ActiveDurabilityMonitor::ActiveDurabilityMonitor(EPStats& stats,
         s->lastTrackedSeqno.reset(
                 pdm.state.wlock()->highCompletedSeqno.lastWriteSeqno);
     }
+    s->highPreparedSeqno.reset(pdm.getHighPreparedSeqno());
+    s->highCompletedSeqno.reset(pdm.getHighCompletedSeqno());
 }
 
 ActiveDurabilityMonitor::~ActiveDurabilityMonitor() = default;
@@ -426,11 +441,7 @@ void ActiveDurabilityMonitor::setReplicationTopology(
 }
 
 int64_t ActiveDurabilityMonitor::getHighPreparedSeqno() const {
-    const auto s = state.rlock();
-    if (!s->firstChain) {
-        return 0;
-    }
-    return s->highPreparedSeqno;
+    return state.rlock()->highPreparedSeqno;
 }
 
 int64_t ActiveDurabilityMonitor::getHighCompletedSeqno() const {

engines/ep/src/durability/active_durability_monitor.h

@@ -27,6 +27,7 @@
 
 class EPStats;
 class PassiveDurabilityMonitor;
+struct vbucket_state;
 class VBucket;
 
 /*
@@ -47,14 +48,17 @@ class ActiveDurabilityMonitor : public DurabilityMonitor {
      * Construct an ActiveDM for the given vBucket, with the specified
      * outstanding prepares as the initial state of the tracked SyncWrites. Used
      * by warmup to restore the state as it was before restart.
+     * @param stats EPStats object for the associated Bucket.
      * @param vb VBucket which owns this Durability Monitor.
+     * @param vbs reference to the vbucket_state found at warmup
      * @param outstandingPrepares In-flight prepares which the DM should take
      *        responsibility for.
      *        These must be ordered by ascending seqno, otherwise
      *        std::invalid_argument will be thrown.
      */
     ActiveDurabilityMonitor(EPStats& stats,
                             VBucket& vb,
+                            const vbucket_state& vbs,
                             std::vector<queued_item>&& outstandingPrepares);
 
     /**

engines/ep/src/durability/passive_durability_monitor.cc

@@ -24,6 +24,7 @@
 #include "statwriter.h"
 #include "stored-value.h"
 #include "vbucket.h"
+#include "vbucket_state.h"
 
 #include <boost/range/adaptor/reversed.hpp>
 #include <gsl.h>
@@ -38,10 +39,22 @@ PassiveDurabilityMonitor::PassiveDurabilityMonitor(VBucket& vb)
     s->highCompletedSeqno = Position(s->trackedWrites.end());
 }
 
-PassiveDurabilityMonitor::PassiveDurabilityMonitor(
-        VBucket& vb, std::vector<queued_item>&& outstandingPrepares)
+PassiveDurabilityMonitor::PassiveDurabilityMonitor(VBucket& vb,
+                                                   int64_t highPreparedSeqno,
+                                                   int64_t highCompletedSeqno)
     : PassiveDurabilityMonitor(vb) {
     auto s = state.wlock();
+    s->highPreparedSeqno.lastWriteSeqno.reset(highPreparedSeqno);
+    s->highCompletedSeqno.lastWriteSeqno.reset(highCompletedSeqno);
+}
+
+PassiveDurabilityMonitor::PassiveDurabilityMonitor(
+        VBucket& vb,
+        int64_t highPreparedSeqno,
+        int64_t highCompletedSeqno,
+        std::vector<queued_item>&& outstandingPrepares)
+    : PassiveDurabilityMonitor(vb, highPreparedSeqno, highCompletedSeqno) {
+    auto s = state.wlock();
     for (auto& prepare : outstandingPrepares) {
         // Any outstanding prepares "grandfathered" into the DM should have
         // already specified a non-default timeout.

engines/ep/src/durability/passive_durability_monitor.h

@@ -22,6 +22,7 @@
 #include <vector>
 
 class RollbackResult;
+struct vbucket_state;
 class VBucket;
 class StoredDocKey;
 
@@ -38,17 +39,31 @@ class PassiveDurabilityMonitor : public DurabilityMonitor {
 public:
     PassiveDurabilityMonitor(VBucket& vb);
 
+    /**
+     * Construct a PassiveDM for the given vBucket, with the specified hps/hcs
+     * @param vb VBucket which owns this Durability Monitor.
+     * @param highPreparedSeqno seqno to use as the initial highPreparedSeqno
+     * @param highCompletedSeqno seqno to use as the initial highCompletedSeqno
+     */
+    PassiveDurabilityMonitor(VBucket& vb,
+                             int64_t highPreparedSeqno,
+                             int64_t highCompletedSeqno);
+
     /**
      * Construct a PassiveDM for the given vBucket, with the specified
      * outstanding prepares as the initial state of the tracked SyncWrites. Used
      * by warmup to restore the state as it was before restart.
      * @param vb VBucket which owns this Durability Monitor.
+     * @param highPreparedSeqno seqno to use as the initial highPreparedSeqno
+     * @param highCompletedSeqno seqno to use as the initial highCompletedSeqno
      * @param outstandingPrepares In-flight prepares which the DM should take
      *        responsibility for.
      *        These must be ordered by ascending seqno, otherwise
      *        std::invalid_argument will be thrown.
      */
     PassiveDurabilityMonitor(VBucket& vb,
+                             int64_t highPreparedSeqno,
+                             int64_t highCompletedSeqno,
                              std::vector<queued_item>&& outstandingPrepares);
 
     ~PassiveDurabilityMonitor();

engines/ep/src/ep_vb.cc

@@ -31,6 +31,7 @@
 #include "stored_value_factories.h"
 #include "tasks.h"
 #include "vbucket_bgfetch_item.h"
+#include "vbucket_state.h"
 #include "vbucketdeletiontask.h"
 #include <folly/lang/Assume.h>
 
@@ -743,6 +744,7 @@ MutationStatus EPVBucket::insertFromWarmup(Item& itm,
 }
 
 void EPVBucket::restoreOutstandingPreparesFromWarmup(
+        const vbucket_state& vbs,
         std::vector<queued_item>&& outstandingPrepares) {
     // About to change the durabilityMonitor object, which is guarded by
     // stateLock.
@@ -763,19 +765,17 @@ void EPVBucket::restoreOutstandingPreparesFromWarmup(
     switch (getState()) {
     case vbucket_state_active: {
         durabilityMonitor = std::make_unique<ActiveDurabilityMonitor>(
-                stats, *this, std::move(outstandingPrepares));
-        auto topology = getReplicationTopology();
-        if (!topology.is_null()) {
-            dynamic_cast<ActiveDurabilityMonitor*>(durabilityMonitor.get())
-                    ->setReplicationTopology(topology);
-        }
+                stats, *this, vbs, std::move(outstandingPrepares));
         return;
     }
     case vbucket_state_replica:
     case vbucket_state_pending:
     case vbucket_state_dead:
         durabilityMonitor = std::make_unique<PassiveDurabilityMonitor>(
-                *this, std::move(outstandingPrepares));
+                *this,
+                vbs.highPreparedSeqno,
+                vbs.highCompletedSeqno,
+                std::move(outstandingPrepares));
         return;
     }
 }

engines/ep/src/ep_vb.h

@@ -180,10 +180,12 @@ class EPVBucket : public VBucket {
      * Populates the HashTable and the DurabilityMonitor with the given
      * set of queued_items.
      *
+     * @param vbs The vbucket_state read during warmup
      * @param outstandingPrepares Sequence of prepared_sync_writes, sorted by
      *        seqno in ascending order.
      */
     void restoreOutstandingPreparesFromWarmup(
+            const vbucket_state& vbs,
             std::vector<queued_item>&& outstandingPrepares);
 
     size_t getNumPersistedDeletes() const override;

engines/ep/src/vbucket.cc

@@ -662,7 +662,15 @@ void VBucket::setupSyncReplication(const nlohmann::json& topology) {
             return;
         }
         // Current DM (if exists) is not Passive; replace it with a Passive one.
-        durabilityMonitor = std::make_unique<PassiveDurabilityMonitor>(*this);
+        if (durabilityMonitor) {
+            durabilityMonitor = std::make_unique<PassiveDurabilityMonitor>(
+                    *this,
+                    durabilityMonitor->getHighPreparedSeqno(),
+                    durabilityMonitor->getHighCompletedSeqno());
+        } else {
+            durabilityMonitor =
+                    std::make_unique<PassiveDurabilityMonitor>(*this);
+        }
         return;
     case vbucket_state_dead:
         // No DM in dead state.

engines/ep/src/vbucket_state.cc

@@ -139,3 +139,10 @@ void from_json(const nlohmann::json& j, vbucket_state& vbs) {
     // Note: We don't track on disk prepares pre-6.5
     vbs.onDiskPrepares = std::stoll(j.value("on_disk_prepares", "0"));
 }
+
+std::ostream& operator<<(std::ostream& os, const vbucket_state& vbs) {
+    nlohmann::json j;
+    to_json(j, vbs);
+    os << j;
+    return os;
+}

engines/ep/src/vbucket_state.h

@@ -149,3 +149,5 @@ void to_json(nlohmann::json& json, const vbucket_state& vbs);
 
 /// Method to allow nlohmann::json to convert from JSON to vbucket_state.
 void from_json(const nlohmann::json& j, vbucket_state& vbs);
+
+std::ostream& operator<<(std::ostream& os, const vbucket_state& vbs);

engines/ep/src/warmup.cc

@@ -1011,7 +1011,7 @@ void Warmup::createVBuckets(uint16_t shardId) {
     // objects if they do not already exist.
     for (const auto& itr : shardVbStates[shardId]) {
         Vbid vbid = itr.first;
-        vbucket_state vbs = itr.second;
+        const vbucket_state& vbs = itr.second;
 
         // Collections and sync-repl requires that the VBucket datafiles have
         // 'namespacing' applied to the key space
@@ -1322,7 +1322,17 @@ void Warmup::loadPreparedSyncWrites(uint16_t shardId) {
                   [](const auto& a, const auto& b) {
                       return a->getBySeqno() < b->getBySeqno();
                   });
-        epVb.restoreOutstandingPreparesFromWarmup(std::move(prepares));
+
+        // Need the HPS/HCS so the DurabilityMonitor can be fully resumed
+        auto vbState = shardVbStates[shardId].find(vbid);
+        if (vbState == shardVbStates[shardId].end()) {
+            throw std::logic_error(
+                    "Warmup::loadPreparedSyncWrites: processing " +
+                    vbid.to_string() + ", but found no vbucket_state");
+        }
+        const vbucket_state& vbs = vbState->second;
+
+        epVb.restoreOutstandingPreparesFromWarmup(vbs, std::move(prepares));
     }
 
     if (++threadtask_count == store.vbMap.getNumShards()) {